mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-27 02:44:50 +01:00
feat(runner): stream live Codex turns end to end
The clean-room chat revealed a finished reply instead of writing one: the live session accumulated agent deltas in its turn waiter, the package session server awaited the whole turn and answered with one JSON body, and the browser waited on `response.json()`. Nothing between the provider and the page could show work in progress, so the board saw the answer appear all at once. A turn now streams: - `Phase7LiveSession` publishes bounded, ordered live turn events and mirrors the waiter's assembled text into the transcript entry the reply will finally occupy, so one rendered card grows and the settled text is still assembled in exactly one place. - `src/phase7/turn-stream.ts` defines the NDJSON frame codec and reader. The decoder refuses a replayed, reordered, oversized, late, or truncated frame rather than applying a stale view. - `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. Client disconnect interrupts the provider turn. - The browser client consumes frames and renders each projection in place; the settled payload is applied last and remains the authority. Stop, reset, and `New chat` abandon the stream, and frames from an abandoned turn generation are dropped. Measured against real Codex through real runnerd: headers at 265 ms, 62 growing assistant states from the first character to a 301-character reply at 3.1 s, the last interim state identical to the settled one, and a single assistant card in the settled thread. The scenario explorer and every fixture, preset, and replay route are unchanged and stay deterministic. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5 committed
1 parent
5becbbd216
commit
b73524ec2a
18 files changed
+1947
-69
No files matched your search
@@ -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<void> {
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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,
|
||||
},
|
||||
],
|
||||
});
|
||||
@@ -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<HTMLDivElement | null>(null);
|
||||
const cancelResetRef = useRef<HTMLButtonElement | null>(null);
|
||||
/** Last announced pending request, so the live region tracks transitions. */
|
||||
const announcedPendingRef = useRef<string | null>(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<AbortController | null>(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 (
|
||||
<div
|
||||
className="pit-app"
|
||||
data-thread-state={settled && snapshotSurface === route.surface ? "settled" : "loading"}
|
||||
data-thread-state={
|
||||
streamingTurn
|
||||
? "streaming"
|
||||
: settled && snapshotSurface === route.surface
|
||||
? "settled"
|
||||
: "loading"
|
||||
}
|
||||
data-session-mode={snapshot.mode}
|
||||
data-surface={route.surface}
|
||||
data-connection-state={snapshot.connection.state}
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
import type { Phase7IssueThreadSnapshot } from "../../../src/issue-thread/types";
|
||||
import {
|
||||
PHASE7_TURN_STREAM_ACCEPT,
|
||||
Phase7TurnStreamError,
|
||||
readPhase7TurnStream,
|
||||
} from "../../../src/phase7/turn-stream";
|
||||
|
||||
/**
|
||||
* Browser client for the package session server.
|
||||
@@ -7,6 +12,12 @@ import type { Phase7IssueThreadSnapshot } from "../../../src/issue-thread/types"
|
||||
* posts intents and renders what comes back; it never patches the snapshot
|
||||
* locally, which is what keeps policy and state authority on the server
|
||||
* (contract §11).
|
||||
*
|
||||
* A turn is the one route that answers with many of those projections instead
|
||||
* of one: `send` consumes NDJSON frames as the provider produces them, hands
|
||||
* each interim view to `onFrame`, and resolves with the settled payload. The
|
||||
* authority rule is unchanged — every frame is still a server projection, and
|
||||
* the settled one is final.
|
||||
*/
|
||||
|
||||
const BASE = "/api/phase7/ui";
|
||||
@@ -68,6 +79,42 @@ async function post(path: string, body: unknown): Promise<Phase7LiveResponse> {
|
||||
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<Phase7LiveResponse> {
|
||||
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<Phase7IssueThreadSnapshot, Phase7LiveResponse>(
|
||||
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<Phase7LiveResponse> {
|
||||
const query = sessionId === null ? "" : `?sessionId=${encodeURIComponent(sessionId)}`;
|
||||
@@ -89,8 +136,16 @@ export const phase7LiveClient = {
|
||||
newCleanRoom(sessionId: string | null): Promise<Phase7LiveResponse> {
|
||||
return post("/cleanroom/session", sessionId === null ? {} : { sessionId });
|
||||
},
|
||||
send(sessionId: string, message: string): Promise<Phase7LiveResponse> {
|
||||
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<Phase7LiveResponse> {
|
||||
return postTurnStream("/message", { sessionId, message }, options);
|
||||
},
|
||||
stop(sessionId: string): Promise<Phase7LiveResponse> {
|
||||
return post("/interrupt", { sessionId });
|
||||
|
||||
@@ -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 }) };
|
||||
}
|
||||
+125
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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";
|
||||
|
||||
|
||||
@@ -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";
|
||||
@@ -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.",
|
||||
|
||||
@@ -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<void> {
|
||||
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<CodexRpcNotification> {
|
||||
#values: CodexRpcNotification[] = [];
|
||||
#waiters: Array<(value: IteratorResult<CodexRpcNotification>) => 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<boolean>, attempts = 200): Promise<void> {
|
||||
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<Phase7IssueThreadSnapshot, CleanRoomPayload>;
|
||||
|
||||
/**
|
||||
* 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<Uint8Array>;
|
||||
readonly #decoder = createPhase7TurnStreamDecoder<Phase7IssueThreadSnapshot, CleanRoomPayload>();
|
||||
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<CleanRoomFrame> {
|
||||
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<CleanRoomFrame> {
|
||||
for (;;) {
|
||||
const frame = await this.next();
|
||||
if (predicate(frame)) return frame;
|
||||
}
|
||||
}
|
||||
|
||||
async end(): Promise<void> {
|
||||
await this.#reader.cancel().catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
let server: Server;
|
||||
let origin: string;
|
||||
let middleware: Awaited<ReturnType<typeof createMiddleware>>;
|
||||
@@ -237,7 +351,28 @@ async function createMiddleware(state: FakeProviderState) {
|
||||
});
|
||||
}
|
||||
|
||||
async function call(path: string, body?: unknown): Promise<CleanRoomPayload> {
|
||||
/**
|
||||
* 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<CleanRoomPayload> {
|
||||
return await readPhase7TurnStream<Phase7IssueThreadSnapshot, CleanRoomPayload>(
|
||||
response,
|
||||
(frame) => {
|
||||
frames.push(frame.view);
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
async function call(
|
||||
path: string,
|
||||
body?: unknown,
|
||||
frames?: Phase7IssueThreadSnapshot[],
|
||||
): Promise<CleanRoomPayload> {
|
||||
const url = `${origin}${path}`;
|
||||
requestedUrls.push(url);
|
||||
const response = await fetch(
|
||||
@@ -250,6 +385,9 @@ async function call(path: string, body?: unknown): Promise<CleanRoomPayload> {
|
||||
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");
|
||||
|
||||
@@ -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";
|
||||
@@ -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<Variant, "seq" | "at">
|
||||
: 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<typeof setTimeout>;
|
||||
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<string, unknown> {
|
||||
@@ -381,6 +447,9 @@ export class Phase7LiveSession {
|
||||
#turnWaiter: TurnWaiter | null = null;
|
||||
#pump: Promise<void> | null = null;
|
||||
#persistChain: Promise<void> = Promise.resolve();
|
||||
readonly #listeners = new Set<Phase7LiveTurnListener>();
|
||||
#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<string, unknown>;
|
||||
const terminal = new Promise<Omit<Phase7LiveTurnResult, "snapshot">>((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(
|
||||
|
||||
@@ -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<void> {
|
||||
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<CodexRpcNotification> {
|
||||
#values: CodexRpcNotification[] = [];
|
||||
#waiters: Array<(value: IteratorResult<CodexRpcNotification>) => 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<CodexRpcNotification> {
|
||||
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<string, unknown>): Promise<Record<string, unknown>> {
|
||||
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<CodexRpcNotification> {
|
||||
return this.queue;
|
||||
}
|
||||
|
||||
setServerRequestHandler(_handler: CodexServerRequestHandler): void {}
|
||||
|
||||
async close(): Promise<void> {
|
||||
this.#closed = true;
|
||||
this.queue.close();
|
||||
}
|
||||
|
||||
processInfo() {
|
||||
return { pid: 7100, processGroupId: 7100, exited: this.#closed, exitCode: null, signal: null };
|
||||
}
|
||||
|
||||
async #runTurn(turnId: string): Promise<void> {
|
||||
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<void>;
|
||||
}> {
|
||||
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<void> {
|
||||
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<Phase7TurnStreamFrame<{ text: string }, { done: boolean }>> = [
|
||||
{ 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<Phase7IssueThreadSnapshot, { text: string }>();
|
||||
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/);
|
||||
});
|
||||
});
|
||||
@@ -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<Record<string, string>> = 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<View> {
|
||||
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<Payload> {
|
||||
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<View, Payload> =
|
||||
| Phase7TurnStreamViewFrame<View>
|
||||
| Phase7TurnStreamSettledFrame<Payload>
|
||||
| Phase7TurnStreamErrorFrame;
|
||||
|
||||
export function isPhase7TurnStreamTerminalFrame<View, Payload>(
|
||||
frame: Phase7TurnStreamFrame<View, Payload>,
|
||||
): frame is Phase7TurnStreamSettledFrame<Payload> | Phase7TurnStreamErrorFrame {
|
||||
return frame.type === "settled" || frame.type === "error";
|
||||
}
|
||||
|
||||
export function encodePhase7TurnStreamFrame<View, Payload>(
|
||||
frame: Phase7TurnStreamFrame<View, Payload>,
|
||||
): string {
|
||||
return `${JSON.stringify(frame)}\n`;
|
||||
}
|
||||
|
||||
export class Phase7TurnStreamError extends Error {
|
||||
constructor(
|
||||
readonly code: string,
|
||||
message: string,
|
||||
) {
|
||||
super(message);
|
||||
this.name = "Phase7TurnStreamError";
|
||||
}
|
||||
}
|
||||
|
||||
function parseFrame<View, Payload>(line: string): Phase7TurnStreamFrame<View, Payload> {
|
||||
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<Phase7TurnStreamFrame<View, Payload>>;
|
||||
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<View, Payload>;
|
||||
}
|
||||
|
||||
export interface Phase7TurnStreamDecoder<View, Payload> {
|
||||
/** Frames completed by this chunk, in wire order. */
|
||||
push(chunk: string): Array<Phase7TurnStreamFrame<View, Payload>>;
|
||||
/** Frames completed by the end of the body; also rejects a truncated tail. */
|
||||
flush(): Array<Phase7TurnStreamFrame<View, Payload>>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<View, Payload>(): Phase7TurnStreamDecoder<
|
||||
View,
|
||||
Payload
|
||||
> {
|
||||
let buffer = "";
|
||||
let lastSeq = 0;
|
||||
let terminal = false;
|
||||
|
||||
function accept(line: string): Phase7TurnStreamFrame<View, Payload> | 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<View, Payload>(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<Phase7TurnStreamFrame<View, Payload>> = [];
|
||||
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<View, Payload>(
|
||||
response: Phase7TurnStreamResponseLike,
|
||||
onFrame: (frame: Phase7TurnStreamViewFrame<View>) => void,
|
||||
): Promise<Payload> {
|
||||
if (response.body === null) {
|
||||
throw new Phase7TurnStreamError("stream_unreadable", "The turn response had no readable body.");
|
||||
}
|
||||
const decoder = createPhase7TurnStreamDecoder<View, Payload>();
|
||||
const text = new TextDecoder();
|
||||
const reader = response.body.getReader();
|
||||
let settled: Payload | undefined;
|
||||
let settledSeen = false;
|
||||
|
||||
function apply(frames: Array<Phase7TurnStreamFrame<View, Payload>>): 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;
|
||||
}
|
||||
@@ -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",
|
||||
},
|
||||
});
|
||||
Reference in new issue
Block a user