From f916cb1d70ca3408083ed15dcbf26a23a4ff6c03 Mon Sep 17 00:00:00 2001
From: Ramnique Singh <30795890+ramnique@users.noreply.github.com>
Date: Thu, 9 Jul 2026 14:59:22 +0530
Subject: [PATCH 1/3] refactor(x): finish turn-spine adoption for headless
 surfaces
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Background-task and live-note transcript views move from fetch-once to
the shared useAgentRunTranscript hook (useTurn over the turns:events
spine, with a one-shot legacy runs:fetch fallback for pre-migration run
ids), so an in-flight run's transcript now streams live. The headless
runner drains its execution stream — delivery rides the turn event bus,
and an unconsumed HotStream buffered every durable event until settle.
Deletes the caller-less runHeadlessTurn duplicate (HeadlessAgentRunner
is the implementation; session-design.md now says so) and the orphaned
web-search skill file that was never registered.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../renderer/src/components/bg-tasks-view.tsx | 29 ++-------
 .../src/components/live-note-sidebar.tsx      | 39 +++---------
 .../src/hooks/use-agent-run-transcript.ts     | 61 +++++++++++++++++++
 apps/x/apps/renderer/src/hooks/use-turn.ts    | 28 ++++++---
 .../renderer/src/lib/turn-follower.test.ts    | 15 +++++
 apps/x/apps/renderer/src/lib/turn-follower.ts |  8 ++-
 apps/x/packages/core/docs/session-design.md   |  5 +-
 apps/x/packages/core/src/agents/headless.ts   | 12 ++++
 .../assistant/skills/web-search/skill.ts      | 52 ----------------
 apps/x/packages/core/src/sessions/headless.ts | 41 -------------
 apps/x/packages/core/src/sessions/index.ts    |  1 -
 .../core/src/sessions/sessions.test.ts        | 21 -------
 12 files changed, 132 insertions(+), 180 deletions(-)
 create mode 100644 apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts
 delete mode 100644 apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts
 delete mode 100644 apps/x/packages/core/src/sessions/headless.ts

diff --git a/apps/x/apps/renderer/src/components/bg-tasks-view.tsx b/apps/x/apps/renderer/src/components/bg-tasks-view.tsx
index 45e23169..c0c3610f 100644
--- a/apps/x/apps/renderer/src/components/bg-tasks-view.tsx
+++ b/apps/x/apps/renderer/src/components/bg-tasks-view.tsx
@@ -15,7 +15,8 @@ import { useBackgroundTaskAgentStatus } from '@/hooks/use-bg-task-agent-status'
 import { formatRelativeTime } from '@/lib/relative-time'
 import { toast } from '@/lib/toast'
 import type { ConversationItem } from '@/lib/chat-conversation'
-import { fetchAgentRunTranscript, type AgentRunTranscript } from '@/lib/agent-transcript'
+import { fetchAgentRunTranscript } from '@/lib/agent-transcript'
+import { useAgentRunTranscript } from '@/hooks/use-agent-run-transcript'
 import { CompactConversation } from '@/components/compact-conversation'
 import { RichMarkdownViewer } from '@/components/rich-markdown-viewer'
 import { HtmlFileViewer } from '@/components/html-file-viewer'
@@ -1116,29 +1117,9 @@ function RunTranscriptView({
     isInFlight: boolean
     onBack: () => void
 }) {
-    const [transcript, setTranscript] = useState<AgentRunTranscript | null>(null)
-    const [loading, setLoading] = useState(true)
-    const [error, setError] = useState<string | null>(null)
-
-    useEffect(() => {
-        let cancelled = false
-        setLoading(true)
-        setError(null)
-        void (async () => {
-            try {
-                const t = await fetchAgentRunTranscript(runId)
-                if (cancelled) return
-                setTranscript(t)
-            } catch (err) {
-                if (cancelled) return
-                setError(err instanceof Error ? err.message : String(err))
-                setTranscript(null)
-            } finally {
-                if (!cancelled) setLoading(false)
-            }
-        })()
-        return () => { cancelled = true }
-    }, [runId])
+    // Live via the turns:events spine: an in-flight run's transcript streams
+    // in as the agent works; settled runs render from one snapshot fetch.
+    const { transcript, loading, error } = useAgentRunTranscript(runId)
 
     const summary = transcript ?? undefined
     const items: ConversationItem[] = transcript?.items ?? []
diff --git a/apps/x/apps/renderer/src/components/live-note-sidebar.tsx b/apps/x/apps/renderer/src/components/live-note-sidebar.tsx
index 18654729..f863e6cb 100644
--- a/apps/x/apps/renderer/src/components/live-note-sidebar.tsx
+++ b/apps/x/apps/renderer/src/components/live-note-sidebar.tsx
@@ -13,7 +13,7 @@ import {
 import { LiveNoteSchema, type LiveNote, type Triggers } from '@x/shared/dist/live-note.js'
 import { useLiveNoteAgentStatus } from '@/hooks/use-live-note-agent-status'
 import { formatRelativeTime } from '@/lib/relative-time'
-import { fetchAgentRunTranscript, type AgentRunTranscript } from '@/lib/agent-transcript'
+import { useAgentRunTranscript } from '@/hooks/use-agent-run-transcript'
 import { CompactConversation } from '@/components/compact-conversation'
 
 export type OpenLiveNotePanelDetail = {
@@ -659,37 +659,14 @@ function SectionRegion({ label, children }: { label?: string; children: React.Re
 }
 
 function LastRunTab({ live }: { live: LiveNote }) {
-  const [transcript, setTranscript] = useState<AgentRunTranscript | null>(null)
-  const [loadingRun, setLoadingRun] = useState(false)
-  const [fetchError, setFetchError] = useState<string | null>(null)
-
   const runId = live.lastRunId ?? null
-
-  useEffect(() => {
-    if (!runId) {
-      setTranscript(null)
-      setFetchError(null)
-      setLoadingRun(false)
-      return
-    }
-    let cancelled = false
-    setLoadingRun(true)
-    setFetchError(null)
-    void (async () => {
-      try {
-        const t = await fetchAgentRunTranscript(runId)
-        if (cancelled) return
-        setTranscript(t)
-      } catch (err) {
-        if (cancelled) return
-        setFetchError(err instanceof Error ? err.message : String(err))
-        setTranscript(null)
-      } finally {
-        if (!cancelled) setLoadingRun(false)
-      }
-    })()
-    return () => { cancelled = true }
-  }, [runId])
+  // Live via the turns:events spine: an in-flight run's transcript streams
+  // in as the agent works; settled runs render from one snapshot fetch.
+  const {
+    transcript,
+    loading: loadingRun,
+    error: fetchError,
+  } = useAgentRunTranscript(runId)
 
   if (!runId) {
     return (
diff --git a/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts
new file mode 100644
index 00000000..eaa2b6ca
--- /dev/null
+++ b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts
@@ -0,0 +1,61 @@
+import { useEffect, useMemo, useState } from 'react'
+import {
+  runToTranscript,
+  turnStateToTranscript,
+  type AgentRunTranscript,
+} from '@/lib/agent-transcript'
+import { useTurn } from '@/hooks/use-turn'
+
+export interface UseAgentRunTranscriptResult {
+  transcript: AgentRunTranscript | null
+  loading: boolean
+  error: string | null
+}
+
+// Live transcript of a headless agent run (background task, live note) by
+// run id. Run ids are turn ids since the runs→turns migration, so useTurn
+// keeps the transcript live via the turns:events spine; ids whose turn
+// snapshot definitively fails (pre-migration legacy files) fall back to one
+// runs:fetch. Callers open this for runs that already exist, so a failed
+// snapshot means "not a turn", not "not created yet" — maxRetries 0 makes
+// the legacy fallback immediate.
+export function useAgentRunTranscript(
+  runId: string | undefined | null,
+): UseAgentRunTranscriptResult {
+  const id = runId ?? undefined
+  const { state, error: turnError, snapshotFailed } = useTurn(id, { maxRetries: 0 })
+  const [legacy, setLegacy] = useState<AgentRunTranscript | null>(null)
+  const [legacyError, setLegacyError] = useState<string | null>(null)
+
+  useEffect(() => {
+    setLegacy(null)
+    setLegacyError(null)
+    if (!id || !snapshotFailed) {
+      return
+    }
+    let alive = true
+    window.ipc
+      .invoke('runs:fetch', { runId: id })
+      .then((run) => {
+        if (alive) setLegacy(runToTranscript(run))
+      })
+      .catch((err: unknown) => {
+        if (alive) setLegacyError(err instanceof Error ? err.message : String(err))
+      })
+    return () => {
+      alive = false
+    }
+  }, [id, snapshotFailed])
+
+  const transcript = useMemo(() => {
+    if (id && state) {
+      return turnStateToTranscript(id, state)
+    }
+    return legacy
+  }, [id, state, legacy])
+
+  const error = transcript ? null : (turnError ?? legacyError)
+  const loading = !!id && !transcript && !error
+
+  return { transcript, loading, error }
+}
diff --git a/apps/x/apps/renderer/src/hooks/use-turn.ts b/apps/x/apps/renderer/src/hooks/use-turn.ts
index e619fcc2..da1c5ed4 100644
--- a/apps/x/apps/renderer/src/hooks/use-turn.ts
+++ b/apps/x/apps/renderer/src/hooks/use-turn.ts
@@ -1,4 +1,4 @@
-import { useEffect, useState } from 'react'
+import { useEffect, useRef, useState } from 'react'
 import type { TurnState } from '@x/shared/src/turns.js'
 import { subscribeTurnFeed } from '@/lib/turn-feed'
 import { followTurn } from '@/lib/turn-follower'
@@ -6,6 +6,10 @@ import { followTurn } from '@/lib/turn-follower'
 export interface UseTurnResult {
   state: TurnState | null
   error: string | null
+  // Snapshot retries exhausted with no state — the id may not be a turn at
+  // all (e.g. a pre-migration legacy run id). Clears if a later feed event
+  // recovers the turn.
+  snapshotFailed: boolean
 }
 
 // Live view of one turn by id: snapshot via sessions:getTurn, then durable
@@ -14,20 +18,25 @@ export interface UseTurnResult {
 // sub-agents.
 export function useTurn(
   turnId: string | undefined,
-  opts?: { enabled?: boolean },
+  opts?: { enabled?: boolean; maxRetries?: number },
 ): UseTurnResult {
   const enabled = opts?.enabled ?? true
+  const maxRetries = opts?.maxRetries
   const [state, setState] = useState<TurnState | null>(null)
   const [error, setError] = useState<string | null>(null)
+  const [snapshotFailed, setSnapshotFailed] = useState(false)
+  const lastTurnId = useRef<string | undefined>(undefined)
 
   useEffect(() => {
-    if (!turnId) {
+    // A different turn means the previous turn's state is stale; a mere
+    // enabled toggle keeps the last rendered state while hidden.
+    if (turnId !== lastTurnId.current) {
+      lastTurnId.current = turnId
       setState(null)
       setError(null)
-      return
+      setSnapshotFailed(false)
     }
-    if (!enabled) {
-      // Keep the last rendered state while hidden; re-enabling refetches.
+    if (!turnId || !enabled) {
       return
     }
     return followTurn(turnId, {
@@ -36,10 +45,13 @@ export function useTurn(
       onState: (next) => {
         setState(next)
         setError(null)
+        setSnapshotFailed(false)
       },
       onError: (message) => setError(message),
+      onSnapshotFailed: () => setSnapshotFailed(true),
+      ...(maxRetries === undefined ? {} : { maxRetries }),
     })
-  }, [turnId, enabled])
+  }, [turnId, enabled, maxRetries])
 
-  return { state, error }
+  return { state, error, snapshotFailed }
 }
diff --git a/apps/x/apps/renderer/src/lib/turn-follower.test.ts b/apps/x/apps/renderer/src/lib/turn-follower.test.ts
index b17adbdd..b8dae0dd 100644
--- a/apps/x/apps/renderer/src/lib/turn-follower.test.ts
+++ b/apps/x/apps/renderer/src/lib/turn-follower.test.ts
@@ -42,6 +42,7 @@ function harness(fetches: Array<TEvent[] | Error>) {
   let unsubscribed = false
   const states: TurnState[] = []
   const errors: string[] = []
+  const snapshotFailures: string[] = []
   const deps: TurnFollowerDeps = {
     fetchTurn,
     subscribe: (fn) => {
@@ -53,6 +54,7 @@ function harness(fetches: Array<TEvent[] | Error>) {
     },
     onState: (state) => states.push(state),
     onError: (message) => errors.push(message),
+    onSnapshotFailed: (message) => snapshotFailures.push(message),
     maxRetries: 0,
   }
   return {
@@ -60,6 +62,7 @@ function harness(fetches: Array<TEvent[] | Error>) {
     fetchTurn,
     states,
     errors,
+    snapshotFailures,
     emit: (event: TurnBusEvent) => listener?.(event),
     get unsubscribed() {
       return unsubscribed
@@ -155,6 +158,18 @@ describe('followTurn', () => {
     detach()
   })
 
+  it('reports snapshot failure once retries are exhausted', async () => {
+    const h = harness([new Error('turn not found: legacy run id')])
+    const detach = followTurn(TURN, h.deps)
+    await flush()
+
+    // maxRetries 0: the first failure is definitive (legacy-run fallback
+    // hinges on this signal firing).
+    expect(h.snapshotFailures).toEqual(['turn not found: legacy run id'])
+    expect(h.states).toHaveLength(0)
+    detach()
+  })
+
   it('stops delivering after detach and unsubscribes from the feed', async () => {
     const full = log()
     const h = harness([full.slice(0, 2)])
diff --git a/apps/x/apps/renderer/src/lib/turn-follower.ts b/apps/x/apps/renderer/src/lib/turn-follower.ts
index 7c289263..e0a2878c 100644
--- a/apps/x/apps/renderer/src/lib/turn-follower.ts
+++ b/apps/x/apps/renderer/src/lib/turn-follower.ts
@@ -23,6 +23,10 @@ export interface TurnFollowerDeps {
   subscribe: (listener: (event: TurnBusEvent) => void) => () => void
   onState: (state: TurnState) => void
   onError: (message: string) => void
+  // Fired when snapshot retries are exhausted (the id may not be a turn at
+  // all — e.g. a pre-migration legacy run). A later feed event still retries
+  // and can recover.
+  onSnapshotFailed?: (message: string) => void
   retryDelayMs?: number
   maxRetries?: number
 }
@@ -89,7 +93,7 @@ export function followTurn(turnId: string, deps: TurnFollowerDeps): () => void {
         }
       }
       publish()
-    } catch {
+    } catch (err) {
       if (!alive) return
       // Turn file may not exist yet (subscription raced creation) or the
       // fetch failed transiently.
@@ -97,6 +101,8 @@ export function followTurn(turnId: string, deps: TurnFollowerDeps): () => void {
       if (retries < maxRetries) {
         retries += 1
         retryTimer = setTimeout(() => void fetchSnapshot(), retryDelayMs)
+      } else {
+        deps.onSnapshotFailed?.(err instanceof Error ? err.message : String(err))
       }
     } finally {
       fetching = false
diff --git a/apps/x/packages/core/docs/session-design.md b/apps/x/packages/core/docs/session-design.md
index 2694f9e9..f2ba6ccc 100644
--- a/apps/x/packages/core/docs/session-design.md
+++ b/apps/x/packages/core/docs/session-design.md
@@ -376,7 +376,10 @@ which governs mutation of live logs, not their removal.
 ## 11. Headless standalone turns
 
 A helper covers the non-session callers (background tasks, live notes,
-knowledge pipelines, scheduled agents):
+knowledge pipelines, scheduled agents). Implemented as
+`HeadlessAgentRunner` in `agents/headless.ts` (start/run handle with
+turn id, reduced state, and final assistant text); the shape below is
+the contract it fulfils:
 
 ```ts
 function runHeadlessTurn(input: {
diff --git a/apps/x/packages/core/src/agents/headless.ts b/apps/x/packages/core/src/agents/headless.ts
index 0f17eae4..0372d977 100644
--- a/apps/x/packages/core/src/agents/headless.ts
+++ b/apps/x/packages/core/src/agents/headless.ts
@@ -163,6 +163,18 @@ export class HeadlessAgentRunner implements IHeadlessAgentRunner {
         const execution = this.turnRuntime.advanceTurn(turnId, undefined, {
             signal: options.signal,
         });
+        // Drain the execution stream: live delivery rides the turn event bus,
+        // and an unconsumed HotStream would buffer every durable event in
+        // memory until the turn settles.
+        void (async () => {
+            try {
+                for await (const event of execution.events) {
+                    void event;
+                }
+            } catch {
+                // Infrastructure failures surface through the outcome.
+            }
+        })();
         const done = execution.outcome.then(async (outcome) => {
             const state = reduceTurn(
                 (await this.turnRuntime.getTurn(turnId)).events,
diff --git a/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts b/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts
deleted file mode 100644
index d3d0e6e1..00000000
--- a/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts
+++ /dev/null
@@ -1,52 +0,0 @@
-export const skill = String.raw`
-# Web Search Skill
-
-You have access to two search tools for finding information on the internet. Choose the right one based on the user's intent.
-
-## Tools
-
-### web-search (Brave Search)
-Quick, general-purpose web search. Returns titles, URLs, and short descriptions.
-
-**Best for:**
-- Quick lookups for things that change ("current price of Bitcoin", "weather in SF")
-- Current events and breaking news
-- Finding a specific website or page
-- Simple questions with direct answers
-- Checking a fact or date
-
-### research-search (Exa Search)
-Deep, research-oriented search. Returns full article text, highlights, and metadata (author, published date).
-
-**Best for:**
-- Exploring a topic in depth ("what are the latest advances in CRISPR")
-- Finding articles, blog posts, papers, and quality sources
-- Discovering companies, people, or organizations
-- Research where you need rich context, not just links
-- When the user says "research", "find articles about", "look into", "deep dive"
-
-**Category filter:** Use the category parameter when the user's intent clearly maps to one: company, research paper, news, tweet, personal site, financial report, people.
-
-## How Many Searches to Do
-
-**CRITICAL: Always start with exactly ONE search call.** Pick the single best tool (\`web-search\` or \`research-search\`) and make one request. Wait for the result before deciding if more searches are needed.
-
-**NEVER call multiple search tools simultaneously.** No parallel web-search + research-search. No firing off two web-searches at once. Always sequential: one search at a time.
-
-Only make a follow-up search if:
-- The first search returned truly uninformative or irrelevant results
-- The query has clearly distinct sub-topics that the first search couldn't cover (e.g., "compare X and Y" after getting results for X only)
-- The user explicitly asks you to dig deeper
-
-One good search is almost always enough. Default to one and stop.
-
-## Choosing Between the Two
-
-If both tools are attached, prefer:
-- \`web-search\` when the user wants a quick answer or specific link
-- \`research-search\` when the user wants to learn, explore, or gather sources
-
-If only one is attached, use whichever is available.
-`;
-
-export default skill;
diff --git a/apps/x/packages/core/src/sessions/headless.ts b/apps/x/packages/core/src/sessions/headless.ts
deleted file mode 100644
index 071d7fd4..00000000
--- a/apps/x/packages/core/src/sessions/headless.ts
+++ /dev/null
@@ -1,41 +0,0 @@
-import type { z } from "zod";
-import type { UserMessage } from "@x/shared/dist/message.js";
-import type {
-    ConversationMessage,
-    RequestedAgent,
-} from "@x/shared/dist/turns.js";
-import type { ITurnRuntime, TurnOutcome } from "../turns/api.js";
-
-// Standalone turns for non-session callers (background tasks, live notes,
-// knowledge pipelines, scheduled agents): sessionId null, automatic
-// permissions, no human. Never appears in the session index; callers keep
-// the turnId if they need history.
-export async function runHeadlessTurn(
-    turnRuntime: ITurnRuntime,
-    input: {
-        agent: z.infer<typeof RequestedAgent>;
-        context?: Array<z.infer<typeof ConversationMessage>>;
-        input: z.infer<typeof UserMessage>;
-        maxModelCalls?: number;
-        signal?: AbortSignal;
-    },
-): Promise<{ turnId: string; outcome: TurnOutcome }> {
-    const turnId = await turnRuntime.createTurn({
-        agent: input.agent,
-        sessionId: null,
-        context: input.context ?? [],
-        input: input.input,
-        config: {
-            autoPermission: true,
-            humanAvailable: false,
-            ...(input.maxModelCalls === undefined
-                ? {}
-                : { maxModelCalls: input.maxModelCalls }),
-        },
-    });
-    const execution = turnRuntime.advanceTurn(turnId, undefined, {
-        signal: input.signal,
-    });
-    const outcome = await execution.outcome;
-    return { turnId, outcome };
-}
diff --git a/apps/x/packages/core/src/sessions/index.ts b/apps/x/packages/core/src/sessions/index.ts
index df906d01..7e7e994c 100644
--- a/apps/x/packages/core/src/sessions/index.ts
+++ b/apps/x/packages/core/src/sessions/index.ts
@@ -1,7 +1,6 @@
 export * from "./api.js";
 export * from "./bus.js";
 export * from "./fs-repo.js";
-export * from "./headless.js";
 export * from "./in-memory-session-repo.js";
 export * from "./repo.js";
 export * from "./session-index.js";
diff --git a/apps/x/packages/core/src/sessions/sessions.test.ts b/apps/x/packages/core/src/sessions/sessions.test.ts
index 93fc1a40..9bca5360 100644
--- a/apps/x/packages/core/src/sessions/sessions.test.ts
+++ b/apps/x/packages/core/src/sessions/sessions.test.ts
@@ -17,7 +17,6 @@ import type {
 import { TurnInputError } from "../turns/api.js";
 import { HotStream } from "../turns/stream.js";
 import { TurnNotSettledError } from "./api.js";
-import { runHeadlessTurn } from "./headless.js";
 import { InMemorySessionRepo } from "./in-memory-session-repo.js";
 import type { ISessionRepo } from "./repo.js";
 import { SessionsImpl } from "./sessions.js";
@@ -823,26 +822,6 @@ describe("startup scan (13.6)", () => {
     });
 });
 
-describe("headless standalone turns (13.8)", () => {
-    it("runs a standalone turn with sessionId null, auto permission, no human", async () => {
-        const fake = new FakeTurnRuntime();
-        const { turnId, outcome } = await runHeadlessTurn(fake, {
-            agent: { agentId: "background-task-agent" },
-            input: user("summarize"),
-            maxModelCalls: 3,
-        });
-        expect(fake.createTurnInputs[0]).toEqual({
-            agent: { agentId: "background-task-agent" },
-            sessionId: null,
-            context: [],
-            input: user("summarize"),
-            config: { autoPermission: true, humanAvailable: false, maxModelCalls: 3 },
-        });
-        expect(outcome.status).toBe("completed");
-        expect(fake.advanceCalls).toEqual([{ turnId, input: undefined }]);
-    });
-});
-
 describe("active-skill carry-forward", () => {
     const skillTool = {
         toolId: "builtin:file-writeText",

From 876bc35e9e2494e8ad376321f4c61f91ab2bbb86 Mon Sep 17 00:00:00 2001
From: Ramnique Singh <30795890+ramnique@users.noreply.github.com>
Date: Thu, 9 Jul 2026 15:47:21 +0530
Subject: [PATCH 2/3] =?UTF-8?q?refactor(x):=20one=20delivery=20path=20for?=
 =?UTF-8?q?=20turn=20events=20=E2=80=94=20chat,=20channels,=20deltas?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Completes the turn-spine unification. Chat's SessionChatStore now
consumes turns:events like every other surface, joining live turns by
file offset (drop covered, append contiguous, refetch on gap) instead
of blind-appending session-bus events. Text/reasoning deltas cross IPC
only for turns a window subscribed to (turns:subscribe/unsubscribe, a
per-webContents registry that also survives window teardown), so
headless pipeline chatter never reaches windows that aren't watching.
The channels bridge settles turns off the turn event bus instead of
filtering the session broadcast. With no turn-event consumers left,
SessionsImpl stops forwarding entirely (it drains its execution streams
and keeps outcome/index handling) and sessions:events shrinks to
index-changed entries — one channel for turn events, one for session
metadata.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 apps/x/apps/main/src/ipc.ts                   | 46 +++++++++--
 .../src/hooks/useSessionChat.test.tsx         | 61 ++++++++++-----
 .../apps/renderer/src/hooks/useSessionChat.ts | 15 +++-
 apps/x/apps/renderer/src/hooks/useSessions.ts |  6 +-
 .../src/lib/session-chat/store.test.ts        | 64 +++++++++++++--
 .../renderer/src/lib/session-chat/store.ts    | 77 +++++++++++++++----
 apps/x/packages/core/docs/session-design.md   | 17 ++--
 .../packages/core/docs/turn-runtime-design.md |  7 +-
 .../packages/core/src/channels/bridge.test.ts | 10 +--
 apps/x/packages/core/src/channels/bridge.ts   | 10 ++-
 apps/x/packages/core/src/channels/service.ts  |  4 +-
 .../core/src/sessions/sessions.test.ts        | 18 ++---
 apps/x/packages/core/src/sessions/sessions.ts | 17 ++--
 apps/x/packages/shared/src/ipc.ts             | 18 ++++-
 apps/x/packages/shared/src/sessions.ts        | 29 +++----
 15 files changed, 284 insertions(+), 115 deletions(-)

diff --git a/apps/x/apps/main/src/ipc.ts b/apps/x/apps/main/src/ipc.ts
index 4ae5b8c4..e47d7593 100644
--- a/apps/x/apps/main/src/ipc.ts
+++ b/apps/x/apps/main/src/ipc.ts
@@ -697,9 +697,32 @@ export function startSessionsWatcher(): void {
 // runtime executes (session chat, headless background/knowledge runners,
 // spawned sub-agents), tagged with sessionId and the event's file offset so
 // consumers can join a live turn against a sessions:getTurn snapshot without
-// gaps or duplicates. Deltas are deliberately not broadcast here in v1 —
-// session chat already streams them via sessions:events; scoping deltas to
-// subscribed turns is a follow-up.
+// gaps or duplicates. Durable events are broadcast to every window;
+// text/reasoning deltas are high-volume and ephemeral, so they are sent only
+// to windows that subscribed to that turn via turns:subscribe.
+const turnDeltaSubs = new Map<Electron.WebContents, Set<string>>();
+
+export function subscribeTurnDeltas(sender: Electron.WebContents, turnId: string): void {
+  let turnIds = turnDeltaSubs.get(sender);
+  if (!turnIds) {
+    turnIds = new Set();
+    turnDeltaSubs.set(sender, turnIds);
+    sender.once('destroyed', () => turnDeltaSubs.delete(sender));
+  }
+  turnIds.add(turnId);
+}
+
+export function unsubscribeTurnDeltas(sender: Electron.WebContents, turnId: string): void {
+  const turnIds = turnDeltaSubs.get(sender);
+  if (!turnIds) {
+    return;
+  }
+  turnIds.delete(turnId);
+  if (turnIds.size === 0) {
+    turnDeltaSubs.delete(sender);
+  }
+}
+
 let turnEventsWatcher: (() => void) | null = null;
 export function startTurnEventsWatcher(): void {
   if (turnEventsWatcher) {
@@ -707,10 +730,15 @@ export function startTurnEventsWatcher(): void {
   }
   const hub = container.resolve<TurnEventHub>('turnEventBus');
   turnEventsWatcher = hub.subscribeAll((event) => {
-    if (!isDurableTurnEvent(event.event)) {
+    if (isDurableTurnEvent(event.event)) {
+      broadcastToWindows('turns:events', event);
       return;
     }
-    broadcastToWindows('turns:events', event);
+    for (const [sender, turnIds] of turnDeltaSubs) {
+      if (turnIds.has(event.turnId) && !sender.isDestroyed()) {
+        sender.send('turns:events', event);
+      }
+    }
   });
 }
 
@@ -1003,6 +1031,14 @@ export function setupIpcHandlers() {
       await container.resolve<ISessions>('sessions').deleteSession(args.sessionId);
       return { success: true };
     },
+    'turns:subscribe': async (event, args) => {
+      subscribeTurnDeltas(event.sender, args.turnId);
+      return { success: true };
+    },
+    'turns:unsubscribe': async (event, args) => {
+      unsubscribeTurnDeltas(event.sender, args.turnId);
+      return { success: true };
+    },
     'sessions:downloadLog': async (event, args) => {
       // Concatenate the session's turn logs into one JSONL for debugging.
       const sessions = container.resolve<ISessions>('sessions');
diff --git a/apps/x/apps/renderer/src/hooks/useSessionChat.test.tsx b/apps/x/apps/renderer/src/hooks/useSessionChat.test.tsx
index b1b57124..44b84881 100644
--- a/apps/x/apps/renderer/src/hooks/useSessionChat.test.tsx
+++ b/apps/x/apps/renderer/src/hooks/useSessionChat.test.tsx
@@ -1,9 +1,8 @@
 import { StrictMode } from 'react'
 import { act, renderHook, waitFor } from '@testing-library/react'
 import { describe, expect, it } from 'vitest'
-import type { SessionBusEvent } from '@x/shared/src/sessions.js'
+import type { TurnBusEvent } from '@x/shared/src/turns.js'
 import type { SessionsClient } from '@/lib/session-chat/client'
-import type { SessionFeedListener } from '@/lib/session-chat/feed'
 import {
   assistantText,
   completed,
@@ -21,8 +20,9 @@ const S1 = 'sess-1'
 
 function makeDeps() {
   const calls: Array<{ method: string; args: unknown[] }> = []
-  let emit: SessionFeedListener = () => undefined
+  let emit: (event: TurnBusEvent) => void = () => undefined
   let unsubscribed = 0
+  const deltaSubs: string[] = []
   const sessions = new Map([[S1, sessionState(S1, ['turn-1'])]])
   const turns = new Map([['turn-1', completedTurnLog('turn-1', S1, 'q1', 'a1')]])
   const client: SessionsClient = {
@@ -54,22 +54,46 @@ function makeDeps() {
   return {
     deps: {
       client,
-      subscribeFeed: (listener: SessionFeedListener) => {
+      subscribeTurnFeed: (listener: (event: TurnBusEvent) => void) => {
         emit = listener
         return () => {
           unsubscribed += 1
         }
       },
+      subscribeDeltas: (turnId: string) => {
+        deltaSubs.push(turnId)
+        return () => {
+          const i = deltaSubs.indexOf(turnId)
+          if (i >= 0) deltaSubs.splice(i, 1)
+        }
+      },
     },
     calls,
-    emit: (event: SessionBusEvent) => emit(event),
+    emit: (event: TurnBusEvent) => emit(event),
+    deltaSubs,
     getUnsubscribed: () => unsubscribed,
   }
 }
 
+function durable(
+  turnId: string,
+  event: TurnBusEvent['event'],
+  offset: number,
+): TurnBusEvent {
+  return { turnId, sessionId: S1, event, offset }
+}
+
+function delta(turnId: string, text: string): TurnBusEvent {
+  return {
+    turnId,
+    sessionId: S1,
+    event: { type: 'text_delta', turnId, modelCallIndex: 0, delta: text },
+  }
+}
+
 describe('useSessionChat', () => {
   it('seeds from the session, follows live events, and routes actions', async () => {
-    const { deps, calls, emit } = makeDeps()
+    const { deps, calls, emit, deltaSubs } = makeDeps()
     const { result } = renderHook(() => useSessionChat(S1, deps), { wrapper: StrictMode })
 
     await waitFor(() => {
@@ -78,25 +102,23 @@ describe('useSessionChat', () => {
     expect(
       result.current.chatState?.conversation.filter(isChatMessage).map((m) => m.content),
     ).toEqual(['q1', 'a1'])
+    // The window subscribed to the latest turn's deltas.
+    expect(deltaSubs).toEqual(['turn-1'])
 
-    // A new turn streams in over the feed.
+    // A new turn streams in over the turns:events spine.
     act(() => {
-      emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: created('turn-2', S1, user('q2')) })
-      emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: requested('turn-2', 0) })
-      emit({
-        kind: 'turn-event',
-        sessionId: S1,
-        turnId: 'turn-2',
-        event: { type: 'text_delta', turnId: 'turn-2', modelCallIndex: 0, delta: 'a2…' },
-      })
+      emit(durable('turn-2', created('turn-2', S1, user('q2')), 1))
+      emit(durable('turn-2', requested('turn-2', 0), 2))
+      emit(delta('turn-2', 'a2…'))
     })
     expect(result.current.latestTurnId).toBe('turn-2')
     expect(result.current.chatState?.currentAssistantMessage).toBe('a2…')
     expect(result.current.chatState?.isProcessing).toBe(true)
+    expect(deltaSubs).toEqual(['turn-2'])
 
     act(() => {
-      emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: completed('turn-2', 0, assistantText('a2')) })
-      emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: turnCompleted('turn-2', 'a2') })
+      emit(durable('turn-2', completed('turn-2', 0, assistantText('a2')), 3))
+      emit(durable('turn-2', turnCompleted('turn-2', 'a2'), 4))
     })
     expect(result.current.chatState?.isProcessing).toBe(false)
 
@@ -111,13 +133,14 @@ describe('useSessionChat', () => {
   })
 
   it('unsubscribes from the feed on unmount (StrictMode double-mounts included)', async () => {
-    const { deps, getUnsubscribed } = makeDeps()
+    const { deps, getUnsubscribed, deltaSubs } = makeDeps()
     const { unmount } = renderHook(() => useSessionChat(S1, deps), {
       wrapper: StrictMode,
     })
     unmount()
     // StrictMode's simulated cleanup plus the real unmount: every subscribe
-    // was matched by an unsubscribe.
+    // was matched by an unsubscribe, and no delta subscription leaks.
     expect(getUnsubscribed()).toBe(2)
+    expect(deltaSubs).toEqual([])
   })
 })
diff --git a/apps/x/apps/renderer/src/hooks/useSessionChat.ts b/apps/x/apps/renderer/src/hooks/useSessionChat.ts
index 6f8ee5ce..363c7300 100644
--- a/apps/x/apps/renderer/src/hooks/useSessionChat.ts
+++ b/apps/x/apps/renderer/src/hooks/useSessionChat.ts
@@ -1,11 +1,22 @@
 import { useEffect, useMemo, useState, useSyncExternalStore } from 'react'
 import { ipcSessionsClient } from '@/lib/session-chat/client'
-import { subscribeSessionFeed } from '@/lib/session-chat/feed'
+import { subscribeTurnFeed } from '@/lib/turn-feed'
 import { SessionChatStore, type SessionChatStoreDeps } from '@/lib/session-chat/store'
 
+// Declare "this window is watching turn X" so main forwards its deltas.
+// Fire-and-forget on both edges: a lost subscribe only degrades streaming
+// granularity (durable events still arrive), never correctness.
+function subscribeDeltas(turnId: string): () => void {
+  void window.ipc.invoke('turns:subscribe', { turnId }).catch(() => undefined)
+  return () => {
+    void window.ipc.invoke('turns:unsubscribe', { turnId }).catch(() => undefined)
+  }
+}
+
 const defaultDeps: SessionChatStoreDeps = {
   client: ipcSessionsClient,
-  subscribeFeed: subscribeSessionFeed,
+  subscribeTurnFeed,
+  subscribeDeltas,
 }
 
 // Thin subscription over SessionChatStore — all logic (seeding, feed events,
diff --git a/apps/x/apps/renderer/src/hooks/useSessions.ts b/apps/x/apps/renderer/src/hooks/useSessions.ts
index 9c4a7213..6c1bad1f 100644
--- a/apps/x/apps/renderer/src/hooks/useSessions.ts
+++ b/apps/x/apps/renderer/src/hooks/useSessions.ts
@@ -1,16 +1,16 @@
 import { useEffect, useMemo, useState, useSyncExternalStore } from 'react'
 import { ipcSessionsClient } from '@/lib/session-chat/client'
 import { subscribeSessionFeed } from '@/lib/session-chat/feed'
-import { SessionListStore, type SessionChatStoreDeps } from '@/lib/session-chat/store'
+import { SessionListStore, type SessionListStoreDeps } from '@/lib/session-chat/store'
 
-const defaultDeps: SessionChatStoreDeps = {
+const defaultDeps: SessionListStoreDeps = {
   client: ipcSessionsClient,
   subscribeFeed: subscribeSessionFeed,
 }
 
 // The session list (chat history sidebar): seeded from sessions:list, kept
 // current by index-changed feed events. Logic lives in SessionListStore.
-export function useSessions(deps: SessionChatStoreDeps = defaultDeps) {
+export function useSessions(deps: SessionListStoreDeps = defaultDeps) {
   const [store] = useState(() => new SessionListStore(deps))
   useEffect(() => {
     const disconnect = store.connect()
diff --git a/apps/x/apps/renderer/src/lib/session-chat/store.test.ts b/apps/x/apps/renderer/src/lib/session-chat/store.test.ts
index 10955cea..1224faa2 100644
--- a/apps/x/apps/renderer/src/lib/session-chat/store.test.ts
+++ b/apps/x/apps/renderer/src/lib/session-chat/store.test.ts
@@ -1,5 +1,6 @@
 import { describe, expect, it } from 'vitest'
-import type { SessionBusEvent, SessionState } from '@x/shared/src/sessions.js'
+import type { SessionState } from '@x/shared/src/sessions.js'
+import { isDurableTurnEvent, type TurnBusEvent } from '@x/shared/src/turns.js'
 import { isChatMessage } from '@/lib/chat-conversation'
 import type { SessionsClient } from './client'
 import type { SessionFeedListener } from './feed'
@@ -86,12 +87,16 @@ class FakeClient implements SessionsClient {
 
 function makeStore() {
   const client = new FakeClient()
-  let emit: SessionFeedListener = () => undefined
+  let emit: (event: TurnBusEvent) => void = () => undefined
   let subscribed = 0
   let unsubscribed = 0
+  // Live delta subscriptions by turn id (multiset semantics not needed: the
+  // store holds at most one).
+  const deltaSubs: string[] = []
+  offsetByTurn.clear()
   const store = new SessionChatStore({
     client,
-    subscribeFeed: (listener) => {
+    subscribeTurnFeed: (listener) => {
       subscribed += 1
       emit = listener
       return () => {
@@ -99,26 +104,44 @@ function makeStore() {
         emit = () => undefined
       }
     },
+    subscribeDeltas: (turnId) => {
+      deltaSubs.push(turnId)
+      return () => {
+        const i = deltaSubs.indexOf(turnId)
+        if (i >= 0) deltaSubs.splice(i, 1)
+      }
+    },
   })
   const disconnect = store.connect()
   return {
     client,
     store,
     disconnect,
-    emit: (event: SessionBusEvent) => emit(event),
+    emit: (event: TurnBusEvent) => emit(event),
+    deltaSubs,
     getSubscribed: () => subscribed,
     getUnsubscribed: () => unsubscribed,
   }
 }
 
+// Tags an event for the turn feed; durable events get their 1-based per-turn
+// offset assigned automatically (mirroring the file line index the runtime
+// stamps on the bus envelope).
+const offsetByTurn = new Map<string, number>()
+
 function turnEvent(
   sessionId: string,
   turnId: string,
   event:
     | TEvent
     | { type: 'text_delta' | 'reasoning_delta'; turnId: string; modelCallIndex: number; delta: string },
-): SessionBusEvent {
-  return { kind: 'turn-event', sessionId, turnId, event }
+): TurnBusEvent {
+  if (!isDurableTurnEvent(event)) {
+    return { turnId, sessionId, event }
+  }
+  const offset = (offsetByTurn.get(turnId) ?? 0) + 1
+  offsetByTurn.set(turnId, offset)
+  return { turnId, sessionId, event, offset }
 }
 
 describe('SessionChatStore', () => {
@@ -203,6 +226,35 @@ describe('SessionChatStore', () => {
     ).toEqual(['q1', 'a1', 'q2'])
   })
 
+  it('keeps exactly one delta subscription, following the latest turn', async () => {
+    const { client, store, emit, deltaSubs, disconnect } = makeStore()
+    client.sessions.set(S1, sessionState(S1, ['turn-1']))
+    client.turns.set('turn-1', completedTurnLog('turn-1', S1, 'q1', 'a1'))
+    await store.setSession(S1)
+    expect(deltaSubs).toEqual(['turn-1'])
+
+    emit(turnEvent(S1, 'turn-2', created('turn-2', S1, user('q2'))))
+    expect(deltaSubs).toEqual(['turn-2'])
+
+    disconnect()
+    expect(deltaSubs).toEqual([])
+  })
+
+  it('drops duplicate feed events already covered by the snapshot', async () => {
+    const { client, store, emit } = makeStore()
+    client.sessions.set(S1, sessionState(S1, []))
+    await store.setSession(S1)
+    emit(turnEvent(S1, 'turn-1', created('turn-1', S1, user('go'))))
+    emit(turnEvent(S1, 'turn-1', requested('turn-1', 0)))
+    // A replay of line 2 (e.g. snapshot/live race) must not double-apply —
+    // the reducer would throw on the impossible history.
+    emit({ turnId: 'turn-1', sessionId: S1, event: requested('turn-1', 0), offset: 2 })
+    emit(turnEvent(S1, 'turn-1', completed('turn-1', 0, assistantText('done'))))
+    emit(turnEvent(S1, 'turn-1', turnCompleted('turn-1')))
+    expect(store.getSnapshot().error).toBeNull()
+    expect(store.getSnapshot().chatState?.isProcessing).toBe(false)
+  })
+
   it('ignores events for other sessions', async () => {
     const { client, store, emit } = makeStore()
     client.sessions.set(S1, sessionState(S1, []))
diff --git a/apps/x/apps/renderer/src/lib/session-chat/store.ts b/apps/x/apps/renderer/src/lib/session-chat/store.ts
index 2bd0b4b4..5b94b0da 100644
--- a/apps/x/apps/renderer/src/lib/session-chat/store.ts
+++ b/apps/x/apps/renderer/src/lib/session-chat/store.ts
@@ -2,11 +2,12 @@ import type { z } from 'zod'
 import type { UserMessage } from '@x/shared/src/message.js'
 import type { SessionBusEvent, SessionIndexEntry } from '@x/shared/src/sessions.js'
 import {
+  isDurableTurnEvent,
   reduceTurn,
   type JsonValue,
+  type TurnBusEvent,
   type TurnEvent,
   type TurnState,
-  type TurnStreamEvent,
 } from '@x/shared/src/turns.js'
 import type { SendMessageConfig, SessionsClient } from './client'
 import type { SessionFeedListener } from './feed'
@@ -30,7 +31,12 @@ export interface SessionChatSnapshot {
 
 export interface SessionChatStoreDeps {
   client: SessionsClient
-  subscribeFeed: (listener: SessionFeedListener) => () => void
+  // The turns:events spine (durable events with offsets, plus deltas for
+  // turns this window subscribed to).
+  subscribeTurnFeed: (listener: (event: TurnBusEvent) => void) => () => void
+  // Declares "this window is watching turn X" so main forwards its deltas;
+  // returns the unsubscribe.
+  subscribeDeltas: (turnId: string) => () => void
 }
 
 // Framework-agnostic controller for one active session's chat. Owns all the
@@ -39,7 +45,8 @@ export interface SessionChatStoreDeps {
 // hook is a thin useSyncExternalStore subscription over it.
 export class SessionChatStore {
   private readonly client: SessionsClient
-  private readonly subscribeFeed: (listener: SessionFeedListener) => () => void
+  private readonly subscribeTurnFeed: (listener: (event: TurnBusEvent) => void) => () => void
+  private readonly subscribeDeltas: (turnId: string) => () => void
   private feedDisconnect: (() => void) | null = null
   private readonly listeners = new Set<() => void>()
 
@@ -53,6 +60,9 @@ export class SessionChatStore {
   private error: string | null = null
   // Guards stale async loads after a session switch.
   private generation = 0
+  // The turn whose deltas this window currently receives.
+  private deltaTurnId: string | null = null
+  private deltaUnsub: (() => void) | null = null
 
   private snapshot: SessionChatSnapshot = {
     sessionId: null,
@@ -64,7 +74,8 @@ export class SessionChatStore {
 
   constructor(deps: SessionChatStoreDeps) {
     this.client = deps.client
-    this.subscribeFeed = deps.subscribeFeed
+    this.subscribeTurnFeed = deps.subscribeTurnFeed
+    this.subscribeDeltas = deps.subscribeDeltas
   }
 
   // Feed attachment is effect-managed and idempotent so React StrictMode's
@@ -72,11 +83,28 @@ export class SessionChatStore {
   // subscription would be torn down by the first cleanup and never restored).
   connect(): () => void {
     if (!this.feedDisconnect) {
-      this.feedDisconnect = this.subscribeFeed(this.onFeedEvent)
+      this.feedDisconnect = this.subscribeTurnFeed(this.onTurnEvent)
+      this.syncDeltas()
     }
     return () => {
       this.feedDisconnect?.()
       this.feedDisconnect = null
+      this.syncDeltas()
+    }
+  }
+
+  // Keep exactly one delta subscription: the latest turn, while connected.
+  private syncDeltas(): void {
+    const want =
+      this.feedDisconnect && this.latestEvents
+        ? this.latestEvents[0].turnId
+        : null
+    if (want === this.deltaTurnId) return
+    this.deltaUnsub?.()
+    this.deltaUnsub = null
+    this.deltaTurnId = want
+    if (want) {
+      this.deltaUnsub = this.subscribeDeltas(want)
     }
   }
 
@@ -99,6 +127,7 @@ export class SessionChatStore {
     this.overlay = emptyOverlay()
     this.error = null
     this.loading = sessionId !== null
+    this.syncDeltas()
     this.emit()
     if (sessionId === null) return
 
@@ -112,6 +141,7 @@ export class SessionChatStore {
       this.priorTurns = reduced.slice(0, -1)
       this.latestEvents = turns.length > 0 ? turns[turns.length - 1].events : null
       this.loading = false
+      this.syncDeltas()
       this.emit()
     } catch (error) {
       if (generation !== this.generation) return
@@ -122,23 +152,37 @@ export class SessionChatStore {
     }
   }
 
-  private onFeedEvent: SessionFeedListener = (event: SessionBusEvent) => {
-    if (event.kind !== 'turn-event' || event.sessionId !== this.sessionId) return
+  private onTurnEvent = (event: TurnBusEvent): void => {
+    if (this.sessionId === null || event.sessionId !== this.sessionId) return
     const turnEvent = event.event
-    if (isDurable(turnEvent)) {
+    if (isDurableTurnEvent(turnEvent) && event.offset !== undefined) {
       if (turnEvent.type === 'turn_created') {
         // A new turn started for this session: freeze the previous latest.
         this.freezeLatest()
         this.latestEvents = [turnEvent]
         this.overlay = emptyOverlay()
-      } else if (this.latestEvents && this.latestEvents[0].turnId === turnEvent.turnId) {
-        this.latestEvents.push(turnEvent)
+        this.syncDeltas()
+      } else if (this.latestEvents && this.latestEvents[0].turnId === event.turnId) {
+        // Offset join against the local log: drop already-known events,
+        // append the contiguous next line, refetch on a gap.
+        if (event.offset <= this.latestEvents.length) {
+          return
+        }
+        if (event.offset === this.latestEvents.length + 1) {
+          this.latestEvents.push(turnEvent)
+        } else {
+          void this.reloadTurn(event.turnId)
+          return
+        }
       } else {
         // An event for a turn we haven't seen (missed turn_created, e.g. the
         // feed attached mid-turn): reconcile by refetching that turn.
         void this.reloadTurn(event.turnId)
         return
       }
+    } else if (!this.latestEvents || this.latestEvents[0].turnId !== event.turnId) {
+      // Deltas are ephemeral: only the latest turn's deltas paint the overlay.
+      return
     }
     this.overlay = applyOverlay(this.overlay, turnEvent)
     this.emit()
@@ -164,6 +208,7 @@ export class SessionChatStore {
         this.freezeLatest()
       }
       this.latestEvents = turn.events
+      this.syncDeltas()
       this.emit()
     } catch (error) {
       // The next snapshot-worthy event will retry.
@@ -237,10 +282,6 @@ export class SessionChatStore {
   }
 }
 
-function isDurable(event: TurnStreamEvent): event is TEvent {
-  return event.type !== 'text_delta' && event.type !== 'reasoning_delta'
-}
-
 // ---------------------------------------------------------------------------
 // Session list store
 // ---------------------------------------------------------------------------
@@ -250,6 +291,12 @@ export interface SessionListSnapshot {
   loading: boolean
 }
 
+export interface SessionListStoreDeps {
+  client: SessionsClient
+  // sessions:events — index-changed entries for the session list.
+  subscribeFeed: (listener: SessionFeedListener) => () => void
+}
+
 export class SessionListStore {
   private readonly client: SessionsClient
   private readonly subscribeFeed: (listener: SessionFeedListener) => () => void
@@ -259,7 +306,7 @@ export class SessionListStore {
   private loading = true
   private snapshot: SessionListSnapshot = { sessions: [], loading: true }
 
-  constructor(deps: SessionChatStoreDeps) {
+  constructor(deps: SessionListStoreDeps) {
     this.client = deps.client
     this.subscribeFeed = deps.subscribeFeed
   }
diff --git a/apps/x/packages/core/docs/session-design.md b/apps/x/packages/core/docs/session-design.md
index f2ba6ccc..ab4f122d 100644
--- a/apps/x/packages/core/docs/session-design.md
+++ b/apps/x/packages/core/docs/session-design.md
@@ -357,18 +357,23 @@ which governs mutation of live logs, not their removal.
 
 ## 10. Event forwarding and live UI
 
-1. For every `advanceTurn` it initiates, the session layer consumes
-   `TurnExecution.events` and forwards each `TurnStreamEvent` — tagged with
-   `sessionId` — through the application bus to renderer windows over one
-   IPC channel (`sessions:events`).
+1. Live turn delivery is not the session layer's job: the turn runtime
+   publishes every turn's events to the process-wide turn event bus
+   (turn-runtime-design.md §17.1), which the app layer bridges to renderer
+   windows over one IPC channel (`turns:events` — durable events broadcast
+   with their file offsets; deltas only to windows subscribed to that turn).
+   The session layer drains each `TurnExecution.events` it initiates so an
+   unconsumed stream never buffers, and `sessions:events` carries only
+   `session-index-changed` entries.
 2. When `outcome` settles, the session layer updates the index entry and
    publishes `session-index-changed`.
 3. The renderer follows the turn spec's historical/live pattern: fetch
    turns via `getTurn`, run the shared `reduceTurn` per turn, compose the
    session timeline turn-by-turn (each turn renders its input and its own
    activity; the referenced prefix is never re-rendered from context),
-   append live durable events and re-reduce, and keep text/reasoning deltas
-   in an ephemeral overlay cleared by canonical responses.
+   join live durable events by file offset (drop covered, append
+   contiguous, refetch on gap) and re-reduce, and keep text/reasoning
+   deltas in an ephemeral overlay cleared by canonical responses.
 4. Pending approvals and ask-human prompts render from the suspended turn's
    reduced state, so they survive restarts without any session-layer
    bookkeeping.
diff --git a/apps/x/packages/core/docs/turn-runtime-design.md b/apps/x/packages/core/docs/turn-runtime-design.md
index 5a4139c1..c3ef6be9 100644
--- a/apps/x/packages/core/docs/turn-runtime-design.md
+++ b/apps/x/packages/core/docs/turn-runtime-design.md
@@ -1435,9 +1435,10 @@ stream for the initiating caller.
   events with `offset <= snapshot.length` — no gaps, no duplicates, no
   sequence numbers in the durable schema.
 - Text/reasoning deltas are published without an offset (they are not
-  durable). The app-layer IPC bridge (`turns:events`) forwards durable
-  events only in v1; deltas still reach session chat via the session-layer
-  forwarding. Scoping delta delivery to subscribed turns is future work.
+  durable). The app-layer IPC bridge (`turns:events`) broadcasts durable
+  events to every window; deltas are forwarded only to windows that
+  declared they are watching that turn (`turns:subscribe` /
+  `turns:unsubscribe`, a per-webContents registry in the app layer).
 - The bus is ephemeral and observational, like the lifecycle bus: listener
   errors are swallowed, nothing durable depends on delivery, and a crash
   losing listeners accurately reflects that no execution is known active.
diff --git a/apps/x/packages/core/src/channels/bridge.test.ts b/apps/x/packages/core/src/channels/bridge.test.ts
index 2c78f90a..c2b3adb7 100644
--- a/apps/x/packages/core/src/channels/bridge.test.ts
+++ b/apps/x/packages/core/src/channels/bridge.test.ts
@@ -1,7 +1,7 @@
 import { describe, expect, it, vi } from "vitest";
 import type { SessionIndexEntry } from "@x/shared/dist/sessions.js";
 import type { TurnStreamEvent } from "@x/shared/dist/turns.js";
-import { EmitterSessionBus } from "../sessions/bus.js";
+import { TurnEventHub } from "../turns/event-hub.js";
 import { TurnInputError } from "../turns/api.js";
 import type { ISessions } from "../sessions/api.js";
 import { ChannelBridge, type ModelChoice } from "./bridge.js";
@@ -50,7 +50,7 @@ function askEvent(turnId: string, question: string, options?: string[]): TurnStr
 
 interface Harness {
     bridge: ChannelBridge;
-    bus: EmitterSessionBus;
+    bus: TurnEventHub;
     replies: string[];
     reply: (text: string) => Promise<void>;
     sessions: {
@@ -72,9 +72,9 @@ const MODELS: ModelChoice[] = [
 ];
 
 function harness(entries: SessionIndexEntry[] = []): Harness {
-    const bus = new EmitterSessionBus();
+    const bus = new TurnEventHub();
     const publish = (turnId: string, event: TurnStreamEvent) =>
-        bus.publish({ kind: "turn-event", sessionId: "s1", turnId, event });
+        bus.publish({ turnId, sessionId: "s1", event });
     const sessions = {
         createSession: vi.fn(async () => "s1"),
         sendMessage: vi.fn(async () => ({ turnId: "t1" })),
@@ -86,7 +86,7 @@ function harness(entries: SessionIndexEntry[] = []): Harness {
     const listModels = vi.fn(async () => MODELS);
     const bridge = new ChannelBridge({
         sessions: sessions as unknown as ISessions,
-        sessionBus: bus,
+        turnEventBus: bus,
         listModels,
     });
     const replies: string[] = [];
diff --git a/apps/x/packages/core/src/channels/bridge.ts b/apps/x/packages/core/src/channels/bridge.ts
index e53be4d8..69e84e9e 100644
--- a/apps/x/packages/core/src/channels/bridge.ts
+++ b/apps/x/packages/core/src/channels/bridge.ts
@@ -4,7 +4,7 @@ import { assistantText, lastAssistantText } from "../agents/headless.js";
 import { TurnInputError } from "../turns/api.js";
 import { ASK_HUMAN_TOOL } from "../turns/bridges/real-agent-resolver.js";
 import { TurnNotSettledError, type ISessions } from "../sessions/api.js";
-import type { EmitterSessionBus } from "../sessions/bus.js";
+import type { TurnEventHub } from "../turns/event-hub.js";
 
 // Transport-agnostic command layer: inbound texts from a messaging channel
 // are parsed into commands (list / resume / new / stop / status) or forwarded
@@ -136,7 +136,7 @@ export class ChannelBridge {
     constructor(
         private readonly deps: {
             sessions: ISessions;
-            sessionBus: EmitterSessionBus;
+            turnEventBus: TurnEventHub;
             listModels: () => Promise<ModelChoice[]>;
         },
     ) {}
@@ -502,8 +502,10 @@ export class ChannelBridge {
         const buffered: Array<{ turnId: string; settled: Settled }> = [];
         let waiter: { turnId: string; resolve: (settled: Settled) => void } | null = null;
         let cancelTimer: (() => void) | null = null;
-        const unsubscribe = this.deps.sessionBus.subscribe((event) => {
-            if (event.kind !== "turn-event") return;
+        // The turn event spine carries every turn's events (deltas included
+        // for subscribed windows, but settleOf ignores those); subscribeAll
+        // because the turnId is unknown until sendMessage returns.
+        const unsubscribe = this.deps.turnEventBus.subscribeAll((event) => {
             const settled = settleOf(event.event);
             if (!settled) return;
             if (waiter) {
diff --git a/apps/x/packages/core/src/channels/service.ts b/apps/x/packages/core/src/channels/service.ts
index 097d415a..a0bed3f4 100644
--- a/apps/x/packages/core/src/channels/service.ts
+++ b/apps/x/packages/core/src/channels/service.ts
@@ -5,7 +5,7 @@ import type { ChannelsConfig, ChannelsStatus } from "@x/shared/dist/channels.js"
 import container from "../di/container.js";
 import { WorkDir } from "../config/config.js";
 import type { ISessions } from "../sessions/api.js";
-import type { EmitterSessionBus } from "../sessions/bus.js";
+import type { TurnEventHub } from "../turns/event-hub.js";
 import { isSignedIn } from "../account/account.js";
 import { listGatewayModels } from "../models/gateway.js";
 import { listOnboardingModels } from "../models/models-dev.js";
@@ -98,7 +98,7 @@ function ensureBridge(): ChannelBridge {
     if (!bridge) {
         bridge = new ChannelBridge({
             sessions: container.resolve<ISessions>("sessions"),
-            sessionBus: container.resolve<EmitterSessionBus>("sessionBus"),
+            turnEventBus: container.resolve<TurnEventHub>("turnEventBus"),
             listModels: listBridgeModels,
         });
     }
diff --git a/apps/x/packages/core/src/sessions/sessions.test.ts b/apps/x/packages/core/src/sessions/sessions.test.ts
index 9bca5360..e1252a43 100644
--- a/apps/x/packages/core/src/sessions/sessions.test.ts
+++ b/apps/x/packages/core/src/sessions/sessions.test.ts
@@ -644,7 +644,10 @@ describe("stopTurn and resumeTurn", () => {
 });
 
 describe("event forwarding and index maintenance (13.6, 13.7)", () => {
-    it("forwards stream events to the bus tagged with sessionId in order", async () => {
+    it("drains the execution stream and publishes only index updates", async () => {
+        // Live turn delivery is the turn event bus's job (published by the
+        // runtime itself); the session bus carries index changes only. The
+        // stream must still be consumed so it never buffers until settle.
         const { sessions, fake, bus } = makeSessions();
         const streamed: TurnStreamEvent[] = [
             { type: "text_delta", turnId: "x", modelCallIndex: 0, delta: "he" },
@@ -652,19 +655,12 @@ describe("event forwarding and index maintenance (13.6, 13.7)", () => {
         ];
         fake.script = () => ({ events: streamed, outcome: completedOutcome() });
         const sessionId = await sessions.createSession();
-        const { turnId } = await sessions.sendMessage(sessionId, user("go"), {
+        await sessions.sendMessage(sessionId, user("go"), {
             agent: { agentId: "copilot" },
         });
         await flush();
-        const turnEvents = bus.events.filter((e) => e.kind === "turn-event");
-        expect(turnEvents).toEqual(
-            streamed.map((event) => ({
-                kind: "turn-event",
-                sessionId,
-                turnId,
-                event,
-            })),
-        );
+        expect(bus.events.length).toBeGreaterThan(0);
+        expect(bus.events.every((e) => e.kind === "index-changed")).toBe(true);
     });
 
     it("outcome settlement updates the index entry's latest turn status", async () => {
diff --git a/apps/x/packages/core/src/sessions/sessions.ts b/apps/x/packages/core/src/sessions/sessions.ts
index acd9d7ef..d42640b1 100644
--- a/apps/x/packages/core/src/sessions/sessions.ts
+++ b/apps/x/packages/core/src/sessions/sessions.ts
@@ -343,9 +343,11 @@ export class SessionsImpl implements ISessions {
         await execution.outcome;
     }
 
-    // Every advance this layer initiates: forward its stream to the bus
-    // tagged with sessionId, keep the abort controller for stopTurn, and
-    // update the index entry when the outcome settles.
+    // Every advance this layer initiates: keep the abort controller for
+    // stopTurn and update the index entry when the outcome settles. Live
+    // event delivery is not this layer's job — the runtime publishes every
+    // event to the turn event bus; the execution stream is drained so an
+    // unconsumed HotStream never buffers events until settle.
     private startTrackedAdvance(
         sessionId: string | null,
         turnId: string,
@@ -368,14 +370,7 @@ export class SessionsImpl implements ISessions {
         void (async () => {
             try {
                 for await (const event of execution.events) {
-                    if (sessionId !== null) {
-                        this.sessionBus.publish({
-                            kind: "turn-event",
-                            sessionId,
-                            turnId,
-                            event,
-                        });
-                    }
+                    void event;
                 }
             } catch {
                 // Infrastructure failures surface through the outcome.
diff --git a/apps/x/packages/shared/src/ipc.ts b/apps/x/packages/shared/src/ipc.ts
index cf6dd880..c00220bc 100644
--- a/apps/x/packages/shared/src/ipc.ts
+++ b/apps/x/packages/shared/src/ipc.ts
@@ -461,7 +461,7 @@ const ipcSchemas = {
   },
   // ── New runtime: sessions + turns (session-design.md) ────────────────────
   // Turn-mutating calls return quickly; the renderer follows progress through
-  // the sessions:events feed and the shared reduceTurn reducer.
+  // the turns:events feed and the shared reduceTurn reducer.
   'sessions:create': {
     req: z.object({ title: z.string().optional() }),
     res: z.object({ sessionId: z.string() }),
@@ -545,12 +545,24 @@ const ipcSchemas = {
   },
   // Process-wide turn event spine: every turn's durable events (with file
   // offsets), regardless of who started the turn — session chat, headless
-  // background/knowledge runners, spawned sub-agents. Deltas are not
-  // broadcast here in v1; session chat streams them via sessions:events.
+  // background/knowledge runners, spawned sub-agents. Text/reasoning deltas
+  // ride the same channel but only reach windows that subscribed to that
+  // turn via turns:subscribe.
   'turns:events': {
     req: z.custom<TurnBusEvent>(),
     res: z.null(),
   },
+  // Per-window delta subscription: deltas are high-volume and ephemeral, so
+  // they cross IPC only for turns this window declared it is watching.
+  // Durable events are always broadcast regardless.
+  'turns:subscribe': {
+    req: z.object({ turnId: z.string() }),
+    res: z.object({ success: z.literal(true) }),
+  },
+  'turns:unsubscribe': {
+    req: z.object({ turnId: z.string() }),
+    res: z.object({ success: z.literal(true) }),
+  },
   'services:events': {
     req: ServiceEvent,
     res: z.null(),
diff --git a/apps/x/packages/shared/src/sessions.ts b/apps/x/packages/shared/src/sessions.ts
index 1a52d38f..08435650 100644
--- a/apps/x/packages/shared/src/sessions.ts
+++ b/apps/x/packages/shared/src/sessions.ts
@@ -1,9 +1,5 @@
 import { z } from "zod";
-import {
-    ModelDescriptor,
-    type TurnStatus,
-    type TurnStreamEvent,
-} from "./turns.js";
+import { ModelDescriptor, type TurnStatus } from "./turns.js";
 
 // Durable session contract for the session layer (see
 // packages/core/docs/session-design.md). A session is an append-only chain
@@ -171,21 +167,14 @@ export interface SessionIndexEntry {
     error?: string;
 }
 
-// What the renderer's single feed consumer receives over IPC: live turn
-// stream events (durable events + ephemeral deltas) and index updates.
-// entry: null signals deletion.
-export type SessionBusEvent =
-    | {
-          kind: "turn-event";
-          sessionId: string;
-          turnId: string;
-          event: TurnStreamEvent;
-      }
-    | {
-          kind: "index-changed";
-          sessionId: string;
-          entry: SessionIndexEntry | null;
-      };
+// What the renderer's session-feed consumer receives over IPC: session index
+// updates only. Turn events (durable + deltas) travel on the turns:events
+// spine (TurnBusEvent in turns.ts). entry: null signals deletion.
+export type SessionBusEvent = {
+    kind: "index-changed";
+    sessionId: string;
+    entry: SessionIndexEntry | null;
+};
 
 export function sessionIndexEntry(
     state: SessionState,

From 51b11101f040a6041decee87bb63c468b77e96de Mon Sep 17 00:00:00 2001
From: Ramnique Singh <30795890+ramnique@users.noreply.github.com>
Date: Thu, 9 Jul 2026 15:56:45 +0530
Subject: [PATCH 3/3] test(x): cover useTurn and useAgentRunTranscript hook
 behavior

The follower's join protocol was already unit-tested, but the hooks
grew stateful behavior of their own: useTurn's reset-on-turnId-change
vs keep-while-disabled, the snapshotFailed signal and feed-event
recovery, and useAgentRunTranscript's legacy runs:fetch fallback and
loading/error derivation. Tests drive the real turn-feed singleton
through a stubbed window.ipc preload surface.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../hooks/use-agent-run-transcript.test.tsx   |  77 +++++++++
 .../apps/renderer/src/hooks/use-turn.test.tsx | 148 ++++++++++++++++++
 2 files changed, 225 insertions(+)
 create mode 100644 apps/x/apps/renderer/src/hooks/use-agent-run-transcript.test.tsx
 create mode 100644 apps/x/apps/renderer/src/hooks/use-turn.test.tsx

diff --git a/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.test.tsx b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.test.tsx
new file mode 100644
index 00000000..eda45598
--- /dev/null
+++ b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.test.tsx
@@ -0,0 +1,77 @@
+import { renderHook, waitFor } from '@testing-library/react'
+import { beforeEach, describe, expect, it } from 'vitest'
+import { completedTurnLog } from '@/lib/session-chat/test-fixtures'
+import { useAgentRunTranscript } from './use-agent-run-transcript'
+
+// Same preload stub as use-turn.test.tsx: the hook rides useTurn (real
+// turn-feed singleton) plus a runs:fetch legacy fallback.
+let handlers: Record<string, (args: unknown) => Promise<unknown>> = {}
+
+;(window as unknown as { ipc: unknown }).ipc = {
+  on: () => () => undefined,
+  invoke: (channel: string, args: unknown) => {
+    const handler = handlers[channel]
+    return handler ? handler(args) : Promise.resolve({ success: true })
+  },
+}
+
+beforeEach(() => {
+  handlers = {}
+})
+
+describe('useAgentRunTranscript', () => {
+  it('renders a turn-backed transcript', async () => {
+    handlers['sessions:getTurn'] = async () => ({
+      turnId: 'run-1',
+      events: completedTurnLog('run-1', 's1', 'do the task', 'task done'),
+    })
+    const { result } = renderHook(() => useAgentRunTranscript('run-1'))
+
+    await waitFor(() => expect(result.current.transcript).not.toBeNull())
+    expect(result.current.transcript?.id).toBe('run-1')
+    expect(result.current.transcript?.summary).toBe('task done')
+    expect(result.current.loading).toBe(false)
+    expect(result.current.error).toBeNull()
+  })
+
+  it('falls back to legacy runs:fetch when the id is not a turn', async () => {
+    handlers['sessions:getTurn'] = async () => {
+      throw new Error('turn not found: legacy id')
+    }
+    handlers['runs:fetch'] = async (args) => ({
+      id: (args as { runId: string }).runId,
+      createdAt: '2026-07-01T00:00:00Z',
+      subUseCase: 'cron',
+      log: [],
+    })
+    const { result } = renderHook(() => useAgentRunTranscript('legacy-run'))
+
+    await waitFor(() => expect(result.current.transcript).not.toBeNull())
+    expect(result.current.transcript?.id).toBe('legacy-run')
+    expect(result.current.transcript?.trigger).toBe('cron')
+    expect(result.current.loading).toBe(false)
+    expect(result.current.error).toBeNull()
+  })
+
+  it('surfaces an error when both the turn and legacy paths fail', async () => {
+    handlers['sessions:getTurn'] = async () => {
+      throw new Error('turn not found')
+    }
+    handlers['runs:fetch'] = async () => {
+      throw new Error('run not found either')
+    }
+    const { result } = renderHook(() => useAgentRunTranscript('ghost-run'))
+
+    await waitFor(() => expect(result.current.error).not.toBeNull())
+    expect(result.current.error).toMatch(/run not found either/)
+    expect(result.current.transcript).toBeNull()
+    expect(result.current.loading).toBe(false)
+  })
+
+  it('is idle without a run id', () => {
+    const { result } = renderHook(() => useAgentRunTranscript(null))
+    expect(result.current.transcript).toBeNull()
+    expect(result.current.loading).toBe(false)
+    expect(result.current.error).toBeNull()
+  })
+})
diff --git a/apps/x/apps/renderer/src/hooks/use-turn.test.tsx b/apps/x/apps/renderer/src/hooks/use-turn.test.tsx
new file mode 100644
index 00000000..60e58f98
--- /dev/null
+++ b/apps/x/apps/renderer/src/hooks/use-turn.test.tsx
@@ -0,0 +1,148 @@
+import { act, renderHook, waitFor } from '@testing-library/react'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import type { TurnBusEvent } from '@x/shared/src/turns.js'
+import {
+  assistantText,
+  completed,
+  completedTurnLog,
+  created,
+  requested,
+  turnCompleted,
+  user,
+  type TEvent,
+} from '@/lib/session-chat/test-fixtures'
+import { useTurn } from './use-turn'
+
+// The hook wires the real turn-feed singleton and window.ipc, so the tests
+// stub the preload surface: `on` captures the feed listener (the feed
+// attaches lazily on first subscribe), `invoke` routes by channel through a
+// per-test handler map.
+let feedListener: ((event: TurnBusEvent) => void) | null = null
+let handlers: Record<string, (args: unknown) => Promise<unknown>> = {}
+
+;(window as unknown as { ipc: unknown }).ipc = {
+  on: (channel: string, handler: (event: TurnBusEvent) => void) => {
+    if (channel === 'turns:events') feedListener = handler
+    return () => undefined
+  },
+  invoke: (channel: string, args: unknown) => {
+    const handler = handlers[channel]
+    return handler ? handler(args) : Promise.resolve({ success: true })
+  },
+}
+
+function emit(event: TurnBusEvent): void {
+  act(() => feedListener?.(event))
+}
+
+function durable(turnId: string, event: TEvent, offset: number): TurnBusEvent {
+  return { turnId, sessionId: null, event, offset }
+}
+
+function serveTurns(turns: Record<string, TEvent[]>): void {
+  handlers['sessions:getTurn'] = async (args) => {
+    const { turnId } = args as { turnId: string }
+    const events = turns[turnId]
+    if (!events) throw new Error(`turn not found: ${turnId}`)
+    return { turnId, events: [...events] }
+  }
+}
+
+beforeEach(() => {
+  handlers = {}
+})
+
+describe('useTurn', () => {
+  it('fetches the snapshot and applies contiguous feed events', async () => {
+    const T = 'turn-live'
+    const full: TEvent[] = [
+      created(T, 's1', user('go')),
+      requested(T, 0),
+      completed(T, 0, assistantText('done')),
+      turnCompleted(T, 'done'),
+    ]
+    serveTurns({ [T]: full.slice(0, 2) })
+    const { result } = renderHook(() => useTurn(T))
+
+    await waitFor(() => expect(result.current.state).not.toBeNull())
+    expect(result.current.state?.terminal).toBeUndefined()
+
+    emit(durable(T, full[2], 3))
+    emit(durable(T, full[3], 4))
+    expect(result.current.state?.terminal?.type).toBe('turn_completed')
+    expect(result.current.error).toBeNull()
+  })
+
+  it('resets state when turnId changes instead of showing the stale turn', async () => {
+    let release!: () => void
+    const gate = new Promise<void>((resolve) => {
+      release = resolve
+    })
+    serveTurns({ 'turn-a': completedTurnLog('turn-a', 's1', 'qa', 'aa') })
+    const base = handlers['sessions:getTurn']
+    handlers['sessions:getTurn'] = async (args) => {
+      if ((args as { turnId: string }).turnId === 'turn-b') {
+        await gate
+        return { turnId: 'turn-b', events: completedTurnLog('turn-b', 's1', 'qb', 'ab') }
+      }
+      return base(args)
+    }
+
+    const { result, rerender } = renderHook(({ id }) => useTurn(id), {
+      initialProps: { id: 'turn-a' },
+    })
+    await waitFor(() => expect(result.current.state).not.toBeNull())
+
+    rerender({ id: 'turn-b' })
+    // turn-b's snapshot is gated: the hook must show nothing, not turn-a.
+    expect(result.current.state).toBeNull()
+
+    release()
+    await waitFor(() =>
+      expect(result.current.state?.definition.turnId).toBe('turn-b'),
+    )
+  })
+
+  it('keeps the last state while disabled and refetches on re-enable', async () => {
+    const getTurn = vi.fn(async () => ({
+      turnId: 'turn-a',
+      events: completedTurnLog('turn-a', 's1', 'q', 'a'),
+    }))
+    handlers['sessions:getTurn'] = getTurn
+
+    const { result, rerender } = renderHook(
+      ({ enabled }) => useTurn('turn-a', { enabled }),
+      { initialProps: { enabled: true } },
+    )
+    await waitFor(() => expect(result.current.state).not.toBeNull())
+    const fetches = getTurn.mock.calls.length
+
+    rerender({ enabled: false })
+    expect(result.current.state).not.toBeNull() // kept while hidden
+
+    rerender({ enabled: true })
+    await waitFor(() =>
+      expect(getTurn.mock.calls.length).toBeGreaterThan(fetches),
+    )
+    expect(result.current.state).not.toBeNull()
+  })
+
+  it('reports snapshotFailed after retries and recovers via a feed event', async () => {
+    const T = 'turn-late'
+    const full = completedTurnLog(T, 's1', 'q', 'a')
+    let available = false
+    handlers['sessions:getTurn'] = async () => {
+      if (!available) throw new Error('not created yet')
+      return { turnId: T, events: [...full] }
+    }
+
+    const { result } = renderHook(() => useTurn(T, { maxRetries: 0 }))
+    await waitFor(() => expect(result.current.snapshotFailed).toBe(true))
+    expect(result.current.state).toBeNull()
+
+    available = true
+    emit(durable(T, full[full.length - 1], full.length))
+    await waitFor(() => expect(result.current.state).not.toBeNull())
+    expect(result.current.snapshotFailed).toBe(false)
+  })
+})