Merge pull request #705 from rowboatlabs/runtime-refactor

refactor(x): complete the turn-spine unification — headless surfaces, chat, deltas, channels

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');

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 ?? []

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 (

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()
+  })
+})

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 }
+}

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)
+  })
+})

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 }
 }

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([])
   })
 })

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,

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()

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, []))

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
   }

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)])

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

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.
@@ -376,7 +381,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: {

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.

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,

apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts

@@ -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;

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[] = [];

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) {

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,
         });
     }

apps/x/packages/core/src/sessions/headless.ts

@@ -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 };
-}

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

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";
@@ -645,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" },
@@ -653,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 () => {
@@ -823,26 +818,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",

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.

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(),

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,