diff --git a/CHANGELOG.md b/CHANGELOG.md index 175d29f..afcd3c7 100644 --- a/CHANGELOG.md +++ b/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 }`); diff --git a/docs/architecture/serve.md b/docs/architecture/serve.md index 78056d9..7460c16 100644 --- a/docs/architecture/serve.md +++ b/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 diff --git a/src/serve/server.ts b/src/serve/server.ts index a8f5ebd..65c06b5 100644 --- a/src/serve/server.ts +++ b/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}` : ""; } diff --git a/test/serve_prompts.test.ts b/test/serve_prompts.test.ts index b384e30..6d93968 100644 --- a/test/serve_prompts.test.ts +++ b/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((resolve) => { + release_hang = resolve; + }); + let hang_started!: () => void; + const started = new Promise((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 => + ((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");