CL-6380: in-flight reply survives navigation and resumes on return

apps/hub/src/index.ts

@@ -103,6 +103,7 @@ import {
 import {
   applyInsightsMigrations,
   createDrizzleRunTraceReader,
+  createDrizzleTurnTextSnapshotReader,
   createInsightsRoutes,
   createPostgresTurnLatencyStore,
   createPostgresUsageStore,
@@ -1288,6 +1289,8 @@ export async function createHub(config: HubConfig) {
     tenancy: chatTenancy,
     threads: threadStore,
     agentTurns,
+    turnTextSnapshot: (input) =>
+      createDrizzleTurnTextSnapshotReader(db).read(input),
     blockResponses: blockResponseStore,
     reactions: reactionStore,
     pins: pinStore,

packages/chat-ui/src/api.ts

@@ -1236,6 +1236,81 @@ export function patchBenchChatSettings(
   );
 }
 
+// The turn projection's read surface (CL-6329/CL-6380): what a client
+// reattaching to a workbench (page navigation, tab refocus, a dropped SSE
+// connection) uses to find whether a turn is still running and, if so,
+// replay whatever text it has already committed before the live stream's
+// tail resumes — see `GET /workbenches/:id/turns[/:turnId]` in
+// `packages/chat/src/routes.ts`.
+const AgentTurnWire = type({
+  id: "string",
+  workbenchId: "string",
+  agentAddress: "string",
+  childRunId: "string",
+  status: "'running' | 'completed' | 'failed'",
+  "replyMessageId?": "string | null",
+});
+export type AgentTurnSummary = typeof AgentTurnWire.infer;
+
+const AgentTurnsListWire = type({ items: AgentTurnWire.array() });
+
+function turnsPath(tenantId: string, workbenchId: string): string {
+  return `/api/tenants/${tenantId}/chat/workbenches/${workbenchId}/turns`;
+}
+
+export function listWorkbenchTurns(
+  tenantId: string,
+  workbenchId: string,
+): Promise<readonly AgentTurnSummary[]> {
+  return request(turnsPath(tenantId, workbenchId), AgentTurnsListWire).then(
+    (body) => body.items,
+  );
+}
+
+const AgentTurnDetailWire = AgentTurnWire.and({
+  "textSnapshot?": "string | null",
+});
+export type AgentTurnDetail = typeof AgentTurnDetailWire.infer;
+
+export function getWorkbenchTurn(
+  tenantId: string,
+  workbenchId: string,
+  turnId: string,
+): Promise<AgentTurnDetail> {
+  return request(
+    `${turnsPath(tenantId, workbenchId)}/${turnId}`,
+    AgentTurnDetailWire,
+  );
+}
+
+/**
+ * The newest still-`running` turn for `agentAddress`, or `null` if none —
+ * what a remounting workbench asks on mount to know whether to hydrate its
+ * streaming indicator immediately rather than wait for the next live event.
+ * A 404 (no turn store injected on this deployment) reads the same as "no
+ * running turn": the feature is simply unavailable, never an error the
+ * caller needs to handle.
+ */
+export async function fetchRunningTurn(
+  tenantId: string,
+  workbenchId: string,
+  agentAddress: string,
+): Promise<AgentTurnDetail | null> {
+  let turns: readonly AgentTurnSummary[];
+  try {
+    turns = await listWorkbenchTurns(tenantId, workbenchId);
+  } catch (cause) {
+    if (cause instanceof ChatApiError && cause.status === 404) return null;
+    throw cause;
+  }
+  const running = turns.find(
+    (turn) => turn.status === "running" && turn.agentAddress === agentAddress,
+  );
+  if (running === undefined) return null;
+  const detail = await getWorkbenchTurn(tenantId, workbenchId, running.id);
+  return { ...detail, textSnapshot: detail.textSnapshot ?? null };
+}
+
 /**
  * A readable name for a run, since the runs listing carries no name field:
  * the asset id's final path segment with any extension stripped, e.g.

packages/chat-ui/src/chat-workspace.tsx

@@ -28,6 +28,7 @@ import {
   workbenchesQueryKey,
   workbenchesQueryKeyPrefix,
   describeChatError,
+  fetchRunningTurn,
   inviteAgent,
   listWorkbenches,
   listInvitableDefinitions,
@@ -626,6 +627,7 @@ function ChatWorkspaceInner({
     replyTimedOut,
     handleStreamEvent: handleStreamingReplyEvent,
     noteAwaitingReply,
+    resumeFromTurn,
   } = useStreamingReply(activeWorkbenchId);
   const { activity: turnActivity, handleStreamEvent: handleTurnActivityEvent } =
     useTurnActivity(activeWorkbenchId);
@@ -790,6 +792,34 @@ function ChatWorkspaceInner({
     (participant) => isAgentAddress(participant.address),
   );
 
+  const resumeAgentAddress = (activeWorkbench?.participants ?? []).find(
+    (participant) => isAgentAddress(participant.address),
+  )?.address;
+
+  // CL-6380: a turn runs entirely server-side — this component mounting or
+  // unmounting never starts or stops it (see `useWorkbenchStream`'s own
+  // header: unmount only closes the `EventSource`, nothing server-side).
+  // So a fresh mount (first visit, or a return after navigating away while
+  // a reply was still streaming) asks once whether the agent has a turn
+  // still running and, if so, replays its committed text immediately
+  // rather than showing nothing until the next live token arrives. Any
+  // live event that beats this fetch back always wins — see
+  // `resumeFromTurn`'s own guard.
+  useEffect(() => {
+    if (activeWorkbenchId === null || resumeAgentAddress === undefined) {
+      return;
+    }
+    let cancelled = false;
+    fetchRunningTurn(tenantId, activeWorkbenchId, resumeAgentAddress)
+      .then((runningTurn) => {
+        if (!cancelled) resumeFromTurn(runningTurn);
+      })
+      .catch(() => undefined);
+    return () => {
+      cancelled = true;
+    };
+  }, [tenantId, activeWorkbenchId, resumeAgentAddress]);
+
   const { pendingSends, handleSend, retryPendingSend, discardPendingSend } =
     useOptimisticSends({
       tenantId,

packages/chat-ui/src/streaming-reply.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, test } from "bun:test";
 
 import {
+  hydrateStreamingReplyFromTurn,
   nextStreamingReplyState,
   openPendingReply,
   typingAgentNames,
@@ -209,3 +210,21 @@ describe("typingAgentNames", () => {
     expect(typingAgentNames({ text: "" }, [HUMAN])).toEqual([]);
   });
 });
+
+describe("hydrateStreamingReplyFromTurn (CL-6380: reattach snapshot)", () => {
+  test("no running turn resumes to nothing", () => {
+    expect(hydrateStreamingReplyFromTurn(null)).toBeNull();
+  });
+
+  test("a running turn with committed text opens the reply carrying it", () => {
+    expect(
+      hydrateStreamingReplyFromTurn({ textSnapshot: "streamed so far" }),
+    ).toEqual({ text: "streamed so far" });
+  });
+
+  test("a running turn with no text yet opens the same empty pending pulse as openPendingReply", () => {
+    expect(hydrateStreamingReplyFromTurn({ textSnapshot: null })).toEqual({
+      text: "",
+    });
+  });
+});

packages/chat-ui/src/streaming-reply.ts

@@ -109,6 +109,23 @@ export function openPendingReply(
   return current ?? { text: "" };
 }
 
+/**
+ * The catch-up snapshot a client reattaching mid-turn (a fresh mount after
+ * navigating away and back, CL-6380) hydrates its streaming reply with,
+ * before the live SSE tail resumes: a running turn with committed text
+ * opens the reply already carrying it; a running turn with none yet (still
+ * in its first inference call) opens the same empty pending state
+ * `openPendingReply` would; no running turn at all means there's nothing to
+ * resume. Never called once a live event has already produced state — see
+ * `resumeFromTurn`'s own guard below.
+ */
+export function hydrateStreamingReplyFromTurn(
+  runningTurn: { readonly textSnapshot?: string | null } | null,
+): StreamingReplyState {
+  if (runningTurn === null) return null;
+  return { text: runningTurn.textSnapshot ?? "" };
+}
+
 /** How long an empty pending reply may sit with no tokens before the
  * indicator clears itself — the backstop for a turn whose stream events
  * never arrive (agent down, SSE dropped mid-reconnect). */
@@ -133,6 +150,10 @@ export function useStreamingReply(
   readonly replyTimedOut: boolean;
   readonly handleStreamEvent: (eventType: string, data: unknown) => void;
   readonly noteAwaitingReply: () => void;
+  /** See `resumeFromTurn`'s own doc comment below. */
+  readonly resumeFromTurn: (
+    runningTurn: { readonly textSnapshot?: string | null } | null,
+  ) => void;
 } {
   const [streamingReply, setStreamingReply] =
     useState<StreamingReplyState>(null);
@@ -220,11 +241,32 @@ export function useStreamingReply(
     );
   }
 
+  /**
+   * Applies a fetched turn-state snapshot (see `api.ts`'s
+   * `fetchRunningTurn`) on a fresh mount, before any live event has
+   * arrived. Guarded to only ever fill an empty (`null`) state — a stream
+   * event that already opened or grew the reply always wins, since it is
+   * strictly newer than a snapshot fetched moments earlier over a separate
+   * request. A `null` turn (nothing running) is a no-op, not a reset: it
+   * must never clear a reply a fast SSE `reactor.start` already opened
+   * while the snapshot fetch was in flight.
+   */
+  function resumeFromTurn(
+    runningTurn: { readonly textSnapshot?: string | null } | null,
+  ) {
+    if (runningTurn === null) return;
+    setReplyTimedOut(false);
+    setStreamingReply(
+      (current) => current ?? hydrateStreamingReplyFromTurn(runningTurn),
+    );
+  }
+
   return {
     streamingReply,
     replyTimedOut,
     handleStreamEvent,
     noteAwaitingReply,
+    resumeFromTurn,
   };
 }
 

packages/chat-ui/test/use-streaming-reply.test.tsx

@@ -26,6 +26,9 @@ function mount(
   let send: (eventType: string, data: unknown) => void = () => {};
   let setWorkbenchId: (id: string | null) => void = () => {};
   let awaitReply: () => void = () => {};
+  let resume: (
+    runningTurn: { readonly textSnapshot: string | null } | null,
+  ) => void = () => {};
 
   function Host() {
     const [workbenchId, updateWorkbenchId] = useState(initialWorkbenchId);
@@ -35,11 +38,13 @@ function mount(
       replyTimedOut,
       handleStreamEvent,
       noteAwaitingReply,
+      resumeFromTurn,
     } = useStreamingReply(workbenchId, clearMs, minVisibleMs);
     latestState = streamingReply;
     latestTimedOut = replyTimedOut;
     send = handleStreamEvent;
     awaitReply = noteAwaitingReply;
+    resume = resumeFromTurn;
     return null;
   }
 
@@ -60,6 +65,12 @@ function mount(
       act(() => {
         awaitReply();
       }),
+    resumeFromTurn: (
+      runningTurn: { readonly textSnapshot: string | null } | null,
+    ) =>
+      act(() => {
+        resume(runningTurn);
+      }),
     settle: (ms: number) => act(() => sleep(ms)),
     get: () => latestState,
     timedOut: () => latestTimedOut,
@@ -186,6 +197,49 @@ describe("useStreamingReply's reply-timeout backstop (CL-6252 #6)", () => {
   });
 });
 
+describe("useStreamingReply.resumeFromTurn (CL-6380: catch-up on remount)", () => {
+  test("hydrates the reply from a running turn's committed text on a fresh mount", () => {
+    const harness = mount("chan_a");
+    expect(harness.get()).toBeNull();
+
+    harness.resumeFromTurn({ textSnapshot: "already streamed so far" });
+    expect(harness.get()).toEqual({ text: "already streamed so far" });
+    harness.unmount();
+  });
+
+  test("a running turn with no text yet opens the same empty pending pulse as noteAwaitingReply", () => {
+    const harness = mount("chan_a");
+    harness.resumeFromTurn({ textSnapshot: null });
+    expect(harness.get()).toEqual({ text: "" });
+    harness.unmount();
+  });
+
+  test("no running turn is a no-op, not a reset", () => {
+    const harness = mount("chan_a");
+    harness.send("chat.agent", {
+      type: "inference.start",
+      seq: 0,
+      data: { model: "x" },
+    });
+    harness.send("chat.agent", delta("hi"));
+    expect(harness.get()).toEqual({ text: "hi" });
+
+    harness.resumeFromTurn(null);
+    expect(harness.get()).toEqual({ text: "hi" });
+    harness.unmount();
+  });
+
+  test("a live event that already opened the reply wins over a slower snapshot fetch", () => {
+    const harness = mount("chan_a");
+    harness.send("chat.agent", delta("live wins"));
+    expect(harness.get()).toEqual({ text: "live wins" });
+
+    harness.resumeFromTurn({ textSnapshot: "stale snapshot" });
+    expect(harness.get()).toEqual({ text: "live wins" });
+    harness.unmount();
+  });
+});
+
 describe("useStreamingReply's typing-pulse floor", () => {
   test("a token arriving immediately still leaves the empty pulse up until minVisibleMs", async () => {
     const harness = mount("chan_a", undefined, 40);

packages/chat/src/routes.ts

@@ -209,6 +209,19 @@ export type CreateChatRoutesDeps = {
    * contract `pins` and `blockResponses` already follow.
    */
   agentTurns?: AgentTurnStore;
+  /**
+   * The visible text a still-running turn has committed to the platform's
+   * own `turn_part` rows so far (CL-6380) — what a client reattaching
+   * mid-turn replays as its catch-up snapshot before the live stream's tail
+   * resumes. Omitted, a running turn's detail simply carries no
+   * `textSnapshot` field and the client falls back to showing only the
+   * indicator until the next live event arrives — the same "no store, no
+   * feature" contract every other optional dep here follows.
+   */
+  turnTextSnapshot?: (input: {
+    readonly tenantId: string;
+    readonly runId: string;
+  }) => Promise<string | null>;
   /**
    * Poll/form response storage — see `./block-responses.ts`. Omitted
    * entirely, the response routes 404 rather than silently accepting
@@ -3651,7 +3664,17 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono<TenantEnv> {
       if (turn === undefined || turn.workbenchId !== workbenchId) {
         return c.json(ErrorEnvelope("not_found", "turn not found"), 404);
       }
-      return c.json(turn);
+      // Only a still-running turn gets a catch-up snapshot attached — a
+      // settled turn's reply already lives in the timeline as an ordinary
+      // message, and re-deriving its text here would just duplicate it.
+      const textSnapshot =
+        turn.status === "running" && deps.turnTextSnapshot !== undefined
+          ? await deps.turnTextSnapshot({
+              tenantId: access.ownerTenantId,
+              runId: turn.childRunId,
+            })
+          : null;
+      return c.json({ ...turn, textSnapshot });
     },
   );