fix(serve): wait to resume a transcript an in-flight submit is writing

CHANGELOG.md

@@ -7,6 +7,9 @@
   sessions or session-less frames such as `health` on the same socket. Frames
   for one session keep their order; replies across sessions may interleave,
   so clients match them by JSON-RPC `id` (Ossuary already does).
+  `session.resume` shares the lane of `params.id`, so resuming the transcript
+  a submit is still writing waits until that run finishes instead of forking
+  a mid-run history that drops the in-flight user turn.
 - Gateway (#144): sender allowlists and the gateway toolset are one
   `GatewayPolicy`, built from config at startup (`src/gateway/access.ts`).
   Platform adapters declare `capabilities` (`{ kind: "text", max_reply_chars }`);

docs/architecture/serve.md

@@ -118,6 +118,9 @@ same WebSocket that issued `prompt.submit` while the call is still in flight.
   `params.session_id` match are handled in arrival order and get in-order
   replies, while other sessions, and frames that name no session (such as
   `health`, `session.list` or `session.create`), do not wait behind them.
+  `session.resume` is queued by `params.id` instead, so resuming the transcript
+  an in-flight `prompt.submit` is still writing waits for that run. A resume
+  of a different id does not.
   Replies from different sessions can therefore interleave; match them by
   JSON-RPC `id`.
   `prompt.abort` is dispatched immediately so it can cancel an in-flight

src/serve/server.ts

@@ -236,6 +236,7 @@ function attach_client(
   // Serialize frames per session on this connection: frames for one session_id
   // get in-order replies, while different sessions (and session-less frames
   // such as health or session.list) no longer wait behind each other's runs.
+  // session.resume is keyed by params.id so it waits on the transcript it reads.
   // Replies across lanes can interleave; clients match them by JSON-RPC id.
   // prompt.abort is on no lane. It must run while prompt.submit is still
   // awaiting the model; waiting would deadlock a hung run with its own cancel.
@@ -271,13 +272,23 @@ function attach_client(
   });
 }
 
-/** Lane for a frame: its params.session_id, or "" for frames that name no session. */
+/**
+ * Lane for a frame. `prompt.submit` / `session.clear` use `params.session_id`.
+ * `session.resume` uses `params.id` (the transcript being read) so a resume of
+ * the file an in-flight submit is appending waits until that run finishes.
+ * Other frames share "".
+ */
 function frame_lane(data: RawData): string {
-  const params = parse_frame_object(data)?.params;
+  const body = parse_frame_object(data);
+  const params = body?.params;
   if (typeof params !== "object" || params === null || Array.isArray(params)) {
     return "";
   }
-  const session_id = (params as { session_id?: unknown }).session_id;
+  const record = params as { session_id?: unknown; id?: unknown };
+  if (body?.method === "session.resume" && typeof record.id === "string" && record.id.length > 0) {
+    return `session:${record.id}`;
+  }
+  const session_id = record.session_id;
   return typeof session_id === "string" && session_id.length > 0 ? `session:${session_id}` : "";
 }
 

test/serve_prompts.test.ts

@@ -1204,6 +1204,82 @@ describe("serve prompt over websocket", () => {
     }
   });
 
+  it("waits to resume the transcript an in-flight submit is writing", async () => {
+    const work_dir = await make_temp_dir("serve-ws-resume-lane");
+    const session_dir = path.join(work_dir, "sessions");
+    let release_hang!: () => void;
+    const hang = new Promise<void>((resolve) => {
+      release_hang = resolve;
+    });
+    let hang_started!: () => void;
+    const started = new Promise<void>((resolve) => {
+      hang_started = resolve;
+    });
+    const fetch_fn: typeof fetch = async (_url, init) => {
+      const body = JSON.parse(String(init?.body)) as { messages: Array<{ content?: string }> };
+      if (body.messages.some((message) => message.content === "hang")) {
+        hang_started();
+        await hang;
+      }
+      return new Response(
+        JSON.stringify({
+          model: "mock-model",
+          choices: [{ message: { role: "assistant", content: "done" }, finish_reason: "stop" }],
+          usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 },
+        }),
+        { status: 200 },
+      );
+    };
+    const agent = mock_agent(work_dir, fetch_fn);
+    const server = create_serve_server({ port: 0, boot_stdout: null, session_dir, agent, version: "9.9.9" });
+    servers.push(server);
+    const boot = await server.start();
+    const ws = await open_ws(`ws://127.0.0.1:${boot.port}/?token=${encodeURIComponent(boot.token)}`);
+    try {
+      const create = async (id: number): Promise<string> =>
+        ((await request_rpc(ws, { jsonrpc: "2.0", id, method: "session.create", params: { source: "test" } })).result as {
+          session_id: string;
+        }).session_id;
+      const live = await create(1);
+      const other = await create(2);
+      const slow = request_rpc(ws, {
+        jsonrpc: "2.0",
+        id: 3,
+        method: "prompt.submit",
+        params: { session_id: live, text: "hang" },
+      });
+      await started;
+      let resumed_live = false;
+      const resume_live = request_rpc(ws, {
+        jsonrpc: "2.0",
+        id: 4,
+        method: "session.resume",
+        params: { id: live },
+      }).then((body) => {
+        resumed_live = true;
+        return body;
+      });
+      const resume_other = await request_rpc(ws, {
+        jsonrpc: "2.0",
+        id: 5,
+        method: "session.resume",
+        params: { id: other },
+      });
+      expect(resume_other.result).toMatchObject({ resumed_id: other, message_count: 0 });
+      await new Promise((resolve) => setTimeout(resolve, 200));
+      expect(resumed_live).toBe(false);
+      release_hang();
+      const resumed = await resume_live;
+      expect((await slow).result).toMatchObject({ session_id: live, stopped_reason: "final" });
+      const result = resumed.result as { message_count: number; resumed_id: string };
+      expect(result.resumed_id).toBe(live);
+      expect(result.message_count).toBeGreaterThanOrEqual(2);
+    } finally {
+      release_hang();
+      ws.close();
+    }
+  });
+
   it("stop() aborts an in-flight prompt instead of draining the model call", async () => {
     const work_dir = await make_temp_dir("serve-stop-abort");
     const session_dir = path.join(work_dir, "sessions");