From 2f8cf71f528fa96c482aef240bc1598e9f7137c2 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Sat, 5 Sep 2026 21:33:13 +0300 Subject: [PATCH] Wave 2: 409 merge-retry transport, dead pendingJob code, evidence-based timing, prompt-size guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - components/chat-transport.ts (NEW): createChatTransport(chatId) — the chat.tsx seam is now a one-line swap. Its fetch turns a 409 on POST /api/chat (a learner send racing the wake pipeline's ~1s adoption window) into: wait out the in-flight turn (poll turnInFlight, 30s cap), refetch the record, MERGE the outgoing messages onto the fresh server list, re-POST. A bare retry would clobber the steer + wake messages via the write-ahead persist (steerDelivered makes that loss permanent) — merging is the whole point. Unit-tested: pass-through, wait+merge+reorder, dedupe by id, repeated-409 loop, bounded give-up. - src/core/pending.ts: findPendingJobCall deleted (dead since wave 1 — job tools resolve server-side with an ack, nothing pends client-side). JOB_TOOLS kept (runner.ts still uses it). Header doc rewritten. - src/server/jobs.ts: GRACE_MS 1500 → 750, with the evidence in the comment: the race window is only the learner's answer→POST flight (sendAutomaticallyWhen dispatches synchronously; ~10ms localhost, <300ms realistic RTT — 750 is 2.5× worst case) and misses self-heal via the pump's gate recheck, the 409 backoff ramp, and the new transport merge-retry. POLL_MS=400 kept (sampling granularity only). - tests/jobs.test.ts: prompt-size guard — every fragment ≤8k chars, composed prompt ≤60k — the README-leak class (815 bytes ≈ 200 tokens per teacher turn before wave 1's fix) can't silently recur. Gates: typecheck, 279/279 tests, lint, build — all green. --- components/chat-transport.ts | 117 ++++++++++++++++++++++++++++++ src/core/pending.ts | 12 ++- src/server/jobs.ts | 16 +++- tests/chat-transport.test.ts | 137 +++++++++++++++++++++++++++++++++++ tests/jobs.test.ts | 15 +++- 5 files changed, 287 insertions(+), 10 deletions(-) create mode 100644 components/chat-transport.ts create mode 100644 tests/chat-transport.test.ts diff --git a/components/chat-transport.ts b/components/chat-transport.ts new file mode 100644 index 0000000..1a5d7a3 --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + return new DefaultChatTransport({ + api: "/api/chat", + body: { chatId }, + fetch: retryingFetch(chatId), + }); +} \ No newline at end of file diff --git a/src/core/pending.ts b/src/core/pending.ts index 37c7f4a..1c65b15 100644 --- a/src/core/pending.ts +++ b/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 " 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); -} + diff --git a/src/server/jobs.ts b/src/server/jobs.ts index bde2032..160fb60 100644 --- a/src/server/jobs.ts +++ b/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; diff --git a/tests/chat-transport.test.ts b/tests/chat-transport.test.ts new file mode 100644 index 0000000..b7b18be --- /dev/null +++ b/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/ → 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}`); + }); +}); \ No newline at end of file diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index 6dab6ba..2b4712e 100644 --- a/tests/jobs.test.ts +++ b/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"); + }); });