From 5919244b07e56714872774686f60afc0b9c792a7 Mon Sep 17 00:00:00 2001 From: worker Date: Fri, 23 Oct 2026 12:47:20 +0800 Subject: [PATCH 1/2] fix(turn): keep a resumed session stream live for a running turn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The agent-transcript integration case was red on CI about half the time: a queued message that drained while the page was reattaching never showed its answer, and the queue row it left behind never cleared. Two product defects, both about the page missing what it was not told. The runtime decided whether a resumed session stream is live or a replay from the session's persisted status. That status is an asynchronous projection of the runtime's status events, so it still read "idle" while a turn ran — for a queue-drained turn that is exactly what happens. The client had already asked whether a turn was running (`getActiveTurn`), the runtime said yes, and the stream it was then given replayed the history and closed, dropping the answer the drained turn committed a moment later. The runtime now reads the authoritative active turn as well, so the attach and the stream agree. A stream that waits for a terminal must not wait for one that can no longer arrive: the stream service records the turn of the last terminal it published, and a resume that follows that turn ends at its replay instead. The browser's event channel is a separate socket that can take seconds to open, and every event published before it opens is lost — the runtime publishes them, the page never sees them. The page therefore re-reads the two things those events would have told it when the channel becomes ready: the transcript head (a turn that ran and settled in that window is in no probe's answer) and the queue (an item that drained in it leaves a stale row). The guarantee is unchanged and strengthened: a queued message is delivered in order, and the queue empties. --- .../conversation/conversation-application.ts | 14 +- .../stream/session-stream-service.test.ts | 140 ++++++++++++++++++ .../stream/session-stream-service.ts | 41 +++++ .../client/application/create-application.ts | 11 ++ release/public-source.json | 1 + test/vitest-suites.json | 1 + 6 files changed, 207 insertions(+), 1 deletion(-) create mode 100644 packages/local-runtime-v2/src/service/session-system/stream/session-stream-service.test.ts diff --git a/packages/local-runtime-v2/src/application/conversation/conversation-application.ts b/packages/local-runtime-v2/src/application/conversation/conversation-application.ts index cce59a0c1..2c763d49d 100644 --- a/packages/local-runtime-v2/src/application/conversation/conversation-application.ts +++ b/packages/local-runtime-v2/src/application/conversation/conversation-application.ts @@ -290,6 +290,15 @@ export class ConversationApplication { `Session not found: ${req.id}`, ); } + // The active turn is the authoritative answer to "is a turn running", and + // it is the answer `getActiveTurn` already gave the client that is + // attaching. The session's persisted status is an asynchronous projection + // of the runtime's status events, so it still reads "idle" while a turn + // runs — for a queue-drained turn that is exactly what happens, and the + // resumed stream would then replay the history and close, dropping the + // answer the drained turn commits a moment later. Read both, and let the + // turn decide. + const activeTurnId = this.options.turnInspection?.activeTurnId(req.id); const frames = this.options.stream.resume({ sessionId: req.id, ...(req.afterMsgId ? { afterMsgId: req.afterMsgId } : {}), @@ -297,7 +306,9 @@ export class ConversationApplication { waitForTerminal: shouldWaitForTerminal( req.drainQueued === true || req.continuePausedQueue === true, session.status, + activeTurnId !== undefined, ), + ...(activeTurnId ? { followTurnId: activeTurnId } : {}), }); try { if (req.continuePausedQueue) { @@ -513,8 +524,9 @@ function queueContinueApplicationError( function shouldWaitForTerminal( drainQueued: boolean | undefined, status: SessionStatus, + turnRunning: boolean, ): boolean { - return drainQueued === true || status === "started"; + return drainQueued === true || status === "started" || turnRunning; } function turnContinuationStateView( diff --git a/packages/local-runtime-v2/src/service/session-system/stream/session-stream-service.test.ts b/packages/local-runtime-v2/src/service/session-system/stream/session-stream-service.test.ts new file mode 100644 index 000000000..5965716ec --- /dev/null +++ b/packages/local-runtime-v2/src/service/session-system/stream/session-stream-service.test.ts @@ -0,0 +1,140 @@ +// The resumed session stream's liveness. +// +// A resume that follows a turn is live: it must stay open until that turn +// publishes its terminal, because the turn's answer is committed after the +// resume and a stream that closed at the replay would drop it. The one case +// that has to end is the turn that is already over — it will publish no other +// terminal, so a stream that waited for one would wait forever. + +import { describe, expect, it } from "vitest"; + +import { SessionStreamService } from "./session-stream-service.js"; + +const SESSION_ID = "session-1"; + +function createService() { + return new SessionStreamService({ + messages: { listAfter: async () => ({ status: "ok", messages: [] }) }, + }); +} + +function writeTurnFrame( + service: SessionStreamService, + input: { readonly identity: string; readonly turnId: string; readonly kind?: "runtime-event" | "turn-terminal" }, +): void { + service.write({ + identity: input.identity, + sessionId: SESSION_ID, + kind: input.kind ?? "runtime-event", + data: {}, + turnId: input.turnId, + }); +} + +/** Drain the resume until it ends, and fail if it does not. */ +async function drainToEnd( + service: SessionStreamService, + input: { readonly afterMsgId: string; readonly followTurnId: string }, +): Promise { + const iterator = service.resume({ + sessionId: SESSION_ID, + afterMsgId: input.afterMsgId, + waitForTerminal: true, + followTurnId: input.followTurnId, + }); + const identities: string[] = []; + const ended = await Promise.race([ + (async () => { + for (;;) { + const step = await iterator.next(); + if (step.done) return true; + identities.push(step.value.identity); + } + })(), + new Promise((resolve) => { + const timer = setTimeout(() => resolve(false), 250); + timer.unref?.(); + }), + ]); + if (!ended) { + throw new Error("the resumed stream did not end"); + } + return identities; +} + +describe("SessionStreamService · the resumed stream's liveness", () => { + it("stays live for a turn that has not published its terminal", async () => { + const service = createService(); + writeTurnFrame(service, { identity: "frame-1", turnId: "turn-1" }); + + const iterator = service.resume({ + sessionId: SESSION_ID, + afterMsgId: "msg-0", + waitForTerminal: true, + followTurnId: "turn-1", + }); + + // The replay carries what the ring already holds. + const first = await iterator.next(); + expect(first.done).toBe(false); + expect(first.value?.identity).toBe("frame-1"); + + // The turn ends while the stream is open: the terminal closes it. + writeTurnFrame(service, { + identity: "terminal-1", + turnId: "turn-1", + kind: "turn-terminal", + }); + const second = await iterator.next(); + expect(second.done).toBe(false); + expect(second.value?.kind).toBe("turn-terminal"); + + const third = await iterator.next(); + expect(third.done).toBe(true); + }); + + it("ends at the replay for a turn that already published its terminal", async () => { + const service = createService(); + writeTurnFrame(service, { identity: "frame-1", turnId: "turn-1" }); + writeTurnFrame(service, { + identity: "terminal-1", + turnId: "turn-1", + kind: "turn-terminal", + }); + + // The follower names the turn it follows. That turn is over, so the + // stream must end rather than wait for a terminal that cannot arrive. + const identities = await drainToEnd(service, { + afterMsgId: "msg-0", + followTurnId: "turn-1", + }); + expect(identities).toEqual(["frame-1"]); + }); + + it("keeps waiting for the turn it follows while that turn runs", async () => { + const service = createService(); + writeTurnFrame(service, { identity: "frame-1", turnId: "turn-1" }); + // Turn 0 ended; turn 1 is the one being followed and it is still running. + writeTurnFrame(service, { + identity: "terminal-0", + turnId: "turn-0", + kind: "turn-terminal", + }); + + const iterator = service.resume({ + sessionId: SESSION_ID, + afterMsgId: "msg-0", + waitForTerminal: true, + followTurnId: "turn-1", + }); + const ended = await Promise.race([ + iterator.next().then((step) => step.done), + new Promise((resolve) => { + const timer = setTimeout(() => resolve(false), 250); + timer.unref?.(); + }), + ]); + expect(ended).toBe(false); + await iterator.return?.(); + }); +}); diff --git a/packages/local-runtime-v2/src/service/session-system/stream/session-stream-service.ts b/packages/local-runtime-v2/src/service/session-system/stream/session-stream-service.ts index 1d88ee962..951623fc4 100644 --- a/packages/local-runtime-v2/src/service/session-system/stream/session-stream-service.ts +++ b/packages/local-runtime-v2/src/service/session-system/stream/session-stream-service.ts @@ -42,6 +42,15 @@ export interface ResumeSessionInput { * always live-only. Defaults to true for internal callers. */ readonly waitForTerminal?: boolean; + /** + * The turn this stream follows. + * + * A turn that already published its terminal will publish no other, so a + * stream that waits for one has to end at the replay instead. The caller + * names the turn it is following so the stream can tell "this turn is still + * running" from "this turn is already over". + */ + readonly followTurnId?: string; } export interface SessionStreamReservation { @@ -60,6 +69,14 @@ interface SessionStreamState { hasFrames: boolean; reservationCount: number; deleted: boolean; + /** + * The turn of the last terminal this session published. + * + * A resume that follows that turn must end at its replay: the turn has + * published its terminal and will publish no other, so waiting for one + * would wait forever. + */ + lastTerminalTurnId?: string; } type SessionRingEntry = RingEntry; @@ -83,6 +100,11 @@ export class SessionStreamService implements SessionStreamWriter { write(input: SessionFrameInput): SessionFrameWriteResult { const state = this.state(input.sessionId); + // Recorded before the frame is published, so a resume that subscribes in + // the same tick sees the terminal as already published. + if (input.kind === 'turn-terminal' && input.turnId) { + state.lastTerminalTurnId = input.turnId; + } const value: UnsequencedSessionFrame = { identity: input.identity, sessionId: input.sessionId, @@ -246,6 +268,12 @@ class SessionResumeIterator implements AsyncIterableIterator { private async initialize(): Promise { const waitForTerminal = this.input.waitForTerminal ?? true; if (this.initializeUnanchored(waitForTerminal)) return; + // The anchored paths deliver a replay, so the stream can end at it. The + // turn this stream follows already published its terminal: it will publish + // no other, so waiting for one would wait forever. + if (waitForTerminal && this.followedTurnAlreadyTerminated()) { + this.closeAfterReplay = true; + } const initial = this.state.ring.replay(this.input.afterCursor); if (initial.status !== 'ok') { this.finishWithResync(initial.status); @@ -258,6 +286,19 @@ class SessionResumeIterator implements AsyncIterableIterator { await this.initializeDurable(initial.entries, waitForTerminal); } + /** + * Whether the turn this stream follows has already published its terminal. + * + * Read against the same state the subscription was taken from, so the two + * cannot disagree: a terminal published before the subscription is already + * recorded, and one published after it is delivered to the subscriber. + */ + private followedTurnAlreadyTerminated(): boolean { + const followTurnId = this.input.followTurnId; + if (!followTurnId) return false; + return this.state.lastTerminalTurnId === followTurnId; + } + private async initializeDurable( initial: readonly SessionRingEntry[], waitForTerminal: boolean, diff --git a/packages/webui/src/client/application/create-application.ts b/packages/webui/src/client/application/create-application.ts index 431ca06f9..8fc8a80bc 100644 --- a/packages/webui/src/client/application/create-application.ts +++ b/packages/webui/src/client/application/create-application.ts @@ -579,7 +579,18 @@ export function createWebuiApplication( dirtyResources.delete(sessionId); reconcileSession(sessionId); turnCommandsFor(sessionId).recover(); + // The queue is a projection of events this channel could have missed: + // an item that drained while the channel was down leaves a stale row + // behind, and the authoritative read is what clears it. + composerController?.refreshQueue(sessionId); } + // The transcript is the same story: the events that would have said a + // turn started were missed, and a turn that ran and settled in that + // window is in no probe's answer. The head page is the authoritative + // read of what the session holds, and the view that presents it is the + // one that reads it. + const presented = store.getSelectedSessionId(); + if (presented) void transcriptHistory.refreshPage(presented); notify(channelReadyListeners, undefined); }, // The one writer of the working tree tells the reader of it, and the link diff --git a/release/public-source.json b/release/public-source.json index 84d66b6c7..9e47e593a 100644 --- a/release/public-source.json +++ b/release/public-source.json @@ -1425,6 +1425,7 @@ "packages/local-runtime-v2/src/service/session-system/stream/index.ts", "packages/local-runtime-v2/src/service/session-system/stream/session-frame-retention.ts", "packages/local-runtime-v2/src/service/session-system/stream/session-frame.ts", + "packages/local-runtime-v2/src/service/session-system/stream/session-stream-service.test.ts", "packages/local-runtime-v2/src/service/session-system/stream/session-stream-service.ts", "packages/local-runtime-v2/src/service/session-system/usage/commit-signal.ts", "packages/local-runtime-v2/src/service/session-system/usage/index.ts", diff --git a/test/vitest-suites.json b/test/vitest-suites.json index 3b36f3dc3..abc0b90f4 100644 --- a/test/vitest-suites.json +++ b/test/vitest-suites.json @@ -200,6 +200,7 @@ "packages/tui/test/unit/tui/controller/side-session-flow.test.ts", "packages/tui/test/unit/observability.test.ts", "packages/local-runtime-v2/src/service/session-system/fork/side-history-boundary.test.ts", + "packages/local-runtime-v2/src/service/session-system/stream/session-stream-service.test.ts", "packages/tui/test/unit/host-tui-settings.test.ts", "packages/tui/test/unit/tui/features/settings/theme-picker.test.ts", "packages/tui/test/unit/tui/theme/custom-themes.test.ts", From c4c0997aa4be61ab030149aa093b23048b015da8 Mon Sep 17 00:00:00 2001 From: probe Date: Fri, 23 Oct 2026 13:41:14 +0800 Subject: [PATCH 2/2] test(webui): wait for the client's first screen on the boot deadline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The macOS CI leg of this gate failed at the first screen: `#webui-root` was still the empty mount div when the default 5s deadline expired, so the case reported a red about a shell that had not booted yet. The client boots by asking the host for its version and rendering the shell when the answer arrives, and that request carries the wire's own 30s deadline (`DEFAULT_REQUEST_TIMEOUT_MS`): past it the client renders the startup fallback instead of the shell, so no shell can arrive later than that. The wait therefore moves to the composer — the page's own input surface, which can only be on screen once the shell has rendered — and is given that same deadline. The mount node's own assertion stays. The macOS CI run of the whole 213-file webui group measured this case at 26.7s with the shell still not up when the 5s deadline expired, against 12.0s for the same case on ubuntu; the group itself measured 160.9s on macOS against 76.9s on ubuntu. The case's own 120s budget still has room for the rest of the case: about 21.6s of setup ran before that wait, and the rest of the case needs about 25s at the same ratio, so a wait that runs to its deadline still ends the case around 77s — the budget is never the thing that catches the failure. The reloaded page is a second cold start of the same client, so it waits on the same condition with the same deadline. No expectation changes: the queue drain this case is about is asserted exactly as before. --- .../test/integration/agent-transcript.test.ts | 42 +++++++++++++++++-- 1 file changed, 39 insertions(+), 3 deletions(-) diff --git a/packages/webui/test/integration/agent-transcript.test.ts b/packages/webui/test/integration/agent-transcript.test.ts index 94cbb0fad..2baeda607 100644 --- a/packages/webui/test/integration/agent-transcript.test.ts +++ b/packages/webui/test/integration/agent-transcript.test.ts @@ -13,6 +13,33 @@ const MODEL_ID = "stub-model"; const PROVIDER_NAME = "stub"; const PROVIDER_ID = `custom_provider:${PROVIDER_NAME}`; +/** + * The deadline for a wait whose completion depends on the browser client + * booting and rendering the page, not on the runtime's own work. + * + * The client boots by asking the host for its version and rendering the shell + * when the answer arrives, and that request carries the wire's own 30s + * deadline (`DEFAULT_REQUEST_TIMEOUT_MS` in `platform/wire.ts`): past it the + * client renders the startup fallback instead of the shell, so no shell can + * arrive later than this. The deadline is therefore the product's own + * tolerance — a case that waits to it and still sees no composer has learned + * that the client did not boot, which is the product's failure to report and + * not this gate's to paper over. + * + * The default 5s is shorter than that on a loaded runner. The macOS CI run of + * the whole 213-file webui group measured this file's case at 26.7s with the + * shell still not up when the default deadline expired, against 12.0s for the + * same case on ubuntu; the group itself measured 160.9s on macOS against + * 76.9s on ubuntu. 30s is also the budget the sibling real-browser gates in + * this directory already give (`browser-panel-live-navigation.test.ts`, + * `workspace-browser-tab-acceptance.test.ts`), and it leaves this case's own + * 120s budget enough for the rest of the case: about 21.6s of setup ran + * before this wait on that macOS run and the rest of the case needs about 25s + * at the same ratio, so a wait that runs to its deadline still ends the case + * around 77s — the budget is never the thing that catches the failure. + */ +const FIRST_SCREEN_TIMEOUT_MS = 30_000; + function frame(delta: Record, finishReason: string | null = null) { return `data: ${JSON.stringify({ id: "chatcmpl-agent-transcript", @@ -204,9 +231,13 @@ it("a real agent turn resumes after refresh and drains a queued message in order }); const url = `http://127.0.0.1:${info.tcpPort}/?token=${info.credential.token}#session=${sessionId}`; await page.goto(url); - await browserExpect(page.locator("#webui-root")).toBeVisible(); + // The first screen is the composer, not the mount node: `#webui-root` is an + // empty div until React commits, so waiting for it to be "visible" only + // proves the client has not finished booting. The deadline is the + // product's own handshake deadline — see `FIRST_SCREEN_TIMEOUT_MS`. const composer = page.getByPlaceholder("输入消息…（输入 / 唤起命令）"); - await browserExpect(composer).toBeVisible(); + await browserExpect(composer).toBeVisible({ timeout: FIRST_SCREEN_TIMEOUT_MS }); + await browserExpect(page.locator("#webui-root")).toBeVisible(); await composer.fill("Please reason, then answer."); await page.getByRole("button", { name: "发送" }).click(); @@ -219,8 +250,13 @@ it("a real agent turn resumes after refresh and drains a queued message in order request.body?.content === "Send this after the active answer.")).toBe(true); await browserExpect(page.locator("[data-webui-queue-item]")).toHaveCount(1); await page.reload(); + // The reload is a second cold start of the same client, so it waits on the + // same condition and with the same deadline. The transcript's first user + // message is part of that condition: it can only be on screen once the + // reloaded client has resumed the session and rendered what it read back. + await browserExpect(composer).toBeVisible({ timeout: FIRST_SCREEN_TIMEOUT_MS }); await browserExpect(page.locator("#webui-root")).toBeVisible(); - await browserExpect(page.getByText("Please reason, then answer.")).toBeVisible(); + await browserExpect(page.getByText("Please reason, then answer.")).toBeVisible({ timeout: FIRST_SCREEN_TIMEOUT_MS }); await browserExpect.poll(() => requests.some((request) => request.operation === "resumeSession")) .toBe(true); const resumeRequest = requests.find((request) => request.operation === "resumeSession");