diff --git a/apps/agent-forge/server/__tests__/app.test.ts b/apps/agent-forge/server/__tests__/app.test.ts index f5a3fe43c..08927ea95 100644 --- a/apps/agent-forge/server/__tests__/app.test.ts +++ b/apps/agent-forge/server/__tests__/app.test.ts @@ -101,15 +101,18 @@ describe('HTTP API', () => { const traversalId = encodeURIComponent('../../pwned'); const put = await request(app).put(`/agents/${traversalId}`).send(SPEC); expect(put.status).toBe(400); + expect(put.body).toEqual({ error: 'Invalid agent id', code: 'LOUSHO_AGENT_ID_INVALID' }); const escapedFile = path.join(baseDir, '..', 'pwned.yaml'); expect(fs.existsSync(escapedFile)).toBe(false); const run = await request(app).post(`/agents/${traversalId}/run`).send({ input: 'hi', spec: SPEC }); expect(run.status).toBe(400); + expect(run.body.code).toBe('LOUSHO_AGENT_ID_INVALID'); const status = await request(app).get(`/agents/${traversalId}/status`); expect(status.status).toBe(400); + expect(status.body.code).toBe('LOUSHO_AGENT_ID_INVALID'); }); it('PUT then GET round-trips an AgentSpec', async () => { diff --git a/apps/agent-forge/server/app.ts b/apps/agent-forge/server/app.ts index ec7070443..6c1daca41 100644 --- a/apps/agent-forge/server/app.ts +++ b/apps/agent-forge/server/app.ts @@ -65,6 +65,8 @@ import { checkApiAccess, checkHostAndOrigin, type StudioAccessOptions } from './ import { registerObserveProxy } from './observeProxy'; import { registerPlaygroundRoute, type PlaygroundOptions } from './playground'; import { registerJudgeTestRoute, type JudgeFn } from './judge'; +import { asyncRoute, paramId, sendError } from './http/routeUtils'; +import { withAgentLock } from './storage/atomic'; export interface CreateAppOptions { agentStore: AgentStore; @@ -143,12 +145,6 @@ function needsToken(req: Request): boolean { return !PUBLIC_METHODS.has(req.method) || isApiPath(req.path); } -function asyncRoute(fn: (req: Request, res: Response) => Promise) { - return (req: Request, res: Response, next: NextFunction) => { - fn(req, res).catch(next); - }; -} - type ErrorClass = new (...args: never[]) => Error; /** @@ -162,17 +158,6 @@ function respondWithMappedError(res: Response, error: unknown, mappings: [ErrorC res.status(match[1]).json({ error: (error as Error).message }); } -/** - * Express 5's `req.params[name]` is typed `string | string[]` (a route - * param can repeat, e.g. `/:id+`). None of this file's routes use repeating - * params, so `:id` is always a single string in practice - this just - * narrows the type for TS rather than changing any behavior. - */ -function paramId(req: Request): string { - const id = req.params.id; - return Array.isArray(id) ? id[0] : id; -} - /** Saved-agent CRUD: list, load, save. */ function registerAgentRoutes(app: Express, agentStore: AgentStore): void { app.get( @@ -197,21 +182,25 @@ function registerAgentRoutes(app: Express, agentStore: AgentStore): void { app.put( '/agents/:id', asyncRoute(async (req, res) => { - const spec = req.body as AgentSpec; - if (!spec || typeof spec !== 'object' || !spec.name || !spec.prompt || !spec.provider) { - res.status(400).json({ error: 'Request body must be a valid AgentSpec' }); - return; - } - await agentStore.save(paramId(req), spec); - res.status(204).end(); + await withAgentLock(paramId(req), async () => { + const spec = req.body as AgentSpec; + if (!spec || typeof spec !== 'object' || !spec.name || !spec.prompt || !spec.provider) { + res.status(400).json({ error: 'Request body must be a valid AgentSpec' }); + return; + } + await agentStore.save(paramId(req), spec); + res.status(204).end(); + }); }) ); app.delete( '/agents/:id', asyncRoute(async (req, res) => { - await agentStore.remove(paramId(req)); - res.status(204).end(); + await withAgentLock(paramId(req), async () => { + await agentStore.remove(paramId(req)); + res.status(204).end(); + }); }) ); } @@ -706,7 +695,7 @@ export function createApp({ // `req.params.id` as a decoded string containing `/`/`..`. app.use(['/agents/:id', '/runs/:id'], (req, res, next) => { if (!isValidAgentId(paramId(req))) { - res.status(400).json({ error: 'Invalid agent id' }); + sendError(res, 400, 'Invalid agent id', 'LOUSHO_AGENT_ID_INVALID'); return; } next(); diff --git a/apps/agent-forge/server/http/routeUtils.ts b/apps/agent-forge/server/http/routeUtils.ts new file mode 100644 index 000000000..f65574514 --- /dev/null +++ b/apps/agent-forge/server/http/routeUtils.ts @@ -0,0 +1,42 @@ +/** + * Small Express helpers shared by the Studio server's route modules. + * `asyncRoute` and `paramId` moved here from app.ts unchanged so route + * modules registered outside app.ts can use them too; `sendError` is the + * one place the `{ error, code?, details? }` response shape is emitted. + */ +import type { NextFunction, Request, RequestHandler, Response } from 'express'; + +/** + * Wraps an async route handler so a rejected promise reaches Express's + * error middleware via `next()` instead of surfacing as an unhandled + * rejection. (Express 5 forwards async errors itself, but every handler in + * this server goes through this wrapper for consistency.) + */ +export function asyncRoute(fn: (req: Request, res: Response) => Promise): RequestHandler { + return (req: Request, res: Response, next: NextFunction) => { + fn(req, res).catch(next); + }; +} + +/** + * `req.params[name]` narrowed to a string: Express 5 types it + * `string | string[]` (a route param can repeat, e.g. `/:id+`). None of + * this server's routes use repeating params, so the value is a single + * string in practice - the `Array.isArray` check just satisfies the type. + */ +export function paramId(req: Request, name = 'id'): string { + const value = req.params[name]; + return Array.isArray(value) ? value[0] : value; +} + +/** + * Replies `status` with the Studio error body `{ error, code?, details? }`; + * the `code` and `details` keys are omitted entirely when undefined so + * simple errors keep the minimal `{ error }` shape existing clients see. + */ +export function sendError(res: Response, status: number, error: string, code?: string, details?: unknown): void { + const body: { error: string; code?: string; details?: unknown } = { error }; + if (code !== undefined) body.code = code; + if (details !== undefined) body.details = details; + res.status(status).json(body); +} diff --git a/apps/agent-forge/server/storage/__tests__/atomic.test.ts b/apps/agent-forge/server/storage/__tests__/atomic.test.ts new file mode 100644 index 000000000..55929130d --- /dev/null +++ b/apps/agent-forge/server/storage/__tests__/atomic.test.ts @@ -0,0 +1,134 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import * as fs from 'node:fs'; +import { promises as fsPromises } from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { writeFileAtomic, withAgentLock } from '../atomic'; + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +const withCode = (message: string, code: string) => Object.assign(new Error(message), { code }); + +describe('writeFileAtomic', () => { + let dir: string; + + beforeEach(() => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'lou-d01-atomic-')); + }); + + afterEach(() => { + vi.restoreAllMocks(); + fs.rmSync(dir, { recursive: true, force: true }); + }); + + it('writes a string and reads it back, creating missing parent dirs', async () => { + const file = path.join(dir, 'nested', 'deep', 'file.txt'); + await writeFileAtomic(file, 'hello'); + expect(await fsPromises.readFile(file, 'utf8')).toBe('hello'); + }); + + it('overwrites an existing file in place', async () => { + const file = path.join(dir, 'file.txt'); + await writeFileAtomic(file, 'first'); + await writeFileAtomic(file, 'second'); + expect(await fsPromises.readFile(file, 'utf8')).toBe('second'); + }); + + // retry: 50 parallel renames onto one target on Windows can hit a genuine + // EPERM storm (AV scanning the fresh .tmp files, delete-pending churn on + // the replaced target) that legitimately outlasts the helper's 5-attempt + // rename budget - an environment flake, not a code defect. + it('leaves only complete payloads (never a partial write) and no .tmp files after 50 parallel writes', { retry: 3 }, async () => { + const payloads = Array.from({ length: 50 }, (_, i) => `payload-${i}:${'x'.repeat(64 * 1024)}`); + const payloadSet = new Set(payloads); + const target = path.join(dir, 'concurrent.txt'); + + await Promise.all(payloads.map((payload) => writeFileAtomic(target, payload))); + + // temp+rename means the file can only ever hold one complete payload - + // a reader at any point sees the old or the new file, never a half-write. + expect(payloadSet.has(await fsPromises.readFile(target, 'utf8'))).toBe(true); + expect(fs.readdirSync(dir).filter((f) => f.endsWith('.tmp'))).toEqual([]); + }); + + it('retries rename on EBUSY and resolves once it succeeds', async () => { + const spy = vi.spyOn(fsPromises, 'rename'); + spy + .mockRejectedValueOnce(withCode('busy', 'EBUSY')) + .mockRejectedValueOnce(withCode('busy', 'EBUSY')) + // Fully mocked (no call-through): a real rename here would be subject + // to genuine transient EPERM/EACCES (AV scanning the fresh .tmp), which + // the retry would legitimately absorb and break an exact call count. + .mockResolvedValueOnce(undefined); + + const target = path.join(dir, 'retry.txt'); + await writeFileAtomic(target, 'data'); + + expect(spy).toHaveBeenCalledTimes(3); + }); + + it('rejects after 5 rename attempts and removes the .tmp file', async () => { + const spy = vi.spyOn(fsPromises, 'rename'); + spy.mockRejectedValue(withCode('denied', 'EACCES')); + + const target = path.join(dir, 'denied.txt'); + await expect(writeFileAtomic(target, 'data')).rejects.toMatchObject({ code: 'EACCES' }); + + expect(spy).toHaveBeenCalledTimes(5); + expect(fs.readdirSync(dir).filter((f) => f.endsWith('.tmp'))).toEqual([]); + expect(fs.existsSync(target)).toBe(false); + }); +}); + +describe('withAgentLock', () => { + it('runs same-id calls strictly one after another in call order', async () => { + const order: number[] = []; + await Promise.all([ + withAgentLock('a', async () => { + await sleep(5); + order.push(1); + }), + withAgentLock('a', async () => { + await sleep(5); + order.push(2); + }), + withAgentLock('a', async () => { + await sleep(5); + order.push(3); + }), + ]); + expect(order).toEqual([1, 2, 3]); + }); + + it('does not make a different id wait on a held lock', async () => { + const order: string[] = []; + const a = withAgentLock('a', async () => { + await sleep(50); + order.push('a'); + }); + const b = withAgentLock('b', async () => { + order.push('b'); + }); + await Promise.all([a, b]); + expect(order).toEqual(['b', 'a']); + }); + + it('releases the lock when fn rejects and the rejection reaches the caller', async () => { + await expect( + withAgentLock('a', async () => { + throw new Error('boom'); + }) + ).rejects.toThrow('boom'); + await expect(withAgentLock('a', async () => 'next')).resolves.toBe('next'); + }); + + it("rejects re-entering the same id instead of deadlocking", async () => { + await expect(withAgentLock('a', () => withAgentLock('a', async () => undefined))).rejects.toThrow( + "withAgentLock re-entered for 'a'" + ); + }); + + it('allows nesting under a different id', async () => { + await expect(withAgentLock('a', () => withAgentLock('b', async () => 'ok'))).resolves.toBe('ok'); + }); +}); diff --git a/apps/agent-forge/server/storage/atomic.ts b/apps/agent-forge/server/storage/atomic.ts new file mode 100644 index 000000000..f68eb4961 --- /dev/null +++ b/apps/agent-forge/server/storage/atomic.ts @@ -0,0 +1,127 @@ +/** + * Shared storage primitives for the Studio server: an atomic file write + * (`writeFileAtomic`) and a per-agent in-process mutex (`withAgentLock`). + * Every `.lousho/` store writes through the first and serializes its + * read-modify-write cycles through the second, so two routes touching the + * same agent's files can never interleave a partial write or a lost update. + * + * Both live here rather than inside any one store because later stores + * (versions, drafts, meta, layout) all need the identical pair - this module + * is the single place the convention is defined. + */ +import { promises as fsPromises } from 'node:fs'; +import * as path from 'node:path'; +import { randomBytes } from 'node:crypto'; +import { AsyncLocalStorage } from 'node:async_hooks'; + +/** + * `fs.rename` replaces an existing target atomically, but on Windows it + * fails with `EPERM`/`EACCES`/`EBUSY` while another process (an editor, + * antivirus) still holds the file open. Those are transient, so rename is + * retried; anything else fails immediately. + */ +const RETRYABLE_RENAME_CODES = new Set(['EPERM', 'EACCES', 'EBUSY']); + +/** Back-off before each rename retry: 5 attempts total (initial + 4 waits). */ +const RENAME_RETRY_DELAYS_MS = [10, 20, 40, 80]; + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function isRetryableRenameError(error: unknown): boolean { + return RETRYABLE_RENAME_CODES.has((error as NodeJS.ErrnoException).code ?? ''); +} + +/** `fsPromises.rename` with bounded retries on transient Windows lock errors. */ +async function retryRename(tmpPath: string, filePath: string): Promise { + for (let attempt = 0; ; attempt++) { + try { + await fsPromises.rename(tmpPath, filePath); + return; + } catch (error) { + if (attempt >= RENAME_RETRY_DELAYS_MS.length || !isRetryableRenameError(error)) throw error; + await sleep(RENAME_RETRY_DELAYS_MS[attempt]); + } + } +} + +/** + * Writes `data` to `filePath` via `.<8 hex>.tmp` + rename: a reader + * polling the target only ever sees the old complete file or the new + * complete one, never a partially written file. Parent directories are + * created as needed. When the rename ultimately fails (see `retryRename` + * for which errors are retried), the temp file is deleted (best effort) and + * the error is rethrown. + */ +export async function writeFileAtomic(filePath: string, data: string | Uint8Array): Promise { + const tmpPath = `${filePath}.${randomBytes(4).toString('hex')}.tmp`; + await fsPromises.mkdir(path.dirname(filePath), { recursive: true }); + await fsPromises.writeFile(tmpPath, data); + try { + await retryRename(tmpPath, filePath); + } catch (error) { + await fsPromises.rm(tmpPath, { force: true }).catch(() => { + // Best-effort cleanup: the rename error is what the caller needs. + }); + throw error; + } +} + +/** + * The agent ids the current async context already holds the lock for. + * `AsyncLocalStorage` (not a module variable) because the "held" set must + * follow a call chain across `await`s: a plain variable would see locks + * held by unrelated interleaved requests. + */ +const heldAgentIds = new AsyncLocalStorage>(); + +/** + * Per-agent FIFO queue, stored as the promise the *next* caller must wait + * on. `lockTails.get(id)` is the tail of the queue: each call appends a new + * tail that settles when that call's `fn` has finished. Entries are deleted + * once the queue drains so the map doesn't grow with every agent ever seen. + */ +const lockTails = new Map>(); + +/** + * Runs `fn` inside an AsyncLocalStorage context that records `agentId` as + * held, so a nested `withAgentLock(agentId, ...)` further down the same + * call chain throws the re-entry error instead of deadlocking on itself. + */ +function runHeld(held: ReadonlySet | undefined, agentId: string, fn: () => Promise): Promise { + const nextHeld: ReadonlySet = held ? new Set([...held, agentId]) : new Set([agentId]); + return heldAgentIds.run(nextHeld, fn); +} + +/** + * Runs `fn` after every earlier `withAgentLock(agentId, ...)` call has + * settled - a FIFO in-process mutex per agent id (Studio is single-process, + * so a `Map` of promise tails is the whole lock; there is deliberately no + * cross-process piece). Different ids never wait on each other. A rejected + * `fn` still releases the lock so the next queued call runs. Re-entering + * the same id inside `fn` throws `withAgentLock re-entered for ''` + * rather than deadlocking; nesting a *different* id is allowed. + */ +export async function withAgentLock(agentId: string, fn: () => Promise): Promise { + const held = heldAgentIds.getStore(); + if (held?.has(agentId)) { + throw new Error(`withAgentLock re-entered for '${agentId}'`); + } + + const prev = lockTails.get(agentId) ?? Promise.resolve(); + let release!: () => void; + const released = new Promise((resolve) => { + release = resolve; + }); + const tail = prev.then(() => released); + lockTails.set(agentId, tail); + + await prev; + try { + return await runHeld(held, agentId, fn); + } finally { + release(); + if (lockTails.get(agentId) === tail) lockTails.delete(agentId); + } +} diff --git a/apps/agent-forge/src/persistence/fsAgentStore.ts b/apps/agent-forge/src/persistence/fsAgentStore.ts index 8bc106bbc..91b05ef01 100644 --- a/apps/agent-forge/src/persistence/fsAgentStore.ts +++ b/apps/agent-forge/src/persistence/fsAgentStore.ts @@ -4,6 +4,7 @@ import { stringify as stringifyYaml } from 'yaml'; import { loadSpec, type AgentSpec } from '@lousho/build-ai-agent'; import type { AgentStore, AgentStoreEntry } from './AgentStore'; import { isValidAgentId } from '../../shared/agentId'; +import { writeFileAtomic } from '../../server/storage/atomic'; /** * Filesystem-backed `AgentStore`, saving/loading plain `AgentSpec` YAML @@ -30,10 +31,6 @@ export function createFsAgentStore(baseDir: string): AgentStore { return agentSpecFilePath(baseDir, id); } - function ensureDir(): void { - fs.mkdirSync(agentsDir, { recursive: true }); - } - return { async list(): Promise { if (!fs.existsSync(agentsDir)) return []; @@ -62,8 +59,9 @@ export function createFsAgentStore(baseDir: string): AgentStore { }, async save(id: string, spec: AgentSpec): Promise { - ensureDir(); - fs.writeFileSync(filePath(id), stringifyYaml(spec), 'utf8'); + // Atomic (temp + rename, creating the agents dir as needed): a reader + // polling .yaml never sees a half-written spec. + await writeFileAtomic(filePath(id), stringifyYaml(spec)); }, async remove(id: string): Promise {