Merge branch 'main' into feat/ui-polish

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";
+
+/** 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),
+	});
+}

components/chat.tsx

@@ -2,7 +2,8 @@
 
 import { createContext, Fragment, useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } from "react";
 import { useChat } from "@ai-sdk/react";
-import { DefaultChatTransport, isToolUIPart, lastAssistantMessageIsCompleteWithToolCalls } from "ai";
+import { isToolUIPart, lastAssistantMessageIsCompleteWithToolCalls } from "ai";
+import { createChatTransport } from "@/components/chat-transport";
 import { useStickToBottomContext } from "use-stick-to-bottom";
 import {
 	Conversation,
@@ -81,7 +82,7 @@ export function Chat({
 		useChat<TeacherUIMessage>({
 			id: chatId,
 			messages: initialMessages,
-			transport: new DefaultChatTransport({ api: "/api/chat", body: { chatId } }),
+			transport: createChatTransport(chatId),
 			// After the learner answers (addToolOutput), automatically send the
 			// continuation request so the teacher reacts to the answer.
 			sendAutomaticallyWhen: lastAssistantMessageIsCompleteWithToolCalls,

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