Merge feat/sub-agents wave 2: 409 merge-retry transport, dead pendingJob deletion, evidence-based timing, prompt-size guard

The 409 fix is a merge-retry, not a bare retry: a 409 means the server
record is AHEAD (wake turn persisted the steer), and the write-ahead
persist makes the request body canonical — a naive re-POST would clobber
steer+wake messages and steerDelivered makes the loss permanent. The
transport waits for the racing turn (500ms poll, 30s cap), refetches the
record, appends the client's unsent messages by id, and re-POSTs. The
integration seam (one line in chat.tsx) follows in the next commit.

components/chat-transport.ts

@@ -0,0 +1,117 @@
+/**
+ * Chat transport with 409 merge-retry — the learner's send can race the
+ * server's wake pipeline (contracts/jobs.ts "THE ASYNC PROTOCOL"): the wake
+ * turn starts server-side and the client only notices via the /api/jobs poll
+ * up to a second later. A send in that window gets 409'd ("a turn is already
+ * running") and, before this file, surfaced as an error banner.
+ *
+ * Why NOT a bare retry: every 409 means the server record is AHEAD of the
+ * client — the wake turn persisted the steer message (and, once it finishes,
+ * its response) that the client hasn't seen. The chat handler's write-ahead
+ * persist makes the request body the canonical transcript, so re-POSTing the
+ * stale client list would clobber the steer + wake messages out of the
+ * record (and the job already flipped done — steerDelivered makes
+ * re-delivery impossible: permanent loss). The retry must therefore MERGE:
+ * wait for the in-flight turn to finish, refetch the record, append the
+ * client's unsent messages (by id) to the server's list, and re-POST the
+ * merged transcript.
+ *
+ * Client-side gap after a merged retry: useChat's local list still lacks the
+ * steer/wake messages until the next syncFromServer (job settle, remount, or
+ * a future "server-ahead" heal in the poller). Rare path; self-heals.
+ */
+import { DefaultChatTransport } from "ai";
+import type { TeacherUIMessage } from "@/src/core/ui-message";
+
+/** How often to poll turnInFlight while the racing turn runs. */
+const TURN_POLL_MS = 500;
+/** Give up waiting (a wake turn can run long) and surface the 409 as before. */
+const TURN_WAIT_MAX_MS = 30_000;
+
+/** Injectable timing (tests). */
+export interface RetryFetchOptions {
+	pollMs?: number;
+	waitMaxMs?: number;
+}
+
+function delay(ms: number, signal?: AbortSignal): Promise<void> {
+	return new Promise((resolve, reject) => {
+		const t = setTimeout(resolve, ms);
+		signal?.addEventListener("abort", () => {
+			clearTimeout(t);
+			reject(new DOMException("Aborted", "AbortError"));
+		}, { once: true });
+	});
+}
+
+async function turnInFlight(chatId: string): Promise<boolean | null> {
+	try {
+		const res = await fetch(`/api/jobs?chatId=${encodeURIComponent(chatId)}`);
+		if (!res.ok) return null;
+		return ((await res.json()) as { turnInFlight?: boolean }).turnInFlight === true;
+	} catch {
+		return null; // network hiccup: keep waiting, the deadline bounds us
+	}
+}
+
+async function serverMessages(chatId: string): Promise<TeacherUIMessage[] | null> {
+	try {
+		const res = await fetch(`/api/chats/${encodeURIComponent(chatId)}`);
+		if (!res.ok) return null;
+		const record = (await res.json()) as { messages?: TeacherUIMessage[] };
+		return Array.isArray(record.messages) ? record.messages : null;
+	} catch {
+		return null;
+	}
+}
+
+/**
+ * A fetch that turns a 409 on POST /api/chat into: wait for the running turn,
+ * merge the outgoing messages onto the fresh server record, re-POST. Non-409
+ * responses pass through untouched.
+ */
+/** Exported for tests. */
+export function retryingFetch(chatId: string, opts: RetryFetchOptions = {}): typeof globalThis.fetch {
+	const pollMs = opts.pollMs ?? TURN_POLL_MS;
+	const waitMaxMs = opts.waitMaxMs ?? TURN_WAIT_MAX_MS;
+	return async (input, init) => {
+		let response = await fetch(input, init);
+		let waited = 0;
+
+		while (response.status === 409) {
+			// Free the 409 body's socket before waiting on the racing turn.
+			await response.body?.cancel().catch(() => {});
+			if (init?.signal?.aborted || waited >= waitMaxMs) return response;
+
+			await delay(pollMs, init?.signal ?? undefined);
+			waited += pollMs;
+			if (init?.signal?.aborted) return response;
+			if ((await turnInFlight(chatId)) !== false) continue;
+
+			// Turn finished: merge onto the fresh record, then re-POST.
+			const record = await serverMessages(chatId);
+			if (!record || typeof init?.body !== "string") return response;
+			const serverIds = new Set(record.map((m) => m.id));
+			const body = JSON.parse(init.body) as { messages?: TeacherUIMessage[] };
+			if (!Array.isArray(body.messages)) return response;
+			const outgoing = body.messages.filter((m) => m.id && !serverIds.has(m.id));
+			const merged = JSON.stringify({ ...body, messages: [...record, ...outgoing] });
+
+			response = await fetch(input, { ...init, body: merged }); // may 409 again → loop
+		}
+
+		return response;
+	};
+}
+
+/**
+ * The chat.tsx one-liner: `transport: createChatTransport(chatId)` replaces
+ * `new DefaultChatTransport({ api: "/api/chat", body: { chatId } })`.
+ */
+export function createChatTransport(chatId: string): DefaultChatTransport<TeacherUIMessage> {
+	return new DefaultChatTransport<TeacherUIMessage>({
+		api: "/api/chat",
+		body: { chatId },
+		fetch: retryingFetch(chatId),
+	});
+}

src/core/pending.ts

@@ -5,9 +5,10 @@ import { isToolUIPart, type UIMessage } from "ai";
  * blocks the composer — one question at a time, mirroring the teacher's
  * prompt contract. Derived client state; the server enforces nothing.
  *
- * Job tool parts (research/demo/visual) are NOT returned here — they gate
- * sending separately with a "waiting for <job>" affordance (see contracts.ts
- * "THE ASYNC PROTOCOL" §5; use findPendingJobCall).
+ * Job tool parts (research/demo/visual) never pend client-side — the async
+ * protocol resolves them server-side with an immediate ack (see
+ * contracts/jobs.ts "THE ASYNC PROTOCOL"), so job tools are NOT returned
+ * here. This file carries only the question gate and the shared scan helper.
  */
 export interface PendingToolCall {
 	toolCallId: string;
@@ -42,7 +43,4 @@ export function findPendingToolCall(messages: UIMessage[]): PendingToolCall | nu
 	return findPending(messages, QUESTION_TOOLS);
 }
 
-/** The running background job, if any (input-available research/demo/visual part). */
-export function findPendingJobCall(messages: UIMessage[]): PendingToolCall | null {
-	return findPending(messages, JOB_TOOLS);
-}
+

src/server/jobs.ts

@@ -87,8 +87,20 @@ export function setJobRespawn(respawn: JobRespawn): void {
 
 const POLL_MS = 400;
 /** After the gates open, wait a beat before firing — a client that just
- *  answered a question is about to send its continuation; don't 409 it. */
-const GRACE_MS = 1500;
+ *  answered a question is about to send its continuation; don't 409 it.
+ *
+ *  Evidence for 750: the race window is exactly the learner's answer→POST
+ *  flight — sendAutomaticallyWhen dispatches synchronously on addToolOutput,
+ *  the answer hits the record only at the write-ahead persist, and beginTurn
+ *  flips the turn registry immediately after, all before the grace can fire
+ *  (measured localhost RTT ~10ms; typical LAN/production RTT <300ms). 750ms
+ *  is 2.5× the worst realistic flight. A miss self-heals: the pump rechecks
+ *  the gates after the grace AND the wake pipeline retries 409s with a
+ *  backoff ramp — and since the transport's merge-retry (chat-transport.ts)
+ *  made client 409s non-fatal, even a pathological RTT degrades gracefully.
+ *  (Was 1500 — pure guess; 750 claws back 750ms of post-answer delivery
+ *  latency without weakening the guard.) */
+const GRACE_MS = 750;
 /** 409 backoff: capped ramp so a busy client can't spin the pump. */
 const RETRY_BACKOFF_MS = [500, 1000, 2000, 5000] as const;
 

tests/chat-transport.test.ts

@@ -0,0 +1,137 @@
+/**
+ * 409 merge-retry transport tests (components/chat-transport.ts).
+ *
+ * Every 409 means the server record is AHEAD (a wake turn persisted the
+ * steer): the retry must wait out the turn, merge the outgoing messages onto
+ * the fresh record, and re-POST the merged list — a bare retry would clobber
+ * the steer/wake messages via the write-ahead persist.
+ */
+import assert from "node:assert/strict";
+import { afterEach, describe, it } from "node:test";
+import { retryingFetch } from "../components/chat-transport.ts";
+import type { TeacherUIMessage } from "../src/core/ui-message.ts";
+
+const realFetch = globalThis.fetch;
+afterEach(() => {
+	globalThis.fetch = realFetch;
+});
+
+function msg(id: string, role: "user" | "assistant", text: string): TeacherUIMessage {
+	return { id, role, metadata: {}, parts: [{ type: "text", text }] };
+}
+
+interface Call {
+	url: string;
+	body: string | undefined;
+}
+
+function installFetch(routes: {
+	/** POST /api/chat → Response; receives every attempt. */
+	chatPOST: (call: Call, attempt: number) => Response;
+	/** GET /api/jobs?chatId=… → turnInFlight value per poll (poll index). */
+	jobsGET?: (poll: number) => boolean;
+	/** GET /api/chats/<id> → messages array. */
+	chatGET?: () => TeacherUIMessage[];
+}): Call[] {
+	const calls: Call[] = [];
+	let poll = 0;
+	globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => {
+		const url = String(input);
+		calls.push({ url, body: init?.body as string | undefined });
+		if (url === "/api/chat" && init?.method === "POST") {
+			return routes.chatPOST(calls.at(-1)!, calls.filter((c) => c.url === "/api/chat").length);
+		}
+		if (url.startsWith("/api/jobs")) return Response.json({ turnInFlight: routes.jobsGET?.(++poll) ?? false });
+		if (url.startsWith("/api/chats/")) return Response.json({ messages: routes.chatGET?.() ?? [] });
+		throw new Error(`unexpected fetch: ${url}`);
+	}) as typeof globalThis.fetch;
+	return calls;
+}
+
+function outgoing(clientOnly: TeacherUIMessage[]): string {
+	return JSON.stringify({ chatId: "chat_x", messages: [...clientOnly] });
+}
+
+describe("retryingFetch (409 merge-retry)", { skip: process.env.NO_SLOW ? true : undefined }, () => {
+	it("passes non-409 responses through untouched", async () => {
+		const calls = installFetch({
+			chatPOST: () => new Response("stream", { status: 200 }),
+		});
+		const fetch = retryingFetch("chat_x", { pollMs: 10, waitMaxMs: 100 });
+		const res = await fetch("/api/chat", { method: "POST", body: outgoing([msg("c1", "user", "hi")]) });
+		assert.equal(res.status, 200);
+		assert.equal(calls.length, 1, "no polling, no merge, single call");
+	});
+
+	it("waits out the turn, merges onto the fresh record, re-POSTs the merged list", async () => {
+		const server = [msg("steer_1", "user", "Research complete"), msg("s1", "assistant", "wake turn")];
+		const fresh = [msg("c1", "user", "my question while the wake turn ran")];
+		const calls = installFetch({
+			chatPOST: (_call, n) => (n === 1 ? new Response("busy", { status: 409 }) : new Response("stream", { status: 200 })),
+			jobsGET: (poll) => poll <= 1, // first poll: still running; second: done
+			chatGET: () => server,
+		});
+		const fetch = retryingFetch("chat_x", { pollMs: 10, waitMaxMs: 100 });
+		const res = await fetch("/api/chat", { method: "POST", body: outgoing(fresh) });
+		assert.equal(res.status, 200);
+
+		// second POST carries server truth + the client's unsent message, in order
+		const retry = JSON.parse(calls.filter((c) => c.url === "/api/chat")[1]!.body!);
+		assert.deepEqual(
+			retry.messages.map((m: { id: string }) => m.id),
+			["steer_1", "s1", "c1"],
+		);
+	});
+
+	it("drops client messages the server already has (no duplicates in the merge)", async () => {
+		const server = [msg("steer_1", "user", "Research complete"), msg("c1", "user", "my question")];
+		const calls = installFetch({
+			chatPOST: (_call, n) => (n === 1 ? new Response("busy", { status: 409 }) : new Response("stream", { status: 200 })),
+			chatGET: () => server,
+		});
+		const fetch = retryingFetch("chat_x", { pollMs: 10, waitMaxMs: 100 });
+		const res = await fetch("/api/chat", {
+			method: "POST",
+			body: outgoing([msg("steer_1", "user", "Research complete"), msg("c1", "user", "my question")]),
+		});
+		assert.equal(res.status, 200);
+		const retry = JSON.parse(calls.filter((c) => c.url === "/api/chat")[1]!.body!);
+		assert.deepEqual(
+			retry.messages.map((m: { id: string }) => m.id),
+			["steer_1", "c1"],
+		);
+	});
+
+	it("loops on repeated 409s, re-merging fresh each time", async () => {
+		const server = [msg("steer_1", "user", "Research complete")];
+		const calls = installFetch({
+			chatPOST: (_call, n) => (n <= 2 ? new Response("busy", { status: 409 }) : new Response("stream", { status: 200 })),
+			chatGET: () => server,
+		});
+		const fetch = retryingFetch("chat_x", { pollMs: 10, waitMaxMs: 100 });
+		const res = await fetch("/api/chat", { method: "POST", body: outgoing([msg("c1", "user", "q")]) });
+		assert.equal(res.status, 200);
+		const posts = calls.filter((c) => c.url === "/api/chat");
+		assert.equal(posts.length, 3);
+		const last = JSON.parse(posts[2]!.body!);
+		assert.deepEqual(
+			last.messages.map((m: { id: string }) => m.id),
+			["steer_1", "c1"],
+		);
+	});
+
+	it("gives up after TURN_WAIT_MAX_MS and returns the last 409 (banner as before)", async () => {
+		// turnInFlight stuck true; two polls (~1s) then force the deadline path
+		// by asserting we eventually get the 409 back — bounded by poll count.
+		let polls = 0;
+		const calls = installFetch({
+			chatPOST: () => new Response("busy", { status: 409 }),
+			jobsGET: () => true,
+		});
+		const fetch = retryingFetch("chat_x", { pollMs: 10, waitMaxMs: 100 });
+		const res = await fetch("/api/chat", { method: "POST", body: outgoing([msg("c1", "user", "q")]) });
+		assert.equal(res.status, 409);
+		polls = calls.filter((c) => c.url.startsWith("/api/jobs")).length;
+		assert.ok(polls <= 15, `expected bounded polling (~10×10ms), got ${polls}`);
+	});
+});

tests/jobs.test.ts

@@ -700,7 +700,20 @@ describe("prompt composition", () => {
 		assert.ok(instructions.startsWith(base), "base prompt first");
 		for (const f of fs.readdirSync("src/agent/prompts").filter((f) => /^\d+-[\w.-]+\.md$/.test(f)).sort()) {
 			const body = fs.readFileSync(`src/agent/prompts/${f}`, "utf-8").trim();
-		assert.ok(instructions.includes(body), `fragment ${f} must compose verbatim`);
+			assert.ok(instructions.includes(body), `fragment ${f} must compose verbatim`);
 		}
 	});
+
+	it("guards prompt size — no fragment can silently balloon (the README-leak class)", () => {
+		// The wave-1 leak: README.md (815 bytes / ~200 tokens) rode into every
+		// teacher turn until the numeric-prefix filter stopped it. This guard
+		// catches the sibling failure mode — an accidental dump pasted into a
+		// fragment. Thresholds: largest real fragment is 40-demos.md at ~3.2k
+		// chars; 8k is generous headroom. Composed total ≤ 60k chars.
+		for (const f of fs.readdirSync("src/agent/prompts").filter((f) => /^\d+-[\w.-]+\.md$/.test(f))) {
+			const body = fs.readFileSync(`src/agent/prompts/${f}`, "utf-8").trim();
+			assert.ok(body.length <= 8_000, `fragment ${f} is ${body.length} chars — over the 8k guard`);
+		}
+		assert.ok(composeInstructions().length <= 60_000, "composed system prompt over the 60k guard");
+	});
 });