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/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"); 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",