D01: Atomic write helper, per-agent lock and shared route utils

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 () => {

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<void>) {
-  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();

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<void>): 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);
+}

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');
+  });
+});