Background turn runner: every turn is a server-side process, clients are views

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

@@ -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);

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

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.

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";

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<Uint8Array>;
+	/** 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<Uint8Array>({
-		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<PiTurnSubscriber>();
 
-			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<Uint8Array>({
+				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",

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<string, PiTurnHandle>;
+}
+
+function runs(): Map<string, PiTurnHandle> {
+	const g = globalThis as typeof globalThis & LhTurnRunsGlobal;
+	return (g.__lhTurnRuns ??= new Map<string, PiTurnHandle>());
+}
+
+/** 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",
+		},
+	});
+}

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<string> {
 	return res.text(); // consumes the SSE stream to completion (fires onFinish)
 }
 
+function waitFor(cond: () => boolean, timeoutMs = 8000, what = "condition"): Promise<void> {
+	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<void>): 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<void>((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<void>((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<void>((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<string, Promise<string>>();
+		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`);
+		}
+	});
+});