diff --git a/packages/paperclip-runner/devtools/issue-thread/issue-thread.spec.ts b/packages/paperclip-runner/devtools/issue-thread/issue-thread.spec.ts index eafa527ccb7..c170a8fdc64 100644 --- a/packages/paperclip-runner/devtools/issue-thread/issue-thread.spec.ts +++ b/packages/paperclip-runner/devtools/issue-thread/issue-thread.spec.ts @@ -525,6 +525,33 @@ function cleanRoomPayload(identifier: string, withTurn = false) { }; } +/** + * A turn now answers with the NDJSON stream, so a stubbed turn has to speak it. + * These stubs deliver one settled frame: they exercise what a turn *produced*, + * while incremental delivery is proved against the real server below. + */ +function turnStreamBody(payload: unknown, interim: unknown[] = []): string { + const frames = interim.map((view, index) => ({ + schema: "paperclip.phase7.turn-stream.v1", + type: "frame", + seq: index + 1, + reason: "delta", + turnId: "turn-1", + view, + })); + return [ + ...frames, + { + schema: "paperclip.phase7.turn-stream.v1", + type: "settled", + seq: frames.length + 1, + payload, + }, + ] + .map((frame) => `${JSON.stringify(frame)}\n`) + .join(""); +} + async function stubCleanRoom(page: Page, identifiers: string[]) { let opened = 0; await page.route(CLEAN_ROOM_API, async (route) => { @@ -588,8 +615,8 @@ test.describe("Phase 7M clean-room chat", () => { await page.route("**/api/phase7/ui/message", async (route) => { await route.fulfill({ status: 200, - contentType: "application/json", - body: JSON.stringify({ + contentType: "application/x-ndjson", + body: turnStreamBody({ sessionId: "session-MCK-1000", surface: "cleanroom", identity: cleanRoomPayload("MCK-1000").identity, @@ -815,8 +842,8 @@ test.describe("pending-request live region", () => { await page.route("**/api/phase7/ui/message", async (route) => { await route.fulfill({ status: 200, - contentType: "application/json", - body: JSON.stringify({ + contentType: "application/x-ndjson", + body: turnStreamBody({ ...cleanRoomPayload("MCK-1000"), view: interactionView("MCK-1000", "answered", { extraTurn: true }), }), @@ -866,3 +893,120 @@ test.describe("pending-request live region", () => { await expect(live).toHaveText("The pending request is resolved. Withdrawn."); }); }); + +/* -------------------------------------------- Phase 7Q streamed live turn */ + +/** + * Streaming is proved against the real package server, not a stub: port 4185 + * runs the same built bundle and the same session middleware with a scripted + * Codex provider (`vite.issue-thread-stream.config.ts`), so the NDJSON turn + * stream, the live session, the projection, and the browser client are all the + * shipped code. A `route.fulfill` stub cannot show this — it can only deliver a + * body in one piece, which is exactly the behaviour under repair. + */ + +const STREAM_ORIGIN = "http://127.0.0.1:4185"; +const STREAM_REPLY = + "Reading the clean-room issue. It is blank, with one mock agent and one mock task. " + + "Recording a first status against the mock control plane. Done — every record stayed in the mock port."; + +async function openStreamingCleanRoom(page: Page): Promise { + await page.addInitScript(() => window.localStorage.clear()); + await page.goto(`${STREAM_ORIGIN}/#/chat`); + await expect(page.locator('[data-thread-state="settled"]')).toBeVisible({ timeout: 60_000 }); +} + +test.describe("Phase 7Q streamed live turn", () => { + test("one assistant card grows while the turn POST is still open", async ({ page }) => { + await openStreamingCleanRoom(page); + + const states: Array<{ id: string; body: string }> = []; + let pendingWhileGrowing = 0; + let settledWhileGrowing = 0; + let turnFinished = false; + // Sample the rendered card, not the network: what has to be proved is that + // a reader sees the reply grow, and only the DOM can say that. + const sampler = (async () => { + while (!turnFinished) { + const sample = await page.evaluate(() => { + const card = document.querySelector('[data-thread-item="agent_message"]'); + const app = document.querySelector(".pit-app"); + return { + id: card?.id ?? null, + body: card?.querySelector(".pit-card-body")?.textContent ?? null, + threadState: app?.getAttribute("data-thread-state") ?? null, + composer: + document.querySelector("[data-composer-state]")?.getAttribute("data-composer-state") ?? + null, + }; + }); + if (sample.id !== null && sample.body !== null && sample.body.length > 0) { + if (states.at(-1)?.body !== sample.body) states.push({ id: sample.id, body: sample.body }); + if (sample.body !== STREAM_REPLY) { + pendingWhileGrowing += 1; + if (sample.threadState === "settled") settledWhileGrowing += 1; + } + } + await page.waitForTimeout(60); + } + })(); + + await page.locator("#composer-input").fill("Read this issue and record a status."); + await page.getByTestId("composer-send").click(); + + // The whole reply only appears at the end; the card gets there in pieces. + await expect(page.locator('[data-thread-item="agent_message"]')).toContainText(STREAM_REPLY, { + timeout: 60_000, + }); + turnFinished = true; + await sampler; + + const bodies = states.map((state) => state.body); + expect(bodies.length).toBeGreaterThanOrEqual(3); + // One card: every observed state belonged to the same transcript entry. + expect(new Set(states.map((state) => state.id)).size).toBe(1); + for (let index = 1; index < bodies.length; index += 1) { + expect(bodies[index]!.startsWith(bodies[index - 1]!)).toBe(true); + } + expect(bodies.at(-1)).toBe(STREAM_REPLY); + // Partial states were observed, and the surface never claimed to be + // settled while one of them was on screen. + expect(pendingWhileGrowing).toBeGreaterThanOrEqual(2); + expect(settledWhileGrowing).toBe(0); + + // Settled arrives only after the terminal frame. + await expect(page.locator('[data-thread-state="settled"]')).toBeVisible(); + await expect(page.locator('[data-composer-state="ready"]')).toHaveCount(1); + await expect(page.locator('[data-thread-item="agent_message"]')).toHaveCount(1); + await expect(page.locator('[data-thread-item="agent_message"]')).toHaveAttribute( + "data-streaming", + "false", + ); + }); + + test("Stop interrupts the streamed turn and keeps the partial reply", async ({ page }) => { + await openStreamingCleanRoom(page); + + await page.locator("#composer-input").fill("Start a long answer so I can stop it."); + await page.getByTestId("composer-send").click(); + + const card = page.locator('[data-thread-item="agent_message"]'); + await expect(card).toBeVisible({ timeout: 30_000 }); + await expect(card).toHaveAttribute("data-streaming", "true"); + const partial = ((await card.locator(".pit-card-body").textContent()) ?? "").trim(); + expect(partial.length).toBeGreaterThan(0); + expect(STREAM_REPLY.startsWith(partial)).toBe(true); + + await page.getByTestId("composer-stop").click(); + + // The stopped turn settles on what it had said, and nothing lands later. + await expect(page.locator('[data-thread-state="settled"]')).toBeVisible({ timeout: 30_000 }); + await expect(page.getByTestId("stopped-marker")).toBeVisible(); + const stopped = ((await card.locator(".pit-card-body").textContent()) ?? "").trim(); + expect(STREAM_REPLY.startsWith(stopped)).toBe(true); + expect(stopped).not.toBe(STREAM_REPLY); + await page.waitForTimeout(1_500); + expect(((await card.locator(".pit-card-body").textContent()) ?? "").trim()).toBe(stopped); + await expect(page.locator('[data-thread-item="agent_message"]')).toHaveCount(1); + }); +}); diff --git a/packages/paperclip-runner/devtools/issue-thread/playwright.config.ts b/packages/paperclip-runner/devtools/issue-thread/playwright.config.ts index 39b26f659da..a7b841d9b9c 100644 --- a/packages/paperclip-runner/devtools/issue-thread/playwright.config.ts +++ b/packages/paperclip-runner/devtools/issue-thread/playwright.config.ts @@ -24,11 +24,23 @@ export default defineConfig({ : { executablePath: process.env.PAPERCLIP_RUNNER_CHROMIUM_PATH }), }, }, - webServer: { - command: - "pnpm exec vite preview --config vite.issue-thread.config.ts --host 127.0.0.1 --port 4184", - url: "http://127.0.0.1:4184", - reuseExistingServer: false, - timeout: 60_000, - }, + webServer: [ + { + command: + "pnpm exec vite preview --config vite.issue-thread.config.ts --host 127.0.0.1 --port 4184", + url: "http://127.0.0.1:4184", + reuseExistingServer: false, + timeout: 60_000, + }, + { + // Same app and same package server, with a scripted Codex provider, so + // the streaming checks can drive a real NDJSON turn without a provider + // process. See `vite.issue-thread-stream.config.ts`. + command: + "pnpm exec vite preview --config vite.issue-thread-stream.config.ts --host 127.0.0.1 --port 4185", + url: "http://127.0.0.1:4185", + reuseExistingServer: false, + timeout: 60_000, + }, + ], }); diff --git a/packages/paperclip-runner/devtools/issue-thread/src/App.tsx b/packages/paperclip-runner/devtools/issue-thread/src/App.tsx index 41f0a770446..1abd3bb98a2 100644 --- a/packages/paperclip-runner/devtools/issue-thread/src/App.tsx +++ b/packages/paperclip-runner/devtools/issue-thread/src/App.tsx @@ -154,10 +154,28 @@ export function App() { const [showJump, setShowJump] = useState(false); const [reconnected, setReconnected] = useState(false); const [playing, setPlaying] = useState(false); + /** True while a turn stream is open, so the surface never reads as settled. */ + const [streamingTurn, setStreamingTurn] = useState(false); const scrollRef = useRef(null); const cancelResetRef = useRef(null); /** Last announced pending request, so the live region tracks transitions. */ const announcedPendingRef = useRef(null); + /** + * Turn generation. Reset, `New chat`, and leaving the surface all bump it, so + * a frame from a turn that has since been abandoned is dropped instead of + * appended to a thread it no longer belongs to. + */ + const turnGenerationRef = useRef(0); + const turnAbortRef = useRef(null); + + /** Abandons any open turn stream: the server sees the disconnect and stops. */ + const abandonTurn = useCallback(() => { + turnGenerationRef.current += 1; + const controller = turnAbortRef.current; + turnAbortRef.current = null; + setStreamingTurn(false); + controller?.abort(); + }, []); /* --------------------------------------------------------------- loading */ @@ -199,8 +217,11 @@ export function App() { } return () => { cancelled = true; + // Leaving the surface (or reloading it) must not leave a turn streaming + // into a thread that is no longer on screen. + abandonTurn(); }; - }, [chat, route.mode, route.shot, route.fixtureProfile, loadNonce]); + }, [abandonTurn, chat, route.mode, route.shot, route.fixtureProfile, loadNonce]); /* ------------------------------------------------------- deep-link params */ @@ -362,39 +383,66 @@ export function App() { [route.mode], ); + /** + * Sends one message and renders the turn as it arrives. + * + * Each frame is a whole server projection, so the surface still never patches + * state locally — it just gets more than one projection per turn. The settled + * payload is applied last and stays the authority. + */ const send = useCallback( (message: string) => { + if (snapshot === null) return; + if (route.mode !== "live") return; setActionError(null); - setSnapshot((current) => { - if (current === null) return current; - if (route.mode === "live") { - void phase7LiveClient - .send(current.sessionId, message) - .then((next) => setSnapshot(next.view)) - .catch((cause) => { - setActionError(describe(cause)); - // A refused turn must not strand the composer in `sending`. - setSnapshot((stale) => - stale === null - ? stale - : { ...stale, composer: { ...stale.composer, state: "ready" } }, - ); - }); - return { ...current, composer: { ...current.composer, state: "sending" } }; - } - return current; - }); + abandonTurn(); + const generation = turnGenerationRef.current; + const controller = new AbortController(); + turnAbortRef.current = controller; + setStreamingTurn(true); + setSnapshot((current) => + current === null ? current : { ...current, composer: { ...current.composer, state: "sending" } }, + ); + void phase7LiveClient + .send(snapshot.sessionId, message, { + signal: controller.signal, + onFrame: (view) => { + if (turnGenerationRef.current === generation) setSnapshot(view); + }, + }) + .then((next) => { + if (turnGenerationRef.current === generation) setSnapshot(next.view); + }) + .catch((cause) => { + if (turnGenerationRef.current !== generation || controller.signal.aborted) return; + setActionError(describe(cause)); + // A refused turn must not strand the composer in `sending`. + setSnapshot((stale) => + stale === null ? stale : { ...stale, composer: { ...stale.composer, state: "ready" } }, + ); + }) + .finally(() => { + if (turnGenerationRef.current !== generation) return; + turnAbortRef.current = null; + setStreamingTurn(false); + }); }, - [route.mode], + [abandonTurn, route.mode, snapshot], ); const stop = useCallback(() => { setSnapshot((current) => { if (current === null) return current; if (route.mode === "live") { + // Stop reaches the provider interrupt through its own request; the open + // turn stream keeps rendering and its terminal frame — not this + // response — decides what the stopped turn finally looks like. + const streaming = turnAbortRef.current !== null; void phase7LiveClient .stop(current.sessionId) - .then((next) => setSnapshot(next.view)) + .then((next) => { + if (!streaming && turnAbortRef.current === null) setSnapshot(next.view); + }) .catch((cause) => setActionError(describe(cause))); return current; } @@ -427,15 +475,19 @@ export function App() { const newChat = useCallback(() => { setConfirmReset(false); + // The room this turn belongs to is about to be retired, so the stream is + // dropped before the request that retires it. + abandonTurn(); setAnnouncement("Starting a new clean-room chat…"); void phase7LiveClient .newCleanRoom(snapshot?.sessionId ?? null) .then(adoptCleanRoom) .catch((cause) => setActionError(describe(cause))); - }, [adoptCleanRoom, snapshot]); + }, [abandonTurn, adoptCleanRoom, snapshot]); const reset = useCallback(() => { setConfirmReset(false); + abandonTurn(); if (chat && snapshot !== null) { void phase7LiveClient .reset(snapshot.sessionId) @@ -455,7 +507,7 @@ export function App() { } setSnapshot(phase7IssueThreadFixture("thread-baseline", route.fixtureProfile)); setAnnouncement("Scenario reset. The mock state is back to its clean seed."); - }, [adoptCleanRoom, chat, route.fixtureProfile, route.mode, snapshot]); + }, [abandonTurn, adoptCleanRoom, chat, route.fixtureProfile, route.mode, snapshot]); const retry = useCallback(() => { if (route.mode === "live" && snapshot !== null) { @@ -582,7 +634,13 @@ export function App() { return (
{ return (await response.json()) as Phase7LiveResponse; } +/** Interim projection delivered while the turn is still running. */ +export type Phase7TurnFrameHandler = (view: Phase7IssueThreadSnapshot) => void; + +export interface Phase7SendOptions { + onFrame?: Phase7TurnFrameHandler; + /** Abort the turn stream — used when the surface is torn down or rotated. */ + signal?: AbortSignal; +} + +async function postTurnStream( + path: string, + body: unknown, + options: Phase7SendOptions, +): Promise { + const response = await fetch(`${BASE}${path}`, { + method: "POST", + headers: { "content-type": "application/json", accept: PHASE7_TURN_STREAM_ACCEPT }, + body: JSON.stringify(body), + ...(options.signal === undefined ? {} : { signal: options.signal }), + }); + // Admission failures still answer with a JSON body and a real status code, + // because they are decided before the first frame is written. + if (!response.ok) throw await readError(response, path); + try { + return await readPhase7TurnStream( + response, + (frame) => options.onFrame?.(frame.view), + ); + } catch (cause) { + if (cause instanceof Phase7TurnStreamError) { + throw new Phase7LiveError(cause.code, cause.message); + } + throw cause; + } +} + export const phase7LiveClient = { async load(sessionId: string | null): Promise { const query = sessionId === null ? "" : `?sessionId=${encodeURIComponent(sessionId)}`; @@ -89,8 +136,16 @@ export const phase7LiveClient = { newCleanRoom(sessionId: string | null): Promise { return post("/cleanroom/session", sessionId === null ? {} : { sessionId }); }, - send(sessionId: string, message: string): Promise { - return post("/message", { sessionId, message }); + /** + * Runs one turn. `onFrame` fires for every interim projection the server + * writes while the POST is open; the resolved value is the settled payload. + */ + send( + sessionId: string, + message: string, + options: Phase7SendOptions = {}, + ): Promise { + return postTurnStream("/message", { sessionId, message }, options); }, stop(sessionId: string): Promise { return post("/interrupt", { sessionId }); diff --git a/packages/paperclip-runner/devtools/issue-thread/stream-fixture-transport.mjs b/packages/paperclip-runner/devtools/issue-thread/stream-fixture-transport.mjs new file mode 100644 index 00000000000..97c48bec702 --- /dev/null +++ b/packages/paperclip-runner/devtools/issue-thread/stream-fixture-transport.mjs @@ -0,0 +1,154 @@ +/** + * Scripted Codex transport for the browser streaming test (track 7Q). + * + * Everything below the provider is the real thing: the same package server + * middleware, the same `Phase7LiveSession`, the same NDJSON turn stream, and + * the same built browser bundle. Only the provider is scripted, and only so the + * test can decide when a delta arrives — a real Codex process cannot be asked + * to emit exactly four deltas 220 ms apart, and a browser assertion needs that + * to distinguish "streaming" from "arrived all at once". + * + * This module is loaded by `vite.issue-thread-stream.config.ts` only. It is not + * reachable from the shipped server, the deployed surface, or any route: the + * production plugin takes no transport override from a request or an + * environment variable. + */ + +const DELTA_INTERVAL_MS = 220; + +/** Four deltas: enough for a browser to observe growth twice over, and short. */ +export const PHASE7_STREAM_FIXTURE_DELTAS = [ + "Reading the clean-room issue. ", + "It is blank, with one mock agent and one mock task. ", + "Recording a first status against the mock control plane. ", + "Done — every record stayed in the mock port.", +]; + +export const PHASE7_STREAM_FIXTURE_REPLY = PHASE7_STREAM_FIXTURE_DELTAS.join(""); + +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +class Notifications { + #values = []; + #waiters = []; + #closed = false; + + push(value) { + const waiter = this.#waiters.shift(); + if (waiter) waiter({ value, done: false }); + else this.#values.push(value); + } + + close() { + this.#closed = true; + for (const waiter of this.#waiters.splice(0)) waiter({ value: undefined, done: true }); + } + + [Symbol.asyncIterator]() { + return { + next: async () => { + const value = this.#values.shift(); + if (value) return { value, done: false }; + if (this.#closed) return { value: undefined, done: true }; + return new Promise((resolve) => this.#waiters.push(resolve)); + }, + }; + } +} + +class StreamFixtureTransport { + #queue = new Notifications(); + #closed = false; + #turns = 0; + #interrupted = new Set(); + + async request(method, params) { + if (method === "initialize") return { user: { sessionId: "stream-fixture-session" } }; + if (method === "thread/start" || method === "thread/read" || method === "thread/resume") { + return { thread: { id: "stream-fixture-thread", sessionId: "stream-fixture-session" } }; + } + if (method === "turn/start") { + this.#turns += 1; + const turnId = `stream-turn-${this.#turns}`; + void this.#runTurn(turnId); + return { turn: { id: turnId, status: "inProgress" } }; + } + if (method === "turn/interrupt") { + // A stopped turn keeps whatever it had already streamed, so the surface + // can be checked for a coherent partial reply. + this.#interrupted.add(String(params.turnId)); + this.#queue.push({ + method: "turn/completed", + params: { + threadId: "stream-fixture-thread", + turn: { id: String(params.turnId), status: "interrupted" }, + }, + }); + return {}; + } + throw new Error(`unsupported scripted Codex method ${method}`); + } + + notify() {} + + notifications() { + return this.#queue; + } + + setServerRequestHandler() {} + + async close() { + this.#closed = true; + this.#queue.close(); + } + + processInfo() { + return { pid: 7100, processGroupId: 7100, exited: this.#closed, exitCode: null, signal: null }; + } + + async #runTurn(turnId) { + this.#queue.push({ + method: "turn/started", + params: { threadId: "stream-fixture-thread", turn: { id: turnId, status: "inProgress" } }, + }); + for (const delta of PHASE7_STREAM_FIXTURE_DELTAS) { + await sleep(DELTA_INTERVAL_MS); + if (this.#closed || this.#interrupted.has(turnId)) return; + this.#queue.push({ + method: "item/agentMessage/delta", + params: { threadId: "stream-fixture-thread", turnId, delta }, + }); + } + await sleep(DELTA_INTERVAL_MS); + if (this.#closed || this.#interrupted.has(turnId)) return; + this.#queue.push({ + method: "item/completed", + params: { + threadId: "stream-fixture-thread", + turnId, + item: { id: `message-${turnId}`, type: "agentMessage", text: PHASE7_STREAM_FIXTURE_REPLY }, + }, + }); + this.#queue.push({ + method: "turn/completed", + params: { threadId: "stream-fixture-thread", turn: { id: turnId, status: "completed" } }, + }); + } +} + +export function phase7StreamFixtureTransportFactory(options = {}) { + const evidence = { + runnerPid: 7100, + runnerProcessGroupId: 7100, + codexPid: 7200, + runnerExited: false, + runnerExitCode: null, + runnerSignal: null, + childEnvironmentKeys: ["CODEX_HOME", "HOME", "PATH"], + diagnostics: [], + }; + options.onEvidence?.(evidence); + return { transport: new StreamFixtureTransport(), evidence: () => ({ ...evidence }) }; +} diff --git a/packages/paperclip-runner/knowledge/evidence/2026-08-10-phase-07q-turn-streaming-verification.md b/packages/paperclip-runner/knowledge/evidence/2026-08-10-phase-07q-turn-streaming-verification.md new file mode 100644 index 00000000000..e7f81ec3838 --- /dev/null +++ b/packages/paperclip-runner/knowledge/evidence/2026-08-10-phase-07q-turn-streaming-verification.md @@ -0,0 +1,125 @@ +--- +type: Verification Evidence +title: Phase 7Q live turn streaming verification +description: Command-and-result record for streaming real Codex output end to end through the package session server and the browser — live turn events, the NDJSON turn stream, incremental rendering, stop behaviour, and the regression set. +tags: [native-runner, phase-7, clean-room, live-codex, streaming, verification, evidence] +status: stable +generated: { by: anthropic/claude-opus-5, at: 2026-08-10T18:40:00Z } +entry_kind: evidence +phase: "7" +--- + +# Scope + +Track 7Q (PAP-16981) fixes the defect the board reported against the clean-room +chat: "it needs to stream — it seems like the response just simply shows up". +It did. A turn was awaited whole in `Phase7LiveSession`, answered as one JSON +body by the package session server, and consumed with `response.json()` in the +browser, so the earliest a reply could appear was after it was finished. + +The rejected surface is `#/chat` in `devtools/issue-thread`, served by +`scripts/phase7-issue-thread-server.mjs` over `Phase7LiveSessionService`. The +scenario explorer (`examples/phase7-explorer` with `phase7i-demo-server.ts`) is +a different, deterministic path and is unchanged: its preset routes and its +staged `?stage=streaming` screenshot state are untouched, and neither is treated +here as evidence of live streaming. + +# What changed + +* `Phase7LiveSession` publishes bounded, ordered live turn events — assistant + deltas, tool call and result activity, stop requests, terminal, and error — + and mirrors the turn waiter's assembled text into the transcript entry the + reply will finally occupy. The waiter's buffer stays the only assembly point, + so the streamed draft cannot diverge from the settled reply. +* `src/phase7/turn-stream.ts` defines the NDJSON frame codec: `frame` for an + interim projection, `settled` for the authoritative payload, `error` for a + named failure. The decoder rejects a replayed, reordered, oversized, late, or + truncated frame. +* `POST /api/phase7/ui/message` flushes headers, then writes one frame per + event-loop tick of provider activity. Admission checks still answer with a + JSON body and their own status code, because they are decided before the + first frame. +* The browser client reads the frames and hands each projection to the surface, + which renders it in place. One assistant card grows; the settled payload is + applied last. + +# Commands and results + +All commands were run in `packages/paperclip-runner` on branch +`PAP-16679-paperclip-runner`. + +| Command | Result | +| --- | --- | +| `vitest run src/phase7/turn-stream.test.ts` | 6 passed — codec round trip under one-character chunks, replay/late/truncation refusals, three gated deltas, interrupted partial retention, settled-payload read, error frame | +| `vitest run src/phase7/clean-room-server.test.ts` | 11 passed (9 pre-existing + 2 new) — including frames read before the gated turn is released, and provider interrupt on client abort | +| `vitest run` (package) | 40 files, 363 passed | +| `pnpm test:phase7` | 15 files, 167 passed | +| `playwright test --config devtools/issue-thread/playwright.config.ts` | 62 passed (60 pre-existing + 2 new streaming tests) | +| `playwright test --config examples/playwright.phase7.config.ts` | 43 passed — scenario explorer unchanged | +| `node scripts/check-forbidden-imports.mjs` | passed | +| `node scripts/check-phase7-inventory.mjs` | passed | +| `node scripts/check-browser-tokens.mjs` | passed | +| `node scripts/validate-doc-links.mjs` / `validate-okf.mjs` | passed | +| `pnpm build` | passed (TypeScript, Rust, browser, phase 5/6/7, issue thread; strict Phase 7I CSP check green) | +| `node scripts/phase7-clean-room-smoke.mjs --json` (real Codex) | 16/16 assertions true, including the new `streamedIncrementally` | + +## Live Codex streaming measurement + +A single clean-room turn against real Codex through real `paperclip-runnerd`, +asking for a four-sentence answer and no tool call: + +``` +headers at +265ms: application/x-ndjson; charset=utf-8 no-store, no-transform +frames: 64 (62 carrying assistant text) + +2184ms 1 char "I" + +2188ms 3 chars "I’m" + +2193ms 11 chars "I’m working" + +2194ms 16 chars "I’m working on a" + ... + +3066ms 301 chars (terminal) +monotonic: true +last interim state === settled reply: true +assistant cards in the settled thread: 1 +``` + +Response headers arrive at 265 ms — roughly two seconds before the provider's +first token — and the reply is delivered in 62 growing states rather than one. +The last interim state is byte-identical to the settled reply, and the settled +thread holds exactly one assistant card, so the streamed draft was finalised +rather than duplicated. + +# Behaviour under stop and abandonment + +* Stop during a stream reaches the Codex interrupt through the existing + `/interrupt` route. The stopped turn keeps what it had streamed, the turn is + marked stopped by the user, and nothing is appended afterwards + (`src/phase7/turn-stream.test.ts`, plus the browser test that stops a live + stream and re-reads the card after a delay). +* Abandoning the stream — reload, navigation, `New chat`, `Reset`, or an + aborted fetch — disconnects the socket, and the server interrupts the turn + rather than leaving it running. The session returns to `ready`, which an + active turn would not (`clean-room-server.test.ts`). +* The client drops frames from a turn generation that has been abandoned, so a + late chunk cannot land in a thread that has since been rotated. + +# Preserved properties + +* Mock-only control plane, the live-only projection guard on every frame, the + real-API block record, redaction, per-session turn and message limits, + bounded sessions, and the credential-absence probes are unchanged and still + asserted by the clean-room suite on the streamed path. +* The frame count per turn is bounded; past the bound the turn still runs and + still settles, only interim views stop. +* `Cache-Control: no-store, no-transform` and `X-Accel-Buffering: no` are set on + the turn response, and no `Content-Length` is sent, so an intermediary cannot + compress or buffer the frames back into a single body. +* Replay, preset, and fixture routes are unchanged and remain deterministic. + +# Known limitation + +The byte-stable 24-PNG issue-thread screenshot matrix +(`node scripts/record-phase7-ui-evidence.mjs --check`) reports a wholesale +mismatch on this host. It reproduces identically with the UI sources reverted to +`HEAD`, so it is an environment fingerprint difference rather than a regression +from this work; the recorded set is pinned to the Chromium build that produced +the committed PNGs. diff --git a/packages/paperclip-runner/knowledge/evidence/index.md b/packages/paperclip-runner/knowledge/evidence/index.md index 03fa0813ac5..acd8e21ebd0 100644 --- a/packages/paperclip-runner/knowledge/evidence/index.md +++ b/packages/paperclip-runner/knowledge/evidence/index.md @@ -20,3 +20,4 @@ * [Phase 7G issue-thread UI verification](2026-08-10-phase-07g-issue-thread-ui-verification.md) - View-model tests, browser coverage, the axe gate on 12 slugs × 2 viewports, the byte-stable 24-PNG matrix, and the live runnerd/Codex smoke. * [Phase 7 final evidence manifest (Revision 3/4)](2026-08-10-phase-07i-final-evidence.md) - Authoritative Phase 7I manifest for the final candidate build `865d1a72`: offline and live command results, conformance and approved security results, screenshot eligibility, and the Revision 2 ineligibility record. * [Phase 7M clean-room live chat verification](2026-08-10-phase-07m-clean-room-verification.md) - Blank-seed enforcement, deterministic exposure profile, allowed/denied semantic calls, durable multi-turn session, identity rotation on reset and new chat, the real-API block record, browser and axe coverage, and the live Codex smoke. +* [Phase 7Q live turn streaming verification](2026-08-10-phase-07q-turn-streaming-verification.md) - Live turn events, the NDJSON turn stream, incremental browser rendering on one growing assistant card, stop and abandonment behaviour, the preserved security properties, and a measured real Codex turn. diff --git a/packages/paperclip-runner/scripts/phase7-clean-room-smoke.mjs b/packages/paperclip-runner/scripts/phase7-clean-room-smoke.mjs index 7b035fd64f9..43a5204a872 100644 --- a/packages/paperclip-runner/scripts/phase7-clean-room-smoke.mjs +++ b/packages/paperclip-runner/scripts/phase7-clean-room-smoke.mjs @@ -19,6 +19,10 @@ import { once } from "node:events"; import { createPhase7IssueThreadMiddleware } from "./phase7-issue-thread-server.mjs"; +const { readPhase7TurnStream, PHASE7_TURN_STREAM_ACCEPT } = await import( + new URL("../dist/index.js", import.meta.url).href +); + const CREDENTIAL_PATTERNS = [ /bearer\s+[a-z0-9._-]+/i, // Anchored so a mock id such as `task-cleanroom-3f2a9c11` cannot masquerade @@ -42,6 +46,22 @@ function items(view) { return view.turns.flatMap((turn) => turn.items); } +function assistantText(view) { + return items(view) + .filter((item) => item.kind === "agent_message") + .map((item) => item.body) + .join(""); +} + +/** + * Interim assistant states are monotonic when each one extends the last. That + * is the property the streamed turn has to hold against a real provider, not + * just against a scripted one. + */ +function monotonic(states) { + return states.every((state, index) => index === 0 || state.startsWith(states[index - 1])); +} + async function main() { const asJson = process.argv.includes("--json"); const middleware = createPhase7IssueThreadMiddleware({ bindHost: "127.0.0.1" }); @@ -65,6 +85,24 @@ async function main() { body: JSON.stringify(body), }) ).json(); + /** + * Runs one turn and records what the browser would have rendered along the + * way, so the smoke proves incremental delivery from a real Codex turn rather + * than only the settled reply. + */ + const turn = async (body) => { + const response = await fetch(`${origin}/api/phase7/ui/message`, { + method: "POST", + headers: { "content-type": "application/json", accept: PHASE7_TURN_STREAM_ACCEPT }, + body: JSON.stringify(body), + }); + if (!response.ok) throw new Error(`turn failed with HTTP ${response.status}`); + const states = []; + const settled = await readPhase7TurnStream(response, (frame) => { + states.push(assistantText(frame.view)); + }); + return { ...settled, states }; + }; const assertions = {}; let summary = {}; @@ -86,18 +124,21 @@ async function main() { opened.view.issue.identifier === opened.identity.identifier && opened.view.issue.identifier.startsWith("MCK-"); - const first = await post("/api/phase7/ui/message", { + const first = await turn({ sessionId: opened.sessionId, message: "Read this issue with get_task_context, then call report_progress with a one-sentence status. Do not ask me anything.", }); const firstItems = items(first.view); + const growing = first.states.filter((state) => state.length > 0); + assertions.streamedIncrementally = + growing.length >= 2 && monotonic(growing) && growing.at(-1) === assistantText(first.view); assertions.realCodexTurn = firstItems.some((item) => item.kind === "agent_message"); assertions.semanticToolCalled = firstItems.some((item) => item.kind === "tool_activity"); assertions.mockStateMutated = firstItems.some((item) => item.kind === "durable_comment"); assertions.authorizationRecorded = first.view.evidence.authorization.length > 0; - const second = await post("/api/phase7/ui/message", { + const second = await turn({ sessionId: opened.sessionId, message: "In one line, what status did you just record?", }); @@ -131,6 +172,8 @@ async function main() { newChatSessionId: fresh.sessionId, newChatIssue: fresh.view.issue.identifier, turns: second.view.turns.length, + streamedStates: first.states.length, + streamedPrefixes: first.states.filter((state) => state.length > 0).slice(0, 3), toolCalls: second.view.evidence.calls.map((call) => ({ operationId: call.operationId, outcome: call.outcome, diff --git a/packages/paperclip-runner/scripts/phase7-issue-thread-server.mjs b/packages/paperclip-runner/scripts/phase7-issue-thread-server.mjs index 4bb454a5cb6..9f8a1b6babe 100644 --- a/packages/paperclip-runner/scripts/phase7-issue-thread-server.mjs +++ b/packages/paperclip-runner/scripts/phase7-issue-thread-server.mjs @@ -32,6 +32,12 @@ const MAX_CLEAN_ROOM_SESSIONS = 4; /** Bounded output: a clean room is a demo, not a long-lived agent. */ const MAX_TURNS_PER_SESSION = 24; const MAX_MESSAGE_BYTES = 8 * 1024; +/** + * Bounded frames per streamed turn. Past this the turn keeps running and still + * settles with the authoritative payload; only the interim views stop, so a + * chatty provider cannot turn one turn into unbounded socket writes. + */ +const MAX_TURN_STREAM_FRAMES = 600; async function loadRunner() { return import(new URL("../dist/index.js", import.meta.url).href); @@ -256,6 +262,125 @@ export function createPhase7IssueThreadMiddleware(options = {}) { : { sessionId: entry.session.id, surface: "issue", view: view(runner, entry) }; } + /** The clean room keeps its live-only guard on every frame, not just the last. */ + function frameView(runner, entry) { + return entry.surface === "cleanroom" ? liveView(runner, entry) : view(runner, entry); + } + + /** + * Streams one turn as NDJSON frames (track 7Q). + * + * The turn used to be awaited whole and answered once, so the browser could + * only ever reveal a finished reply. Now every provider delta, tool call, and + * tool result the live session announces is projected and written while the + * POST is still open. The final `settled` frame carries exactly the payload + * the single JSON response used to carry, so the terminal projection — not + * any interim frame — remains the authority. + * + * Interim frames are coalesced onto the next event-loop turn: a burst of + * deltas that arrives in one tick becomes one frame, while deltas separated + * by real provider I/O each get their own. That bounds the write rate without + * a timer, and without inventing a cadence the provider did not have. + */ + async function streamTurn(runner, entry, message, request, response) { + response.statusCode = 200; + for (const [name, value] of Object.entries(runner.PHASE7_TURN_STREAM_HEADERS)) { + response.setHeader(name, value); + } + // Headers before the first frame: a client that waits for them must not be + // held until the provider speaks. + response.flushHeaders(); + + let seq = 0; + let frames = 0; + let finished = false; + let scheduled = null; + let pendingReason = null; + let pendingTurnId = null; + + const write = (frame) => { + if (response.writableEnded || response.destroyed) return; + seq += 1; + response.write(runner.encodePhase7TurnStreamFrame({ ...frame, seq })); + }; + + const flush = () => { + scheduled = null; + if (finished || pendingReason === null) return; + if (frames >= MAX_TURN_STREAM_FRAMES) return; + const reason = pendingReason; + const turnId = pendingTurnId; + pendingReason = null; + pendingTurnId = null; + let projected; + try { + projected = frameView(runner, entry); + } catch { + // A projection failure is reported by the terminal frame; dropping an + // interim view must never abort a turn that is still running. + return; + } + frames += 1; + write({ + schema: runner.PHASE7_TURN_STREAM_SCHEMA, + type: "frame", + reason, + turnId, + view: projected, + }); + }; + + const unsubscribe = entry.session.subscribe((event) => { + if (finished || event.kind === "terminal" || event.kind === "error") return; + pendingReason = event.kind === "delta" ? "delta" : event.reason === "stop_requested" ? "stop_requested" : "activity"; + pendingTurnId = event.turnId; + if (scheduled === null) scheduled = setImmediate(flush); + }); + + // A browser that navigates away, reloads, or aborts the fetch mid-turn is a + // stop: the provider turn is interrupted rather than left running against a + // socket nobody is reading. + const onDisconnect = () => { + if (finished) return; + void entry.session.interrupt("client disconnected").catch(() => undefined); + }; + request.on("aborted", onDisconnect); + response.on("close", onDisconnect); + + // `sendMessage` records the user message and marks the session running + // before its first await, so starting it and *then* flushing makes the + // opening frame already show what was sent and a live composer. + const turn = entry.session.sendMessage(message); + pendingReason = "open"; + pendingTurnId = null; + flush(); + + try { + await turn; + finished = true; + write({ + schema: runner.PHASE7_TURN_STREAM_SCHEMA, + type: "settled", + payload: payload(runner, entry), + }); + } catch (error) { + finished = true; + write({ + schema: runner.PHASE7_TURN_STREAM_SCHEMA, + type: "error", + error: error instanceof RouteError ? error.code : "turn_failed", + message: String(error instanceof Error ? error.message : error), + }); + } finally { + finished = true; + if (scheduled !== null) clearImmediate(scheduled); + unsubscribe(); + request.off("aborted", onDisconnect); + response.off("close", onDisconnect); + response.end(); + } + } + const middleware = async function phase7IssueThreadMiddleware(request, response, next) { const url = new URL(request.url ?? "/", "http://phase7.local"); if (!url.pathname.startsWith(`${ROUTE_PREFIX}/`)) { @@ -327,7 +452,10 @@ export function createPhase7IssueThreadMiddleware(options = {}) { ); } entry.turns += 1; - await entry.session.sendMessage(message); + // Every admission check has already answered with its own status code; + // from here the response is a stream, so a failure is a framed error. + await streamTurn(runner, entry, message, request, response); + return; } else if (route === "interrupt") { await entry.session.interrupt("operator stopped the turn"); } else if (route === "reconnect") { @@ -370,6 +498,12 @@ export function createPhase7IssueThreadMiddleware(options = {}) { send(response, 200, payload(runner, entry)); } catch (error) { + if (response.headersSent) { + // A streamed turn reports its own failures as a framed error; there is + // no status code left to send once the first frame is on the wire. + if (!response.writableEnded) response.end(); + return; + } if (error instanceof RouteError) { send(response, error.status, { error: error.code, message: error.message }); return; diff --git a/packages/paperclip-runner/scripts/phase7-issue-thread-smoke.mjs b/packages/paperclip-runner/scripts/phase7-issue-thread-smoke.mjs index 90d653360bc..0e88c5b6cf8 100644 --- a/packages/paperclip-runner/scripts/phase7-issue-thread-smoke.mjs +++ b/packages/paperclip-runner/scripts/phase7-issue-thread-smoke.mjs @@ -16,6 +16,25 @@ import { once } from "node:events"; import { createPhase7IssueThreadMiddleware } from "./phase7-issue-thread-server.mjs"; +const { readPhase7TurnStream, PHASE7_TURN_STREAM_ACCEPT } = await import( + new URL("../dist/index.js", import.meta.url).href +); + +/** Runs one turn over the NDJSON stream and returns its settled payload. */ +async function runTurn(origin, body) { + const response = await fetch(`${origin}/api/phase7/ui/message`, { + method: "POST", + headers: { "content-type": "application/json", accept: PHASE7_TURN_STREAM_ACCEPT }, + body: JSON.stringify(body), + }); + if (!response.ok) throw new Error(`turn failed with HTTP ${response.status}`); + let frames = 0; + const settled = await readPhase7TurnStream(response, () => { + frames += 1; + }); + return { ...settled, frames }; +} + const CREDENTIAL_PATTERNS = [ /bearer\s+[a-z0-9._-]+/i, // Anchored so a mock id such as `task-cleanroom-3f2a9c11` cannot masquerade @@ -62,17 +81,12 @@ async function main() { assertions.mockIdentifier = created.view.issue.identifier.startsWith("MCK-"); assertions.toolsProjected = created.view.evidence.tools.length >= 0; - const first = await ( - await fetch(`${origin}/api/phase7/ui/message`, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ - sessionId, - message: - "Call get_task_context, then call report_progress with a one-sentence status. Do not ask me anything.", - }), - }) - ).json(); + const first = await runTurn(origin, { + sessionId, + message: + "Call get_task_context, then call report_progress with a one-sentence status. Do not ask me anything.", + }); + assertions.turnStreamed = first.frames >= 2; const firstItems = first.view.turns.flatMap((turn) => turn.items); assertions.userMessageRendered = firstItems.some((item) => item.kind === "user_message"); assertions.toolActivityRendered = firstItems.some((item) => item.kind === "tool_activity"); @@ -85,13 +99,10 @@ async function main() { (row) => row.disposition === "control_plane_owned", ); - const second = await ( - await fetch(`${origin}/api/phase7/ui/message`, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ sessionId, message: "Summarise the mock task in one line." }), - }) - ).json(); + const second = await runTurn(origin, { + sessionId, + message: "Summarise the mock task in one line.", + }); assertions.multiTurnThread = second.view.turns.length > first.view.turns.length; assertions.composerReady = second.view.composer.state === "ready"; diff --git a/packages/paperclip-runner/src/index.ts b/packages/paperclip-runner/src/index.ts index 47251c18386..e459a25be8c 100644 --- a/packages/paperclip-runner/src/index.ts +++ b/packages/paperclip-runner/src/index.ts @@ -34,5 +34,6 @@ export * as acceptedPhase7SemanticTools from "./semantic-tools/index.js"; export * from "./phase7/clean-room.js"; export * from "./phase7/live-session.js"; export * from "./phase7/runnerd-codex-transport.js"; +export * from "./phase7/turn-stream.js"; export { projectPhase7IssueThread } from "./issue-thread/live-projection.js"; export * as phase7IssueThread from "./issue-thread/index.js"; diff --git a/packages/paperclip-runner/src/issue-thread/live-projection.ts b/packages/paperclip-runner/src/issue-thread/live-projection.ts index 1851bc6c3e2..4aa4e42b099 100644 --- a/packages/paperclip-runner/src/issue-thread/live-projection.ts +++ b/packages/paperclip-runner/src/issue-thread/live-projection.ts @@ -495,7 +495,10 @@ function composerModel( pendingInteractionId, }; } - if (snapshot.activeTurnId !== null) { + // `running` covers the window between accepting a message and the provider + // reporting a turn id. A streamed turn is observable in that window, so the + // composer has to read as streaming there rather than briefly as ready. + if (snapshot.activeTurnId !== null || snapshot.status === "running") { return { state: "streaming", helper: "Codex is working — send to steer, or stop the turn.", diff --git a/packages/paperclip-runner/src/phase7/clean-room-server.test.ts b/packages/paperclip-runner/src/phase7/clean-room-server.test.ts index 35e774f73ca..8d8673c23f5 100644 --- a/packages/paperclip-runner/src/phase7/clean-room-server.test.ts +++ b/packages/paperclip-runner/src/phase7/clean-room-server.test.ts @@ -12,6 +12,11 @@ import type { } from "../drivers/codex/app-server-transport.js"; import type { Phase7IssueThreadSnapshot, Phase7ThreadItem } from "../issue-thread/types.js"; import type { Phase7LiveTransportFactory } from "./live-session.js"; +import { + createPhase7TurnStreamDecoder, + readPhase7TurnStream, + type Phase7TurnStreamFrame, +} from "./turn-stream.js"; /** * Clean-room chat, driven through the same HTTP routes the browser uses. @@ -29,8 +34,33 @@ interface FakeProviderState { providerSessionId: string; nextTurn: number; transports: FakeCleanRoomTransport[]; + /** Releases one step of a gated streaming turn (`stream` messages). */ + gate: Gate; } +/** Single-slot signal, so a streamed turn advances only when the test says so. */ +class Gate { + #waiters: Array<() => void> = []; + #ready = 0; + + wait(): Promise { + if (this.#ready > 0) { + this.#ready -= 1; + return Promise.resolve(); + } + return new Promise((resolve) => this.#waiters.push(resolve)); + } + + release(): void { + const waiter = this.#waiters.shift(); + if (waiter === undefined) this.#ready += 1; + else waiter(); + } +} + +/** Assistant deltas the gated turn emits, one release at a time. */ +const STREAM_DELTAS = ["Reading ", "the clean-room ", "issue."] as const; + class AsyncNotifications implements AsyncIterable { #values: CodexRpcNotification[] = []; #waiters: Array<(value: IteratorResult) => void> = []; @@ -131,6 +161,30 @@ class FakeCleanRoomTransport implements CodexAppServerTransport { await new Promise((resolve) => setTimeout(resolve, 0)); let assistantText = `Nothing to do for "${message}".`; if (message.includes("hang")) return; + if (message.includes("stream")) { + for (const delta of STREAM_DELTAS) { + await this.state.gate.wait(); + this.queue.push({ + method: "item/agentMessage/delta", + params: { threadId: this.state.threadId, turnId, delta }, + }); + } + await this.state.gate.wait(); + assistantText = STREAM_DELTAS.join(""); + this.queue.push({ + method: "item/completed", + params: { + threadId: this.state.threadId, + turnId, + item: { id: `message-${turnId}`, type: "agentMessage", text: assistantText }, + }, + }); + this.queue.push({ + method: "turn/completed", + params: { threadId: this.state.threadId, turn: { id: turnId, status: "completed" } }, + }); + return; + } // The resumed turn carries the typed interaction result as its input, so it // is matched before the keyword branches it would otherwise trip. if (message.includes("semantic-interaction-result")) { @@ -222,6 +276,66 @@ function items(view: Phase7IssueThreadSnapshot): Phase7ThreadItem[] { return view.turns.flatMap((turn) => turn.items); } +/** Assistant text a reader would see in the projected thread. */ +function assistantText(view: Phase7IssueThreadSnapshot): string { + return items(view) + .map((item) => (item.kind === "agent_message" ? item.body : "")) + .join(""); +} + +/** Polls until `ready` holds, so nothing in the suite waits on a fixed delay. */ +async function settle(ready: () => boolean | Promise, attempts = 200): Promise { + for (let attempt = 0; attempt < attempts; attempt += 1) { + if (await ready()) return; + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("the clean-room stream never reached the expected state"); +} + +type CleanRoomFrame = Phase7TurnStreamFrame; + +/** + * Pulls frames off an open turn response one at a time. Reading incrementally + * — rather than awaiting the whole body — is the point: it is what proves the + * server answers before the turn is released. + */ +class FrameReader { + readonly #reader: ReadableStreamDefaultReader; + readonly #decoder = createPhase7TurnStreamDecoder(); + readonly #text = new TextDecoder(); + #queue: CleanRoomFrame[] = []; + + constructor(response: Response) { + if (response.body === null) throw new Error("the turn response had no body"); + this.#reader = response.body.getReader(); + } + + async next(): Promise { + while (this.#queue.length === 0) { + const chunk = await this.#reader.read(); + if (chunk.done) { + this.#queue.push(...this.#decoder.flush()); + break; + } + this.#queue.push(...this.#decoder.push(this.#text.decode(chunk.value, { stream: true }))); + } + const frame = this.#queue.shift(); + if (frame === undefined) throw new Error("the turn stream ended without another frame"); + return frame; + } + + async nextMatching(predicate: (frame: CleanRoomFrame) => boolean): Promise { + for (;;) { + const frame = await this.next(); + if (predicate(frame)) return frame; + } + } + + async end(): Promise { + await this.#reader.cancel().catch(() => undefined); + } +} + let server: Server; let origin: string; let middleware: Awaited>; @@ -237,7 +351,28 @@ async function createMiddleware(state: FakeProviderState) { }); } -async function call(path: string, body?: unknown): Promise { +/** + * Reads a turn response. `/message` answers with the NDJSON turn stream, so the + * helper collects every interim view and returns the settled payload — the same + * value the route used to answer with in one shot. + */ +async function readTurn( + response: Response, + frames: Phase7IssueThreadSnapshot[] = [], +): Promise { + return await readPhase7TurnStream( + response, + (frame) => { + frames.push(frame.view); + }, + ); +} + +async function call( + path: string, + body?: unknown, + frames?: Phase7IssueThreadSnapshot[], +): Promise { const url = `${origin}${path}`; requestedUrls.push(url); const response = await fetch( @@ -250,6 +385,9 @@ async function call(path: string, body?: unknown): Promise { body: JSON.stringify(body), }, ); + if (response.headers.get("content-type")?.includes("x-ndjson") === true) { + return await readTurn(response, frames); + } const payload = (await response.json()) as CleanRoomPayload & { error?: string }; if (!response.ok) throw new Error(`${path} → ${response.status} ${payload.error ?? ""}`); return payload; @@ -277,6 +415,7 @@ beforeEach(async () => { providerSessionId: "codex-provider-clean-room", nextTurn: 0, transports: [], + gate: new Gate(), }; middleware = await createMiddleware(providerState); server = createServer((request, response) => { @@ -504,6 +643,99 @@ describe("Phase 7 clean-room chat server", () => { expect(body.message).toContain("Start a new chat"); }); + it("streams the turn as frames while the POST is still open", async () => { + const opened = await call("/api/phase7/ui/cleanroom/session"); + const response = await fetch(`${origin}/api/phase7/ui/message`, { + method: "POST", + headers: { "content-type": "application/json", accept: "application/x-ndjson" }, + body: JSON.stringify({ sessionId: opened.sessionId, message: "stream your answer" }), + }); + + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toContain("application/x-ndjson"); + // A cache or proxy directive that permitted transformation could coalesce + // the frames back into one body, which is the failure being fixed. + expect(response.headers.get("cache-control")).toContain("no-transform"); + expect(response.headers.get("content-length")).toBeNull(); + + const reader = new FrameReader(response); + // The body is readable before the gated turn produces anything at all. + const open = await reader.next(); + expect(open.type).toBe("frame"); + if (open.type !== "frame") return; + expect(open.reason).toBe("open"); + expect(assistantText(open.view)).toBe(""); + expect(open.view.turns.at(-1)?.items.some((item) => item.kind === "user_message")).toBe(true); + + const streamed: string[] = []; + for (let index = 0; index < STREAM_DELTAS.length; index += 1) { + providerState.gate.release(); + const frame = await reader.nextMatching( + (candidate) => + candidate.type !== "frame" || assistantText(candidate.view) === STREAM_DELTAS.slice(0, index + 1).join(""), + ); + expect(frame.type).toBe("frame"); + if (frame.type !== "frame") return; + streamed.push(assistantText(frame.view)); + expect(frame.view.composer.state).toBe("streaming"); + } + + expect(streamed).toEqual([ + STREAM_DELTAS[0], + STREAM_DELTAS.slice(0, 2).join(""), + STREAM_DELTAS.join(""), + ]); + + providerState.gate.release(); + const settled = await reader.nextMatching((candidate) => candidate.type === "settled"); + expect(settled.type).toBe("settled"); + if (settled.type !== "settled") return; + // The terminal payload is the authority and is the same shape the route + // used to answer with in one JSON response. + expect(settled.payload.sessionId).toBe(opened.sessionId); + expect(settled.payload.surface).toBe("cleanroom"); + expect(assistantText(settled.payload.view)).toBe(STREAM_DELTAS.join("")); + expect(settled.payload.view.composer.state).toBe("ready"); + const messages = items(settled.payload.view).filter((item) => item.kind === "agent_message"); + expect(messages).toHaveLength(1); + await reader.end(); + + // The stream left the session usable rather than wedged mid-turn. + const next = await call("/api/phase7/ui/cleanroom/session", { sessionId: opened.sessionId }); + expect(next.sessionId).not.toBe(opened.sessionId); + }); + + it("interrupts the provider turn when the client abandons the stream", async () => { + const opened = await call("/api/phase7/ui/cleanroom/session"); + const controller = new AbortController(); + const response = await fetch(`${origin}/api/phase7/ui/message`, { + method: "POST", + headers: { "content-type": "application/json", accept: "application/x-ndjson" }, + body: JSON.stringify({ sessionId: opened.sessionId, message: "stream your answer" }), + signal: controller.signal, + }); + const reader = new FrameReader(response); + await reader.next(); + providerState.gate.release(); + await reader.nextMatching( + (candidate) => candidate.type === "frame" && assistantText(candidate.view).length > 0, + ); + + controller.abort(); + // The abandoned turn is interrupted rather than left running: the session + // returns to `ready`, which a still-active turn would never do. + await settle(async () => { + const probe = await call(`/api/phase7/ui/cleanroom/session?sessionId=${opened.sessionId}`); + return probe.sessionId === opened.sessionId && probe.view.composer.state === "ready"; + }); + const after = await call("/api/phase7/ui/message", { + sessionId: opened.sessionId, + message: "Read the issue and report progress.", + }); + expect(after.sessionId).toBe(opened.sessionId); + expect(items(after.view).some((item) => item.kind === "durable_comment")).toBe(true); + }); + it("keeps the preset scenario surface working and isolated from the clean room", async () => { const scenario = await call("/api/phase7/ui/session?scenario=hb-baseline"); const cleanRoom = await call("/api/phase7/ui/cleanroom/session"); diff --git a/packages/paperclip-runner/src/phase7/index.ts b/packages/paperclip-runner/src/phase7/index.ts index 6529ba01494..969b26d3391 100644 --- a/packages/paperclip-runner/src/phase7/index.ts +++ b/packages/paperclip-runner/src/phase7/index.ts @@ -2,3 +2,4 @@ export * from "./browser-index.js"; export * from "./clean-room.js"; export * from "./live-session.js"; export * from "./runnerd-codex-transport.js"; +export * from "./turn-stream.js"; diff --git a/packages/paperclip-runner/src/phase7/live-session.ts b/packages/paperclip-runner/src/phase7/live-session.ts index d0ac60a1b09..60b3f161b1f 100644 --- a/packages/paperclip-runner/src/phase7/live-session.ts +++ b/packages/paperclip-runner/src/phase7/live-session.ts @@ -138,6 +138,65 @@ export interface Phase7LiveTurnResult { snapshot: Phase7LiveSessionSnapshot; } +/** + * Live turn events (track 7Q). + * + * A turn used to be observable only through its resolved result, so everything + * downstream — the package server, the browser — could do nothing but wait. + * These events expose the same work as it happens without changing what the + * turn finally assembles: `assistantText` on a `delta` event is the running + * value of the very buffer `sendMessage` resolves with, not a parallel copy. + * + * The stream is ordered by `seq` (monotonic per session, from 1) and bounded: + * a turn emits at most `PHASE7_LIVE_TURN_EVENT_LIMIT` events, after which only + * terminal and error events are delivered. A pathological provider can make a + * turn long; it cannot make the event stream unbounded. + */ +export type Phase7LiveTurnEvent = + | { + seq: number; + at: string; + turnId: string | null; + kind: "delta"; + /** Assembled assistant text so far — monotonic within one agent message. */ + assistantText: string; + } + | { + seq: number; + at: string; + turnId: string | null; + kind: "activity"; + /** `turn_started`, `tool_call`, `tool_result`, or `stop_requested`. */ + reason: string; + } + | { + seq: number; + at: string; + turnId: string | null; + kind: "terminal"; + status: Phase7LiveTurnResult["status"]; + assistantText: string; + } + | { + seq: number; + at: string; + turnId: string | null; + kind: "error"; + message: string; + }; + +export type Phase7LiveTurnListener = (event: Phase7LiveTurnEvent) => void; + +/** Distributive `Omit`, so each event variant keeps its own discriminated shape. */ +type Phase7LiveTurnEventInput = Phase7LiveTurnEvent extends infer Variant + ? Variant extends Phase7LiveTurnEvent + ? Omit + : never + : never; + +/** Bound on interim events per turn; terminal and error always get through. */ +export const PHASE7_LIVE_TURN_EVENT_LIMIT = 4_000; + export interface Phase7InteractionResolution { interactionId: string; outcome: "answered" | "accepted" | "rejected" | "expired"; @@ -183,6 +242,13 @@ interface TurnWaiter { reject: (error: Error) => void; timer: ReturnType; assistantText: string; + /** + * Transcript entry the assistant reply is being written into while the turn + * runs, or `null` before the first delta. Streaming into a real transcript + * entry rather than a side buffer is what lets one rendered card grow: the + * projection sees the same entry id from the first delta to the last. + */ + draftId: string | null; } function record(value: unknown): Record { @@ -381,6 +447,9 @@ export class Phase7LiveSession { #turnWaiter: TurnWaiter | null = null; #pump: Promise | null = null; #persistChain: Promise = Promise.resolve(); + readonly #listeners = new Set(); + #eventSeq = 0; + #turnEventCount = 0; private constructor(options: OpenSessionOptions & { snapshot?: Phase7LiveSessionSnapshot }) { this.#port = options.port; @@ -444,6 +513,80 @@ export class Phase7LiveSession { return structuredClone(pendingInteractions(this.#port)); } + /** + * Observe live turn events. Returns the unsubscribe function; a listener that + * throws is dropped rather than allowed to break the provider pump, because a + * broken consumer must not be able to stall or fail the turn it is watching. + */ + subscribe(listener: Phase7LiveTurnListener): () => void { + this.#listeners.add(listener); + return () => { + this.#listeners.delete(listener); + }; + } + + #emit(event: Phase7LiveTurnEventInput): void { + const bounded = event.kind === "delta" || event.kind === "activity"; + if (bounded) { + if (this.#turnEventCount >= PHASE7_LIVE_TURN_EVENT_LIMIT) return; + this.#turnEventCount += 1; + } + this.#eventSeq += 1; + const published = { + ...event, + seq: this.#eventSeq, + at: this.#now().toISOString(), + } as Phase7LiveTurnEvent; + for (const listener of [...this.#listeners]) { + try { + listener(published); + } catch { + this.#listeners.delete(listener); + } + } + } + + /** + * Mirrors the turn waiter's assembled text into its transcript draft. The + * waiter's buffer stays the single assembly point, so the draft can never + * diverge from what the turn will finally resolve with. + */ + #syncAssistantDraft(turnId: string | null): void { + const waiter = this.#turnWaiter; + if (waiter === null || waiter.assistantText.length === 0) return; + if (waiter.draftId === null) { + waiter.draftId = this.#appendTranscript("assistant", waiter.assistantText, turnId); + return; + } + const draft = this.#transcript.find((entry) => entry.id === waiter.draftId); + if (draft === undefined) return; + draft.text = waiter.assistantText; + if (draft.turnId === null && turnId !== null) draft.turnId = turnId; + } + + /** + * Settles the streamed draft against the authoritative final text. An + * interrupted turn keeps whatever it had streamed; a turn that produced + * nothing leaves no empty card behind. + */ + #settleAssistantDraft(waiter: TurnWaiter, assistantText: string, turnId: string): void { + const index = + waiter.draftId === null + ? -1 + : this.#transcript.findIndex((entry) => entry.id === waiter.draftId); + if (index < 0) { + if (assistantText.length > 0) this.#appendTranscript("assistant", assistantText, turnId); + return; + } + if (assistantText.length === 0) { + this.#transcript.splice(index, 1); + return; + } + const draft = this.#transcript[index]!; + draft.text = assistantText; + draft.turnId = draft.turnId ?? turnId; + } + snapshot(): Phase7LiveSessionSnapshot { const records = [ ...this.#priorAuthorizationRecords, @@ -484,14 +627,26 @@ export class Phase7LiveSession { throw new Error("Phase 7 live session already has an active turn"); } this.#status = "running"; - this.#appendTranscript("user", value, null); + this.#turnEventCount = 0; + // Held by id, not by position: a provider can start streaming before + // `turn/start` resolves, in which case the assistant draft is already the + // last transcript entry and the user message would never learn its turn. + const userEntryId = this.#appendTranscript("user", value, null); let response: Record; const terminal = new Promise>((resolve, reject) => { const timer = setTimeout(() => { + const waiter = this.#turnWaiter; this.#turnWaiter = null; + if (waiter !== null) { + this.#emit({ + turnId: this.#activeTurnId, + kind: "error", + message: `Phase 7 Codex turn timed out after ${this.#config.turnTimeoutMs}ms`, + }); + } reject(new Error(`Phase 7 Codex turn timed out after ${this.#config.turnTimeoutMs}ms`)); }, this.#config.turnTimeoutMs); - this.#turnWaiter = { resolve, reject, timer, assistantText: "" }; + this.#turnWaiter = { resolve, reject, timer, assistantText: "", draftId: null }; }); try { response = await this.#transport.request("turn/start", { @@ -516,8 +671,12 @@ export class Phase7LiveSession { throw new Error("Codex turn identity changed during start"); } if (this.#turnWaiter !== null) this.#activeTurnId = turnId; - const lastUser = this.#transcript.at(-1); - if (lastUser?.role === "user" && lastUser.turnId === null) lastUser.turnId = turnId; + const sent = this.#transcript.find((entry) => entry.id === userEntryId); + if (sent !== undefined && sent.turnId === null) sent.turnId = turnId; + const draftId = (this.#turnWaiter as TurnWaiter | null)?.draftId ?? null; + const draft = + draftId === null ? undefined : this.#transcript.find((entry) => entry.id === draftId); + if (draft !== undefined && draft.turnId === null) draft.turnId = turnId; await this.#persist(); const result = await terminal; await this.#persist(); @@ -529,6 +688,7 @@ export class Phase7LiveSession { this.#status = "stopping"; const turnId = this.#activeTurnId; this.#appendEvidence("session", turnId, { action: "stop_requested", reason }); + this.#emit({ turnId, kind: "activity", reason: "stop_requested" }); await this.#transport.request("turn/interrupt", { threadId: this.#providerThreadId, turnId, @@ -733,6 +893,7 @@ export class Phase7LiveSession { input: jsonValue(request.params.arguments), beforeRevision, }); + this.#emit({ turnId, kind: "activity", reason: "tool_call" }); const result = await this.#dispatcher.dispatch({ runId: this.#authority.runId, callId, @@ -746,6 +907,7 @@ export class Phase7LiveSession { beforeRevision, afterRevision: result.stateRevision, }); + this.#emit({ turnId, kind: "activity", reason: "tool_result" }); await this.#persist(); return this.#codexToolResponse(result); } @@ -789,12 +951,17 @@ export class Phase7LiveSession { }); if (notification.method === "turn/started") { if (turnId.length > 0) this.#activeTurnId = turnId; + this.#emit({ turnId: turnId || this.#activeTurnId, kind: "activity", reason: "turn_started" }); } else if (notification.method === "item/agentMessage/delta") { - if (this.#turnWaiter !== null) this.#turnWaiter.assistantText += text(params.delta); + if (this.#turnWaiter !== null) { + this.#turnWaiter.assistantText += text(params.delta); + this.#publishAssistantProgress(turnId || this.#activeTurnId); + } } else if (notification.method === "item/completed" && text(item.type) === "agentMessage") { const assistantText = text(item.text); if (assistantText.length > 0 && this.#turnWaiter !== null) { this.#turnWaiter.assistantText = assistantText; + this.#publishAssistantProgress(turnId || this.#activeTurnId); } } else if (notification.method === "turn/completed") { this.#completeTurn(turnId, terminalStatus(turn.status)); @@ -802,15 +969,26 @@ export class Phase7LiveSession { void this.#persist(); } + /** Streams the assembled text into the transcript draft and announces it. */ + #publishAssistantProgress(turnId: string | null): void { + this.#syncAssistantDraft(turnId); + this.#emit({ + turnId, + kind: "delta", + assistantText: this.#turnWaiter?.assistantText ?? "", + }); + } + #completeTurn(turnId: string, status: Phase7LiveTurnResult["status"]): void { const waiter = this.#turnWaiter; if (waiter === null) return; clearTimeout(waiter.timer); this.#turnWaiter = null; const assistantText = waiter.assistantText.trim(); - if (assistantText.length > 0) this.#appendTranscript("assistant", assistantText, turnId); + this.#settleAssistantDraft(waiter, assistantText, turnId); this.#activeTurnId = null; this.#status = "idle"; + this.#emit({ turnId, kind: "terminal", status, assistantText }); waiter.resolve({ turnId, status, assistantText }); } @@ -819,20 +997,35 @@ export class Phase7LiveSession { if (waiter === null) return; clearTimeout(waiter.timer); this.#turnWaiter = null; + const turnId = this.#activeTurnId; this.#activeTurnId = null; this.#status = "failed"; + // Whatever the turn had already streamed stays in the transcript; a failed + // turn should read as an interrupted reply, not as one that never spoke. + if (turnId !== null) this.#settleAssistantDraft(waiter, waiter.assistantText.trim(), turnId); + this.#emit({ + turnId, + kind: "error", + message: redactCodexDiagnostic(error instanceof Error ? error.message : String(error)), + }); waiter.reject(error instanceof Error ? error : new Error(String(error))); } - #appendTranscript(role: Phase7LiveTranscriptEntry["role"], message: string, turnId: string | null): void { + #appendTranscript( + role: Phase7LiveTranscriptEntry["role"], + message: string, + turnId: string | null, + ): string { this.#entryCounter += 1; + const id = `transcript-${String(this.#entryCounter).padStart(6, "0")}`; this.#transcript.push({ - id: `transcript-${String(this.#entryCounter).padStart(6, "0")}`, + id, role, text: message, turnId, at: this.#now().toISOString(), }); + return id; } #appendEvidence( diff --git a/packages/paperclip-runner/src/phase7/turn-stream.test.ts b/packages/paperclip-runner/src/phase7/turn-stream.test.ts new file mode 100644 index 00000000000..8485a1042bd --- /dev/null +++ b/packages/paperclip-runner/src/phase7/turn-stream.test.ts @@ -0,0 +1,408 @@ +import { describe, expect, it } from "vitest"; + +import type { + CodexAppServerTransport, + CodexRpcNotification, + CodexServerRequestHandler, +} from "../drivers/codex/app-server-transport.js"; +import { projectPhase7IssueThread } from "../issue-thread/live-projection.js"; +import type { Phase7IssueThreadSnapshot } from "../issue-thread/types.js"; +import { createPhase7CleanRoomSessionInput } from "./clean-room.js"; +import { + Phase7LiveSessionService, + type Phase7LiveSession, + type Phase7LiveTransportFactory, + type Phase7LiveTurnEvent, +} from "./live-session.js"; +import { + createPhase7TurnStreamDecoder, + encodePhase7TurnStreamFrame, + PHASE7_TURN_STREAM_SCHEMA, + Phase7TurnStreamError, + readPhase7TurnStream, + type Phase7TurnStreamFrame, +} from "./turn-stream.js"; + +/** + * Transport-level proof that a turn streams (track 7Q). + * + * Three assistant deltas are released one at a time, so nothing here depends on + * a delay, a poll, or the order the event loop happens to choose. What is under + * test is the whole path a frame takes — live session event, projected view, + * NDJSON encode, chunked decode — against the two properties the board's + * complaint reduces to: the reply is visible while it is still being written, + * and what it finally says is exactly what was streamed. + */ + +const DELTAS = ["Reading the ", "clean-room issue ", "and recording a status."] as const; +const FINAL_TEXT = DELTAS.join(""); + +/** Single-slot signal: the test releases one provider step at a time. */ +class Gate { + #waiters: Array<() => void> = []; + #ready = 0; + + wait(): Promise { + if (this.#ready > 0) { + this.#ready -= 1; + return Promise.resolve(); + } + return new Promise((resolve) => this.#waiters.push(resolve)); + } + + release(): void { + const waiter = this.#waiters.shift(); + if (waiter === undefined) this.#ready += 1; + else waiter(); + } +} + +class AsyncNotifications implements AsyncIterable { + #values: CodexRpcNotification[] = []; + #waiters: Array<(value: IteratorResult) => void> = []; + #closed = false; + + push(value: CodexRpcNotification): void { + const waiter = this.#waiters.shift(); + if (waiter) waiter({ value, done: false }); + else this.#values.push(value); + } + + close(): void { + this.#closed = true; + for (const waiter of this.#waiters.splice(0)) waiter({ value: undefined, done: true }); + } + + [Symbol.asyncIterator](): AsyncIterator { + return { + next: async () => { + const value = this.#values.shift(); + if (value) return { value, done: false }; + if (this.#closed) return { value: undefined, done: true }; + return new Promise((resolve) => this.#waiters.push(resolve)); + }, + }; + } +} + +class GatedTransport implements CodexAppServerTransport { + readonly queue = new AsyncNotifications(); + #closed = false; + + constructor(readonly gate: Gate) {} + + async request(method: string, params: Record): Promise> { + if (method === "initialize") return { user: { sessionId: "provider-session" } }; + if (method === "thread/start" || method === "thread/read" || method === "thread/resume") { + return { thread: { id: "provider-thread", sessionId: "provider-session" } }; + } + if (method === "turn/start") { + void this.#runTurn("turn-1"); + return { turn: { id: "turn-1", status: "inProgress" } }; + } + if (method === "turn/interrupt") { + this.queue.push({ + method: "turn/completed", + params: { threadId: "provider-thread", turn: { id: String(params.turnId), status: "interrupted" } }, + }); + return {}; + } + throw new Error(`unsupported gated Codex method ${method}`); + } + + notify(): void {} + + notifications(): AsyncIterable { + return this.queue; + } + + setServerRequestHandler(_handler: CodexServerRequestHandler): void {} + + async close(): Promise { + this.#closed = true; + this.queue.close(); + } + + processInfo() { + return { pid: 7100, processGroupId: 7100, exited: this.#closed, exitCode: null, signal: null }; + } + + async #runTurn(turnId: string): Promise { + this.queue.push({ + method: "turn/started", + params: { threadId: "provider-thread", turn: { id: turnId, status: "inProgress" } }, + }); + for (const delta of DELTAS) { + await this.gate.wait(); + this.queue.push({ + method: "item/agentMessage/delta", + params: { threadId: "provider-thread", turnId, delta }, + }); + } + await this.gate.wait(); + this.queue.push({ + method: "item/completed", + params: { + threadId: "provider-thread", + turnId, + item: { id: "message-1", type: "agentMessage", text: FINAL_TEXT }, + }, + }); + this.queue.push({ + method: "turn/completed", + params: { threadId: "provider-thread", turn: { id: turnId, status: "completed" } }, + }); + } +} + +function gatedTransportFactory(gate: Gate): Phase7LiveTransportFactory { + return (options) => { + const evidence = { + runnerPid: 7100, + runnerProcessGroupId: 7100, + codexPid: 7200, + runnerExited: false, + runnerExitCode: null, + runnerSignal: null, + childEnvironmentKeys: ["CODEX_HOME", "HOME", "PATH"], + diagnostics: [], + }; + options.onEvidence?.(evidence); + return { transport: new GatedTransport(gate), evidence: () => ({ ...evidence }) }; + }; +} + +async function openGatedSession(gate: Gate): Promise<{ + session: Phase7LiveSession; + close: () => Promise; +}> { + const service = new Phase7LiveSessionService({ transportFactory: gatedTransportFactory(gate) }); + const { input } = createPhase7CleanRoomSessionInput({ workingDirectory: process.cwd() }); + const session = await service.create(input); + return { session, close: () => service.shutdown(session.id, "test complete") }; +} + +/** Runs the event loop until `ready` holds, so no assertion waits on a delay. */ +async function settle(ready: () => boolean, attempts = 200): Promise { + for (let attempt = 0; attempt < attempts; attempt += 1) { + if (ready()) return; + await new Promise((resolve) => setTimeout(resolve, 1)); + } + throw new Error("the stream never reached the expected state"); +} + +/** Assistant text the projected view renders, which is what a reader sees. */ +function assistantText(view: Phase7IssueThreadSnapshot): string { + return view.turns + .flatMap((turn) => turn.items) + .filter((item) => item.kind === "agent_message") + .map((item) => (item.kind === "agent_message" ? item.body : "")) + .join(""); +} + +describe("Phase 7 turn stream framing", () => { + it("round-trips frames split across arbitrary chunk boundaries", () => { + const frames: Array> = [ + { schema: PHASE7_TURN_STREAM_SCHEMA, type: "frame", seq: 1, reason: "open", turnId: null, view: { text: "" } }, + { schema: PHASE7_TURN_STREAM_SCHEMA, type: "frame", seq: 2, reason: "delta", turnId: "t", view: { text: "ab\ncd" } }, + { schema: PHASE7_TURN_STREAM_SCHEMA, type: "settled", seq: 3, payload: { done: true } }, + ]; + const wire = frames.map(encodePhase7TurnStreamFrame).join(""); + const decoder = createPhase7TurnStreamDecoder<{ text: string }, { done: boolean }>(); + const decoded = []; + // One character at a time is the worst case a socket can produce. + for (const character of wire) decoded.push(...decoder.push(character)); + decoded.push(...decoder.flush()); + + expect(decoded).toEqual(frames); + }); + + it("refuses a replayed frame, a frame after the terminal, and a truncated stream", () => { + const frame = (seq: number): string => + encodePhase7TurnStreamFrame({ + schema: PHASE7_TURN_STREAM_SCHEMA, + type: "frame", + seq, + reason: "delta", + turnId: "t", + view: {}, + }); + const settled = encodePhase7TurnStreamFrame({ + schema: PHASE7_TURN_STREAM_SCHEMA, + type: "settled", + seq: 5, + payload: {}, + }); + + const replayed = createPhase7TurnStreamDecoder(); + replayed.push(frame(2)); + expect(() => replayed.push(frame(2))).toThrow(Phase7TurnStreamError); + + const late = createPhase7TurnStreamDecoder(); + late.push(settled); + expect(() => late.push(frame(6))).toThrow(/after the terminal/); + + const truncated = createPhase7TurnStreamDecoder(); + truncated.push(frame(1)); + expect(() => truncated.flush()).toThrow(/before a terminal frame/); + }); + + it("streams three gated deltas into growing client states that match the final text", async () => { + const gate = new Gate(); + const { session, close } = await openGatedSession(gate); + try { + const events: Phase7LiveTurnEvent[] = []; + // The wire between the server and the client: every live event is + // projected, framed, and handed back through the decoder, so what the + // assertions inspect is what a browser would have received. + const decoder = createPhase7TurnStreamDecoder(); + const clientStates: string[] = []; + let terminalSeen = false; + let statesBeforeTerminal = 0; + let seq = 0; + + const unsubscribe = session.subscribe((event) => { + events.push(event); + if (event.kind !== "delta") return; + seq += 1; + const wire = encodePhase7TurnStreamFrame({ + schema: PHASE7_TURN_STREAM_SCHEMA, + type: "frame", + seq, + reason: "delta", + turnId: event.turnId, + view: projectPhase7IssueThread({ snapshot: session.snapshot() }), + }); + // Split every frame so the decoder is exercised on partial chunks. + const half = Math.floor(wire.length / 2); + for (const frame of [...decoder.push(wire.slice(0, half)), ...decoder.push(wire.slice(half))]) { + if (frame.type !== "frame") continue; + clientStates.push(assistantText(frame.view)); + if (!terminalSeen) statesBeforeTerminal += 1; + } + }); + + const turn = session.sendMessage("Read the issue and record a status."); + // One delta at a time, each observed by the client before the next is + // produced. This is the property the board asked for: a reply that is + // readable while it is still being written, not one revealed at the end. + for (const [index, _delta] of DELTAS.entries()) { + const before = clientStates.length; + gate.release(); + await settle(() => clientStates.length > before); + expect(clientStates.at(-1)).toBe(DELTAS.slice(0, index + 1).join("")); + expect(session.snapshot().activeTurnId).not.toBeNull(); + } + gate.release(); + const result = await turn; + terminalSeen = true; + unsubscribe(); + + const nonEmpty = clientStates.filter((text) => text.length > 0); + expect(statesBeforeTerminal).toBeGreaterThanOrEqual(2); + expect(nonEmpty.length).toBeGreaterThanOrEqual(2); + for (let index = 1; index < nonEmpty.length; index += 1) { + // Monotonic: a later state always contains its predecessor as a prefix. + expect(nonEmpty[index]!.startsWith(nonEmpty[index - 1]!)).toBe(true); + expect(nonEmpty[index]!.length).toBeGreaterThanOrEqual(nonEmpty[index - 1]!.length); + } + // The intermediate states are real prefixes of the reply, not one + // finished block delivered late. + expect(nonEmpty[0]).toBe(DELTAS[0]); + expect(nonEmpty.at(-1)).toBe(FINAL_TEXT); + expect(DELTAS.join("")).toBe(result.assistantText); + + const terminal = events.at(-1); + expect(terminal?.kind).toBe("terminal"); + if (terminal?.kind === "terminal") { + expect(terminal.status).toBe("completed"); + expect(terminal.assistantText).toBe(FINAL_TEXT); + } + // Ordered: sequence numbers never repeat or go backwards. + const sequences = events.map((event) => event.seq); + expect(sequences).toEqual([...sequences].sort((left, right) => left - right)); + expect(new Set(sequences).size).toBe(sequences.length); + + // The settled transcript holds exactly one assistant entry, so the + // streamed draft was finalised rather than duplicated. + const finalView = projectPhase7IssueThread({ snapshot: result.snapshot }); + const messages = finalView.turns + .flatMap((turn) => turn.items) + .filter((item) => item.kind === "agent_message"); + expect(messages).toHaveLength(1); + expect(assistantText(finalView)).toBe(FINAL_TEXT); + } finally { + await close(); + } + }); + + it("keeps the partial reply when a stopped turn never finishes its message", async () => { + const gate = new Gate(); + const { session, close } = await openGatedSession(gate); + try { + const turn = session.sendMessage("Start writing, then I will stop you."); + gate.release(); + // Wait for the first delta to reach the transcript before interrupting. + await new Promise((resolve) => setTimeout(resolve, 5)); + await session.interrupt("operator stopped the turn"); + const result = await turn; + + expect(result.status).toBe("interrupted"); + expect(result.assistantText).toBe(DELTAS[0]!.trim()); + const view = projectPhase7IssueThread({ snapshot: result.snapshot }); + expect(assistantText(view)).toBe(DELTAS[0]!.trim()); + expect(view.turns.at(-1)?.stoppedByUser).toBe(true); + // Nothing further is appended once the turn is terminal. + gate.release(); + gate.release(); + await new Promise((resolve) => setTimeout(resolve, 5)); + expect(assistantText(projectPhase7IssueThread({ snapshot: session.snapshot() }))).toBe( + DELTAS[0]!.trim(), + ); + } finally { + await close(); + } + }); + + it("reads a whole stream body into its settled payload", async () => { + const wire = [ + encodePhase7TurnStreamFrame({ + schema: PHASE7_TURN_STREAM_SCHEMA, + type: "frame", + seq: 1, + reason: "delta", + turnId: "t", + view: { text: "partial" }, + }), + encodePhase7TurnStreamFrame({ + schema: PHASE7_TURN_STREAM_SCHEMA, + type: "settled", + seq: 2, + payload: { text: "final" }, + }), + ].join(""); + const body = new Response(wire); + const views: string[] = []; + const payload = await readPhase7TurnStream<{ text: string }, { text: string }>( + body as unknown as { body: never }, + (frame) => views.push(frame.view.text), + ); + + expect(views).toEqual(["partial"]); + expect(payload).toEqual({ text: "final" }); + }); + + it("raises a server error frame as a typed stream error", async () => { + const wire = encodePhase7TurnStreamFrame({ + schema: PHASE7_TURN_STREAM_SCHEMA, + type: "error", + seq: 1, + error: "turn_failed", + message: "the provider went away", + }); + + await expect( + readPhase7TurnStream(new Response(wire) as unknown as { body: never }, () => undefined), + ).rejects.toThrow(/the provider went away/); + }); +}); diff --git a/packages/paperclip-runner/src/phase7/turn-stream.ts b/packages/paperclip-runner/src/phase7/turn-stream.ts new file mode 100644 index 00000000000..6f11937fa28 --- /dev/null +++ b/packages/paperclip-runner/src/phase7/turn-stream.ts @@ -0,0 +1,264 @@ +/** + * NDJSON turn-stream framing for the Phase 7 live chat (track 7Q). + * + * One POST carries a whole turn. Before this the server awaited the terminal + * artifact and answered once, so the browser could only reveal a finished + * reply — which is exactly what the board rejected. The turn now answers with + * an ordered sequence of newline-delimited frames written as the provider + * produces them, and the final `settled` frame carries the same payload the + * single JSON response used to carry. The last frame stays the authority: an + * interim frame is a view of work in progress, never a substitute for it. + * + * NDJSON rather than SSE because the request is already a POST the client + * consumes with `fetch`. `EventSource` cannot POST, and framing `data:` lines + * by hand would buy nothing over one JSON object per line. + * + * This module is deliberately dependency-free — no Node built-ins, no view + * types — so the browser bundle, the package server, and the smoke scripts can + * all agree on the wire format without any of them dragging in the others. + */ + +export const PHASE7_TURN_STREAM_SCHEMA = "paperclip.phase7.turn-stream.v1" as const; + +/** `no-transform` is the half that keeps an intermediary from coalescing frames. */ +export const PHASE7_TURN_STREAM_HEADERS: Readonly> = Object.freeze({ + "content-type": "application/x-ndjson; charset=utf-8", + "cache-control": "no-store, no-transform", + // Nginx and friends buffer proxied responses by default, which would hold + // every frame until the turn ended and reproduce the reported symptom. + "x-accel-buffering": "no", +}); + +export const PHASE7_TURN_STREAM_ACCEPT = "application/x-ndjson" as const; + +/** + * A frame line longer than this means the peer is not speaking this protocol + * (or is trying to make the reader buffer without bound), so the decoder fails + * rather than growing. + */ +export const PHASE7_TURN_STREAM_MAX_FRAME_BYTES = 4 * 1024 * 1024; + +/** + * Why the server emitted an interim frame. Carried for observability only: the + * client renders `view` the same way whatever produced it. + */ +export type Phase7TurnStreamReason = + | "open" + | "delta" + | "activity" + | "stop_requested" + | "terminal"; + +export interface Phase7TurnStreamViewFrame { + schema: typeof PHASE7_TURN_STREAM_SCHEMA; + type: "frame"; + /** Monotonic within one turn stream, starting at 1. */ + seq: number; + reason: Phase7TurnStreamReason; + turnId: string | null; + view: View; +} + +export interface Phase7TurnStreamSettledFrame { + schema: typeof PHASE7_TURN_STREAM_SCHEMA; + type: "settled"; + seq: number; + payload: Payload; +} + +export interface Phase7TurnStreamErrorFrame { + schema: typeof PHASE7_TURN_STREAM_SCHEMA; + type: "error"; + seq: number; + error: string; + message: string; +} + +export type Phase7TurnStreamFrame = + | Phase7TurnStreamViewFrame + | Phase7TurnStreamSettledFrame + | Phase7TurnStreamErrorFrame; + +export function isPhase7TurnStreamTerminalFrame( + frame: Phase7TurnStreamFrame, +): frame is Phase7TurnStreamSettledFrame | Phase7TurnStreamErrorFrame { + return frame.type === "settled" || frame.type === "error"; +} + +export function encodePhase7TurnStreamFrame( + frame: Phase7TurnStreamFrame, +): string { + return `${JSON.stringify(frame)}\n`; +} + +export class Phase7TurnStreamError extends Error { + constructor( + readonly code: string, + message: string, + ) { + super(message); + this.name = "Phase7TurnStreamError"; + } +} + +function parseFrame(line: string): Phase7TurnStreamFrame { + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch { + throw new Phase7TurnStreamError("stream_frame_invalid", "A turn frame was not valid JSON."); + } + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) { + throw new Phase7TurnStreamError("stream_frame_invalid", "A turn frame was not an object."); + } + const frame = parsed as Partial>; + if (frame.schema !== PHASE7_TURN_STREAM_SCHEMA) { + throw new Phase7TurnStreamError("stream_schema_unknown", "A turn frame used an unknown schema."); + } + if (frame.type !== "frame" && frame.type !== "settled" && frame.type !== "error") { + throw new Phase7TurnStreamError("stream_frame_invalid", "A turn frame used an unknown type."); + } + if (typeof frame.seq !== "number" || !Number.isInteger(frame.seq) || frame.seq < 1) { + throw new Phase7TurnStreamError("stream_frame_invalid", "A turn frame omitted its sequence."); + } + return frame as Phase7TurnStreamFrame; +} + +export interface Phase7TurnStreamDecoder { + /** Frames completed by this chunk, in wire order. */ + push(chunk: string): Array>; + /** Frames completed by the end of the body; also rejects a truncated tail. */ + flush(): Array>; +} + +/** + * Incremental NDJSON reader. A chunk boundary can land anywhere, including + * mid-frame, so the tail is held until its newline arrives; ordering is the + * wire order, and `seq` is checked to be strictly increasing so a reordered or + * replayed frame is a failure rather than a silently applied stale view. + */ +export function createPhase7TurnStreamDecoder(): Phase7TurnStreamDecoder< + View, + Payload +> { + let buffer = ""; + let lastSeq = 0; + let terminal = false; + + function accept(line: string): Phase7TurnStreamFrame | null { + const trimmed = line.trim(); + if (trimmed.length === 0) return null; + if (terminal) { + throw new Phase7TurnStreamError( + "stream_after_terminal", + "A turn frame arrived after the terminal frame.", + ); + } + const frame = parseFrame(trimmed); + if (frame.seq <= lastSeq) { + throw new Phase7TurnStreamError( + "stream_out_of_order", + "A turn frame arrived out of order.", + ); + } + lastSeq = frame.seq; + if (isPhase7TurnStreamTerminalFrame(frame)) terminal = true; + return frame; + } + + return { + push(chunk) { + buffer += chunk; + if (buffer.length > PHASE7_TURN_STREAM_MAX_FRAME_BYTES) { + throw new Phase7TurnStreamError( + "stream_frame_too_large", + "A turn frame exceeded the stream frame limit.", + ); + } + const frames: Array> = []; + let newline = buffer.indexOf("\n"); + while (newline >= 0) { + const frame = accept(buffer.slice(0, newline)); + buffer = buffer.slice(newline + 1); + if (frame !== null) frames.push(frame); + newline = buffer.indexOf("\n"); + } + return frames; + }, + flush() { + const tail = buffer; + buffer = ""; + const frame = tail.trim().length === 0 ? null : accept(tail); + if (!terminal) { + throw new Phase7TurnStreamError( + "stream_truncated", + "The turn stream ended before a terminal frame.", + ); + } + return frame === null ? [] : [frame]; + }, + }; +} + +export interface Phase7TurnStreamBodyLike { + getReader(): { + read(): Promise<{ done: boolean; value?: Uint8Array | undefined }>; + releaseLock(): void; + }; +} + +export interface Phase7TurnStreamResponseLike { + body: Phase7TurnStreamBodyLike | null; +} + +/** + * Drives `onFrame` from a `fetch` response body and returns the settled + * payload. An `error` frame is raised as a `Phase7TurnStreamError` carrying the + * server's own code, so a stream failure reads the same way as the JSON error + * bodies the pre-turn checks still return. + */ +export async function readPhase7TurnStream( + response: Phase7TurnStreamResponseLike, + onFrame: (frame: Phase7TurnStreamViewFrame) => void, +): Promise { + if (response.body === null) { + throw new Phase7TurnStreamError("stream_unreadable", "The turn response had no readable body."); + } + const decoder = createPhase7TurnStreamDecoder(); + const text = new TextDecoder(); + const reader = response.body.getReader(); + let settled: Payload | undefined; + let settledSeen = false; + + function apply(frames: Array>): void { + for (const frame of frames) { + if (frame.type === "frame") { + onFrame(frame); + } else if (frame.type === "settled") { + settled = frame.payload; + settledSeen = true; + } else { + throw new Phase7TurnStreamError(frame.error, frame.message); + } + } + } + + try { + for (;;) { + const chunk = await reader.read(); + if (chunk.done) break; + if (chunk.value !== undefined) { + apply(decoder.push(text.decode(chunk.value, { stream: true }))); + } + } + apply(decoder.push(text.decode())); + apply(decoder.flush()); + } finally { + reader.releaseLock(); + } + + if (!settledSeen) { + throw new Phase7TurnStreamError("stream_truncated", "The turn stream never settled."); + } + return settled as Payload; +} diff --git a/packages/paperclip-runner/vite.issue-thread-stream.config.ts b/packages/paperclip-runner/vite.issue-thread-stream.config.ts new file mode 100644 index 00000000000..276a7b7b1f8 --- /dev/null +++ b/packages/paperclip-runner/vite.issue-thread-stream.config.ts @@ -0,0 +1,39 @@ +import { resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +import react from "@vitejs/plugin-react"; +import { defineConfig } from "vite"; + +import { phase7StreamFixtureTransportFactory } from "./devtools/issue-thread/stream-fixture-transport.mjs"; +import { phase7IssueThreadServerPlugin } from "./scripts/phase7-issue-thread-server.mjs"; + +/** + * Test-only preview of the issue-thread app for the Phase 7Q streaming check. + * + * Identical to `vite.issue-thread.config.ts` — same root, same built bundle, + * same package server middleware — except that the Codex provider is scripted + * so the browser test can control when a delta lands. Everything the test + * actually asserts (the NDJSON turn stream, the live session, the projection, + * and the browser client) is the shipped code path. + * + * The shipped config takes no such override, and none can be injected through a + * request or an environment variable. + */ + +const packageRoot = fileURLToPath(new URL(".", import.meta.url)); + +export default defineConfig({ + root: resolve(packageRoot, "devtools/issue-thread"), + plugins: [ + react(), + phase7IssueThreadServerPlugin({ transportFactory: phase7StreamFixtureTransportFactory }), + ], + server: { host: "127.0.0.1" }, + preview: { host: "127.0.0.1" }, + optimizeDeps: { esbuildOptions: { target: "esnext" } }, + build: { + outDir: resolve(packageRoot, "dist-issue-thread"), + emptyOutDir: true, + target: "esnext", + }, +});