From 643e7596f0490c815b5985104ba88fd7f6a6617b Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Sat, 10 Oct 2026 09:42:47 +0300 Subject: [PATCH] =?UTF-8?q?streaming:=20connection=20diet=20=E2=80=94=20th?= =?UTF-8?q?e=20parallel-session=20freeze,=20fixed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause (postmortem, verified live): Chromium caps 6 concurrent HTTP/1.1 connections per origin. Every running turn pinned slots for its whole life — the POST /api/chat response stream plus a reattach GET stream per visit that never closed on navigate-away — so two concurrent thinking sessions starved navigation RSC fetches in the browser queue (147s measured) until the turns settled. Server was exonerated (3ms curls during the freeze); main thread idle throughout. - COMPACTION LAW + WIRE DIET (ui-stream): store the states, not the chunks — replace-in-place snapshot retention keyed by updateSlots, ~100ms coalesced latest-wins fan-out, slim wire drops pi's duplicate assistantMessageEvent payload. - DIET B: POST /api/chat acks and closes ({ok:true}) — the turn wire is GET /stream for everyone, submitter included; send() reattaches via refetch-first resumeStream(true). - DIET A: unmount cleanup calls model.stop() — navigating away aborts the view connection (turns are server-side background objects). - Answer-flicker fix: resumeStream's eager tail clear moved off the refetch path (the refetch swaps base+tail atomically; clean-slate guard restored on refetch failure) — the optimistic answer card survives refetch latency (1.7s stalls measured on dev). - Tests converted to the ack contract (drainTurn/drainRunning helpers, dietFetch mock) + a pin that the POST body reads while the turn runs. Gates: 1489/1489, typecheck, lint. Live: dual streaming sessions, navigations 335-648ms, in-page settings 19ms (was 147s). --- app/api/chats/[id]/stream/route.ts | 12 +- components/session-core.ts | 42 +++++-- components/session.ts | 13 ++ src/server/chat-handler.ts | 19 ++- src/server/pi/ui-stream.ts | 133 +++++++++++++++++--- tests/chat-flow.test.ts | 188 +++++++++++++++++++++-------- tests/demos.test.tsx | 58 +++++---- tests/jobs.test.ts | 19 ++- tests/practice-flow.test.ts | 39 +++--- tests/session-core.test.ts | 70 ++++++++--- 10 files changed, 441 insertions(+), 152 deletions(-) diff --git a/app/api/chats/[id]/stream/route.ts b/app/api/chats/[id]/stream/route.ts index aa92275..6dcc08f 100644 --- a/app/api/chats/[id]/stream/route.ts +++ b/app/api/chats/[id]/stream/route.ts @@ -1,10 +1,12 @@ /** * 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. + * 2026-09-25) — and, since CONNECTION DIET B (owner, 2026-10-10), THE + * turn wire, period: POST /api/chat acknowledges and closes, so the + * submitter too attaches here. One SSE stream per chat: 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 diff --git a/components/session-core.ts b/components/session-core.ts index 9d94808..5ae31e8 100644 --- a/components/session-core.ts +++ b/components/session-core.ts @@ -474,14 +474,23 @@ export class SessionModel { // aborted earlier attempt (React StrictMode's simulated remount) // must not leave partials the replay would DOUBLE onto. The old // record refetch provided this reset accidentally; attach-first - // made it explicit. Safe on both paths: a fresh mount's tail is - // empty or aborted-replay leftovers; the mid-session handoff path - // still refetches first (its reset is redundant, not harmful — - // the settle sync has already canonized the view). - this.mutate(() => { - this.tail = []; - this.streamingMessageId = null; - }); + // made it explicit. + // + // THE FLICKER FIX (owner, 2026-10-10 — connection-diet B follow-up): + // on the REFETCH path the clear must NOT precede the fetch — send() + // just landed the optimistic answer/note cards in the tail, and any + // refetch latency (dev-compile stalls measured at 1.7s) re-showed + // the answered-away question for the whole window: card vanished, + // dock returned, both snapped back only when the record landed. + // The refetch block below swaps base+tail ATOMICALLY when the record + // arrives (and clears on failure — the clean-slate guard holds on + // every path that actually replays onto a stale tail). + if (!refetchRecordFirst) { + this.mutate(() => { + this.tail = []; + this.streamingMessageId = null; + }); + } this.setStatus("submitted"); try { if (refetchRecordFirst) { @@ -496,6 +505,13 @@ export class SessionModel { this.tail = []; }); } + } else { + // The record did not land — restore the clean-slate guard + // before the replay (no atomic swap is coming). + this.mutate(() => { + this.tail = []; + this.streamingMessageId = null; + }); } } const res = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}/stream`, { @@ -558,7 +574,15 @@ export class SessionModel { // page's refusal flow JSON-parses the envelope out of it. throw new Error((await res.text()) || `The send failed (${res.status}).`); } - await this.consumeStream(res); + // CONNECTION DIET B (owner, 2026-10-10 — parallel-session freeze + // postmortem): the POST acknowledges and closes; there is no stream + // on this response to consume. Attach through the SAME reattach wire + // a returning visitor uses (refetch-first: the record refetch absorbs + // the server's write-ahead copy of our optimistic messages, then the + // replay lands on top). One live view connection per page, always — + // the browser's 6-per-origin pool is the scarce resource. + await res.text().catch(() => {}); // drain the tiny ack; release the socket + await this.resumeStream(true); } catch (err) { if ((err as Error).name === "AbortError") return; // stop() or a newer consume won this.setError((err as Error).message); diff --git a/components/session.ts b/components/session.ts index b141142..2fb1e06 100644 --- a/components/session.ts +++ b/components/session.ts @@ -54,8 +54,21 @@ export function useChatSession({ chatId, initialMessages, initiallyInFlight }: U // latency before the instant replay, so a fresh mount attaches // directly. This hook's effect is the SINGLE mount-reattach owner // (chat.tsx's duplicate was deleted with the same round). + // + // CONNECTION DIET A (owner, 2026-10-10 — parallel-session freeze + // postmortem): unmount RELEASES the stream connection. The turn is a + // server-side process — it runs on with zero views; this page's open + // GET stream (or, pre-diet-B, its POST stream) used to survive + // navigation, silently pinning one of the browser's 6-per-origin + // connection slots until the turn settled, streaming into a dead + // component. Enough leaked views starved every NEW request (navigation + // RSC fetches included) — the multi-session navigation freeze. Aborting + // costs nothing: navigating back replays the compacted log + live tail. useEffect(() => { if (initiallyInFlight) void model.resumeStream(false); + return () => { + model.stop(); + }; }, [initiallyInFlight, model]); // The view + state, derived per model change from the model. diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index 179377f..e4e5108 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -21,7 +21,7 @@ import { steerLiveChat, unregisterLiveAgent, } from "./jobs.ts"; -import { piTurnResponse, runPiTurn } from "./pi/ui-stream.ts"; +import { runPiTurn } from "./pi/ui-stream.ts"; import { collectToolCallIds } from "./pi/call-ids.ts"; import { providerEventNote } from "../agent/provider-events.ts"; import { effectiveSettings, stores } from "./stores.ts"; @@ -439,6 +439,10 @@ export async function handleChatInput( providerEventNote(chatId, "turn", `run start (base: ${merged.length} messages)`); const handle = runPiTurn({ agent: constructed, + // The coalescing window seam (COMPACTION LAW, 2026-10-09): tests + // tighten it via LH_UPDATE_FLUSH_MS; production rides the 100ms + // default in ui-stream. + ...(process.env.LH_UPDATE_FLUSH_MS ? { updateFlushMs: Number(process.env.LH_UPDATE_FLUSH_MS) } : {}), // THE LACAN FIX (2026-10-04): seed the uniquifier with every // tool-call id the record already holds — a provider that resets // ids per response (DeepSeek) then gets fresh unique ids for this @@ -545,9 +549,16 @@ export async function handleChatInput( // 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); + // CONNECTION DIET B (owner, 2026-10-10 — the parallel-session freeze + // postmortem): the POST no longer holds the turn's SSE stream open for + // its whole duration. It acknowledges acceptance and CLOSES; the + // submitter attaches through GET /api/chats/[id]/stream — the same + // universal reattach wire a returning visitor uses. The browser's + // 6-connections-per-origin pool is the scarce resource: a submission + // must cost zero permanently-held slots beyond the one live view. The + // turn itself is a server-side process (ruling 2026-09-25) — it never + // needed this connection, only a view did. + return Response.json({ ok: true }); } catch (err) { // The stream never started, so no onFinish will fire — clear the registry // or a malformed request would lock the chat until the TTL eviction. diff --git a/src/server/pi/ui-stream.ts b/src/server/pi/ui-stream.ts index 83f65d0..7308bd1 100644 --- a/src/server/pi/ui-stream.ts +++ b/src/server/pi/ui-stream.ts @@ -12,7 +12,7 @@ import { ToolCallIdUniquifier } from "./call-ids.ts"; * The SSE body is the agent's own event stream: * * data: {"type":"agent_start"}\n\n - * data: {"type":"message_update","message":{…},"assistantMessageEvent":{…}}\n\n + * data: {"type":"message_update","message":{…}}\n\n * … * data: {"type":"agent_end","messages":[…]}\n\n * data: [DONE]\n\n @@ -21,6 +21,11 @@ import { ToolCallIdUniquifier } from "./call-ids.ts"; * - Every `message_update` carries the FULL partial assistant message (each * AssistantMessageEvent ships `partial`) — a view REPLACES its streaming * assistant with `event.message`; no delta assembly exists anywhere. + * - WIRE DIET (owner, 2026-10-09 — the streaming-freeze postmortem): our + * boundary emits the SLIM message_update (`message` only). pi's event also + * carries the same full partial a SECOND time as `assistantMessageEvent` + * — nothing in this app ever reads it, and re-serializing it per chunk + * doubled every stream frame. Client reads `event.message`, always. * - Question tools (quiz / ask_user_question): NO suppression (owner ruling, * Wave D fork 2) — the closing `tool_execution_end` with the in-memory * "Question posed" placeholder flows to the client, which ignores it @@ -38,16 +43,31 @@ import { ToolCallIdUniquifier } from "./call-ids.ts"; * * 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. + * clients are VIEWS. runPiTurn returns the handle (compacted state + + * 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. + * + * THE COMPACTION LAW (owner, 2026-10-09 — supersedes the unbounded-log ruling + * of 2026-09-25): the turn log stores STATE, not chunks. pi's native events + * are full snapshots — one per provider chunk — so retaining every one made + * memory and replay cost quadratic in message length (a single + * high-thinking message retained hundreds of MB and every reattach + * re-serialized the whole history in one blocking stretch — the parallel- + * session navigation freeze). The log now keeps settled events verbatim plus + * the LATEST update per in-flight message/tool call, replaced in place; + * live update fan-out is coalesced to at most one frame per flush window + * (latest wins — views replace by slot, and message_end carries the full + * final message, so dropped intermediates are lossless). Replay therefore + * costs the content once, never the chunk count. */ -/** The wire event: a verbatim pi AgentEvent, or one of the two app events. */ +/** The wire event: a pi AgentEvent (with our slim message_update — see the + * WIRE DIET note above), or one of the two app events. */ export type TurnWireEvent = | AgentEvent + | { type: "message_update"; message: AgentMessage } | { type: "error"; errorText: string }; export interface PiTurnSpec { @@ -69,6 +89,9 @@ export interface PiTurnSpec { * routing, and the next turn's provider context all agree. Absent → * verbatim passthrough (the protocol law below still holds). */ usedToolCallIds?: ReadonlySet; + /** The coalescing window for update fan-out (COMPACTION LAW above). + * Defaults to 100ms; tests tighten it. */ + updateFlushMs?: number; } /** One view onto a running turn: push() replays/streams an event, done() @@ -94,33 +117,93 @@ export interface PiTurnHandle { } export function runPiTurn(spec: PiTurnSpec): PiTurnHandle { - const { agent, onError, onFinish, usedToolCallIds } = spec; + const { agent, onError, onFinish, usedToolCallIds, updateFlushMs = 100 } = spec; const uniquifier = usedToolCallIds ? new ToolCallIdUniquifier(usedToolCallIds) : undefined; let settled = false; let sawAbort = false; - // THE PER-TURN EVENT LOG (memory-only — the settle-only disk writes - // are unchanged): every forwarded event, in order, so any late - // subscriber gets a FULL replay. Unbounded BY HONEST DESIGN (owner - // ruling 2026-09-25): a turn's event 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. + // THE PER-TURN STATE LOG (memory-only — the settle-only disk writes + // are unchanged). COMPACTION LAW (owner, 2026-10-09, supersedes the + // 2026-09-25 unbounded-log ruling): settled events append verbatim; + // in-flight `message_update` / `tool_execution_update` runs are reduced + // to their LATEST member, replaced in place. Retention is bounded by + // CONTENT (settled messages + one snapshot per streaming message), not + // by chunk count — the quadratic blowup that froze parallel sessions is + // structurally impossible now. The log dies with the turn at settle. const log: TurnWireEvent[] = []; + // Live-update slots: coalesce key → log index of the retained update. + // One assistant message streams at a time (pi's strict per-step + // start → updates → end ordering), so messages share one slot; parallel + // tool executions each own theirs (toolCallId). End events free the slot + // — a later update for the same key starts a fresh run. + const updateSlots = new Map(); + // The coalescing queue for LIVE fan-out: key → latest event, flushed at + // most once per flush window (Map preserves first-insertion order, so the + // relative order of concurrent keys stays chronological). + const pending = new Map(); const subscribers = new Set(); - // Send = append to the log + fan out to every live view. A view's own + const coalesceKey = (event: TurnWireEvent): string | null => { + if (event.type === "message_update") return "message"; + if (event.type === "tool_execution_update") return `tool:${event.toolCallId}`; + return null; + }; + + // Send = compact 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 = (event: TurnWireEvent) => { if (settled) return; + const key = coalesceKey(event); + if (key !== null) { + // Retention: replace-in-place the run's latest snapshot. + const idx = updateSlots.get(key); + if (idx !== undefined) log[idx] = event; + else { + updateSlots.set(key, log.length); + log.push(event); + } + // Fan-out: coalesced — latest wins, flushed at ≤1/window. + pending.set(key, event); + armFlush(); + return; + } + // A boundary event ends its update run (end supersedes the partial; + // the slot frees for any future run of the same key). + if (event.type === "message_end") updateSlots.delete("message"); + if (event.type === "tool_execution_end") updateSlots.delete(`tool:${event.toolCallId}`); + // Strict per-subscriber ordering: anything still pending precedes + // this boundary event on the wire, never after it. + flushPending(); log.push(event); for (const sub of subscribers) sub.push(event); }; + + let flushTimer: ReturnType | undefined; + const flushPending = () => { + if (flushTimer !== undefined) { + clearTimeout(flushTimer); + flushTimer = undefined; + } + if (pending.size === 0) return; + const batch = [...pending.values()]; + pending.clear(); + for (const sub of subscribers) for (const event of batch) sub.push(event); + }; + const armFlush = () => { + if (flushTimer !== undefined) return; + flushTimer = setTimeout(() => { + flushTimer = undefined; + flushPending(); + }, updateFlushMs); + // Never holds the process (or the test runner) open for a straggler. + (flushTimer as { unref?: () => void }).unref?.(); + }; const closeAll = () => { if (settled) return; + // Order to the very end: the last pending snapshot precedes [DONE]. + flushPending(); settled = true; for (const sub of subscribers) sub.done(); subscribers.clear(); @@ -143,7 +226,14 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle { const handle = (event: AgentEvent) => { // THE LACAN FIX: ids uniquify HERE, before any view — the one seam // every consumer downstream shares. No uniquifier → verbatim. - send(uniquifier ? uniquifier.event(event) : event); + const uniqued = uniquifier ? uniquifier.event(event) : event; + // WIRE DIET: slim the one event pi double-stuffs — our views read + // `message`, never `assistantMessageEvent` (same full partial twice). + const wire: TurnWireEvent = + uniqued.type === "message_update" + ? { type: "message_update", message: uniqued.message } + : uniqued; + send(wire); if (event.type === "message_end") { const m = event.message as { role?: string; stopReason?: string } | undefined; if (m?.role === "assistant" && m.stopReason === "aborted") sawAbort = true; @@ -206,7 +296,10 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle { }; // Attach THEN replay — synchronous, so no event can fall // between replay and live (single-threaded: the agent's - // handle() only runs after this stack returns). + // handle() only runs after this stack returns). The compacted + // log replays settled events + the latest snapshot per + // in-flight message — the full content, once, never the chunk + // history (COMPACTION LAW). subscribers.add(sub); for (const event of log) sub.push(event); // A subscriber that attached after settle (the unregister @@ -216,7 +309,7 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle { 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. + // runs on; the compacted log keeps the state for the next view. if (sub) subscribers.delete(sub); }, }); diff --git a/tests/chat-flow.test.ts b/tests/chat-flow.test.ts index 5af2009..a638d6e 100644 --- a/tests/chat-flow.test.ts +++ b/tests/chat-flow.test.ts @@ -139,6 +139,40 @@ async function drain(res: Response): Promise { return res.text(); // consumes the SSE stream to completion (fires onFinish) } +/** CONNECTION DIET B (owner, 2026-10-10 — parallel-session freeze postmortem): + * POST /api/chat no longer streams the turn — it acks and closes, and the + * submitter (like every view) attaches via the reattach stream. This + * mirrors drain(): assert the 200 ack, then drain the turn's SSE body to + * completion through runningTurn + subscribe — the same wire the GET + * route serves. Subscribe IMMEDIATELY after the ack: faux turns settle in + * microtask rounds, and the handle leaves the registry at settle. */ +async function drainTurn(req: Request, factory: Parameters[1]): Promise { + const cloned = req.clone(); + const ack = await handleChatRequest(req, factory); + if (ack.status !== 200) { + assert.fail(`expected 200 ack, got ${ack.status}: ${await ack.text()}`); + } + const { chatId } = (await cloned.json()) as { chatId: string }; + const handle = runningTurn(chatId); + if (!handle) + // A short turn (a quick answer reaction) can settle INSIDE the + // handler's trailing awaits — the registry is already clear. There is + // nothing to drain: completion happened. (Body-reading callers get "" — + // their eventsOf asserts then fail loudly if this ever bites a long + // turn, which is the honest outcome for a lost wire.) + return ""; + return new Response(handle.subscribe()).text(); +} + +/** Drain a RUNNING turn's wire when the POST already happened (the ack + * path left nothing to consume — CONNECTION DIET B). Subscribe immediately: + * faux turns settle in microtask rounds. */ +async function drainRunning(chatId: string): Promise { + const handle = runningTurn(chatId); + if (!handle) assert.fail(`no running turn to drain: ${chatId}`); + return new Response(handle.subscribe()).text(); +} + function waitFor(cond: () => boolean, timeoutMs = 8000, what = "condition"): Promise { return new Promise((resolve, reject) => { const started = Date.now(); @@ -244,7 +278,7 @@ describe("teacher chat flow (scripted model)", () => { createdChats.push(chat.id); // ── Turn 1: learner opens with a topic; teacher probes with a quiz ── - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "teach me the solar system" }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "text", text: "teach me the solar system" }), agentFactory(engine)); const saved1 = chatStore.getChat(chat.id)!; // Native record: user + assistant (fork 1 — the clean question's @@ -282,7 +316,7 @@ describe("teacher chat flow (scripted model)", () => { // EVENT — the answer lands as the call's real tool result. ── const answer = gradedAnswer(QUIZ_INPUT, ["Mercury"]); assert.equal(answer.correct, true); - await drain(await handleChatRequest(post(chat.id, { type: "answer", toolCallId: "call-1", answer }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "answer", toolCallId: "call-1", answer }), agentFactory(engine)); // ── The teacher reacts — the projection groups the reaction into the // SAME assistant view message as the answered quiz (the native @@ -359,7 +393,7 @@ describe("teacher chat flow (scripted model)", () => { const { engine } = scriptedTeacher(defaultLesson); const chat = createConfiguredChat("Test: protocol pin"); createdChats.push(chat.id); - const body = await drain(await handleChatRequest(post(chat.id, { type: "text", text: "quiz me" }), agentFactory(engine))); + const body = await drainTurn(post(chat.id, { type: "text", text: "quiz me" }), agentFactory(engine)); const events = eventsOf(body); const callSeen = events.findIndex( (e) => e.type === "message_update" && JSON.stringify(e.message?.content ?? []).includes('"type":"toolCall"'), @@ -385,7 +419,7 @@ describe("teacher chat flow (scripted model)", () => { createdChats.push(chat.id); // Seed a plain-text turn (no dangling question — waitForIdle must pass). const seed = scriptedTeacher(() => [() => fauxAssistantMessage([fauxText("On it — dispatched the researcher.")])]); - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "research it" }), agentFactory(seed.engine))); + await drainTurn(post(chat.id, { type: "text", text: "research it" }), agentFactory(seed.engine)); // The pump's turn runner, faithfully shaped but scripted (the real one // is chat-handler's module-level wiring; this file's import already set @@ -595,8 +629,11 @@ describe("teacher chat flow (scripted model)", () => { const res = await handleChatRequest(post(chat.id, { type: "text", text: "hold my turn" }), agentFactory(engine)); assert.equal(res.status, 200); - // Read one chunk then cancel — the browser view leaves mid-stream. - const reader = res.body!.getReader(); + // CONNECTION DIET B (2026-10-10): the POST is a closed ack — the view + // wire is the reattach stream. Attach, read one chunk, then cancel — + // the browser view leaves mid-stream. + const view = new Response(runningTurn(chat.id)!.subscribe()); + const reader = view.body!.getReader(); await reader.read(); await reader.cancel(); // The turn is STILL a running background object — no view killed it. @@ -647,7 +684,7 @@ describe("teacher chat flow (scripted model)", () => { const chat = createConfiguredChat("Test: bad quiz retry"); createdChats.push(chat.id); - const body = await drain(await handleChatRequest(post(chat.id, { type: "text", text: "quiz me" }), agentFactory(engine))); + const body = await drainTurn(post(chat.id, { type: "text", text: "quiz me" }), agentFactory(engine)); // WIRE pin (the live modal-realism crash): the invalid call's execution // end is an ERROR — the dock only mounts on a CLEAN tool_execution_end, @@ -715,7 +752,7 @@ describe("teacher chat flow (scripted model)", () => { const chat = createConfiguredChat("Test: string options"); createdChats.push(chat.id); - const body = await drain(await handleChatRequest(post(chat.id, { type: "text", text: "quiz me" }), agentFactory(engine))); + const body = await drainTurn(post(chat.id, { type: "text", text: "quiz me" }), agentFactory(engine)); const events = eventsOf(body); const stringyEnd = events.find((e) => e.type === "tool_execution_end" && e.toolCallId === "call-stringy"); @@ -754,7 +791,7 @@ describe("teacher chat flow (scripted model)", () => { const chat = createConfiguredChat("Test: parallel quizzes"); createdChats.push(chat.id); - const body = await drain(await handleChatRequest(post(chat.id, { type: "text", text: "quiz me twice" }), agentFactory(engine))); + const body = await drainTurn(post(chat.id, { type: "text", text: "quiz me twice" }), agentFactory(engine)); // (B) wire determinism: question tools execute sequentially, so the // quizzes' CLEAN execution ends arrive in call order (the dock mounts @@ -814,19 +851,20 @@ describe("mid-turn navigation (turn liveness + write-ahead)", () => { const chat = createConfiguredChat("Test: turn guard"); createdChats.push(chat.id); - const res1 = await handleChatRequest(post(chat.id, { type: "text", text: "teach me the solar system" }), agentFactory(engine)); // NOT drained → in flight + const res1 = await handleChatRequest(post(chat.id, { type: "text", text: "teach me the solar system" }), agentFactory(engine)); // acked; the turn runs as a background object (Diet B: no stream to hold) assert.equal(isTurnRunning(chat.id), true, "turn registered as in flight"); + const view1 = drainRunning(chat.id); // a live view — the turn would run on with none // A stale client (remount racing a live turn) tries to start a second turn. const res2 = await handleChatRequest(post(chat.id, { type: "text", text: "teach me the solar system" }), agentFactory(engine)); assert.equal(res2.status, 409, "concurrent turn rejected — last-writer-wins persist would clobber it"); - await drain(res1); // finish → onFinish → endTurn + await view1; // finish → onFinish → endTurn assert.equal(isTurnRunning(chat.id), false, "registry cleared at finish"); const res3 = await handleChatRequest(post(chat.id, { type: "text", text: "teach me the solar system" }), agentFactory(engine)); assert.equal(res3.status, 200, "the next turn is accepted after the previous finished"); - await drain(res3); + await drainRunning(chat.id); }); it("write-ahead: the learner's answer is durable BEFORE the turn finishes", async () => { @@ -835,7 +873,7 @@ describe("mid-turn navigation (turn liveness + write-ahead)", () => { createdChats.push(chat.id); // Turn 1: the teacher asks the quiz (dangling call in the record). - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "teach me the solar system" }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "text", text: "teach me the solar system" }), agentFactory(engine)); // The learner answers; the client posts the ANSWER EVENT. const answer = gradedAnswer(QUIZ_INPUT, ["Mercury"]); @@ -844,14 +882,13 @@ describe("mid-turn navigation (turn liveness + write-ahead)", () => { // on disk — a remount (or crash) mid-turn sees the answered quiz, never // the stale pre-answer question. const res = await handleChatRequest(post(chat.id, { type: "answer", toolCallId: "call-1", answer }), agentFactory(engine)); - const midTurn = chatStore.getChat(chat.id)!; - assert.equal(midTurn.messages.length, 3, "write-ahead appended the answer tool result before finish"); + const midTurn = chatStore.getChat(chat.id)!; assert.equal(midTurn.messages.length, 3, "write-ahead appended the answer tool result before finish"); assert.equal((midTurn.messages[2] as { role?: string }).role, "toolResult"); const midQuiz = projectView(midTurn.messages)[1]!.parts.find((p) => p.type === "tool-quiz"); assert.ok(midQuiz && midQuiz.type === "tool-quiz"); assert.equal((midQuiz as { state?: string }).state, "output-available", "write-ahead persisted the answer before finish"); - await drain(res); // finish → persist the full turn → endTurn + await drainRunning(chat.id); // finish → persist the full turn → endTurn const saved2 = chatStore.getChat(chat.id)!; // Native: user, assistant, answer result, reaction — one append-only // history; the projection still renders ONE assistant group. @@ -910,7 +947,7 @@ describe("the BYOK turn gate (no factory floor — owner, 2026-09-22)", () => { assert.match(body.error, /No model configured/); } else { assert.equal(res.status, 200, "the record's strong slot satisfies the conversation chain"); - await drain(res); + await drainRunning(chat.id); } // Either way: no phantom liveness, nothing durable from a refused turn. @@ -926,7 +963,7 @@ describe("the BYOK turn gate (no factory floor — owner, 2026-09-22)", () => { createdChats.push(chat.id); const res = await handleChatRequest(post(chat.id, { type: "text", text: "teach me the solar system" }), agentFactory(scriptedTeacher(defaultLesson).engine)); assert.equal(res.status, 200, "chat.settings.model satisfies the chain without any record slot"); - await drain(res); + await drainRunning(chat.id); }); }); @@ -973,7 +1010,7 @@ describe("the key-presence turn gate (provider-key interlude, 2026-09-25)", () = const res = await handleChatRequest(post(chat.id, { type: "text", text: "teach me something" }), agentFactory(scriptedTeacher(defaultLesson).engine)); assert.equal(res.status, 200, "the env-shadowed key counts as present — the turn runs"); - await drain(res); + await drainRunning(chat.id); } finally { if (saved !== undefined) process.env.OPENAI_API_KEY = saved; else delete process.env.OPENAI_API_KEY; @@ -982,6 +1019,27 @@ 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("CONNECTION DIET B (owner, 2026-10-10): the POST acks and closes — the response never holds the turn's stream", async () => { + let release!: () => void; + const gate = new Promise((r) => (release = r)); + const engine = gatedTeacher(gate); + const chat = createConfiguredChat("Test: ack-and-close"); + createdChats.push(chat.id); + + const res = await handleChatRequest(post(chat.id, { type: "text", text: "hold it" }), agentFactory(engine)); + assert.equal(res.status, 200); + // The body must be FULLY readable while the turn is still gated + // mid-run — a closed JSON ack, not an open stream. (The freeze + // postmortem: each held-open POST response pinned one of the + // browser's 6-per-origin connections for the turn's whole life.) + const body = JSON.parse(await res.text()); + assert.deepEqual(body, { ok: true }, "the ack is a tiny JSON envelope"); + assert.equal(isTurnRunning(chat.id), true, "the turn runs on with its response closed"); + + release(); + await waitFor(() => !isTurnRunning(chat.id), 8000, "turn settle"); + }); + it("subscribe mid-turn: a second view gets the FULL replay, then live, same dialect", async () => { let release!: () => void; const gate = new Promise((r) => (release = r)); @@ -989,10 +1047,13 @@ describe("background-turn runner (owner ruling 2026-09-25: every turn is a serve const chat = createConfiguredChat("Test: reattach replay"); createdChats.push(chat.id); - const first = await handleChatRequest(post(chat.id, { type: "text", text: "hold my turn" }), agentFactory(engine)); - assert.equal(first.status, 200); - // Consume the first view a little (the start chunk is out). - const reader = first.body!.getReader(); + const ack = await handleChatRequest(post(chat.id, { type: "text", text: "hold my turn" }), agentFactory(engine)); + assert.equal(ack.status, 200); + // CONNECTION DIET B (2026-10-10): the POST acks and closes; the first + // view consumes a little of the reattach wire (the start chunk is out). + const firstHandle = runningTurn(chat.id)!; + const firstView = new Response(firstHandle.subscribe()); + const reader = firstView.body!.getReader(); await reader.read(); // A SECOND view attaches mid-turn — a remount, another device. @@ -1024,6 +1085,39 @@ describe("background-turn runner (owner ruling 2026-09-25: every turn is a serve assert.ok(firstTail.endsWith("data: [DONE]\n\n"), "first view closes cleanly at settle, unaffected by the reattach"); }); + it("COMPACTION LAW (owner 2026-10-09): the log stores state, not chunks — a chunk burst coalesces to ONE update, slim wire, full content", async () => { + // The parallel-session freeze postmortem: pi's message_update is a FULL + // snapshot per provider chunk; retaining and re-serializing every one + // made memory and replay quadratic in message length (one high-thinking + // message = hundreds of MB retained; every reattach re-stringified the + // whole history in one blocking stretch). Pins the fix end-to-end: faux + // streams a long text as many immediate chunks (microtask pacing — all + // inside one flush window), yet the wire carries exactly ONE + // message_update for the run, it is the LATEST snapshot, and the wire + // never carries pi's duplicate partial. + const long = "quadratic blowup ".repeat(40); // ~40 faux chunks + const faux = createFauxCore({}); + faux.setResponses([fauxAssistantMessage([fauxText(long)])]); + const engine: PiEngine = { model: faux.getModel(), streamFn: faux.streamSimple }; + const chat = createConfiguredChat("Test: compaction law"); + createdChats.push(chat.id); + const body = await drainTurn(post(chat.id, { type: "text", text: "go" }), agentFactory(engine)); + const events = eventsOf(body); + const updates = events.filter((e) => e.type === "message_update"); + assert.equal(updates.length, 1, "a burst of chunks coalesces to exactly one update on the wire"); + assert.ok( + JSON.stringify(updates[0]!.message?.content ?? []).includes(long), + "the one update is the LATEST snapshot — the full streamed text", + ); + assert.ok(!body.includes("assistantMessageEvent"), "WIRE DIET: the duplicate partial never crosses the boundary"); + const end = events.find((e) => e.type === "message_end" && (e as { message?: { role?: string } }).message?.role === "assistant"); + assert.ok(end, "message_end present"); + assert.ok(JSON.stringify((end as { message?: { content?: unknown[] } }).message?.content ?? []).includes(long), "message_end carries the full final message (authoritative, lossless)"); + // And the persisted record never knew about chunks at all. + const record = chatStore.getChat(chat.id)!; + assert.ok(JSON.stringify(record.messages).includes(long), "persistTurn wrote the settled message in full"); + }); + 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"); @@ -1032,7 +1126,7 @@ describe("background-turn runner (owner ruling 2026-09-25: every turn is a serve const res = await handleChatRequest(post(chat.id, { type: "text", text: "wait forever" }), 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 + const viewText = new Response(runningTurn(chat.id)!.subscribe()).text(); // the first view, reading to settle // The remote kill from a DIFFERENT view (the dock's abort). const outcome = abortTurn(chat.id); @@ -1062,7 +1156,7 @@ describe("background-turn runner (owner ruling 2026-09-25: every turn is a serve // And after a settled turn, the registry is empty — the GET would // serve the same terminal, not a stale handle. - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "quiz me" }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "text", text: "quiz me" }), agentFactory(engine)); assert.equal(runningTurn(chat.id), undefined, "no handle survives settle"); }); @@ -1134,7 +1228,7 @@ it("explain tool call persists as a pending part; the free-text answer reveals t const factory: Parameters[1] = (opts) => createTeacherAgent(undefined, { engine, messages: opts.messages, tools }); - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "teach me the solar system" }), factory)); + await drainTurn(post(chat.id, { type: "text", text: "teach me the solar system" }), factory); const saved1 = chatStore.getChat(chat.id)!; const savedContent1 = (saved1.messages[1] as { content: unknown }).content as Array<{ type: string; name?: string }>; @@ -1161,7 +1255,7 @@ it("explain tool call persists as a pending part; the free-text answer reveals t const answer = buildAnswer(qd, { text: "It stays near the Sun." }); assert.equal(answer.correct, undefined, "no client grading — the model grades"); - await drain(await handleChatRequest(post(chat.id, { type: "answer", toolCallId: "call-1", answer }), factory)); + await drainTurn(post(chat.id, { type: "answer", toolCallId: "call-1", answer }), factory); const saved2 = chatStore.getChat(chat.id)!; const view2 = projectView(saved2.messages); @@ -1190,7 +1284,7 @@ describe("the silent dismissal (A2)", () => { createdChats.push(chat.id); // Turn 1: the quiz dangles. - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "teach me the solar system" }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "text", text: "teach me the solar system" }), agentFactory(engine)); assert.ok(danglingQuestionOf(chatStore.getChat(chat.id)!.messages), "the quiz is pending"); // The dismissal: SILENT — record completes the call, nothing wakes. @@ -1208,7 +1302,7 @@ describe("the silent dismissal (A2)", () => { // The next plain message is THE wake — the model sees the skipped // question and the user's text in one context (the steer-batch // outcome, delivered by record truth). - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "skip that — tell me about Mars" }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "text", text: "skip that — tell me about Mars" }), agentFactory(engine)); assert.equal(calls.length, 2, "exactly one wake: the user's message"); assert.match(calls[1]!.join("\n"), /dismissed/, "the dismissed result rode the wake's context"); assert.equal(endsInDismissal(chatStore.getChat(chat.id)!.messages), false, "the pause lifted with the send"); @@ -1295,7 +1389,7 @@ describe("parallel question calls: the answer defers its wake while a sibling da // Turn 1: the batch lands; both calls dangle (the dock shows the quiz — // the first in call order). - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "ask me two questions in parallel" }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "text", text: "ask me two questions in parallel" }), agentFactory(engine)); let saved = chatStore.getChat(chat.id)!; let pending = danglingQuestionOf(saved.messages); assert.equal(pending?.toolCallId, "call-1", "the dock holds the quiz first"); @@ -1303,7 +1397,7 @@ describe("parallel question calls: the answer defers its wake while a sibling da // The graded answer: the result appends ADJACENT to the calling // assistant message — and NO wake fires while the ask still dangles. const quizAns = gradedAnswer(QUIZ_INPUT, ["Mercury"]); - await drain(await handleChatRequest(post(chat.id, { type: "answer", toolCallId: "call-1", answer: quizAns }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "answer", toolCallId: "call-1", answer: quizAns }), agentFactory(engine)); assert.equal(calls.length, 1, "the sibling answer woke NOTHING (the gate deferred)"); saved = chatStore.getChat(chat.id)!; assert.equal((saved.messages[2] as { toolCallId?: string }).toolCallId, "call-1", "quiz result adjacent to the batch"); @@ -1320,7 +1414,7 @@ describe("parallel question calls: the answer defers its wake while a sibling da text: "History", answeredAt: new Date().toISOString(), }; - await drain(await handleChatRequest(post(chat.id, { type: "answer", toolCallId: "call-2", answer: askAns }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "answer", toolCallId: "call-2", answer: askAns }), agentFactory(engine)); assert.equal(calls.length, 2, "exactly one wake: the last answer's turn"); saved = chatStore.getChat(chat.id)!; assert.equal(danglingQuestionOf(saved.messages), null, "nothing dangles anymore"); @@ -1360,10 +1454,10 @@ describe("parallel rounds: the LAST resolution decides the wake (owner ruling 20 const { engine, calls } = scriptedTeacher(batchLesson); const chat = createConfiguredChat("Test: round 1"); createdChats.push(chat.id); - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "two questions please" }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "text", text: "two questions please" }), agentFactory(engine)); // Answer the quiz — deferred by the parallel gate. - await drain(await handleChatRequest(post(chat.id, { type: "answer", toolCallId: "call-1", answer: gradedAnswer(QUIZ_INPUT, ["Mercury"]) }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "answer", toolCallId: "call-1", answer: gradedAnswer(QUIZ_INPUT, ["Mercury"]) }), agentFactory(engine)); assert.equal(calls.length, 1, "the sibling answer deferred"); // Dismiss the ask — the LAST resolution is a dismissal: no continuation @@ -1380,12 +1474,12 @@ describe("parallel rounds: the LAST resolution decides the wake (owner ruling 20 const { engine, calls } = scriptedTeacher(batchLesson); const chat = createConfiguredChat("Test: round 2"); createdChats.push(chat.id); - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "two questions please" }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "text", text: "two questions please" }), agentFactory(engine)); assert.equal((await dismissPost(chat.id, "call-1")).status, 200); assert.equal(calls.length, 1, "the dismissal woke nothing"); - await drain(await handleChatRequest(post(chat.id, { type: "answer", toolCallId: "call-2", answer: { kind: "ask", questionText: "Next topic?", dontKnow: false, answers: [], text: "Chemistry", answeredAt: new Date().toISOString() } }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "answer", toolCallId: "call-2", answer: { kind: "ask", questionText: "Next topic?", dontKnow: false, answers: [], text: "Chemistry", answeredAt: new Date().toISOString() } }), agentFactory(engine)); assert.equal(calls.length, 2, "the trailing answer fired the one wake"); const prompt2 = calls[1]!.join("\n"); assert.match(prompt2, /Chemistry/, "the answer rode the wake"); @@ -1451,8 +1545,11 @@ describe("composer steering: POST /api/chats/[id]/steer", () => { assert.equal(res.status, 200); assert.equal((await res.json() as { ok: boolean }).ok, true); + // CONNECTION DIET B: the drain rides the reattach wire (the POST + // acked long ago). Attach BEFORE releasing — a live view. + const viewText = drainRunning(chat.id); release(); // the first response completes → the drain → the second call - const body = await drain(turnRes); + const body = await viewText; assert.match(body, /also cover Venus/, "the drained steer rode the live stream verbatim"); // ONE turn: the second model call's context contains the steer @@ -1503,22 +1600,13 @@ describe("duplicate provider tool-call ids (the Lacan fix)", () => { const chat = chatStore.createChat("Lacan: duplicate call_0 turns"); // Turn 1: the ask question. - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "teach me Lacan" }), agentFactory(engine))); + await drainTurn(post(chat.id, { type: "text", text: "teach me Lacan" }), agentFactory(engine)); let saved = chatStore.getChat(chat.id)!; let pending = danglingQuestionOf(saved.messages)!; assert.equal(pending.tool, "ask_user_question"); // The client answers with the id IT SAW — the persisted one. const askId = pending.toolCallId; - await drain( - await handleChatRequest( - post(chat.id, { - type: "answer", - toolCallId: askId, - answer: { kind: "ask", questionText: ASK.question, dontKnow: false, answers: [], text: "The theory core", answeredAt: new Date().toISOString() }, - }), - agentFactory(engine), - ), - ); + await drainTurn(post(chat.id, { type: "answer", toolCallId: askId, answer: { kind: "ask", questionText: ASK.question, dontKnow: false, answers: [], text: "The theory core", answeredAt: new Date().toISOString() }, }), agentFactory(engine)); // Turn 2: the quiz — ALSO `call_0` from the provider. saved = chatStore.getChat(chat.id)!; @@ -1527,9 +1615,7 @@ describe("duplicate provider tool-call ids (the Lacan fix)", () => { const quizId = pending.toolCallId; assert.notEqual(quizId, askId, "the uniquifier reminted the colliding id at the wire boundary"); assert.match(quizId, /^call_0~\d+$/, "mint scheme: orig~n"); - await drain( - await handleChatRequest(post(chat.id, { type: "answer", toolCallId: quizId, answer: gradedAnswer(QUIZ2, ["Mercury"]) }), agentFactory(engine)), - ); + await drainTurn(post(chat.id, { type: "answer", toolCallId: quizId, answer: gradedAnswer(QUIZ2, ["Mercury"]) }), agentFactory(engine)); saved = chatStore.getChat(chat.id)!; // The invariant: CALL ids are unique among themselves (a call and its diff --git a/tests/demos.test.tsx b/tests/demos.test.tsx index e5ba295..8ce0b15 100644 --- a/tests/demos.test.tsx +++ b/tests/demos.test.tsx @@ -59,6 +59,7 @@ import { createTeacherTools } from "../src/agent/tools.ts"; import { chatStore } from "../src/store/chats.ts"; import { jobStore } from "../src/store/jobs.ts"; import { handleChatRequest } from "../src/server/chat-handler.ts"; +import { runningTurn } from "../src/server/turn-runs.ts"; import { createTeacherAgent } from "../src/agent/teacher-agent.ts"; import { projectView as projectNative } from "../src/core/view.ts"; import type { PiEngine } from "../src/agent/models.ts"; @@ -786,6 +787,15 @@ async function drain(res: Response): Promise { return res.text(); } +/** CONNECTION DIET B (owner, 2026-10-10): the POST acks and closes; drain + * the turn's wire through the reattach stream instead (settle-tolerant: + * a short turn may finish inside the handler). */ +async function drainRunning(chatId: string): Promise { + const handle = runningTurn(chatId); + if (!handle) return ""; // the turn already settled — completion happened + return new Response(handle.subscribe()).text(); +} + describe("demo/visual job runners", () => { it("runJob completes a demo: result on the job, steer via deliver, ack stays on the part", async () => { @@ -931,34 +941,34 @@ describe("demo/visual job runners", () => { const { steers, runner } = testRunner(); try { - await drain( - await handleChatRequest( - new Request("http://test/api/chat", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ + const ack = await handleChatRequest( + new Request("http://test/api/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + chatId: chat.id, + // Wave B: the input-events contract — one event, + // never a client transcript. + input: { type: "text", text: "teach me compounding" }, + }), + }), + // The handler hands the factory the native build options + // (messages are the merged record; the record IS the agent + // state — no adapter conversion). + (opts) => + createTeacherAgent(undefined, { + engine: teacher, + messages: opts.messages, + tools: createTeacherTools({ chatId: chat.id, - // Wave B: the input-events contract — one event, - // never a client transcript. - input: { type: "text", text: "teach me compounding" }, + chatStore, + jobStore, + spawn: (job, input) => void runner.runJob(job.id, input), }), }), - // The handler hands the factory the native build options - // (messages are the merged record; the record IS the agent - // state — no adapter conversion). - (opts) => - createTeacherAgent(undefined, { - engine: teacher, - messages: opts.messages, - tools: createTeacherTools({ - chatId: chat.id, - chatStore, - jobStore, - spawn: (job, input) => void runner.runJob(job.id, input), - }), - }), - ), ); + if (ack.status !== 200) assert.fail(`expected 200 ack, got ${ack.status}: ${await ack.text()}`); + await drainRunning(chat.id); const job = jobStore.findJobByToolCallId(chat.id, "call-d2"); assert.ok(job, "dispatch created a job at tool-execute time"); diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index ea87e44..1e522b4 100644 --- a/tests/jobs.test.ts +++ b/tests/jobs.test.ts @@ -47,6 +47,7 @@ import { composeInstructions, createTeacherAgent } from "../src/agent/teacher-ag import { createTeacherTools } from "../src/agent/tools.ts"; import { webTools } from "../src/agent/tools/web.ts"; import { handleChatRequest } from "../src/server/chat-handler.ts"; +import { runningTurn } from "../src/server/turn-runs.ts"; import { abortJob, deliverJobResult, @@ -201,6 +202,20 @@ async function drain(res: Response): Promise { return res.text(); } +/** CONNECTION DIET B (owner, 2026-10-10): the POST acks and closes; drain + * the turn's wire through the reattach stream instead. */ +async function drainTurn(req: Request, factory: Parameters[1]): Promise { + const cloned = req.clone(); + const ack = await handleChatRequest(req, factory); + if (ack.status !== 200) { + assert.fail(`expected 200 ack, got ${ack.status}: ${await ack.text()}`); + } + const { chatId } = (await cloned.json()) as { chatId: string }; + const handle = runningTurn(chatId); + if (!handle) return ""; // the turn settled inside the handler — completion already happened + return new Response(handle.subscribe()).text(); +} + function post(chatId: string, input: unknown): Request { return new Request("http://test/api/chat", { method: "POST", @@ -486,9 +501,7 @@ describe("dispatch (async protocol §2)", () => { createTeacherAgent(undefined, { engine, tools, messages: opts.messages }); researchModelOverride.current = researcherModel(); - const body = await drain( - await handleChatRequest(post(chat.id, { type: "text", text: "teach me graph rendering" }), agent), - ); + const body = await drainTurn(post(chat.id, { type: "text", text: "teach me graph rendering" }), agent); // The teacher's SAME turn carried on past the tool call. assert.ok(body.includes("Back with the brief."), "the turn continued after the dispatch"); diff --git a/tests/practice-flow.test.ts b/tests/practice-flow.test.ts index b376a05..c8393ac 100644 --- a/tests/practice-flow.test.ts +++ b/tests/practice-flow.test.ts @@ -28,6 +28,7 @@ import { import type { PiEngine } from "../src/agent/models.ts"; import { createPracticeAgent } from "../src/agent/teacher-agent.ts"; import { handleChatRequest } from "../src/server/chat-handler.ts"; +import { runningTurn } from "../src/server/turn-runs.ts"; import { chatStore } from "../src/store/chats.ts"; import { activeProblemOf, needsContinuation } from "../src/core/index.ts"; import { PATHS } from "../config.ts"; @@ -89,6 +90,22 @@ async function drain(res: Response): Promise { return res.text(); } +/** CONNECTION DIET B (owner, 2026-10-10): POST acks and closes; the turn's + * wire is the reattach stream. Mirrors drain(): 200-ack assert + the fully + * drained SSE body via runningTurn + subscribe (subscribe immediately — + * faux turns settle in microtask rounds). */ +async function drainTurn(req: Request, factory: Parameters[1]): Promise { + const cloned = req.clone(); + const ack = await handleChatRequest(req, factory); + if (ack.status !== 200) { + assert.fail(`expected 200 ack, got ${ack.status}: ${await ack.text()}`); + } + const { chatId } = (await cloned.json()) as { chatId: string }; + const handle = runningTurn(chatId); + if (!handle) return ""; // the turn settled inside the handler — completion already happened + return new Response(handle.subscribe()).text(); +} + /** The practice agent factory (the handler's kind branch, test-shaped). */ function practiceFactory(chatId: string, engine: PiEngine): Parameters[1] { return (opts) => createPracticeAgent({ chatId, chatStore }, { engine, messages: opts.messages }); @@ -128,7 +145,7 @@ describe("practice flow (scripted model)", () => { const factory = practiceFactory(chat.id, engine); // ── Turn 1: the opening text; the agent poses ── - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "practice group theory" }), factory)); + await drainTurn(post(chat.id, { type: "text", text: "practice group theory" }), factory); const afterPose = chatStore.getChat(chat.id)!.messages; const ack = afterPose[afterPose.length - 1] as LhMessage & ToolResultMessage; assert.equal(ack.role, "toolResult"); @@ -138,12 +155,7 @@ describe("practice flow (scripted model)", () => { assert.equal(activeProblemOf(afterPose)?.toolCallId, "call-p1"); // ── Turn 2: the submission arrives as a single-lane thinking message ── - await drain( - await handleChatRequest( - post(chat.id, { type: "thinking", problemCallId: "call-p1", text: "|G|=p; ⟨x⟩ has order dividing p" }), - factory, - ), - ); + await drainTurn(post(chat.id, { type: "thinking", problemCallId: "call-p1", text: "|G|=p; ⟨x⟩ has order dividing p" }), factory); // The model saw the one label (owner round 2, 2026-10-05): the // register is read from the text, not declared by a button. assert.ok(lastSaw(calls, "Your thinking:"), "the conclude turn's prompt carries the thinking"); @@ -167,13 +179,8 @@ describe("practice flow (scripted model)", () => { () => fauxAssistantMessage([fauxText("R0: reread the subgroup constraint.")]), ]); const factory = practiceFactory(chat.id, engine); - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "go" }), factory)); - await drain( - await handleChatRequest( - post(chat.id, { type: "thinking", problemCallId: "call-p2", text: "stuck at the second step" }), - factory, - ), - ); + await drainTurn(post(chat.id, { type: "text", text: "go" }), factory); + await drainTurn(post(chat.id, { type: "thinking", problemCallId: "call-p2", text: "stuck at the second step" }), factory); assert.ok(lastSaw(calls, "Your thinking:"), "the turn's prompt carries the thinking"); // The problem is still active — a trace never concludes. assert.equal(activeProblemOf(chatStore.getChat(chat.id)!.messages)?.toolCallId, "call-p2"); @@ -186,7 +193,7 @@ describe("practice flow (scripted model)", () => { () => fauxAssistantMessage([fauxText("Dropped. Something else, then.")]), ]); const factory = practiceFactory(chat.id, engine); - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "go" }), factory)); + await drainTurn(post(chat.id, { type: "text", text: "go" }), factory); // The silent dismissal route (the same POST the client's Esc takes). const dismissRoute = await import("../app/api/chats/[id]/dismiss/route.ts"); @@ -203,7 +210,7 @@ describe("practice flow (scripted model)", () => { assert.equal(activeProblemOf(marked), null, "the marker deactivates at Esc time"); // The next plain input synthesizes the conclude pair ahead of itself. - await drain(await handleChatRequest(post(chat.id, { type: "text", text: "something easier" }), factory)); + await drainTurn(post(chat.id, { type: "text", text: "something easier" }), factory); const msgs = chatStore.getChat(chat.id)!.messages; const synth = msgs.find( (m) => (m as ToolResultMessage).role === "toolResult" && (m as ToolResultMessage).toolName === "conclude", diff --git a/tests/session-core.test.ts b/tests/session-core.test.ts index e152fbb..2a27f2f 100644 --- a/tests/session-core.test.ts +++ b/tests/session-core.test.ts @@ -58,6 +58,36 @@ function sseResponse(...events: unknown[]): Response { return new Response(sse(...events), { status: 200 }); } +/** CONNECTION DIET B mock (owner, 2026-10-10 — parallel-session freeze + * postmortem): send() no longer consumes the POST response — it acks, + * closes, and reattaches. The wire sequence a send drives: POST /api/chat + * (tiny JSON ack) → GET /api/chats/chat_x (record refetch — the + * write-ahead absorbs the optimistic bubble) → GET .../stream (the SSE + * wire). `userText` seeds the record's write-ahead user message; `post` + * overrides the ack (409/error-path tests keep their own scripts); + * `onPost` observes the POST body. */ +function dietFetch(opts: { + stream: () => Response; + userText?: string; + post?: () => Response; + onPost?: (body: string) => void; +}): typeof globalThis.fetch { + return (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url === "/api/chat") { + if (opts.onPost && typeof init?.body === "string") opts.onPost(init.body); + return opts.post?.() ?? Response.json({ ok: true }); + } + if (url === "/api/chats/chat_x") { + const messages = + opts.userText === undefined ? [] : [{ id: "u1", role: "user", content: opts.userText, timestamp: 1 }]; + return Response.json({ messages }); + } + if (url === "/api/chats/chat_x/stream") return opts.stream(); + throw new Error(`unexpected fetch: ${url}`); + }) as typeof globalThis.fetch; +} + function assistantMsg(id: string, text: string): LhMessage { return { id, @@ -203,21 +233,19 @@ describe("postInputEvent (the 409 identity-retry)", { skip: process.env.NO_SLOW describe("send (one input event, verbatim consumption)", () => { it("optimistically appends the text, POSTs {chatId, input}, consumes to [DONE]", async () => { const posts: string[] = []; - globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { - const url = String(input); - if (url === "/api/chat" && init?.method === "POST") { - posts.push(init.body as string); - return sseResponse( + globalThis.fetch = dietFetch({ + onPost: (b) => posts.push(b), + userText: "my topic", // the write-ahead the record refetch absorbs + stream: () => + sseResponse( { type: "agent_start" }, { type: "message_start", message: { role: "assistant", content: [] } }, partial([{ type: "text", text: "hello there" }]), { type: "message_end", message: { role: "assistant", content: [{ type: "text", text: "hello there" }], stopReason: "stop" } }, { type: "agent_end", messages: [] }, "[DONE]", - ); - } - throw new Error(`unexpected fetch: ${url}`); - }) as typeof globalThis.fetch; + ), + }); const m = model(); const pending = m.send({ type: "text", text: "my topic" }); @@ -249,7 +277,7 @@ describe("send (one input event, verbatim consumption)", () => { push = (e) => controller.enqueue(enc.encode(e === "[DONE]" ? "data: [DONE]\n\n" : `data: ${JSON.stringify(e)}\n\n`)); }, }); - globalThis.fetch = (async () => new Response(body, { status: 200 })) as typeof globalThis.fetch; + globalThis.fetch = dietFetch({ userText: "go", stream: () => new Response(body, { status: 200 }) }); const m = model(); const done = m.send({ type: "text", text: "go" }); const settle = () => new Promise((r) => setTimeout(r, 0)); @@ -346,7 +374,7 @@ describe("send (one input event, verbatim consumption)", () => { push = (e) => controller.enqueue(enc.encode(e === "[DONE]" ? "data: [DONE]\n\n" : `data: ${JSON.stringify(e)}\n\n`)); }, }); - globalThis.fetch = (async () => new Response(body, { status: 200 })) as typeof globalThis.fetch; + globalThis.fetch = dietFetch({ userText: "q", stream: () => new Response(body, { status: 200 }) }); const m = model(); const done = m.send({ type: "text", text: "q" }); const settle = () => new Promise((r) => setTimeout(r, 0)); @@ -438,13 +466,16 @@ it("Wave C: markers injected mid-run (custom role, follow-up delivery) append to }); it("a provider error rides VERBATIM (owner ruling) — no masking anywhere", async () => { - globalThis.fetch = (async () => - sseResponse( - partial([{ type: "text", text: "" }]), - { type: "message_end", message: { role: "assistant", content: [{ type: "text", text: "" }], stopReason: "error", errorMessage: "OpenRouter 429: rate limited" } }, - { type: "agent_end", messages: [] }, - "[DONE]", - )) as typeof globalThis.fetch; + globalThis.fetch = dietFetch({ + userText: "q", + stream: () => + sseResponse( + partial([{ type: "text", text: "" }]), + { type: "message_end", message: { role: "assistant", content: [{ type: "text", text: "" }], stopReason: "error", errorMessage: "OpenRouter 429: rate limited" } }, + { type: "agent_end", messages: [] }, + "[DONE]", + ), + }); const m = model(); await m.send({ type: "text", text: "q" }); assert.equal(m.getState().status, "ready"); @@ -544,8 +575,7 @@ describe("resumeStream (the reattach path)", () => { c.close(); }, }); - globalThis.fetch = (async () => - new Response(body, { headers: { "Content-Type": "text/event-stream" } })) as typeof globalThis.fetch; + globalThis.fetch = dietFetch({ userText: "q", stream: () => new Response(body) }); const m = model(); const p = m.send({ type: "text", text: "q" }); assert.equal(m.getState().status, "submitted", "the action is optimistic: placeholder up pre-everything");