Fix/conversation timeline anchoring and dedup (#4604)

app/src/pages/Conversations.tsx

@@ -1807,7 +1807,6 @@ const Conversations = ({
             entries={selectedThreadToolTimeline}
             onViewDetails={openScopedDetail}
             onViewWholeRun={openWholeRunSource}
-            liveResponse={selectedStreamingAssistant?.content}
           />
         ) : (
           // Transcript-only turn: reasoning/narration was streamed but no tool
@@ -2560,8 +2559,7 @@ const Conversations = ({
                   replaces this via addInferenceResponse on chat_done. */}
             {selectedStreamingAssistant &&
               (selectedStreamingAssistant.thinking.length > 0 ||
-                (selectedStreamingAssistant.content.length > 0 &&
-                  (selectedThreadToolTimeline.length === 0 || hideAgentInsights))) && (
+                selectedStreamingAssistant.content.length > 0) && (
                 <div className="flex justify-start">
                   <div className="relative w-fit max-w-[75%]">
                     {selectedStreamingAssistant.thinking.length > 0 && (
@@ -2575,19 +2573,17 @@ const Conversations = ({
                         </pre>
                       </details>
                     )}
-                    {selectedStreamingAssistant.content.length > 0 &&
-                      (selectedThreadToolTimeline.length === 0 || hideAgentInsights) && (
-                        <div className="rounded-2xl rounded-bl-md px-3 py-1.5 bg-surface-strong/80 dark:bg-surface-muted text-content">
-                          <p className="text-xs text-content-secondary font-mono whitespace-pre-wrap break-words leading-snug">
-                            {selectedStreamingAssistant.content.length >
-                              STREAMING_PREVIEW_CHARS && (
-                              <span className="text-content-faint">…</span>
-                            )}
-                            {selectedStreamingAssistant.content.slice(-STREAMING_PREVIEW_CHARS)}
-                            <span className="inline-block w-1 h-3 ml-0.5 align-middle bg-primary-400 animate-pulse" />
-                          </p>
-                        </div>
-                      )}
+                    {selectedStreamingAssistant.content.length > 0 && (
+                      <div className="rounded-2xl rounded-bl-md px-3 py-1.5 bg-surface-strong/80 dark:bg-surface-muted text-content">
+                        <p className="text-xs text-content-secondary font-mono whitespace-pre-wrap break-words leading-snug">
+                          {selectedStreamingAssistant.content.length > STREAMING_PREVIEW_CHARS && (
+                            <span className="text-content-faint">…</span>
+                          )}
+                          {selectedStreamingAssistant.content.slice(-STREAMING_PREVIEW_CHARS)}
+                          <span className="inline-block w-1 h-3 ml-0.5 align-middle bg-primary-400 animate-pulse" />
+                        </p>
+                      </div>
+                    )}
                   </div>
                 </div>
               )}

app/src/pages/conversations/components/ToolTimelineBlock.tsx

@@ -353,6 +353,75 @@ function normalizeToolBody(value?: string): string | undefined {
   return value;
 }
 
+/**
+ * Whether a timeline entry carries any unique body worth its own row — a
+ * sub-agent's live activity, a returned result, a prompt/detail bubble, or a
+ * structured failure. A row with none of these renders as a bare label + status
+ * and is therefore indistinguishable from any sibling with the same title, so it
+ * is safe to coalesce (see {@link coalesceTimelineEntries}). Mirrors the
+ * `expandable` predicate in the row renderer so the two never disagree.
+ */
+function entryHasUniqueBody(entry: ToolTimelineEntry): boolean {
+  const formatted = formatTimelineEntry(entry);
+  const detailContent = normalizeToolBody(formatted.detail) ?? normalizeToolBody(entry.argsBuffer);
+  const resultContent = normalizeToolBody(entry.result);
+  return (
+    detailContent != null ||
+    resultContent != null ||
+    entry.subagent != null ||
+    entry.failure != null
+  );
+}
+
+/** A rendered timeline row: a representative entry plus how many identical,
+ * body-less entries it stands in for (`count === 1` for an ordinary row). */
+interface CoalescedRow {
+  entry: ToolTimelineEntry;
+  count: number;
+}
+
+/**
+ * Collapse runs of consecutive, identical, body-less rows into a single row
+ * carrying an `×N` count. A retry loop (e.g. the orchestrator re-spawning the
+ * integrations agent 25×, each surfacing the same "Checking your connected app"
+ * label with no distinguishing detail) would otherwise flood the timeline with
+ * indistinguishable nodes. Only truly interchangeable rows merge: same title,
+ * same status, no unique body (result/detail/sub-agent/failure), and never the
+ * live `running` row — so no information is lost, only duplication.
+ */
+export function coalesceTimelineEntries(entries: ToolTimelineEntry[]): CoalescedRow[] {
+  const rows: CoalescedRow[] = [];
+  for (const entry of entries) {
+    const mergeable = entry.status !== 'running' && !entryHasUniqueBody(entry);
+    const previous = rows[rows.length - 1];
+    if (
+      mergeable &&
+      previous != null &&
+      previous.entry.status === entry.status &&
+      !entryHasUniqueBody(previous.entry) &&
+      previous.entry.status !== 'running' &&
+      formatTimelineEntry(previous.entry).title === formatTimelineEntry(entry).title
+    ) {
+      previous.count += 1;
+      continue;
+    }
+    rows.push({ entry, count: 1 });
+  }
+  return rows;
+}
+
+/** Compact "×N" badge appended to a coalesced row's label. */
+function RepeatCount({ count }: { count: number }) {
+  if (count <= 1) return null;
+  return (
+    <span
+      className="shrink-0 rounded-full bg-surface-subtle px-1.5 py-0.5 text-[10px] font-medium text-content-muted"
+      data-testid="timeline-repeat-count">
+      ×{count}
+    </span>
+  );
+}
+
 /**
  * Neutral surface tones for an expanded row's body (worker-thread card,
  * detail bubble, code block). Per the Figma "Agentic task insights"
@@ -438,10 +507,14 @@ export function ToolTimelineBlock({
 
   // The rows + the parent's streaming response — shared by both the collapsible
   // (in-flight) and static (settled) header layouts below.
+  // Coalesce runs of identical, body-less rows (e.g. a retry loop that spawns
+  // the same integrations step 25×) into single `×N` rows before rendering.
+  const rows = coalesceTimelineEntries(entries);
+
   const body = (
     <>
       <div className="text-sm text-content-faint">
-        {entries.map((entry, index) => {
+        {rows.map(({ entry, count }, index) => {
           const formatted = formatTimelineEntry(entry);
           const detailContent =
             normalizeToolBody(formatted.detail) ?? normalizeToolBody(entry.argsBuffer);
@@ -465,7 +538,7 @@ export function ToolTimelineBlock({
             <AgentTimelineRail
               key={entry.id}
               isFirst={index === 0}
-              isLast={index === entries.length - 1}>
+              isLast={index === rows.length - 1}>
               {compact ? (
                 // Collapsed step: the whole label is the link — "Run Code →"
                 // opens the full-run panel scoped to this step. A collapsed row
@@ -481,6 +554,7 @@ export function ToolTimelineBlock({
                       className={`text-[13px] font-medium ${nameTone.replace('animate-pulse ', '')} group-hover/details:underline`}>
                       {formatted.title}
                     </span>
+                    <RepeatCount count={count} />
                     <span className="text-[13px] font-medium text-primary-600 dark:text-primary-300">
                       →
                     </span>
@@ -540,8 +614,9 @@ export function ToolTimelineBlock({
                   ) : null}
                 </details>
               ) : (
-                <div className="flex items-center">
+                <div className="flex items-center gap-1.5">
                   <span className={`text-[13px] font-medium ${nameTone}`}>{formatted.title}</span>
+                  <RepeatCount count={count} />
                 </div>
               )}
             </AgentTimelineRail>
@@ -554,40 +629,19 @@ export function ToolTimelineBlock({
 
   // The group header is a static section label — the live "working" state is
   // conveyed by the pulsing agent-name rows, so it never repeats a "Working…"
-  // string. While the run is in flight the group is collapsible; once it
-  // settles the chevron/collapse is dropped and the header renders static —
-  // matching the finished sub-agent steps, which also drop their collapse when
-  // done.
-  if (!isRunning) {
-    return (
-      <div className="mb-2 px-1 py-0" data-testid="agent-task-insights">
-        <div className="mb-1.5 flex items-center">
-          {onViewWholeRun ? (
-            // Settled: the whole title is the link — "Agentic task insights →"
-            // opens the full-run panel (matches the collapsed step rows).
-            <button
-              type="button"
-              onClick={onViewWholeRun}
-              data-testid="view-process-source"
-              className="group/insights-link flex items-center gap-1.5 text-left">
-              <span className="text-[13px] font-medium text-content-muted group-hover/insights-link:underline">
-                {t('conversations.agentTaskInsights.title')}
-              </span>
-              <span className="text-[13px] font-medium text-primary-600 dark:text-primary-300">
-                →
-              </span>
-            </button>
-          ) : (
-            titleLabel
-          )}
-        </div>
-        {body}
-      </div>
-    );
-  }
+  // string. The group is always collapsible: while the run is in flight it is
+  // open so the live activity is visible; once it settles it collapses to a
+  // single-line opener by default so a finished run (which can be dozens of
+  // steps) never dominates the conversation — the rows stay one click away, and
+  // the whole-run side panel is still reachable via the header link. The full
+  // "Agent Process Source" panel forces every row open via `expandAllRows`.
+  const open = isRunning || expandAllRows;
 
   return (
-    <details open className="group/insights mb-2 px-1 py-0" data-testid="agent-task-insights">
+    <details
+      open={open}
+      className="group/insights mb-2 px-1 py-0"
+      data-testid="agent-task-insights">
       <summary className="mb-1.5 flex cursor-pointer list-none items-center gap-1.5 select-none marker:hidden">
         {titleLabel}
         <span className="text-[11px] text-content-faint transition-transform group-open/insights:rotate-90">

app/src/pages/conversations/components/__tests__/ToolTimelineBlock.test.tsx

@@ -329,6 +329,35 @@ describe('ToolTimelineBlock — agentic task insights surface', () => {
     expect(container.querySelector('[data-testid="agent-task-insights"]')).toBeNull();
   });
 
+  it('stays open while running and collapses once settled so a finished run does not dominate', () => {
+    const running: ToolTimelineEntry[] = [
+      { id: 'r', name: 'web_search', round: 1, status: 'running' },
+    ];
+    const { rerender } = renderInStore(<ToolTimelineBlock entries={running} />);
+    // In flight → the group is open so the live activity is visible.
+    expect(screen.getByTestId('agent-task-insights')).toHaveAttribute('open');
+
+    // Settled (no running row) → collapsed by default; the rows stay in the DOM
+    // one click away, but no longer flood the conversation.
+    const settled: ToolTimelineEntry[] = [
+      { id: 'r', name: 'web_search', round: 1, status: 'success' },
+    ];
+    rerender(
+      <Provider store={store}>
+        <ToolTimelineBlock entries={settled} />
+      </Provider>
+    );
+    expect(screen.getByTestId('agent-task-insights')).not.toHaveAttribute('open');
+
+    // The side panel still forces every row open via expandAllRows.
+    rerender(
+      <Provider store={store}>
+        <ToolTimelineBlock entries={settled} expandAllRows />
+      </Provider>
+    );
+    expect(screen.getByTestId('agent-task-insights')).toHaveAttribute('open');
+  });
+
   it('renders the tool result output inside the expanded row', () => {
     const entries: ToolTimelineEntry[] = [
       {
@@ -396,6 +425,52 @@ describe('ToolTimelineBlock — agentic task insights surface', () => {
   });
 });
 
+describe('ToolTimelineBlock — coalescing repeated rows', () => {
+  it('collapses consecutive identical body-less rows into one ×N row', () => {
+    // A retry loop that spawns the same integrations step five times, each
+    // surfacing the generic "Checking your connected app" label with no
+    // distinguishing detail/result/subagent.
+    const entries: ToolTimelineEntry[] = Array.from({ length: 5 }, (_, i) => ({
+      id: `dup-${i}`,
+      name: 'integrations_agent',
+      round: 1,
+      status: 'success' as const,
+    }));
+    renderInStore(<ToolTimelineBlock entries={entries} />);
+    // Five entries render as a single rail row carrying an ×5 badge.
+    expect(screen.getAllByTestId('agent-timeline-row')).toHaveLength(1);
+    expect(screen.getByTestId('timeline-repeat-count').textContent).toBe('×5');
+  });
+
+  it('does not merge across differing status or the live running row', () => {
+    const entries: ToolTimelineEntry[] = [
+      { id: 'a', name: 'integrations_agent', round: 1, status: 'success' },
+      { id: 'b', name: 'integrations_agent', round: 1, status: 'success' },
+      // Different status breaks the run.
+      { id: 'c', name: 'integrations_agent', round: 1, status: 'error' },
+      // The live running row is never folded away.
+      { id: 'd', name: 'integrations_agent', round: 1, status: 'running' },
+    ];
+    renderInStore(<ToolTimelineBlock entries={entries} />);
+    // success×2 (merged) + error (single) + running (single) = 3 rows.
+    expect(screen.getAllByTestId('agent-timeline-row')).toHaveLength(3);
+    const counts = screen.getAllByTestId('timeline-repeat-count');
+    expect(counts).toHaveLength(1);
+    expect(counts[0].textContent).toBe('×2');
+  });
+
+  it('never merges rows that carry a unique result body', () => {
+    const entries: ToolTimelineEntry[] = [
+      { id: 'a', name: 'run_code', round: 1, status: 'success', result: 'exit 0' },
+      { id: 'b', name: 'run_code', round: 1, status: 'success', result: 'exit 1' },
+    ];
+    renderInStore(<ToolTimelineBlock entries={entries} expandAllRows />);
+    // Both keep their own row — distinct results are never coalesced.
+    expect(screen.getAllByTestId('agent-timeline-row')).toHaveLength(2);
+    expect(screen.queryByTestId('timeline-repeat-count')).toBeNull();
+  });
+});
+
 describe('ToolTimelineBlock — subagent rendering', () => {
   it('expands a subagent row even without prompt detail and shows child tool calls', () => {
     const entry: ToolTimelineEntry = {

app/src/providers/ChatRuntimeProvider.tsx

@@ -13,6 +13,7 @@ import {
   type ChatDoneEvent,
   type ChatInferenceHeartbeatEvent,
   type ChatInferenceStartEvent,
+  type ChatInterimEvent,
   type ChatIterationStartEvent,
   type ChatPlanReviewRequestEvent,
   type ChatSegmentEvent,
@@ -1040,6 +1041,39 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
           })
         );
       },
+      onInterim: (event: ChatInterimEvent) => {
+        // One interim per round — `round` is a stable per-turn dedup key that
+        // survives socket reconnect/replay (a re-delivered frame must not
+        // append the narration bubble twice).
+        const eventKey = `interim:${event.thread_id}:${event.request_id}:${event.round}`;
+        if (
+          !markChatEventSeen(eventKey, { threadId: event.thread_id, requestId: event.request_id })
+        )
+          return;
+        const content = event.full_response?.trim() ?? '';
+        if (!content) return;
+        // Persist this round's leading narration as its own interleaved bubble.
+        void dispatch(addInferenceResponse({ content, threadId: event.thread_id }));
+        // The narration has now become a bubble, so drop it from the live
+        // streaming preview (which accumulates across the whole turn under one
+        // request_id) — otherwise the same text lingers in the preview tail and
+        // reads as a duplicate for the full duration of the tool call. Reset
+        // synchronously so the next round's deltas start from an empty buffer.
+        const cr = store.getState().chatRuntime;
+        const existing = cr.streamingAssistantByThread[event.thread_id];
+        if (existing && existing.requestId === event.request_id) {
+          dispatch(
+            setStreamingAssistantForThread({
+              threadId: event.thread_id,
+              streaming: {
+                requestId: existing.requestId,
+                content: '',
+                thinking: existing.thinking,
+              },
+            })
+          );
+        }
+      },
       onTextDelta: event => {
         const cr = store.getState().chatRuntime;
         // A parallel (forked) turn streams into its own lane so it doesn't

app/src/providers/__tests__/ChatRuntimeProvider.test.tsx

@@ -989,6 +989,68 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria
       expect(streaming?.content).toBe('bbb');
     });
 
+    it('persists interim narration as a bubble and clears it from the live preview', async () => {
+      const listeners = renderProvider();
+
+      // Round-0 narration streams into the live preview…
+      act(() => {
+        listeners.onTextDelta?.({
+          thread_id: 't-interim',
+          request_id: 'r1',
+          round: 0,
+          delta: 'Let me check your calendar first.',
+        });
+      });
+      expect(store.getState().chatRuntime.streamingAssistantByThread['t-interim']?.content).toBe(
+        'Let me check your calendar first.'
+      );
+
+      // …then a tool call closes the round → interim flush.
+      act(() => {
+        listeners.onInterim?.({
+          thread_id: 't-interim',
+          request_id: 'r1',
+          round: 0,
+          full_response: 'Let me check your calendar first.',
+        });
+      });
+
+      // The narration is persisted as its own bubble…
+      await waitFor(() =>
+        expect(threadApi.appendMessage).toHaveBeenCalledWith(
+          't-interim',
+          expect.objectContaining({
+            content: 'Let me check your calendar first.',
+            sender: 'agent',
+          })
+        )
+      );
+      // …and cleared from the live preview so it isn't shown twice.
+      expect(store.getState().chatRuntime.streamingAssistantByThread['t-interim']?.content).toBe('');
+    });
+
+    it('dedupes a re-delivered interim event by round', async () => {
+      const listeners = renderProvider();
+
+      act(() => {
+        listeners.onInterim?.({
+          thread_id: 't-interim-dup',
+          request_id: 'r1',
+          round: 1,
+          full_response: 'Working on it now — pulling the data.',
+        });
+        // Reconnect/replay re-delivers the same round.
+        listeners.onInterim?.({
+          thread_id: 't-interim-dup',
+          request_id: 'r1',
+          round: 1,
+          full_response: 'Working on it now — pulling the data.',
+        });
+      });
+
+      await waitFor(() => expect(threadApi.appendMessage).toHaveBeenCalledTimes(1));
+    });
+
     it('sets inference status to thinking on inference_start and clears it on chat_done', () => {
       const listeners = renderProvider();
 

app/src/services/chatService.ts

@@ -137,6 +137,21 @@ export function segmentText(event: ChatSegmentEvent): string {
   return event.full_response;
 }
 
+/**
+ * The parent agent's leading narration for one round, flushed mid-turn when a
+ * tool/subagent call closes that round. Emitted (`chat_interim`) so the
+ * narration persists as its own interleaved chat bubble instead of vanishing
+ * when the turn settles. `round` is the 1-based iteration it belongs to and
+ * makes a stable per-turn dedup key (one interim per round).
+ */
+export interface ChatInterimEvent {
+  thread_id: string;
+  request_id: string;
+  /** Wire name is `full_response`; carries only this round's narration text. */
+  full_response: string;
+  round: number;
+}
+
 export interface ChatErrorEvent {
   thread_id: string;
   request_id?: string;
@@ -532,6 +547,7 @@ export interface ChatEventListeners {
   onSubagentTextDelta?: (event: ChatSubagentTextDeltaEvent) => void;
   onSubagentThinkingDelta?: (event: ChatSubagentThinkingDeltaEvent) => void;
   onSegment?: (event: ChatSegmentEvent) => void;
+  onInterim?: (event: ChatInterimEvent) => void;
   onTextDelta?: (event: ChatTextDeltaEvent) => void;
   onThinkingDelta?: (event: ChatThinkingDeltaEvent) => void;
   onToolArgsDelta?: (event: ChatToolArgsDeltaEvent) => void;
@@ -578,6 +594,7 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void {
     subagentTextDelta: 'subagent_text_delta',
     subagentThinkingDelta: 'subagent_thinking_delta',
     segment: 'chat_segment',
+    interim: 'chat_interim',
     textDelta: 'text_delta',
     thinkingDelta: 'thinking_delta',
     toolArgsDelta: 'tool_args_delta',
@@ -838,6 +855,22 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void {
     handlers.push([EVENTS.segment, cb]);
   }
 
+  if (listeners.onInterim) {
+    const cb = (payload: unknown) => {
+      const e = payload as ChatInterimEvent;
+      chatLog(
+        '%s thread_id=%s request_id=%s round=%d',
+        EVENTS.interim,
+        e.thread_id,
+        e.request_id,
+        e.round
+      );
+      listeners.onInterim?.(e);
+    };
+    socket.on(EVENTS.interim, cb);
+    handlers.push([EVENTS.interim, cb]);
+  }
+
   if (listeners.onTextDelta) {
     const cb = (payload: unknown) => {
       const e = payload as ChatTextDeltaEvent;

docs/plans/per-turn-tool-timeline-history.md

@@ -0,0 +1,167 @@
+# Per-turn tool-timeline history (draft)
+
+**Status:** draft / design. Not yet implemented.
+**Owners:** conversations UI + threads core.
+**Related work (already shipped on this branch):**
+
+- Coalescing repeated timeline rows + collapsing settled insights (`ToolTimelineBlock`).
+- Streamed narration kept as an in-flow bubble (`Conversations.tsx`).
+- `chat_interim` — mid-turn narration persisted as real interleaved chat messages (`progress_bridge.rs` + `ChatRuntimeProvider`).
+
+## Problem
+
+Each answer in a multi-turn thread should keep its own "Agentic task insights"
+trail (the tools/subagents that produced it). Today the trail is lost for every
+turn but the latest:
+
+- The frontend holds **one** timeline array per thread
+  (`toolTimelineByThread[threadId]`, `chatRuntimeSlice.ts:567`) and renders it
+  once, anchored after the last user message (`Conversations.tsx` — the
+  `lastUserMessageId ? agentInsights : null` anchor). Each new send wipes it
+  (`setToolTimelineForThread([])`).
+- The core persists **one** turn-state snapshot per thread, whole-file
+  overwrite — "latest snapshot wins" (`turn_state/store.rs:1-6`,
+  `snapshot_path` keyed by `hex(thread_id)` at `store.rs:210-216`). A
+  `Completed` snapshot is kept only until the next turn on the same thread
+  overwrites it (`turn_state/types.rs:26-30`, `mirror.rs:466-479`).
+
+So on reload, scrolling up shows past answers with **no** process trail, and the
+single current trail always sits at the bottom.
+
+The `chat_interim` work already fixed the narration half of this (narration is
+now a durable thread message). This design covers the remaining half: the
+**tool timeline** per turn.
+
+## Current data flow (anchors)
+
+- Snapshot type: `TurnState { thread_id, request_id, lifecycle, streaming_text,
+  thinking, tool_timeline: Vec<ToolTimelineEntry>, transcript: Vec<TranscriptItem>,
+  task_board }` — `turn_state/types.rs:291-320`. Each turn already carries a
+  unique `request_id` (`TurnState::started` / `TurnStateMirror::new`,
+  `progress_bridge.rs:290-291`).
+- Mirror: `TurnStateMirror::observe` folds `AgentProgress` into the snapshot;
+  `TurnCompleted` marks `lifecycle = Completed` and keeps the snapshot
+  (`mirror.rs:466-479`).
+- Store: `put` whole-file overwrite (`store.rs:40-73`); `get`/`delete`/`list`/
+  `clear_all`/`mark_all_interrupted` (`store.rs:76-194`); path keyed by thread
+  (`store.rs:210-216`).
+- RPC surface: `GetTurnStateRequest/Response`, `ListTurnStatesResponse`,
+  `ClearTurnStateRequest/Response` (`turn_state/types.rs:322-358`,
+  re-exported `turn_state/mod.rs:16-20`).
+- Frontend consumer: `threadApi.getTurnState(threadId)` (`threadApi.ts:125`) →
+  `hydrateRuntimeFromSnapshot` (`chatRuntimeSlice.ts:1482,1672-1681`), which
+  writes the single `toolTimelineByThread[threadId]`.
+
+## Proposed design
+
+Keep a **bounded ring of completed snapshots per thread**, keyed by
+`request_id`, plus the existing single "live/latest" snapshot. Anchor each
+completed turn's timeline to the answer message(s) that turn produced.
+
+### 1. Store: key by turn, keep the latest pointer
+
+- Change `snapshot_path` to a per-turn file:
+  `…/turn_states/<hex(thread_id)>/<request_id>.<ext>` (a directory per thread).
+- `put` writes `<request_id>.json` (atomic tmp+rename, unchanged durability).
+- Add `put_completed` / retention: on writing a `Completed` snapshot, prune the
+  thread's directory to the newest `N` completed turns (propose `N = 20`) by
+  `completed_at`, so history stays bounded (mirrors the timeline
+  `REGISTRY_SOFT_CAP` philosophy — never unbounded).
+- Keep a `latest` pointer for the in-flight/most-recent turn: either a
+  `latest.json` symlink-free copy, or resolve "latest" by scanning the dir for
+  the max `started_at`. A pointer file avoids a dir scan on the hot
+  `get_latest` path.
+- `get(thread_id)` → latest (back-compat for the current single-turn RPC).
+- New `get(thread_id, request_id)` and `list_completed(thread_id)` (metadata
+  only: `request_id`, `lifecycle`, `started_at`, `completed_at`, counts — not
+  the full `tool_timeline`, to keep the list cheap).
+- `mark_all_interrupted` (startup) and `clear`/`delete` operate over the
+  per-thread directory. `mark_all_interrupted` still skips
+  `Completed`/`Interrupted` (`store.rs:170-194`).
+
+### 2. Turn ↔ message anchoring
+
+The frontend must map each completed turn to the answer bubble(s) it produced.
+Two options — pick **B**:
+
+- **A. Anchor by user message.** Render a turn's timeline above the user message
+  that triggered it. Requires recording the triggering user `message_id` on the
+  snapshot. Simple but places the trail above the *question*, not the *answer*.
+- **B. Anchor by produced assistant message (recommended).** Stamp the assistant
+  messages a turn appends with its `request_id`. `addInferenceResponse`
+  (`threadSlice.ts:185`) and the `chat_segment` / `chat_done` / `chat_interim`
+  handlers all know `request_id`; thread it into `extraMetadata.requestId` and
+  persist it (already an open field on `ThreadMessage.extraMetadata`). Then the
+  frontend groups messages by `requestId` and renders that turn's timeline above
+  the first assistant message of the group.
+
+Option B is reload-coherent: messages already reload from the thread store with
+their `extraMetadata`, and each turn's timeline reloads from its per-turn
+snapshot — no divergence between live and reloaded views.
+
+### 3. RPC + frontend
+
+- New RPC `threads.turn_state_list` → `[{ requestId, lifecycle, startedAt,
+  completedAt, toolCount, subagentCount }]` and `threads.turn_state_get`
+  (`threadId`, `requestId`) → full `PersistedTurnState`. Keep the existing
+  `get_turn_state(threadId)` for the live turn.
+- Frontend store: replace `toolTimelineByThread: Record<threadId, Entry[]>` with
+  `toolTimelineByThread: Record<threadId, Record<requestId, Entry[]>>` plus a
+  `liveRequestIdByThread` pointer. `hydrateRuntimeFromSnapshot` writes under the
+  turn's `requestId`; `setToolTimelineForThread` targets the live `requestId`
+  and no longer wipes history on send.
+- Render loop (`Conversations.tsx`): for each assistant message that is the first
+  of its `requestId` group, render `<ToolTimelineBlock entries={byRequest[rid]}>`
+  above it (collapsed-when-settled behavior already shipped). The live turn keeps
+  its current in-flight rendering.
+- On thread open, call `turn_state_list`, then lazily `turn_state_get` a turn's
+  full timeline the first time its (collapsed) insights block is expanded — the
+  list is cheap; full timelines load on demand so opening a long thread doesn't
+  fetch dozens of full snapshots.
+
+### 4. Migration / back-compat
+
+- Old single-file snapshots (`<hex(thread_id)>.json`) are read once and migrated
+  into `<hex(thread_id)>/<request_id>.json` on first access; if `request_id` is
+  absent on a legacy snapshot, key it `legacy` and treat it as the thread's one
+  historical turn.
+- Messages without `extraMetadata.requestId` (pre-migration) fall back to the
+  current single-anchor behavior (render the live/latest timeline once), so old
+  threads degrade gracefully rather than losing their trail.
+
+### 5. Reload coherence (the invariant to protect)
+
+Live and reloaded views must render identically. Achieved because both sides key
+on `requestId`: assistant messages carry it in `extraMetadata`; timelines are
+stored/fetched per `requestId`. No in-session-only state — the failure mode this
+codebase repeatedly warns about (e.g. the `preserveLiveSubagentProse` comments)
+is avoided.
+
+## Testing plan
+
+- **Store (Rust):** per-turn put/get/list, retention prunes to `N`, latest
+  pointer resolves correctly, legacy single-file migration, `mark_all_interrupted`
+  over the per-thread dir. Extend `turn_state/store.rs` tests.
+- **Mirror (Rust):** a `Completed` snapshot is retained under its `request_id`
+  and a subsequent turn does not overwrite it.
+- **Frontend:** `hydrateRuntimeFromSnapshot` writes under `requestId`; the render
+  loop groups messages by `requestId` and renders one timeline per turn above the
+  first assistant bubble; legacy messages (no `requestId`) fall back to the single
+  anchor. Extend `Conversations.render.test.tsx` + `chatRuntimeSlice` tests.
+
+## Rollout
+
+1. Store keyed by `request_id` + retention + legacy migration (Rust, behind the
+   existing single-turn RPC — no UI change yet).
+2. Stamp assistant messages with `extraMetadata.requestId` (frontend, additive).
+3. New `turn_state_list` / `turn_state_get(requestId)` RPCs.
+4. Frontend store keyed by `requestId` + per-turn render, with legacy fallback.
+
+Each step is independently shippable; the UI only changes at step 4.
+
+## Estimated size
+
+Medium-large. Step 1 is the core risk (persistence format + migration); steps
+2–4 are mechanical but touch the render loop and the slice shape. Recommend
+landing steps 1–3 first (invisible), then step 4 behind quick manual QA on a
+multi-turn thread.

src/openhuman/channels/providers/web/progress_bridge.rs

@@ -18,6 +18,55 @@ use super::types::ChatRequestMetadata;
 /// genuine-disconnect error path (6 missed beats before the 120s window lapses).
 const INFERENCE_HEARTBEAT_SECS: u64 = 20;
 
+/// Minimum trimmed length for the parent agent's leading narration to be
+/// surfaced as its own interim chat bubble. Below this a stray "Ok." / "Sure."
+/// is left as transient streaming text rather than persisted as a message.
+const MIN_INTERIM_NARRATION_CHARS: usize = 24;
+
+/// Flush the parent agent's accumulated leading narration (streamed before a
+/// tool call in the current round) as an interim `chat_interim` event, so it
+/// persists as a chat bubble interleaved with the tool activity instead of
+/// vanishing when the turn settles. Clears `buffer` unconditionally; emits
+/// nothing for narration that is empty or too short to stand alone.
+fn flush_interim_narration(
+    buffer: &mut String,
+    round: u32,
+    client_id: &str,
+    thread_id: &str,
+    request_id: &str,
+) {
+    let text = std::mem::take(buffer);
+    let Some(narration) = interim_narration_text(&text) else {
+        return;
+    };
+    log::debug!(
+        "[web_channel][bridge] chat_interim round={} chars={} request_id={}",
+        round,
+        narration.chars().count(),
+        request_id,
+    );
+    publish_web_channel_event(WebChannelEvent {
+        event: "chat_interim".to_string(),
+        client_id: client_id.to_string(),
+        thread_id: thread_id.to_string(),
+        request_id: request_id.to_string(),
+        full_response: Some(narration),
+        round: Some(round),
+        ..Default::default()
+    });
+}
+
+/// The trimmed narration to surface as an interim bubble, or `None` when it is
+/// empty or shorter than [`MIN_INTERIM_NARRATION_CHARS`]. Pure so the threshold
+/// is unit-testable without the global event bus.
+fn interim_narration_text(buffer: &str) -> Option<String> {
+    let trimmed = buffer.trim();
+    if trimmed.chars().count() < MIN_INTERIM_NARRATION_CHARS {
+        return None;
+    }
+    Some(trimmed.to_string())
+}
+
 /// Current wall-clock time as Unix-epoch milliseconds, used to stamp tracing
 /// spans (issue #3886). Saturates to `0` if the clock is before the epoch.
 fn unix_epoch_ms() -> u64 {
@@ -283,6 +332,13 @@ pub(crate) fn spawn_progress_bridge(
         );
         let mut round: u32 = 0;
         let mut parent_max_iterations: u32 = 0;
+        // Accumulates the parent agent's streamed narration for the current
+        // round. When a tool call closes the round, this leading narration is
+        // flushed as an interim chat bubble (`chat_interim`) so it persists in
+        // the thread instead of vanishing on settle — the final answer arrives
+        // separately via `deliver_response` and is never part of this buffer
+        // (it belongs to the terminal round, which ends with no tool call).
+        let mut pending_narration = String::new();
         let mut events_seen: u64 = 0;
         let mut parent_completed = false;
         let mut parent_tool_count: u64 = 0;
@@ -556,6 +612,16 @@ pub(crate) fn spawn_progress_bridge(
                     display_label,
                     display_detail,
                 } => {
+                    // The parent's leading narration for this round is complete
+                    // once it calls a tool — flush it as an interim bubble so it
+                    // persists interleaved with the tool activity.
+                    flush_interim_narration(
+                        &mut pending_narration,
+                        iteration,
+                        &client_id,
+                        &thread_id,
+                        &request_id,
+                    );
                     parent_tool_count += 1;
                     ledger_append_event(
                         &config,
@@ -1168,6 +1234,9 @@ pub(crate) fn spawn_progress_bridge(
                     });
                 }
                 AgentProgress::TextDelta { delta, iteration } => {
+                    // Buffer the round's narration so it can be flushed as an
+                    // interim bubble if a tool call closes this round.
+                    pending_narration.push_str(&delta);
                     publish_web_channel_event(WebChannelEvent {
                         event: "text_delta".to_string(),
                         client_id: client_id.clone(),
@@ -1389,8 +1458,27 @@ pub(crate) fn spawn_progress_bridge(
 
 #[cfg(test)]
 mod tests {
+    use super::interim_narration_text;
     use super::session_profile_user_attribution;
 
+    #[test]
+    fn interim_narration_skips_empty_and_trivial() {
+        assert_eq!(interim_narration_text(""), None);
+        assert_eq!(interim_narration_text("   \n  "), None);
+        // Below the min length → left as transient streaming text.
+        assert_eq!(interim_narration_text("Ok."), None);
+        assert_eq!(interim_narration_text("Sure, one sec"), None);
+    }
+
+    #[test]
+    fn interim_narration_surfaces_and_trims_substantial_text() {
+        let text = "  Let me check your calendar for conflicts first.  ";
+        assert_eq!(
+            interim_narration_text(text),
+            Some("Let me check your calendar for conflicts first.".to_string())
+        );
+    }
+
     #[test]
     fn session_profile_attribution_none_when_signed_out() {
         let tmp = tempfile::TempDir::new().unwrap();