fix(turn): keep a resumed session stream live for a running turn

packages/local-runtime-v2/src/application/conversation/conversation-application.ts

@@ -290,14 +290,25 @@ 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 } : {}),
         ...(req.afterCursor ? { afterCursor: req.afterCursor } : {}),
         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(

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<readonly string[]> {
+  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<false>((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<boolean>((resolve) => {
+        const timer = setTimeout(() => resolve(false), 250);
+        timer.unref?.();
+      }),
+    ]);
+    expect(ended).toBe(false);
+    await iterator.return?.();
+  });
+});

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<RetainedSessionFrame>;
@@ -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<SessionFrame> {
   private async initialize(): Promise<void> {
     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<SessionFrame> {
     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,

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

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

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",