diff --git a/packages/common/src/index.ts b/packages/common/src/index.ts index cc4607e..fd16251 100644 --- a/packages/common/src/index.ts +++ b/packages/common/src/index.ts @@ -2,3 +2,4 @@ export * from './types.js'; export * from './constants.js'; export * from './logger.js'; export * from './wallet.js'; +export * from './json-store.js'; diff --git a/packages/common/src/json-store.test.ts b/packages/common/src/json-store.test.ts new file mode 100644 index 0000000..d0bb789 --- /dev/null +++ b/packages/common/src/json-store.test.ts @@ -0,0 +1,47 @@ +import fs from 'fs'; +import os from 'os'; +import path from 'path'; +import { afterEach, describe, expect, it } from 'vitest'; +import { createJsonWriteQueue, writeJsonSafe } from './json-store.js'; + +const tempDirs: string[] = []; + +function makeTempDir(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'clevercon-json-store-')); + tempDirs.push(dir); + return dir; +} + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +describe('writeJsonSafe', () => { + it('writes formatted JSON through an atomic temp-file rename', () => { + const filePath = path.join(makeTempDir(), 'nested', 'state.json'); + + writeJsonSafe(filePath, { ok: true, count: 2 }); + + expect(JSON.parse(fs.readFileSync(filePath, 'utf8'))).toEqual({ ok: true, count: 2 }); + expect(fs.readdirSync(path.dirname(filePath)).filter((name) => name.endsWith('.tmp'))).toEqual( + [], + ); + }); +}); + +describe('createJsonWriteQueue', () => { + it('serializes writes and leaves the last queued value on disk', () => { + const filePath = path.join(makeTempDir(), 'registry.json'); + const save = createJsonWriteQueue>(filePath); + + save([{ id: 'agent-a' }]); + save([{ id: 'agent-a' }, { id: 'agent-b' }]); + + expect(JSON.parse(fs.readFileSync(filePath, 'utf8'))).toEqual([ + { id: 'agent-a' }, + { id: 'agent-b' }, + ]); + }); +}); diff --git a/packages/common/src/json-store.ts b/packages/common/src/json-store.ts new file mode 100644 index 0000000..6ba68b9 --- /dev/null +++ b/packages/common/src/json-store.ts @@ -0,0 +1,48 @@ +import fs from 'fs'; +import path from 'path'; + +function temporaryPathFor(filePath: string): string { + const directory = path.dirname(filePath); + const basename = path.basename(filePath); + const nonce = Math.random().toString(16).slice(2); + return path.join(directory, `.${basename}.${process.pid}.${Date.now()}.${nonce}.tmp`); +} + +export function writeJsonSafe(filePath: string, data: unknown): void { + const directory = path.dirname(filePath); + fs.mkdirSync(directory, { recursive: true }); + + const tmpPath = temporaryPathFor(filePath); + try { + fs.writeFileSync(tmpPath, `${JSON.stringify(data, null, 2)}\n`, 'utf8'); + fs.renameSync(tmpPath, filePath); + } catch (error) { + try { + if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath); + } catch { + // Preserve the original write/rename error. + } + throw error; + } +} + +export function createJsonWriteQueue(filePath: string): (data: T) => void { + const queue: T[] = []; + let draining = false; + + return (data: T): void => { + queue.push(data); + + if (draining) return; + + draining = true; + try { + while (queue.length > 0) { + const next = queue.shift(); + writeJsonSafe(filePath, next); + } + } finally { + draining = false; + } + }; +} diff --git a/packages/orchestrator/src/activity-store.ts b/packages/orchestrator/src/activity-store.ts index f98c506..a2e1f5d 100644 --- a/packages/orchestrator/src/activity-store.ts +++ b/packages/orchestrator/src/activity-store.ts @@ -6,6 +6,7 @@ import fs from 'fs'; import path from 'path'; +import { writeJsonSafe } from '@clevercon/common'; const __dirname = path.dirname(path.resolve(process.argv[1])); const DATA_DIR = path.join(__dirname, '..', '..', '..', 'data'); @@ -41,7 +42,7 @@ function load(): Log { try { fs.mkdirSync(DATA_DIR, { recursive: true }); if (!fs.existsSync(LOG_PATH)) { - fs.writeFileSync(LOG_PATH, '[]', 'utf8'); + writeJsonSafe(LOG_PATH, []); } cache = JSON.parse(fs.readFileSync(LOG_PATH, 'utf8')) as Log; } catch { @@ -54,7 +55,7 @@ function save(log: Log): void { fs.mkdirSync(DATA_DIR, { recursive: true }); // Keep at most 1000 entries globally const trimmed = log.slice(-1000); - fs.writeFileSync(LOG_PATH, JSON.stringify(trimmed, null, 2), 'utf8'); + writeJsonSafe(LOG_PATH, trimmed); cache = trimmed; } diff --git a/packages/orchestrator/src/orchestrator-store.ts b/packages/orchestrator/src/orchestrator-store.ts index e95e457..638ea4e 100644 --- a/packages/orchestrator/src/orchestrator-store.ts +++ b/packages/orchestrator/src/orchestrator-store.ts @@ -8,6 +8,7 @@ import fs from 'fs'; import path from 'path'; +import { writeJsonSafe } from '@clevercon/common'; const __dirname = path.dirname(path.resolve(process.argv[1])); // process.argv[1] = .../packages/orchestrator/src/server.ts → go up 3 levels to workspace root @@ -34,7 +35,7 @@ function load(): Store { try { fs.mkdirSync(DATA_DIR, { recursive: true }); if (!fs.existsSync(STORE_PATH)) { - fs.writeFileSync(STORE_PATH, '{}', 'utf8'); + writeJsonSafe(STORE_PATH, {}); } cache = JSON.parse(fs.readFileSync(STORE_PATH, 'utf8')) as Store; } catch { @@ -45,7 +46,7 @@ function load(): Store { function save(store: Store): void { fs.mkdirSync(DATA_DIR, { recursive: true }); - fs.writeFileSync(STORE_PATH, JSON.stringify(store, null, 2), 'utf8'); + writeJsonSafe(STORE_PATH, store); cache = store; } diff --git a/packages/orchestrator/src/task-results.ts b/packages/orchestrator/src/task-results.ts index 6168291..f149bdd 100644 --- a/packages/orchestrator/src/task-results.ts +++ b/packages/orchestrator/src/task-results.ts @@ -5,7 +5,7 @@ import fs from 'fs'; import path from 'path'; -import type { TaskResult } from '@clevercon/common'; +import { writeJsonSafe, type TaskResult } from '@clevercon/common'; const __dirname = path.dirname(path.resolve(process.argv[1])); const DATA_DIR = path.join(__dirname, '..', '..', '..', 'data'); @@ -31,7 +31,7 @@ function load(): Store { if (cache) return cache; try { fs.mkdirSync(DATA_DIR, { recursive: true }); - if (!fs.existsSync(RESULTS_PATH)) fs.writeFileSync(RESULTS_PATH, '[]', 'utf8'); + if (!fs.existsSync(RESULTS_PATH)) writeJsonSafe(RESULTS_PATH, []); cache = JSON.parse(fs.readFileSync(RESULTS_PATH, 'utf8')) as Store; } catch { cache = []; @@ -42,7 +42,7 @@ function load(): Store { function save(store: Store): void { fs.mkdirSync(DATA_DIR, { recursive: true }); const trimmed = store.slice(-500); // cap at 500 entries - fs.writeFileSync(RESULTS_PATH, JSON.stringify(trimmed, null, 2), 'utf8'); + writeJsonSafe(RESULTS_PATH, trimmed); cache = trimmed; } diff --git a/packages/orchestrator/src/vault-ledger.ts b/packages/orchestrator/src/vault-ledger.ts index 5c296c4..d1da886 100644 --- a/packages/orchestrator/src/vault-ledger.ts +++ b/packages/orchestrator/src/vault-ledger.ts @@ -5,6 +5,7 @@ import fs from 'fs'; import path from 'path'; +import { writeJsonSafe } from '@clevercon/common'; const __dirname = path.dirname(path.resolve(process.argv[1])); const DATA_DIR = path.join(__dirname, '..', '..', '..', 'data'); @@ -31,7 +32,7 @@ function load(): Ledger { if (cache) return cache; try { fs.mkdirSync(DATA_DIR, { recursive: true }); - if (!fs.existsSync(LEDGER_PATH)) fs.writeFileSync(LEDGER_PATH, '[]', 'utf8'); + if (!fs.existsSync(LEDGER_PATH)) writeJsonSafe(LEDGER_PATH, []); cache = JSON.parse(fs.readFileSync(LEDGER_PATH, 'utf8')) as Ledger; } catch { cache = []; @@ -42,7 +43,7 @@ function load(): Ledger { function save(ledger: Ledger): void { fs.mkdirSync(DATA_DIR, { recursive: true }); const trimmed = ledger.slice(-2000); - fs.writeFileSync(LEDGER_PATH, JSON.stringify(trimmed, null, 2), 'utf8'); + writeJsonSafe(LEDGER_PATH, trimmed); cache = trimmed; } diff --git a/packages/registry/src/store.ts b/packages/registry/src/store.ts index 90c855c..7391002 100644 --- a/packages/registry/src/store.ts +++ b/packages/registry/src/store.ts @@ -1,18 +1,18 @@ /** * JSON-file-backed persistence for the agent registry. * - * All agents are stored as a single array in `data/registry.json`. Each - * exported function reads or rewrites the whole file — there is no write - * locking, so concurrent writers can race (see SECURITY.md "Known limitations"). + * All agents are stored as a single array in `data/registry.json`. Writes are + * serialized through an in-process queue and committed with atomic renames. */ import fs from 'fs'; import path from 'path'; import { fileURLToPath } from 'url'; -import type { AgentRecord } from '@clevercon/common'; +import { createJsonWriteQueue, type AgentRecord } from '@clevercon/common'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); const DATA_DIR = path.join(__dirname, '..', '..', '..', 'data'); const REGISTRY_FILE = path.join(DATA_DIR, 'registry.json'); +const writeAgentsQueued = createJsonWriteQueue(REGISTRY_FILE); function ensureDataDir(): void { if (!fs.existsSync(DATA_DIR)) { @@ -37,7 +37,7 @@ export function loadAgents(): AgentRecord[] { /** Overwrite `data/registry.json` with the given list of agents. */ export function saveAgents(agents: AgentRecord[]): void { ensureDataDir(); - fs.writeFileSync(REGISTRY_FILE, JSON.stringify(agents, null, 2)); + writeAgentsQueued(agents); } /** Find a single agent by its `agent_id`, or `undefined` if not registered. */