From 7cfd8b5bc74529dcbaaa4010c21a0d9f2d93529d Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Fri, 25 Sep 2026 17:56:12 +0300 Subject: [PATCH] Background turn runner: every turn is a server-side process, clients are views MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Owner ruling 2026-09-25. runPiTurn executes a turn decoupled from any HTTP response: an unbounded memory-only event log (finite by construction; the 30-min liveness TTL bounds the pathological case) plus a subscriber set. The POST /api/chat stream and the new GET /api/chats/[id]/stream are both subscribers — identical SSE dialect; late subscribers get FULL replay (starting at the start chunk) then live until finish + [DONE]. A view cancelling (reload, navigate) now detaches ONLY — never aborts the turn. The remote kill is the abort channel: POST /api/activity/abort gains kind: "turn" (id = chatId; not_running 409 like jobs); handle.abort sets sawAbort + agent.abort(), every subscriber closes cleanly, the partial persists through the usual settle (onFinish runs once, then endTurn, then unregisterTurn LAST so a view that finds nothing finds the record current). No running turn on reattach: the cheap terminal — one {"type":"turn-settled"} event + [DONE] (settledTurnResponse). The live smoke caught what the unit suite had masked: the agent_end close() had been left behind in the restructure — ReferenceError every turn, steered into the error path; fixed and pinned (clean turns must never emit an error chunk; finishReason stop present). --- app/api/activity/abort/route.ts | 28 +- app/api/chats/[id]/stream/route.ts | 21 ++ src/server/chat-handler.ts | 25 +- src/server/index.ts | 2 + src/server/pi/ui-stream.ts | 494 +++++++++++++++++------------ src/server/turn-runs.ts | 81 +++++ tests/chat-flow.test.ts | 187 ++++++++++- 7 files changed, 608 insertions(+), 230 deletions(-) create mode 100644 app/api/chats/[id]/stream/route.ts create mode 100644 src/server/turn-runs.ts diff --git a/app/api/activity/abort/route.ts b/app/api/activity/abort/route.ts index 5b5de06..fac8b14 100644 --- a/app/api/activity/abort/route.ts +++ b/app/api/activity/abort/route.ts @@ -1,16 +1,23 @@ /** * POST /api/activity/abort — the dock's END/ABORT control, uniform across - * process kinds (user-decided contract): body {kind: "job"|"extraction", id} - * where id is the jobId for sub-agent jobs and the chatId for extraction - * runs (the pipeline registry is chat-keyed). Mechanisms live in their - * domains — abortJob (server/jobs.ts, agents stream) and abortExtraction - * (the pipeline module, graph stream); this route is the thin composition. + * process kinds (user-decided contract): body {kind: "job"|"extraction"|"turn", id} + * where id is the jobId for sub-agent jobs, the chatId for extraction runs + * (the pipeline registry is chat-keyed), and the chatId for turns + * (background-turn wave, 2026-09-25: a running turn is a background + * object — any client may stop it here; a view leaving NEVER kills it). + * Mechanisms live in their domains — abortJob (server/jobs.ts, agents + * stream), abortExtraction (the pipeline module, graph stream), and + * abortTurn (server/turn-runs.ts, the running-turn registry); this route + * is the thin composition. * * Outcomes: a job's not_found -> 404, not_running -> 409, aborted -> 200. + * A turn mirrors the job semantics: not_running -> 409 (honest "nothing to + * abort" — the caller refetches), aborted -> 200 (partial persists via + * the usual settle path, every subscriber closes cleanly). * Extraction abort is idempotent by design (aborting with nothing in * flight is success — the blob is gone either way) -> 200. */ -import { abortExtraction, abortJob } from "@/src/server"; +import { abortExtraction, abortJob, abortTurn } from "@/src/server"; export async function POST(req: Request) { let body: unknown; @@ -20,8 +27,8 @@ export async function POST(req: Request) { return Response.json({ error: "Invalid JSON body" }, { status: 400 }); } const { kind, id } = (body ?? {}) as { kind?: string; id?: string }; - if ((kind !== "job" && kind !== "extraction") || typeof id !== "string" || id === "") { - return Response.json({ error: 'Body must be {kind: "job"|"extraction", id: string}' }, { status: 400 }); + if ((kind !== "job" && kind !== "extraction" && kind !== "turn") || typeof id !== "string" || id === "") { + return Response.json({ error: 'Body must be {kind: "job"|"extraction"|"turn", id: string}' }, { status: 400 }); } if (!/^[A-Za-z0-9_-]+$/.test(id)) { return Response.json({ error: "Invalid id" }, { status: 400 }); @@ -35,6 +42,11 @@ export async function POST(req: Request) { } return Response.json({ ok: true }); } + if (kind === "turn") { + const outcome = abortTurn(id); + if (!outcome.ok) return Response.json({ ok: false, reason: outcome.reason }, { status: 409 }); + return Response.json({ ok: true }); + } // kind === "extraction": true cancellation; idempotent (nothing in // flight is success — same dock outcome) abortExtraction(id); diff --git a/app/api/chats/[id]/stream/route.ts b/app/api/chats/[id]/stream/route.ts new file mode 100644 index 0000000..aa92275 --- /dev/null +++ b/app/api/chats/[id]/stream/route.ts @@ -0,0 +1,21 @@ +/** + * THE REATTACH STREAM (background-turn runner wave, owner ruling + * 2026-09-25): one SSE stream per chat, same wire dialect as the POST + * /api/chat stream. A running turn → subscribe to the live turn object: + * FULL replay of the event log, then live chunks until settle ([DONE]). + * No running turn → the cheap terminal ({"type":"turn-settled"} + [DONE]): + * the record is current, the client refetches it. + * + * 404 mirrors the POST handler's unknown-chat. Framework route is a thin + * wrapper over the server barrel (the import law); tests drive the seams + * (turn-runs + ui-stream) directly. + */ +import { getChat, runningTurn, piTurnResponse, settledTurnResponse } from "@/src/server"; + +export async function GET(_req: Request, ctx: { params: Promise<{ id: string }> }): Promise { + const { id } = await ctx.params; + if (!getChat(id)) return Response.json({ error: "Unknown chat" }, { status: 404 }); + const handle = runningTurn(id); + if (!handle) return settledTurnResponse(); + return piTurnResponse(handle); +} diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index 5c77728..f6c6c81 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -7,7 +7,7 @@ import { createExtractionPipeline } from "../agent/subagents/extraction.ts"; import type { ModelRole } from "../../config.ts"; import { claimSteers, deliverJobResult, setJobRespawn, setTurnRunner } from "./jobs.ts"; import { recordToPiMessages } from "./pi/record-adapter.ts"; -import { createPiTurnResponse } from "./pi/ui-stream.ts"; +import { piTurnResponse, runPiTurn } from "./pi/ui-stream.ts"; import { effectiveSettings, stores } from "./stores.ts"; import { searchAvailable, searchProvider } from "./search.ts"; import { providerKeyPresent } from "./keys.ts"; @@ -19,6 +19,7 @@ import { } from "../core/settings-resolution.ts"; import { modelsNotConfigured, providerKeyMissing as providerKeyMissingEnvelope, type ProviderKeyMissing } from "../core/settings.ts"; import { beginTurn, endTurn } from "./turn-liveness.ts"; +import { registerTurn, unregisterTurn } from "./turn-runs.ts"; import { persistTurn, triggerExtraction, withMessageIds } from "./turn-service.ts"; import type { CallUsage } from "../core/contracts.ts"; @@ -310,10 +311,14 @@ export async function handleChatRequest( // the factory's subscription (wired above via onUsage). try { assertValidUiMessages(outgoing); - return createPiTurnResponse({ + // THE TURN AS A BACKGROUND OBJECT (owner ruling, 2026-09-25): + // runPiTurn executes decoupled from this response — the POST stream + // is just the FIRST subscriber (a view). A client disconnect can no + // longer abort the turn; the remote kill is the abort channel, and + // late views reattach via GET /api/chats/[id]/stream. + const handle = runPiTurn({ agent: constructed, uiMessages: outgoing, - signal: req.signal, // Stream-level failures are provider/transport internals — masked, // like the SDK defaults. Tool-input validation messages never // route through here: they ride the error-result path and reach the @@ -352,8 +357,22 @@ export async function handleChatRequest( // LAST: a poller that sees not-in-flight must read an already-persisted // record — endTurn only after persistTurn has done its job. endTurn(chatId); + // And only then does the turn stop being a running object + // (background-turn wave, 2026-09-25): a reattach that finds + // nothing must find the persisted record current — the same + // ordering rule as endTurn. + unregisterTurn(chatId); }, }); + // The turn is now a live background object: register it so any view + // can reattach (GET /api/chats/[id]/stream) or remotely abort it + // (POST /api/activity/abort, kind "turn"). Registration after the + // runPiTurn call is race-free: the agent loop only ticks in later + // microtasks, so settle (and thus unregister) cannot beat us here. + registerTurn(chatId, handle); + // The POST response: just this caller's view onto the running turn — + // identical wire dialect to the reattach stream. + return piTurnResponse(handle); } catch (err) { // The stream never started, so no onFinish will fire — clear the registry // or a malformed request would lock the chat until the TTL eviction. diff --git a/src/server/index.ts b/src/server/index.ts index fd23306..3a2088e 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -36,5 +36,7 @@ export * from "./search.ts"; export { authStore } from "../store/auth.ts"; export * from "./settings-body.ts"; export * from "./turn-liveness.ts"; +export * from "./turn-runs.ts"; +export { runPiTurn, piTurnResponse, type PiTurnHandle, type PiTurnSpec } from "./pi/ui-stream.ts"; export * from "./observability.ts"; export * from "./model-catalog.ts"; diff --git a/src/server/pi/ui-stream.ts b/src/server/pi/ui-stream.ts index 257952b..3fda1d1 100644 --- a/src/server/pi/ui-stream.ts +++ b/src/server/pi/ui-stream.ts @@ -26,6 +26,14 @@ import { QUESTION_TOOLS } from "../../core/pending.ts"; * - The persisted message is assembled from the SAME parts the client * assembles (the processUIMessageStream shapes), so every downstream * consumer (phases, plan blocks, pending, extraction) is unchanged. + * + * BACKGROUND-TURN RUNNER (owner ruling, 2026-09-25): a turn executes as a + * server-side background object decoupled from any single HTTP response — + * clients are VIEWS. runPiTurn returns the handle (event log + subscriber + * set); piTurnResponse wraps a subscriber stream as the SSE response, and + * the POST /api/chat response and GET /api/chats/[id]/stream are BOTH + * exactly that. Cancelling a view detaches it ONLY — never aborts the + * turn; the handle's abort() is the one remote kill. */ /** SSE chunk — the flat JSON line shape the client's zod schema accepts. */ @@ -36,14 +44,34 @@ export interface PiTurnSpec { agent: Agent; /** The outgoing request messages (write-ahead persisted already). */ uiMessages: UIMessage[]; - /** Request abort signal — aborts the agent mid-run. */ - signal?: AbortSignal; /** The handler's error-unmasking policy (validation messages stay readable). */ onError?: (error: unknown) => string; /** The canonical end-of-turn seam: persist + usage + extraction + endTurn. */ onFinish: (messages: UIMessage[], isAborted: boolean) => void; } +/** One view onto a running turn: push() replays/streams a chunk, done() + * terminates the view ([DONE] + close). Cheap to be a Set member. */ +export interface PiTurnSubscriber { + push(chunk: Chunk): void; + done(): void; +} + +/** The server-side turn object (background-turn runner wave, 2026-09-25): + * runs the agent decoupled from any single HTTP response and owns the + * in-memory event log + the live subscriber set. The POST response and + * the reattach stream are both subscribers onto it. */ +export interface PiTurnHandle { + /** A fresh subscriber stream: FULL replay of the log, then live events + * until the turn settles (ending [DONE]). Cancelling it detaches the + * view only — the turn runs on. */ + subscribe(): ReadableStream; + /** The remote kill — the ONLY way a running turn dies early: sawAbort + + * agent.abort(); the loop's agent_end does the rest (finish chunk to + * every subscriber, then onFinish persists the partial transcript). */ + abort(): void; +} + /** Map a pi stop reason to the finish chunk's enum. */ function finishReasonOf(stopReason: string | undefined): string { switch (stopReason) { @@ -88,7 +116,7 @@ interface ToolPart { errorText?: string; } -export function createPiTurnResponse(spec: PiTurnSpec): Response { +export function runPiTurn(spec: PiTurnSpec): PiTurnHandle { const { agent, uiMessages, onFinish } = spec; // The streaming target: continuation (same id, parts appended to a copy @@ -117,232 +145,276 @@ export function createPiTurnResponse(spec: PiTurnSpec): Response { let finishReason: string | undefined; let errorMessage: string | undefined; - const stream = new ReadableStream({ - start(controller) { - const encoder = new TextEncoder(); - // Swallow enqueue on a cancelled stream (the client went away): the - // run must still settle — onFinish persists the partial turn. - const send = (chunk: Chunk) => { - if (settled) return; - try { - controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`)); - } catch { - /* reader cancelled — stop emitting, keep running */ - } - }; - const close = () => { - if (settled) return; - settled = true; - try { - controller.enqueue(encoder.encode("data: [DONE]\n\n")); - controller.close(); - } catch { - /* already cancelled */ - } - }; + // THE PER-TURN EVENT LOG (memory-only — the settle-only disk writes + // are unchanged): every emitted chunk, in order, so any late subscriber + // gets a FULL replay. Unbounded BY HONEST DESIGN (owner ruling + // 2026-09-25): a turn's chunk log is finite by construction — the agent + // loop ends (or the 30-minute liveness TTL bounds the pathological + // never-settling case), and concurrency is human-scale (one turn per + // chat, a few chats). No cap, no truncation: replay must never + // silently drop history. The log dies with the turn at settle. + const log: Chunk[] = []; + const subscribers = new Set(); - send({ type: "start", messageId }); + // Send = append to the log + fan out to every live view. A view's own + // enqueue failure (a cancelled reader) is swallowed by the view — the + // turn must still settle; onFinish persists regardless of who watches. + const send = (chunk: Chunk) => { + if (settled) return; + log.push(chunk); + for (const sub of subscribers) sub.push(chunk); + }; + const closeAll = () => { + if (settled) return; + settled = true; + for (const sub of subscribers) sub.done(); + subscribers.clear(); + }; - const onAbort = () => { - sawAbort = true; - agent.abort(); - }; - spec.signal?.addEventListener("abort", onAbort, { once: true }); + send({ type: "start", messageId }); - const settleWith = () => { - const assembled: UIMessage = { id: messageId, role: "assistant", parts }; - // The SDK's onFinish shape: the input copy stays, the extended - // response copy appends (persistTurn's withMessageIds dedupe - // keeps the last occurrence). - onFinish([...uiMessages, assembled], sawAbort || spec.signal?.aborted === true); - }; + const settleWith = () => { + const assembled: UIMessage = { id: messageId, role: "assistant", parts }; + // The SDK's onFinish shape: the input copy stays, the extended + // response copy appends (persistTurn's withMessageIds dedupe + // keeps the last occurrence). + onFinish([...uiMessages, assembled], sawAbort); + }; - const handle = (event: AgentEvent) => { - switch (event.type) { - case "turn_start": - send({ type: "start-step" }); - parts.push({ type: "step-start" }); - break; - case "message_update": { - const streamEvent = event.assistantMessageEvent; - switch (streamEvent.type) { - case "text_start": { - const id = `${messageId}-t${textN++}`; - textIds.set(streamEvent.contentIndex, id); - const part: TextPart = { type: "text", text: "", state: "streaming" }; - activeParts.set(id, part); - parts.push(part as UIMessage["parts"][number]); - send({ type: "text-start", id }); - break; - } - case "text_delta": { - const id = textIds.get(streamEvent.contentIndex); - const part = id !== undefined ? (activeParts.get(id) as TextPart | undefined) : undefined; - if (part) part.text += streamEvent.delta; - if (id !== undefined) send({ type: "text-delta", id, delta: streamEvent.delta }); - break; - } - case "text_end": { - const id = textIds.get(streamEvent.contentIndex); - const part = id !== undefined ? (activeParts.get(id) as TextPart | undefined) : undefined; - if (part) part.state = "done"; - activeParts.delete(id ?? ""); - if (id !== undefined) send({ type: "text-end", id }); - break; - } - case "thinking_start": { - const id = `${messageId}-r${reasoningN++}`; - reasoningIds.set(streamEvent.contentIndex, id); - const part: ReasoningPart = { type: "reasoning", id, text: "", state: "streaming" }; - activeParts.set(id, part); - parts.push(part as UIMessage["parts"][number]); - send({ type: "reasoning-start", id }); - break; - } - case "thinking_delta": { - const id = reasoningIds.get(streamEvent.contentIndex); - const part = id !== undefined ? (activeParts.get(id) as ReasoningPart | undefined) : undefined; - if (part) part.text += streamEvent.delta; - if (id !== undefined) send({ type: "reasoning-delta", id, delta: streamEvent.delta }); - break; - } - case "thinking_end": { - const id = reasoningIds.get(streamEvent.contentIndex); - const part = id !== undefined ? (activeParts.get(id) as ReasoningPart | undefined) : undefined; - if (part) part.state = "done"; - activeParts.delete(id ?? ""); - if (id !== undefined) send({ type: "reasoning-end", id }); - break; - } - case "toolcall_start": { - const block = (event.message as { content: any[] }).content[streamEvent.contentIndex]; - if (block?.type !== "toolCall") break; - toolInputs.set(block.id, { toolName: block.name, input: undefined }); - const part: ToolPart = { - type: `tool-${block.name}`, - toolCallId: block.id, - state: "input-streaming", - input: undefined, - }; - newToolParts.set(block.id, part); - parts.push(part as unknown as UIMessage["parts"][number]); - send({ type: "tool-input-start", toolCallId: block.id, toolName: block.name }); - break; - } - case "toolcall_delta": { - const block = (event.message as { content: any[] }).content[streamEvent.contentIndex]; - if (block?.type !== "toolCall") break; - send({ - type: "tool-input-delta", - toolCallId: block.id, - inputTextDelta: streamEvent.delta, - }); - break; - } - case "toolcall_end": { - const toolCall = streamEvent.toolCall; - toolInputs.set(toolCall.id, { toolName: toolCall.name, input: toolCall.arguments }); - // Input stays input-streaming here: it only becomes available - // once the call clears validation at the execution boundary - // (tool_execution_end) — an invalid input must never reach - // input-available, where the dock would mount on it. - const part = newToolParts.get(toolCall.id); - if (part) part.input = toolCall.arguments; - break; - } - } + const handle = (event: AgentEvent) => { + switch (event.type) { + case "turn_start": + send({ type: "start-step" }); + parts.push({ type: "step-start" }); + break; + case "message_update": { + const streamEvent = event.assistantMessageEvent; + switch (streamEvent.type) { + case "text_start": { + const id = `${messageId}-t${textN++}`; + textIds.set(streamEvent.contentIndex, id); + const part: TextPart = { type: "text", text: "", state: "streaming" }; + activeParts.set(id, part); + parts.push(part as UIMessage["parts"][number]); + send({ type: "text-start", id }); break; } - case "tool_execution_end": { - const known = toolInputs.get(event.toolCallId); - const toolName = event.toolName || known?.toolName || ""; - const input = known?.input; - const part = newToolParts.get(event.toolCallId); - if (event.isError) { - const errorText = resultErrorText(event.result); - if (part) { - part.state = "output-error"; - part.rawInput = input; - part.errorText = errorText; - } - send({ - type: "tool-input-error", - toolCallId: event.toolCallId, - toolName, - input, - errorText, - }); - } else { - // Validation cleared: only NOW does the input become available - // to the client (the InvalidToolInputError gating the SDK owned). - if (part) { - part.state = "input-available"; - part.input = input; - } - send({ type: "tool-input-available", toolCallId: event.toolCallId, toolName, input }); - if ((QUESTION_TOOLS as readonly string[]).includes(toolName)) { - // Client-owned: the answer arrives via addToolOutput; - // the synthetic placeholder result never reaches wire - // or record. - break; - } - const output = (event.result as { details?: unknown } | undefined)?.details; - if (part) { - part.state = "output-available"; - part.output = output; - } - send({ type: "tool-output-available", toolCallId: event.toolCallId, output }); - } + case "text_delta": { + const id = textIds.get(streamEvent.contentIndex); + const part = id !== undefined ? (activeParts.get(id) as TextPart | undefined) : undefined; + if (part) part.text += streamEvent.delta; + if (id !== undefined) send({ type: "text-delta", id, delta: streamEvent.delta }); break; } - case "turn_end": - send({ type: "finish-step" }); + case "text_end": { + const id = textIds.get(streamEvent.contentIndex); + const part = id !== undefined ? (activeParts.get(id) as TextPart | undefined) : undefined; + if (part) part.state = "done"; + activeParts.delete(id ?? ""); + if (id !== undefined) send({ type: "text-end", id }); break; - case "message_end": - if (event.message && (event.message as { role?: string }).role === "assistant") { - const m = event.message as { stopReason?: string; errorMessage?: string }; - finishReason = m.stopReason; - if (m.stopReason === "aborted") sawAbort = true; - if (m.stopReason === "error") errorMessage = m.errorMessage; - } + } + case "thinking_start": { + const id = `${messageId}-r${reasoningN++}`; + reasoningIds.set(streamEvent.contentIndex, id); + const part: ReasoningPart = { type: "reasoning", id, text: "", state: "streaming" }; + activeParts.set(id, part); + parts.push(part as UIMessage["parts"][number]); + send({ type: "reasoning-start", id }); break; - case "agent_end": { - spec.signal?.removeEventListener("abort", onAbort); - if (errorMessage !== undefined) { - send({ type: "error", errorText: spec.onError?.(errorMessage) ?? errorMessage }); - } - send({ type: "finish", finishReason: finishReasonOf(finishReason) }); - close(); - settleWith(); + } + case "thinking_delta": { + const id = reasoningIds.get(streamEvent.contentIndex); + const part = id !== undefined ? (activeParts.get(id) as ReasoningPart | undefined) : undefined; + if (part) part.text += streamEvent.delta; + if (id !== undefined) send({ type: "reasoning-delta", id, delta: streamEvent.delta }); + break; + } + case "thinking_end": { + const id = reasoningIds.get(streamEvent.contentIndex); + const part = id !== undefined ? (activeParts.get(id) as ReasoningPart | undefined) : undefined; + if (part) part.state = "done"; + activeParts.delete(id ?? ""); + if (id !== undefined) send({ type: "reasoning-end", id }); + break; + } + case "toolcall_start": { + const block = (event.message as { content: any[] }).content[streamEvent.contentIndex]; + if (block?.type !== "toolCall") break; + toolInputs.set(block.id, { toolName: block.name, input: undefined }); + const part: ToolPart = { + type: `tool-${block.name}`, + toolCallId: block.id, + state: "input-streaming", + input: undefined, + }; + newToolParts.set(block.id, part); + parts.push(part as unknown as UIMessage["parts"][number]); + send({ type: "tool-input-start", toolCallId: block.id, toolName: block.name }); + break; + } + case "toolcall_delta": { + const block = (event.message as { content: any[] }).content[streamEvent.contentIndex]; + if (block?.type !== "toolCall") break; + send({ + type: "tool-input-delta", + toolCallId: block.id, + inputTextDelta: streamEvent.delta, + }); + break; + } + case "toolcall_end": { + const toolCall = streamEvent.toolCall; + toolInputs.set(toolCall.id, { toolName: toolCall.name, input: toolCall.arguments }); + // Input stays input-streaming here: it only becomes available + // once the call clears validation at the execution boundary + // (tool_execution_end) — an invalid input must never reach + // input-available, where the dock would mount on it. + const part = newToolParts.get(toolCall.id); + if (part) part.input = toolCall.arguments; break; } } - }; - - agent.subscribe(handle); - - void (async () => { - try { - await agent.continue(); - } catch (err) { - // A loop-level failure (not a provider error — those ride the - // stream as stopReason "error"): surface, persist, close. - spec.signal?.removeEventListener("abort", onAbort); - send({ type: "error", errorText: spec.onError?.(err) ?? "An error occurred." }); - send({ type: "finish", finishReason: "error" }); - close(); - settleWith(); + break; + } + case "tool_execution_end": { + const known = toolInputs.get(event.toolCallId); + const toolName = event.toolName || known?.toolName || ""; + const input = known?.input; + const part = newToolParts.get(event.toolCallId); + if (event.isError) { + const errorText = resultErrorText(event.result); + if (part) { + part.state = "output-error"; + part.rawInput = input; + part.errorText = errorText; + } + send({ + type: "tool-input-error", + toolCallId: event.toolCallId, + toolName, + input, + errorText, + }); + } else { + // Validation cleared: only NOW does the input become available + // to the client (the InvalidToolInputError gating the SDK owned). + if (part) { + part.state = "input-available"; + part.input = input; + } + send({ type: "tool-input-available", toolCallId: event.toolCallId, toolName, input }); + if ((QUESTION_TOOLS as readonly string[]).includes(toolName)) { + // Client-owned: the answer arrives via addToolOutput; + // the synthetic placeholder result never reaches wire + // or record. + break; + } + const output = (event.result as { details?: unknown } | undefined)?.details; + if (part) { + part.state = "output-available"; + part.output = output; + } + send({ type: "tool-output-available", toolCallId: event.toolCallId, output }); } - })(); + break; + } + case "turn_end": + send({ type: "finish-step" }); + break; + case "message_end": + if (event.message && (event.message as { role?: string }).role === "assistant") { + const m = event.message as { stopReason?: string; errorMessage?: string }; + finishReason = m.stopReason; + if (m.stopReason === "aborted") sawAbort = true; + if (m.stopReason === "error") errorMessage = m.errorMessage; + } + break; + case "agent_end": { + if (errorMessage !== undefined) { + send({ type: "error", errorText: spec.onError?.(errorMessage) ?? errorMessage }); + } + send({ type: "finish", finishReason: finishReasonOf(finishReason) }); + closeAll(); + settleWith(); + break; + } + } + }; + agent.subscribe(handle); + + void (async () => { + try { + await agent.continue(); + } catch (err) { + // A loop-level failure (not a provider error — those ride the + // stream as stopReason "error"): surface, persist, close. + send({ type: "error", errorText: spec.onError?.(err) ?? "An error occurred." }); + send({ type: "finish", finishReason: "error" }); + closeAll(); + settleWith(); + } + })(); + + return { + subscribe() { + let sub: PiTurnSubscriber | undefined; + const stream = new ReadableStream({ + start(controller) { + const encoder = new TextEncoder(); + let closed = false; + sub = { + push(chunk) { + if (closed) return; + try { + controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`)); + } catch { + /* reader cancelled — stop emitting, keep running */ + closed = true; + } + }, + done() { + if (closed) return; + closed = true; + try { + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + controller.close(); + } catch { + /* already cancelled */ + } + }, + }; + // Attach THEN replay — synchronous, so no chunk can fall + // between replay and live (single-threaded: the agent's + // handle() only runs after this stack returns). + subscribers.add(sub); + for (const chunk of log) sub.push(chunk); + // A subscriber that attached after settle (the unregister + // race) still gets the full history + [DONE]. + if (settled) sub.done(); + }, + cancel() { + // THE OWNER RULING (2026-09-25): a view leaving NEVER kills + // the turn — only the explicit abort channel does. The turn + // runs on; the log keeps everything for the next view. + if (sub) subscribers.delete(sub); + }, + }); + return stream; }, - cancel() { - // The client went away (reload/abort): kill the model call. The - // run still settles — agent_end fires and onFinish persists. + abort() { + if (settled) return; + sawAbort = true; agent.abort(); }, - }); + }; +} - return new Response(stream, { +/** The subscriber-shaped SSE response — the POST /api/chat response and + * GET /api/chats/[id]/stream are BOTH this: same headers, same wire + * dialect, one protocol for the client to consume. */ +export function piTurnResponse(handle: PiTurnHandle): Response { + return new Response(handle.subscribe(), { headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-cache, no-transform", diff --git a/src/server/turn-runs.ts b/src/server/turn-runs.ts new file mode 100644 index 0000000..29bf49a --- /dev/null +++ b/src/server/turn-runs.ts @@ -0,0 +1,81 @@ +/** + * The running-turn registry (background-turn runner wave, owner ruling + * 2026-09-25): chatId → the live PiTurnHandle. THE reattach surface — + * GET /api/chats/[id]/stream resolves here, and the remote abort + * (/api/activity/abort, kind "turn") kills through here. + * + * Same globalThis discipline as turn-liveness.ts (read its header): Next + * dev compiles each route as its own module instance, so a module-level + * Map would be a DIFFERENT registry in the POST /api/chat entry and the + * GET .../stream entry. globalThis spans module instances within the + * process. + * + * Registration lifecycle: chat-handler registers the handle the moment + * the turn starts (after the 409 guard) and its onFinish unregisters LAST + * — after persist, usage, extraction, and endTurn — so anything that sees + * "no running turn" also sees an already-persisted record. The per-chat + * 409 guard means a chatId can hold at most one handle, and beginTurn's + * 30-minute TTL bounds the pathological never-settling case the same way + * it does for the liveness map: if onFinish never fires, the stale entry + * is overwritten by the next turn's registration (Map.set) and garbage + * collected — no unbounded leak beyond the same pathology turn-liveness + * already accepts. + * + * App-quit mid-turn (the process dying with an un-persisted turn) is the + * SAME residual as before the wave: the registry is memory-only, the + * write-ahead record + liveness TTL recovery cover the next mount. + */ + +import type { PiTurnHandle } from "./pi/ui-stream.ts"; + +interface LhTurnRunsGlobal { + __lhTurnRuns?: Map; +} + +function runs(): Map { + const g = globalThis as typeof globalThis & LhTurnRunsGlobal; + return (g.__lhTurnRuns ??= new Map()); +} + +/** Register the chat's running turn (chat-handler, right after the turn + * starts). One per chat by construction — the 409 guard. */ +export function registerTurn(chatId: string, handle: PiTurnHandle): void { + runs().set(chatId, handle); +} + +/** Unregister at settle — chat-handler's onFinish, LAST (after endTurn). */ +export function unregisterTurn(chatId: string): void { + runs().delete(chatId); +} + +/** The live turn handle for a chat, or undefined — the subscribe endpoint's + * and the abort channel's resolution point. */ +export function runningTurn(chatId: string): PiTurnHandle | undefined { + return runs().get(chatId); +} + +/** The remote kill: any client (original sender, reattached viewer, future + * surfaces) may stop a running turn. not_running mirrors abortJob's + * honest 409 — there was nothing to abort, the caller refetches. */ +export function abortTurn(chatId: string): { ok: true } | { ok: false; reason: "not_running" } { + const handle = runs().get(chatId); + if (!handle) return { ok: false, reason: "not_running" }; + handle.abort(); + return { ok: true }; +} + +/** The cheap terminal when GET /api/chats/[id]/stream finds no running + * turn (raced settle, or the client checked turnInFlight a beat too + * late): a same-dialect SSE response whose single chunk tells the client + * "nothing to attach to — the record is current, refetch it". Chosen as + * an EVENT (not a status code) so the reattach path consumes ONE wire + * protocol end to end. */ +export function settledTurnResponse(): Response { + return new Response('data: {"type":"turn-settled"}\n\ndata: [DONE]\n\n', { + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache, no-transform", + "X-Accel-Buffering": "no", + }, + }); +} diff --git a/tests/chat-flow.test.ts b/tests/chat-flow.test.ts index 2f9a825..6c78551 100644 --- a/tests/chat-flow.test.ts +++ b/tests/chat-flow.test.ts @@ -27,6 +27,7 @@ import { createTeacherAgent } from "../src/agent/teacher-agent.ts"; import { handleChatRequest } from "../src/server/chat-handler.ts"; import { effectiveSettings } from "../src/server/stores.ts"; import { isTurnRunning } from "../src/server/turn-liveness.ts"; +import { abortTurn, runningTurn, settledTurnResponse } from "../src/server/turn-runs.ts"; import { recordToPiMessages } from "../src/server/pi/record-adapter.ts"; import { chatStore } from "../src/store/chats.ts"; import { buildAnswer, coerceCorrectAnswer, normalizeOptions, resolveCorrect, type QuestionData } from "../src/core/question-core.ts"; @@ -96,6 +97,46 @@ async function drain(res: Response): Promise { return res.text(); // consumes the SSE stream to completion (fires onFinish) } +function waitFor(cond: () => boolean, timeoutMs = 8000, what = "condition"): Promise { + return new Promise((resolve, reject) => { + const started = Date.now(); + const tick = () => { + if (cond()) return resolve(); + if (Date.now() - started > timeoutMs) return reject(new Error(`timeout waiting for ${what}`)); + setTimeout(tick, 25); + }; + tick(); + }); +} + +/** A teacher whose step 1 blocks on an external gate — a turn held mid-run + * so view-cancel / reattach / abort timing is deterministic. */ +function gatedTeacher(gate: Promise): PiEngine { + const faux = createFauxCore({}); + faux.setResponses([ + async () => { + await gate; + return fauxAssistantMessage([fauxText("Gated reply.")]); + }, + ]); + return { model: faux.getModel(), streamFn: faux.streamSimple }; +} + +/** A teacher that NEVER resolves except on abort (the jobs abortableFactory + * pattern): the only way this turn can end is the remote kill. */ +function abortableTeacher(): PiEngine { + const faux = createFauxCore({}); + faux.setResponses([ + ((_ctx: unknown, opts: { signal?: AbortSignal } | undefined) => + new Promise((resolve) => { + const sig = opts?.signal; + if (sig?.aborted) return resolve(fauxAssistantMessage("late")); + sig?.addEventListener("abort", () => resolve(fauxAssistantMessage("late")), { once: true }); + })) as unknown as FauxResponseStep, + ]); + return { model: faux.getModel(), streamFn: faux.streamSimple }; +} + function post(chatId: string, messages: unknown[]): Request { return new Request("http://test/api/chat", { method: "POST", @@ -227,22 +268,30 @@ describe("teacher chat flow (scripted model)", () => { assert.ok(startIdx < availIdx, "start precedes available"); }); - it("persists even when the stream is aborted (resume keeps partial state)", async () => { - const { engine } = scriptedTeacher(defaultLesson); - const chat = createConfiguredChat("Test: abort"); + it("a cancelled view NEVER aborts the turn — it runs to completion (owner ruling 2026-09-25)", async () => { + let release!: () => void; + const gate = new Promise((r) => (release = r)); + const engine = gatedTeacher(gate); + const chat = createConfiguredChat("Test: view cancel"); createdChats.push(chat.id); - const req = post(chat.id, [{ id: "u1", role: "user", parts: [{ type: "text", text: "quiz me" }] }]); + const req = post(chat.id, [{ id: "u1", role: "user", parts: [{ type: "text", text: "hold my turn" }] }]); const res = await handleChatRequest(req, agentFactory(engine)); assert.equal(res.status, 200); - // Read one chunk then cancel — simulates the browser closing mid-stream. + // Read one chunk then cancel — the browser view leaves mid-stream. const reader = res.body!.getReader(); await reader.read(); await reader.cancel(); - // Give the aborted run a tick to settle (agent_end → onFinish). - await new Promise((r) => setTimeout(r, 200)); + // The turn is STILL a running background object — no view killed it. + assert.equal(isTurnRunning(chat.id), true, "cancel detached the view only — the turn runs on"); + assert.ok(runningTurn(chat.id), "the running-turn registry still holds the handle"); + + // It runs to FULL completion with nobody watching, then settles alone. + release(); + await waitFor(() => !isTurnRunning(chat.id), 8000, "turn settle after view cancel"); const saved = chatStore.getChat(chat.id)!; - assert.ok(saved.messages.length >= 1, "partial transcript persisted"); + assert.equal(saved.messages.length, 2, "complete turn persisted — not a partial"); + assert.equal(runningTurn(chat.id), undefined, "handle unregistered at settle"); }); it("an ungradeable quiz is rejected at the call boundary, the error reaches the model, and the model retries", async () => { @@ -670,3 +719,125 @@ describe("the key-presence turn gate (provider-key interlude, 2026-09-25)", () = } }); }); + +describe("background-turn runner (owner ruling 2026-09-25: every turn is a server-side process, clients are views)", () => { + it("subscribe mid-turn: a second view gets the FULL replay, then live, same dialect", async () => { + let release!: () => void; + const gate = new Promise((r) => (release = r)); + const engine = gatedTeacher(gate); + const chat = createConfiguredChat("Test: reattach replay"); + createdChats.push(chat.id); + + const req = post(chat.id, [{ id: "u1", role: "user", parts: [{ type: "text", text: "hold my turn" }] }]); + const first = await handleChatRequest(req, agentFactory(engine)); + assert.equal(first.status, 200); + // Consume the first view a little (the start chunk is out). + const reader = first.body!.getReader(); + await reader.read(); + + // A SECOND view attaches mid-turn — a remount, another device. + const handle = runningTurn(chat.id)!; + assert.ok(handle, "the registry resolves the live turn"); + const secondText = new Response(handle.subscribe()).text(); + + // Let the turn finish; both views close at settle. + release(); + const text = await secondText; + assert.ok(text.startsWith('data: {"type":"start"'), "replay starts from the very first chunk (the start event)"); + assert.ok(text.includes("Gated reply."), "replay carries the full streamed content"); + assert.ok(text.endsWith("data: [DONE]\n\n"), "live tail ends with the SSE terminal, same dialect as POST"); + // Wire honesty: a normal turn settles via agent_end with NO error chunk + // (the live smoke caught a close()-throw masked into the error path + // — unit asserts never looked; these pins would have). + assert.ok(!text.includes('"type":"error"'), "a clean turn never emits an error chunk"); + assert.ok(text.includes('"finishReason":"stop"'), "the finish chunk carries the stop reason"); + // And the first view also closes cleanly (no hang, no drop): read its + // remaining chunks through the reader we already hold. + const decoder = new TextDecoder(); + let firstTail = ""; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + firstTail += decoder.decode(value, { stream: true }); + } + assert.ok(firstTail.endsWith("data: [DONE]\n\n"), "first view closes cleanly at settle, unaffected by the reattach"); + }); + + it("remote abort: any client can stop a running turn; the partial persists; subscribers close cleanly", async () => { + const engine = abortableTeacher(); + const chat = createConfiguredChat("Test: remote abort"); + createdChats.push(chat.id); + + const req = post(chat.id, [{ id: "u1", role: "user", parts: [{ type: "text", text: "wait forever" }] }]); + const res = await handleChatRequest(req, agentFactory(engine)); + assert.equal(res.status, 200); + assert.ok(runningTurn(chat.id), "turn registered while in flight"); + const viewText = res.text(); // the first view, reading to settle + + // The remote kill from a DIFFERENT view (the dock's abort). + const outcome = abortTurn(chat.id); + assert.deepEqual(outcome, { ok: true }, "abort resolves ok on a live turn"); + + const text = await viewText; + assert.ok(text.endsWith("data: [DONE]\n\n"), "subscriber closes cleanly at the aborted settle"); + + await waitFor(() => !isTurnRunning(chat.id), 8000, "aborted turn settle"); + const saved = chatStore.getChat(chat.id)!; + assert.ok(saved.messages.length >= 2, "the partial transcript persisted via the usual settle path"); + assert.equal(runningTurn(chat.id), undefined, "handle unregistered after the aborted settle"); + + // Honest 409 semantics: there is nothing left to abort. + assert.deepEqual(abortTurn(chat.id), { ok: false, reason: "not_running" }, "second abort is not_running"); + }); + + it("subscribe after settle: the cheap terminal tells the client the record is current", async () => { + const { engine } = scriptedTeacher(defaultLesson); + const chat = createConfiguredChat("Test: settled terminal"); + createdChats.push(chat.id); + + // A quiet chat: no turn ever ran — the stream is the same terminal. + const term = settledTurnResponse(); + assert.equal(await term.text(), 'data: {"type":"turn-settled"}\n\ndata: [DONE]\n\n', "terminal is one event + [DONE]"); + assert.equal(runningTurn(chat.id), undefined); + + // And after a settled turn, the registry is empty — the GET would + // serve the same terminal, not a stale handle. + const req = post(chat.id, [{ id: "u1", role: "user", parts: [{ type: "text", text: "quiz me" }] }]); + await drain(await handleChatRequest(req, agentFactory(engine))); + assert.equal(runningTurn(chat.id), undefined, "no handle survives settle"); + }); + + it("concurrent two-chat reattach: two turns in flight, two views, both settle independently", async () => { + const releaseFns: Array<() => void> = []; + const chats = [1, 2].map((n) => { + const gate = new Promise((r) => (releaseFns.push(r))); + const engine = gatedTeacher(gate); + const chat = createConfiguredChat(`Test: two-chat ${n}`); + createdChats.push(chat.id); + return { n, engine, chat }; + }); + + const views = new Map>(); + for (const { chat, engine } of chats) { + const req = post(chat.id, [{ id: "u1", role: "user", parts: [{ type: "text", text: "hold my turn" }] }]); + const res = await handleChatRequest(req, agentFactory(engine)); + assert.equal(res.status, 200); + const handle = runningTurn(chat.id)!; + assert.ok(handle, `chat ${chat.id} has its own handle`); + views.set(chat.id, new Response(handle.subscribe()).text()); + // The other chat's handle must not be this one (per-chat keying). + const others = chats.filter((c) => c.chat.id !== chat.id).map((c) => runningTurn(c.chat.id)); + assert.ok(!others.includes(handle), "per-chat handles are distinct objects"); + } + + // Release both; both settle independently. + for (const release of releaseFns) release(); + for (const { chat } of chats) { + const text = await views.get(chat.id)!; + assert.ok(text.includes("Gated reply."), `chat ${chat.id} view replayed its own turn`); + assert.ok(text.endsWith("data: [DONE]\n\n"), `chat ${chat.id} view closed at settle`); + await waitFor(() => !isTurnRunning(chat.id), 8000, `chat ${chat.id} settle`); + assert.equal(chatStore.getChat(chat.id)!.messages.length, 2, `chat ${chat.id} persisted its full turn`); + } + }); +});