Merge pull request #704 from rowboatlabs/runtime-refactor

feat(x): turn event spine — one bus for every turn's events

apps/x/apps/main/src/ipc.ts

@@ -29,7 +29,9 @@ let caffeinateBlockerId: number | null = null;
 import { RunEvent } from '@x/shared/dist/runs.js';
 import { ServiceEvent } from '@x/shared/dist/service-events.js';
 import type { SessionBusEvent } from '@x/shared/dist/sessions.js';
+import { isDurableTurnEvent } from '@x/shared/dist/turns.js';
 import type { ISessions, EmitterSessionBus } from '@x/core/dist/sessions/index.js';
+import type { TurnEventHub } from '@x/core/dist/turns/event-hub.js';
 import container from '@x/core/dist/di/container.js';
 import { listOnboardingModels } from '@x/core/dist/models/models-dev.js';
 import { testModelConnection, listModelsForProvider, generateOneShot } from '@x/core/dist/models/models.js';
@@ -486,24 +488,14 @@ let debounceTimer: ReturnType<typeof setTimeout> | null = null;
  * Emit knowledge commit event to all renderer windows
  */
 function emitKnowledgeCommitEvent(): void {
-  const windows = BrowserWindow.getAllWindows();
-  for (const win of windows) {
-    if (!win.isDestroyed() && win.webContents) {
-      win.webContents.send('knowledge:didCommit', {});
-    }
-  }
+  broadcastToWindows('knowledge:didCommit', {});
 }
 
 /**
  * Emit workspace change event to all renderer windows
  */
 function emitWorkspaceChangeEvent(event: z.infer<typeof workspaceShared.WorkspaceChangeEvent>): void {
-  const windows = BrowserWindow.getAllWindows();
-  for (const win of windows) {
-    if (!win.isDestroyed() && win.webContents) {
-      win.webContents.send('workspace:didChange', event);
-    }
-  }
+  broadcastToWindows('workspace:didChange', event);
 }
 
 /**
@@ -620,34 +612,31 @@ export function stopWorkspaceWatcher(): void {
   changeQueue.clear();
 }
 
-function emitRunEvent(event: z.infer<typeof RunEvent>): void {
+// The one renderer fan-out: send a payload to every live window on a channel.
+// All broadcast feeds (runs, services, sessions, turns, code runs, agent
+// status) go through here.
+function broadcastToWindows(channel: string, payload: unknown): void {
   const windows = BrowserWindow.getAllWindows();
   for (const win of windows) {
     if (!win.isDestroyed() && win.webContents) {
-      win.webContents.send('runs:events', event);
+      win.webContents.send(channel, payload);
     }
   }
 }
 
+function emitRunEvent(event: z.infer<typeof RunEvent>): void {
+  broadcastToWindows('runs:events', event);
+}
+
 function emitServiceEvent(event: z.infer<typeof ServiceEvent>): void {
-  const windows = BrowserWindow.getAllWindows();
-  for (const win of windows) {
-    if (!win.isDestroyed() && win.webContents) {
-      win.webContents.send('services:events', event);
-    }
-  }
+  broadcastToWindows('services:events', event);
 }
 
 export function emitOAuthEvent(event: { provider: string; success: boolean; error?: string; userId?: string }): void {
   // Native connection status (e.g. Google) is baked into the Copilot system
   // prompt, so any OAuth state change must rebuild it.
   invalidateCopilotInstructionsCache();
-  const windows = BrowserWindow.getAllWindows();
-  for (const win of windows) {
-    if (!win.isDestroyed() && win.webContents) {
-      win.webContents.send('oauth:didConnect', event);
-    }
-  }
+  broadcastToWindows('oauth:didConnect', event);
 }
 
 async function requireCodeSession(sessionId: string): Promise<CodeSession> {
@@ -667,12 +656,7 @@ export async function startCodeSessionStatusWatcher(): Promise<void> {
   const tracker = container.resolve<CodeSessionStatusTracker>('codeSessionStatusTracker');
   await tracker.start();
   codeSessionStatusWatcher = tracker.onTransition((sessionId, status) => {
-    const windows = BrowserWindow.getAllWindows();
-    for (const win of windows) {
-      if (!win.isDestroyed() && win.webContents) {
-        win.webContents.send('codeSession:status', { sessionId, status });
-      }
-    }
+    broadcastToWindows('codeSession:status', { sessionId, status });
   });
 }
 
@@ -688,12 +672,7 @@ export async function startRunsWatcher(): Promise<void> {
 
 // New runtime: session bus → renderer windows (session-design.md §10).
 function emitSessionEvent(event: SessionBusEvent): void {
-  const windows = BrowserWindow.getAllWindows();
-  for (const win of windows) {
-    if (!win.isDestroyed() && win.webContents) {
-      win.webContents.send('sessions:events', event);
-    }
-  }
+  broadcastToWindows('sessions:events', event);
 }
 
 // Mobile channels: status changes (QR pairing, connect/disconnect) → renderer.
@@ -701,12 +680,7 @@ let channelsWatcher: (() => void) | null = null;
 export function startChannelsWatcher(): void {
   if (channelsWatcher) return;
   channelsWatcher = subscribeChannelsStatus((status) => {
-    const windows = BrowserWindow.getAllWindows();
-    for (const win of windows) {
-      if (!win.isDestroyed() && win.webContents) {
-        win.webContents.send('channels:status', status);
-      }
-    }
+    broadcastToWindows('channels:status', status);
   });
 }
 
@@ -719,6 +693,27 @@ export function startSessionsWatcher(): void {
   sessionsWatcher = sessionBus.subscribe((event) => emitSessionEvent(event));
 }
 
+// Turn event spine → renderer windows: durable events of every turn the
+// 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.
+let turnEventsWatcher: (() => void) | null = null;
+export function startTurnEventsWatcher(): void {
+  if (turnEventsWatcher) {
+    return;
+  }
+  const hub = container.resolve<TurnEventHub>('turnEventBus');
+  turnEventsWatcher = hub.subscribeAll((event) => {
+    if (!isDurableTurnEvent(event.event)) {
+      return;
+    }
+    broadcastToWindows('turns:events', event);
+  });
+}
+
 // Ephemeral code-run stream: CodeRunFeed → all renderer windows. A direct
 // tool→renderer side-channel that bypasses the turn runtime; the durable
 // record is the settle-time code-run-events-batch tool progress.
@@ -729,12 +724,7 @@ export function startCodeRunFeedWatcher(): void {
   }
   const feed = container.resolve<CodeRunFeed>('codeRunFeed');
   codeRunFeedWatcher = feed.subscribe((event) => {
-    const windows = BrowserWindow.getAllWindows();
-    for (const win of windows) {
-      if (!win.isDestroyed() && win.webContents) {
-        win.webContents.send('codeRun:events', event);
-      }
-    }
+    broadcastToWindows('codeRun:events', event);
   });
 }
 
@@ -765,12 +755,7 @@ let liveNoteAgentWatcher: (() => void) | null = null;
 export function startLiveNoteAgentWatcher(): void {
   if (liveNoteAgentWatcher) return;
   liveNoteAgentWatcher = liveNoteBus.subscribe((event) => {
-    const windows = BrowserWindow.getAllWindows();
-    for (const win of windows) {
-      if (!win.isDestroyed() && win.webContents) {
-        win.webContents.send('live-note-agent:events', event);
-      }
-    }
+    broadcastToWindows('live-note-agent:events', event);
   });
 }
 
@@ -778,12 +763,7 @@ let backgroundTaskAgentWatcher: (() => void) | null = null;
 export function startBackgroundTaskAgentWatcher(): void {
   if (backgroundTaskAgentWatcher) return;
   backgroundTaskAgentWatcher = backgroundTaskBus.subscribe((event) => {
-    const windows = BrowserWindow.getAllWindows();
-    for (const win of windows) {
-      if (!win.isDestroyed() && win.webContents) {
-        win.webContents.send('bg-task-agent:events', event);
-      }
-    }
+    broadcastToWindows('bg-task-agent:events', event);
   });
 }
 

apps/x/apps/main/src/main.ts

@@ -2,7 +2,7 @@ import { app, BrowserWindow, desktopCapturer, protocol, net, shell, session, saf
 import path from "node:path";
 import {
   setupIpcHandlers,
-  startRunsWatcher, startSessionsWatcher, markSessionsIndexReady,
+  startRunsWatcher, startSessionsWatcher, startTurnEventsWatcher, markSessionsIndexReady,
   startCodeRunFeedWatcher,
   startChannelsWatcher,
   startCodeSessionStatusWatcher,
@@ -454,6 +454,9 @@ app.whenReady().then(async () => {
     markSessionsIndexReady();
   }
   startSessionsWatcher();
+  // Turn event spine: durable events of every turn (session, headless,
+  // sub-agent) → renderer, for turnId-keyed live views.
+  startTurnEventsWatcher();
   startCodeRunFeedWatcher();
 
   // Mobile channels (WhatsApp/Telegram bridge): needs the session index, so

apps/x/apps/renderer/src/components/sub-agent-block.tsx

@@ -1,44 +1,24 @@
-import { useEffect, useState } from 'react'
+import { useMemo } from 'react'
 import { Tool, ToolContent, ToolHeader } from '@/components/ai-elements/tool'
 import { CompactConversation } from '@/components/compact-conversation'
-import { fetchAgentRunTranscript, type AgentRunTranscript } from '@/lib/agent-transcript'
+import { turnStateToTranscript, type AgentRunTranscript } from '@/lib/agent-transcript'
 import { toToolState, type ToolCall } from '@/lib/chat-conversation'
+import { useTurn } from '@/hooks/use-turn'
 
 // Rendered for a spawn-agent tool call: a collapsed status card that expands
 // into the child turn's live transcript. The child is a standalone turn
-// (sessionId null) whose events never reach the session bus, so while it
-// runs we poll sessions:getTurn via fetchAgentRunTranscript — the file is
-// local and append-only, making this cheap — and do one final fetch when the
-// parent tool call settles.
-
-const POLL_MS = 1000
+// (sessionId null); its durable events arrive on the turns:events spine, so
+// useTurn keeps the transcript live without polling.
 
 function useChildTranscript(
   childTurnId: string | undefined,
-  running: boolean,
   open: boolean,
 ): AgentRunTranscript | null {
-  const [transcript, setTranscript] = useState<AgentRunTranscript | null>(null)
-  useEffect(() => {
-    if (!childTurnId || !open) return
-    let alive = true
-    const fetchOnce = async () => {
-      try {
-        const next = await fetchAgentRunTranscript(childTurnId)
-        if (alive) setTranscript(next)
-      } catch {
-        // Child file may not be readable yet; the next tick retries.
-      }
-    }
-    void fetchOnce()
-    if (!running) return () => { alive = false }
-    const timer = setInterval(() => void fetchOnce(), POLL_MS)
-    return () => {
-      alive = false
-      clearInterval(timer)
-    }
-  }, [childTurnId, running, open])
-  return transcript
+  const { state } = useTurn(childTurnId, { enabled: open })
+  return useMemo(
+    () => (childTurnId && state ? turnStateToTranscript(childTurnId, state) : null),
+    [childTurnId, state],
+  )
 }
 
 // "london-weather" / "meeting_prep" → "London weather" / "Meeting prep".
@@ -67,7 +47,7 @@ export function SubAgentBlock({
   // header truncates with a hover tooltip carrying the full text).
   const title = task ? `${name || 'Agent'}: ${task}` : name || 'Agent'
   const running = item.status === 'pending' || item.status === 'running'
-  const transcript = useChildTranscript(item.subAgent?.childTurnId, running, open)
+  const transcript = useChildTranscript(item.subAgent?.childTurnId, open)
 
   return (
     <Tool open={open} onOpenChange={onOpenChange}>

apps/x/apps/renderer/src/hooks/use-turn.ts

@@ -0,0 +1,45 @@
+import { useEffect, useState } from 'react'
+import type { TurnState } from '@x/shared/src/turns.js'
+import { subscribeTurnFeed } from '@/lib/turn-feed'
+import { followTurn } from '@/lib/turn-follower'
+
+export interface UseTurnResult {
+  state: TurnState | null
+  error: string | null
+}
+
+// Live view of one turn by id: snapshot via sessions:getTurn, then durable
+// events from the turns:events spine (see lib/turn-follower.ts for the join
+// protocol). Works for any turn — session chat, headless runners, spawned
+// sub-agents.
+export function useTurn(
+  turnId: string | undefined,
+  opts?: { enabled?: boolean },
+): UseTurnResult {
+  const enabled = opts?.enabled ?? true
+  const [state, setState] = useState<TurnState | null>(null)
+  const [error, setError] = useState<string | null>(null)
+
+  useEffect(() => {
+    if (!turnId) {
+      setState(null)
+      setError(null)
+      return
+    }
+    if (!enabled) {
+      // Keep the last rendered state while hidden; re-enabling refetches.
+      return
+    }
+    return followTurn(turnId, {
+      fetchTurn: (id) => window.ipc.invoke('sessions:getTurn', { turnId: id }),
+      subscribe: subscribeTurnFeed,
+      onState: (next) => {
+        setState(next)
+        setError(null)
+      },
+      onError: (message) => setError(message),
+    })
+  }, [turnId, enabled])
+
+  return { state, error }
+}

apps/x/apps/renderer/src/lib/agent-transcript.ts

@@ -1,6 +1,6 @@
 import type { z } from 'zod'
 import type { Run } from '@x/shared/src/runs.js'
-import { reduceTurn, type TurnEvent } from '@x/shared/src/turns.js'
+import { reduceTurn, type TurnEvent, type TurnState } from '@x/shared/src/turns.js'
 import type { ConversationItem } from '@/lib/chat-conversation'
 import { runLogToConversation } from '@/lib/run-to-conversation'
 import { buildTurnConversation } from '@/lib/session-chat/turn-view'
@@ -24,7 +24,15 @@ export function turnToTranscript(
   turnId: string,
   events: Array<z.infer<typeof TurnEvent>>,
 ): AgentRunTranscript {
-  const state = reduceTurn(events)
+  return turnStateToTranscript(turnId, reduceTurn(events))
+}
+
+// Reduced-state variant for consumers that already hold a live TurnState
+// (useTurn) and should not re-reduce the event log.
+export function turnStateToTranscript(
+  turnId: string,
+  state: TurnState,
+): AgentRunTranscript {
   const out: AgentRunTranscript = {
     id: turnId,
     createdAt: state.definition.ts,

apps/x/apps/renderer/src/lib/broadcast-feed.ts

@@ -0,0 +1,39 @@
+// One shared consumer per broadcast IPC channel; stores and hooks tap this
+// fan-out instead of each opening their own IPC listener. Factory so tests
+// can drive a fake source.
+
+export type FeedListener<T> = (event: T) => void
+export type FeedSource<T> = (listener: FeedListener<T>) => () => void
+
+export interface BroadcastFeed<T> {
+  subscribe(listener: FeedListener<T>): () => void
+}
+
+export function createBroadcastFeed<T>(source: FeedSource<T>): BroadcastFeed<T> {
+  const listeners = new Set<FeedListener<T>>()
+  let detach: (() => void) | null = null
+
+  const ensureStarted = () => {
+    if (detach) return
+    detach = source((event) => {
+      // Copy so (un)subscribing during dispatch is safe.
+      for (const listener of [...listeners]) {
+        try {
+          listener(event)
+        } catch {
+          // A misbehaving subscriber must never break the feed.
+        }
+      }
+    })
+  }
+
+  return {
+    subscribe(listener: FeedListener<T>): () => void {
+      ensureStarted()
+      listeners.add(listener)
+      return () => {
+        listeners.delete(listener)
+      }
+    },
+  }
+}

apps/x/apps/renderer/src/lib/session-chat/feed.ts

@@ -1,38 +1,18 @@
 import type { SessionBusEvent } from '@x/shared/src/sessions.js'
-
-export type SessionFeedListener = (event: SessionBusEvent) => void
-export type SessionFeedSource = (listener: SessionFeedListener) => () => void
+import {
+  createBroadcastFeed,
+  type FeedListener,
+  type FeedSource,
+} from '@/lib/broadcast-feed'
 
 // One shared consumer of the sessions:events push channel; stores tap this
-// fan-out instead of each opening their own IPC listener. Factory so tests
-// can drive a fake source.
+// fan-out instead of each opening their own IPC listener.
+
+export type SessionFeedListener = FeedListener<SessionBusEvent>
+export type SessionFeedSource = FeedSource<SessionBusEvent>
+
 export function createSessionFeed(source: SessionFeedSource) {
-  const listeners = new Set<SessionFeedListener>()
-  let detach: (() => void) | null = null
-
-  const ensureStarted = () => {
-    if (detach) return
-    detach = source((event) => {
-      // Copy so (un)subscribing during dispatch is safe.
-      for (const listener of [...listeners]) {
-        try {
-          listener(event)
-        } catch {
-          // A misbehaving subscriber must never break the feed.
-        }
-      }
-    })
-  }
-
-  return {
-    subscribe(listener: SessionFeedListener): () => void {
-      ensureStarted()
-      listeners.add(listener)
-      return () => {
-        listeners.delete(listener)
-      }
-    },
-  }
+  return createBroadcastFeed<SessionBusEvent>(source)
 }
 
 const appFeed = createSessionFeed((listener) => window.ipc.on('sessions:events', listener))

apps/x/apps/renderer/src/lib/turn-feed.ts

@@ -0,0 +1,17 @@
+import type { TurnBusEvent } from '@x/shared/src/turns.js'
+import { createBroadcastFeed, type FeedListener } from '@/lib/broadcast-feed'
+
+// The turn event spine: durable events of every turn the runtime executes —
+// session chat, headless background/knowledge runners, spawned sub-agents —
+// tagged with sessionId and the event's 1-based file offset. Consumers join a
+// live turn by subscribing first, fetching the sessions:getTurn snapshot, and
+// discarding feed events whose offset is already covered by the snapshot
+// (see useTurn).
+
+const appFeed = createBroadcastFeed<TurnBusEvent>((listener) =>
+  window.ipc.on('turns:events', listener),
+)
+
+export function subscribeTurnFeed(listener: FeedListener<TurnBusEvent>): () => void {
+  return appFeed.subscribe(listener)
+}

apps/x/apps/renderer/src/lib/turn-follower.test.ts

@@ -0,0 +1,168 @@
+import { describe, expect, it, vi } from 'vitest'
+import type { TurnBusEvent, TurnState } from '@x/shared/src/turns.js'
+import { followTurn, type TurnFollowerDeps } from './turn-follower'
+import {
+  completed,
+  created,
+  requested,
+  turnCompleted,
+  assistantText,
+  user,
+  type TEvent,
+} from './session-chat/test-fixtures'
+
+const TURN = 't1'
+
+function log(): TEvent[] {
+  return [
+    created(TURN, 's1', user('question')),
+    requested(TURN, 0),
+    completed(TURN, 0, assistantText('answer')),
+    turnCompleted(TURN, 'answer'),
+  ]
+}
+
+function bus(event: TEvent, offset: number): TurnBusEvent {
+  return { turnId: TURN, sessionId: 's1', event, offset }
+}
+
+function flush(): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, 0))
+}
+
+function harness(fetches: Array<TEvent[] | Error>) {
+  const fetchTurn = vi.fn(async () => {
+    const next = fetches.shift()
+    if (!next || next instanceof Error) {
+      throw next ?? new Error('no snapshot scripted')
+    }
+    return { events: [...next] }
+  })
+  let listener: ((event: TurnBusEvent) => void) | null = null
+  let unsubscribed = false
+  const states: TurnState[] = []
+  const errors: string[] = []
+  const deps: TurnFollowerDeps = {
+    fetchTurn,
+    subscribe: (fn) => {
+      listener = fn
+      return () => {
+        unsubscribed = true
+        listener = null
+      }
+    },
+    onState: (state) => states.push(state),
+    onError: (message) => errors.push(message),
+    maxRetries: 0,
+  }
+  return {
+    deps,
+    fetchTurn,
+    states,
+    errors,
+    emit: (event: TurnBusEvent) => listener?.(event),
+    get unsubscribed() {
+      return unsubscribed
+    },
+  }
+}
+
+describe('followTurn', () => {
+  it('publishes the snapshot, then extends it with contiguous feed events', async () => {
+    const full = log()
+    const h = harness([full.slice(0, 2)])
+    const detach = followTurn(TURN, h.deps)
+    await flush()
+
+    expect(h.states).toHaveLength(1)
+    expect(h.states[0].modelCalls[0].response).toBeUndefined()
+
+    h.emit(bus(full[2], 3))
+    h.emit(bus(full[3], 4))
+
+    expect(h.states).toHaveLength(3)
+    const final = h.states[2]
+    expect(final.terminal?.type).toBe('turn_completed')
+    expect(h.errors).toEqual([])
+    detach()
+  })
+
+  it('discards feed events already covered by the snapshot', async () => {
+    const full = log()
+    const h = harness([full])
+    const detach = followTurn(TURN, h.deps)
+    await flush()
+
+    // Replays of already-snapshotted lines must not corrupt the reduction.
+    h.emit(bus(full[2], 3))
+    h.emit(bus(full[3], 4))
+
+    expect(h.errors).toEqual([])
+    expect(h.states[h.states.length - 1].terminal?.type).toBe('turn_completed')
+    detach()
+  })
+
+  it('buffers events that arrive before the snapshot and applies them after', async () => {
+    const full = log()
+    let release!: () => void
+    const gate = new Promise<void>((resolve) => {
+      release = resolve
+    })
+    const h = harness([])
+    h.deps.fetchTurn = vi.fn(async () => {
+      await gate
+      return { events: full.slice(0, 3) }
+    })
+    const detach = followTurn(TURN, h.deps)
+
+    h.emit(bus(full[3], 4)) // arrives mid-fetch
+    release()
+    await flush()
+
+    expect(h.states[h.states.length - 1].terminal?.type).toBe('turn_completed')
+    expect(h.errors).toEqual([])
+    detach()
+  })
+
+  it('refetches the snapshot on a contiguity gap', async () => {
+    const full = log()
+    const h = harness([full.slice(0, 2), full])
+    const detach = followTurn(TURN, h.deps)
+    await flush()
+
+    // Offset 4 while the local log has 2 entries: events were missed.
+    h.emit(bus(full[3], 4))
+    await flush()
+
+    expect(h.fetchTurn).toHaveBeenCalledTimes(2)
+    expect(h.states[h.states.length - 1].terminal?.type).toBe('turn_completed')
+    expect(h.errors).toEqual([])
+    detach()
+  })
+
+  it('recovers from a failed snapshot when a feed event arrives', async () => {
+    const full = log()
+    const h = harness([new Error('turn file not created yet'), full])
+    const detach = followTurn(TURN, h.deps)
+    await flush()
+
+    expect(h.states).toHaveLength(0)
+    h.emit(bus(full[3], 4))
+    await flush()
+
+    expect(h.fetchTurn).toHaveBeenCalledTimes(2)
+    expect(h.states[h.states.length - 1].terminal?.type).toBe('turn_completed')
+    detach()
+  })
+
+  it('stops delivering after detach and unsubscribes from the feed', async () => {
+    const full = log()
+    const h = harness([full.slice(0, 2)])
+    const detach = followTurn(TURN, h.deps)
+    await flush()
+    expect(h.states).toHaveLength(1)
+
+    detach()
+    expect(h.unsubscribed).toBe(true)
+  })
+})

apps/x/apps/renderer/src/lib/turn-follower.ts

@@ -0,0 +1,128 @@
+import type { z } from 'zod'
+import {
+  isDurableTurnEvent,
+  reduceTurn,
+  type TurnBusEvent,
+  type TurnEvent,
+} from '@x/shared/src/turns.js'
+import type { TurnState } from '@x/shared/src/turns.js'
+
+// Follows one turn live, regardless of where it runs — session chat, headless
+// background/knowledge runners, spawned sub-agents.
+//
+// Join protocol: subscribe to the turn feed first (buffering), fetch the
+// durable snapshot, then extend the snapshot with feed events. Each durable
+// feed event carries its 1-based file offset, so events already covered by
+// the snapshot are discarded and a contiguity gap (e.g. the subscription
+// raced turn start) falls back to a snapshot refetch. A failed snapshot fetch
+// retries a few times and again on the next feed event for this turn, so a
+// turn whose file is still being created self-heals.
+
+export interface TurnFollowerDeps {
+  fetchTurn: (turnId: string) => Promise<{ events: Array<z.infer<typeof TurnEvent>> }>
+  subscribe: (listener: (event: TurnBusEvent) => void) => () => void
+  onState: (state: TurnState) => void
+  onError: (message: string) => void
+  retryDelayMs?: number
+  maxRetries?: number
+}
+
+const DEFAULT_RETRY_DELAY_MS = 1000
+const DEFAULT_MAX_RETRIES = 3
+
+export function followTurn(turnId: string, deps: TurnFollowerDeps): () => void {
+  const retryDelayMs = deps.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS
+  const maxRetries = deps.maxRetries ?? DEFAULT_MAX_RETRIES
+
+  let alive = true
+  // null until a snapshot lands; feed events buffer in `pending` meanwhile.
+  let events: Array<z.infer<typeof TurnEvent>> | null = null
+  let pending: TurnBusEvent[] = []
+  let fetching = false
+  let retries = 0
+  let retryTimer: ReturnType<typeof setTimeout> | null = null
+
+  const publish = () => {
+    if (!alive || !events) return
+    try {
+      deps.onState(reduceTurn(events))
+    } catch (err) {
+      deps.onError(err instanceof Error ? err.message : String(err))
+    }
+  }
+
+  // Extends `events` with one feed event; false signals a contiguity gap.
+  const apply = (e: TurnBusEvent): boolean => {
+    if (!events || e.offset === undefined || !isDurableTurnEvent(e.event)) {
+      return true
+    }
+    if (e.offset <= events.length) {
+      return true // already covered by the snapshot
+    }
+    if (e.offset === events.length + 1) {
+      events.push(e.event)
+      return true
+    }
+    return false
+  }
+
+  const resync = () => {
+    events = null
+    void fetchSnapshot()
+  }
+
+  const fetchSnapshot = async () => {
+    if (fetching || !alive) return
+    fetching = true
+    try {
+      const turn = await deps.fetchTurn(turnId)
+      if (!alive) return
+      retries = 0
+      events = turn.events
+      const buffered = pending
+      pending = []
+      for (const e of buffered) {
+        if (!apply(e)) {
+          fetching = false
+          resync()
+          return
+        }
+      }
+      publish()
+    } catch {
+      if (!alive) return
+      // Turn file may not exist yet (subscription raced creation) or the
+      // fetch failed transiently.
+      events = null
+      if (retries < maxRetries) {
+        retries += 1
+        retryTimer = setTimeout(() => void fetchSnapshot(), retryDelayMs)
+      }
+    } finally {
+      fetching = false
+    }
+  }
+
+  const unsubscribe = deps.subscribe((e) => {
+    if (!alive || e.turnId !== turnId) return
+    if (!isDurableTurnEvent(e.event) || e.offset === undefined) return
+    if (!events) {
+      pending.push(e)
+      void fetchSnapshot()
+      return
+    }
+    if (apply(e)) {
+      publish()
+    } else {
+      pending = [e]
+      resync()
+    }
+  })
+
+  void fetchSnapshot()
+  return () => {
+    alive = false
+    if (retryTimer) clearTimeout(retryTimer)
+    unsubscribe()
+  }
+}

apps/x/packages/core/docs/turn-runtime-design.md

@@ -1415,6 +1415,33 @@ Lifecycle publication is observational and must not alter durable turn
 semantics. Live durable events and deltas are consumed from `TurnExecution` by
 the application and may be forwarded over the existing bus/IPC bridge.
 
+### 17.1 Turn event bus
+
+Implemented alongside the lifecycle bus: the runtime publishes every turn's
+events to an injected `ITurnEventBus` (`event-hub.ts`), tagged with the
+turn's `sessionId`, regardless of who started the turn — session chat,
+headless runners, spawned sub-agents. This is the process-wide delivery
+spine; `TurnExecution` remains the single-consumer, invocation-scoped
+stream for the initiating caller.
+
+- `turn_created` is published by `createTurn` (it never flows through an
+  execution stream); every other durable event is published by the advance
+  loop immediately after its `stream.push`, inside the serialized commit
+  ritual, so bus order equals file order.
+- Durable events carry `offset`, their 1-based line index in the turn file
+  (`this.events.length` at commit time, which is absolute because each
+  invocation re-reads the full log). A late subscriber joins a live turn by
+  subscribing first, fetching the `getTurn` snapshot, and discarding bus
+  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.
+- 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.
+
 ## 18. Main execution algorithm
 
 At a high level, `advanceTurn` performs:

apps/x/packages/core/src/di/container.ts

@@ -34,6 +34,7 @@ import type { ITurnRepo } from "../turns/repo.js";
 import type { IContextResolver } from "../turns/context-resolver.js";
 import { createContextResolver } from "../turns/context-elision.js";
 import { EmitterTurnLifecycleBus, type ITurnLifecycleBus } from "../turns/bus.js";
+import { TurnEventHub, type ITurnEventBus } from "../turns/event-hub.js";
 import { RealUsageReporter } from "../turns/bridges/real-usage-reporter.js";
 import type { IUsageReporter } from "../turns/usage-reporter.js";
 import { TurnRuntime } from "../turns/runtime.js";
@@ -124,6 +125,9 @@ container.register({
         createContextResolver({ turnRepo }),
     ).singleton(),
     lifecycleBus: asClass<ITurnLifecycleBus>(EmitterTurnLifecycleBus).singleton(),
+    // Process-wide turn event spine: every turn's events, tagged with
+    // sessionId and durable file offsets, regardless of who started the turn.
+    turnEventBus: asClass<ITurnEventBus>(TurnEventHub).singleton(),
     usageReporter: asClass<IUsageReporter>(RealUsageReporter).singleton(),
     agentResolver: asFunction<IAgentResolver>(
         () =>

apps/x/packages/core/src/turns/event-hub.test.ts

@@ -0,0 +1,69 @@
+import { describe, expect, it } from "vitest";
+import type { TurnBusEvent } from "@x/shared/dist/turns.js";
+import { TurnEventHub } from "./event-hub.js";
+
+function busEvent(turnId: string, offset: number): TurnBusEvent {
+    return {
+        turnId,
+        sessionId: null,
+        event: {
+            type: "text_delta",
+            turnId,
+            modelCallIndex: 0,
+            delta: `e${offset}`,
+        },
+        offset,
+    };
+}
+
+describe("TurnEventHub", () => {
+    it("fans out to subscribeAll and turn-scoped listeners in order", () => {
+        const hub = new TurnEventHub();
+        const all: TurnBusEvent[] = [];
+        const scoped: TurnBusEvent[] = [];
+        hub.subscribeAll((e) => all.push(e));
+        hub.subscribe("t1", (e) => scoped.push(e));
+
+        hub.publish(busEvent("t1", 1));
+        hub.publish(busEvent("t2", 1));
+        hub.publish(busEvent("t1", 2));
+
+        expect(all.map((e) => [e.turnId, e.offset])).toEqual([
+            ["t1", 1],
+            ["t2", 1],
+            ["t1", 2],
+        ]);
+        expect(scoped.map((e) => e.offset)).toEqual([1, 2]);
+    });
+
+    it("unsubscribe stops delivery and cleans up turn-scoped sets", () => {
+        const hub = new TurnEventHub();
+        const seen: TurnBusEvent[] = [];
+        const unsubAll = hub.subscribeAll((e) => seen.push(e));
+        const unsubTurn = hub.subscribe("t1", (e) => seen.push(e));
+
+        hub.publish(busEvent("t1", 1));
+        expect(seen).toHaveLength(2);
+
+        unsubAll();
+        unsubTurn();
+        hub.publish(busEvent("t1", 2));
+        expect(seen).toHaveLength(2);
+    });
+
+    it("swallows listener errors and keeps delivering to other listeners", () => {
+        const hub = new TurnEventHub();
+        const seen: TurnBusEvent[] = [];
+        hub.subscribeAll(() => {
+            throw new Error("misbehaving observer");
+        });
+        hub.subscribeAll((e) => seen.push(e));
+        hub.subscribe("t1", () => {
+            throw new Error("misbehaving scoped observer");
+        });
+        hub.subscribe("t1", (e) => seen.push(e));
+
+        expect(() => hub.publish(busEvent("t1", 1))).not.toThrow();
+        expect(seen).toHaveLength(2);
+    });
+});

apps/x/packages/core/src/turns/event-hub.ts

@@ -0,0 +1,63 @@
+import type { TurnBusEvent } from "@x/shared/dist/turns.js";
+
+// Process-wide turn event bus: the runtime publishes every turn's events here
+// (durable events with their file offsets, plus live deltas), regardless of
+// who started the turn — session chat, headless runners, spawned sub-agents.
+// Ephemeral and observational, like the lifecycle bus: nothing durable depends
+// on delivery, listener errors are swallowed, and a process crash losing the
+// listeners accurately reflects that no execution is known active.
+export interface ITurnEventBus {
+    publish(event: TurnBusEvent): void;
+}
+
+type Listener = (event: TurnBusEvent) => void;
+
+export class TurnEventHub implements ITurnEventBus {
+    private readonly all = new Set<Listener>();
+    private readonly byTurn = new Map<string, Set<Listener>>();
+
+    publish(event: TurnBusEvent): void {
+        for (const listener of this.all) {
+            try {
+                listener(event);
+            } catch {
+                // observational only
+            }
+        }
+        const scoped = this.byTurn.get(event.turnId);
+        if (!scoped) {
+            return;
+        }
+        for (const listener of scoped) {
+            try {
+                listener(event);
+            } catch {
+                // observational only
+            }
+        }
+    }
+
+    subscribeAll(listener: Listener): () => void {
+        this.all.add(listener);
+        return () => this.all.delete(listener);
+    }
+
+    subscribe(turnId: string, listener: Listener): () => void {
+        let scoped = this.byTurn.get(turnId);
+        if (!scoped) {
+            scoped = new Set();
+            this.byTurn.set(turnId, scoped);
+        }
+        scoped.add(listener);
+        return () => {
+            const listeners = this.byTurn.get(turnId);
+            if (!listeners) {
+                return;
+            }
+            listeners.delete(listener);
+            if (listeners.size === 0) {
+                this.byTurn.delete(turnId);
+            }
+        };
+    }
+}

apps/x/packages/core/src/turns/index.ts

@@ -4,6 +4,7 @@ export * from "./bus.js";
 export * from "./clock.js";
 export * from "./compose-model-request.js";
 export * from "./context-resolver.js";
+export * from "./event-hub.js";
 export * from "./fs-repo.js";
 export * from "./in-memory-turn-repo.js";
 export * from "./keyed-mutex.js";

apps/x/packages/core/src/turns/runtime.test.ts

@@ -5,8 +5,10 @@ import {
     type JsonValue,
     type ResolvedAgent,
     type ToolDescriptor,
+    type TurnBusEvent,
     type TurnEvent,
     type TurnStreamEvent,
+    isDurableTurnEvent,
     reduceTurn,
 } from "@x/shared/dist/turns.js";
 import type { IAgentResolver } from "./agent-resolver.js";
@@ -276,6 +278,13 @@ class FakeBus {
     }
 }
 
+class FakeTurnEventBus {
+    events: TurnBusEvent[] = [];
+    publish(event: TurnBusEvent): void {
+        this.events.push(event);
+    }
+}
+
 class FakeIdGen {
     private n: number;
     constructor(start = 0) {
@@ -308,6 +317,7 @@ function makeRuntime(opts: {
     const checker = opts.checker ?? new FakePermissionChecker();
     const classifier = opts.classifier ?? new FakePermissionClassifier();
     const bus = new FakeBus();
+    const turnEventBus = new FakeTurnEventBus();
     const usage = new FakeUsageReporter();
     const runtime = new TurnRuntime({
         turnRepo: repo,
@@ -320,9 +330,10 @@ function makeRuntime(opts: {
         permissionChecker: checker,
         permissionClassifier: classifier,
         lifecycleBus: bus,
+        turnEventBus,
         usageReporter: usage,
     });
-    return { runtime, repo, models, checker, classifier, bus, usage };
+    return { runtime, repo, models, checker, classifier, bus, turnEventBus, usage };
 }
 
 async function newTurn(
@@ -2314,3 +2325,74 @@ describe("mid-turn tool extension", () => {
         ).toBe("written");
     });
 });
+
+// ---------------------------------------------------------------------------
+// Turn event bus: process-wide tagged event spine
+// ---------------------------------------------------------------------------
+
+describe("turn event bus", () => {
+    it("publishes every durable event with its file offset and deltas without", async () => {
+        const { runtime, repo, turnEventBus } = makeRuntime({
+            models: [
+                respond(
+                    { type: "text_delta", delta: "do" },
+                    { type: "text_delta", delta: "ne" },
+                    completedResp(assistantText("done")),
+                ),
+            ],
+        });
+        const turnId = await newTurn(runtime, { sessionId: "s1" });
+        await advanceAndSettle(runtime, turnId);
+
+        // Durable bus events mirror the persisted file exactly, in order,
+        // with 1-based line offsets — turn_created included.
+        const log = await persisted(repo, turnId);
+        const durable = turnEventBus.events.filter((e) =>
+            isDurableTurnEvent(e.event),
+        );
+        expect(durable.map((e) => e.event)).toEqual(log);
+        expect(durable.map((e) => e.offset)).toEqual(log.map((_, i) => i + 1));
+
+        // Every bus event is tagged with its origin.
+        for (const event of turnEventBus.events) {
+            expect(event.turnId).toBe(turnId);
+            expect(event.sessionId).toBe("s1");
+        }
+
+        // Deltas ride the bus without offsets (they are not durable).
+        const deltas = turnEventBus.events.filter(
+            (e) => !isDurableTurnEvent(e.event),
+        );
+        expect(deltas.map((e) => e.event.type)).toEqual([
+            "text_delta",
+            "text_delta",
+        ]);
+        expect(deltas.every((e) => e.offset === undefined)).toBe(true);
+    });
+
+    it("continues offsets across suspend/resume invocations of one turn", async () => {
+        const { runtime, repo, turnEventBus } = makeRuntime({
+            models: [
+                respond(completedResp(assistantCalls(toolCallPart("c1", "fetch")))),
+                respond(completedResp(assistantText("done"))),
+            ],
+        });
+        const turnId = await newTurn(runtime);
+        const first = await advanceAndSettle(runtime, turnId);
+        expect(first.outcome?.status).toBe("suspended");
+        const second = await advanceAndSettle(runtime, turnId, {
+            type: "async_tool_result",
+            toolCallId: "c1",
+            result: { output: "ok", isError: false },
+        });
+        expect(second.outcome?.status).toBe("completed");
+
+        const log = await persisted(repo, turnId);
+        const durable = turnEventBus.events.filter((e) =>
+            isDurableTurnEvent(e.event),
+        );
+        expect(durable.map((e) => e.event)).toEqual(log);
+        expect(durable.map((e) => e.offset)).toEqual(log.map((_, i) => i + 1));
+        expect(durable.every((e) => e.sessionId === null)).toBe(true);
+    });
+});

apps/x/packages/core/src/turns/runtime.ts

@@ -41,6 +41,7 @@ import {
 } from "./api.js";
 import type { ITurnLifecycleBus } from "./bus.js";
 import type { IClock } from "./clock.js";
+import type { ITurnEventBus } from "./event-hub.js";
 import { composeModelRequest } from "./compose-model-request.js";
 import type { IUsageReporter } from "./usage-reporter.js";
 import type { IContextResolver } from "./context-resolver.js";
@@ -70,6 +71,7 @@ export interface TurnRuntimeDependencies {
     permissionChecker: IPermissionChecker;
     permissionClassifier: IPermissionClassifier;
     lifecycleBus: ITurnLifecycleBus;
+    turnEventBus: ITurnEventBus;
     usageReporter: IUsageReporter;
 }
 
@@ -86,6 +88,7 @@ export class TurnRuntime implements ITurnRuntime {
     private readonly permissionChecker: IPermissionChecker;
     private readonly permissionClassifier: IPermissionClassifier;
     private readonly lifecycleBus: ITurnLifecycleBus;
+    private readonly turnEventBus: ITurnEventBus;
     private readonly usageReporter: IUsageReporter;
 
     constructor({
@@ -99,6 +102,7 @@ export class TurnRuntime implements ITurnRuntime {
         permissionChecker,
         permissionClassifier,
         lifecycleBus,
+        turnEventBus,
         usageReporter,
     }: TurnRuntimeDependencies) {
         this.turnRepo = turnRepo;
@@ -111,6 +115,7 @@ export class TurnRuntime implements ITurnRuntime {
         this.permissionChecker = permissionChecker;
         this.permissionClassifier = permissionClassifier;
         this.lifecycleBus = lifecycleBus;
+        this.turnEventBus = turnEventBus;
         this.usageReporter = usageReporter;
     }
 
@@ -162,6 +167,14 @@ export class TurnRuntime implements ITurnRuntime {
             },
         });
         await this.turnRepo.create(event);
+        // turn_created never flows through an advance's execution stream, so
+        // publish it here; it is always line 1 of the turn file.
+        this.turnEventBus.publish({
+            turnId,
+            sessionId: event.sessionId,
+            event,
+            offset: 1,
+        });
         return turnId;
     }
 
@@ -277,6 +290,7 @@ export class TurnRuntime implements ITurnRuntime {
             clock: this.clock,
             permissionChecker: this.permissionChecker,
             permissionClassifier: this.permissionClassifier,
+            turnEventBus: this.turnEventBus,
         });
         try {
             return await run.run(input);
@@ -309,6 +323,7 @@ class TurnAdvance {
     private readonly clock: IClock;
     private readonly permissionChecker: IPermissionChecker;
     private readonly permissionClassifier: IPermissionClassifier;
+    private readonly turnEventBus: ITurnEventBus;
 
     // Checker "allowed" outcomes are deliberately not durable: after a crash
     // the checker is simply re-consulted.
@@ -334,6 +349,7 @@ class TurnAdvance {
         clock: IClock;
         permissionChecker: IPermissionChecker;
         permissionClassifier: IPermissionClassifier;
+        turnEventBus: ITurnEventBus;
     }) {
         this.turnId = init.turnId;
         this.events = init.events;
@@ -350,12 +366,23 @@ class TurnAdvance {
         this.clock = init.clock;
         this.permissionChecker = init.permissionChecker;
         this.permissionClassifier = init.permissionClassifier;
+        this.turnEventBus = init.turnEventBus;
     }
 
     private get definition(): TurnState["definition"] {
         return this.state.definition;
     }
 
+    // Deltas ride the execution stream and the process-wide bus, never storage.
+    private pushDelta(delta: Extract<TurnStreamEvent, { type: "text_delta" | "reasoning_delta" }>): void {
+        this.stream.push(delta);
+        this.turnEventBus.publish({
+            turnId: this.turnId,
+            sessionId: this.definition.sessionId,
+            event: delta,
+        });
+    }
+
     private now(): string {
         return this.clock.now();
     }
@@ -400,11 +427,20 @@ class TurnAdvance {
 
     private async commit(batch: TEvent[]): Promise<void> {
         await this.turnRepo.append(this.turnId, batch);
+        // this.events holds the full file history (read at advance start), so
+        // its length is the absolute 1-based line offset of each new event.
+        const base = this.events.length;
         this.events.push(...batch);
         this.state = reduceTurn(this.events);
         this.appended = true;
-        for (const event of batch) {
+        for (const [i, event] of batch.entries()) {
             this.stream.push(event);
+            this.turnEventBus.publish({
+                turnId: this.turnId,
+                sessionId: this.definition.sessionId,
+                event,
+                offset: base + i + 1,
+            });
         }
     }
 
@@ -1107,7 +1143,7 @@ class TurnAdvance {
             })) {
                 switch (event.type) {
                     case "text_delta":
-                        this.stream.push({
+                        this.pushDelta({
                             type: "text_delta",
                             turnId: this.turnId,
                             modelCallIndex: index,
@@ -1115,7 +1151,7 @@ class TurnAdvance {
                         });
                         break;
                     case "reasoning_delta":
-                        this.stream.push({
+                        this.pushDelta({
                             type: "reasoning_delta",
                             turnId: this.turnId,
                             modelCallIndex: index,

apps/x/packages/shared/src/ipc.ts

@@ -14,7 +14,7 @@ import {
     TriggersSchema,
 } from './background-task.js';
 import { UserMessage, UserMessageContent } from './message.js';
-import { RequestedAgent, type TurnEvent } from './turns.js';
+import { RequestedAgent, type TurnBusEvent, type TurnEvent } from './turns.js';
 import type { SessionBusEvent, SessionIndexEntry, SessionState } from './sessions.js';
 import { RowboatApiConfig } from './rowboat-account.js';
 import { ZListToolkitsResponse } from './composio.js';
@@ -543,6 +543,14 @@ const ipcSchemas = {
     req: z.custom<SessionBusEvent>(),
     res: z.null(),
   },
+  // 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.
+  'turns:events': {
+    req: z.custom<TurnBusEvent>(),
+    res: z.null(),
+  },
   'services:events': {
     req: ServiceEvent,
     res: z.null(),

apps/x/packages/shared/src/turns.ts

@@ -470,6 +470,26 @@ export type TurnStreamEvent =
     | TextDelta
     | ReasoningDelta;
 
+// One entry on the process-wide turn event bus: every event of every turn the
+// runtime executes, tagged with its origin so consumers can subscribe without
+// knowing who started the turn. `offset` is the 1-based line index of a
+// durable event in the turn's JSONL file — a late subscriber can fetch the
+// turn snapshot and discard bus events with offset <= snapshot length to join
+// a live turn without gaps or duplicates. Deltas are not durable and carry no
+// offset.
+export interface TurnBusEvent {
+    turnId: string;
+    sessionId: string | null;
+    event: TurnStreamEvent;
+    offset?: number;
+}
+
+export function isDurableTurnEvent(
+    event: TurnStreamEvent,
+): event is z.infer<typeof TurnEvent> {
+    return event.type !== "text_delta" && event.type !== "reasoning_delta";
+}
+
 // ---------------------------------------------------------------------------
 // Derived turn state
 // ---------------------------------------------------------------------------