diff --git a/apps/x/apps/main/src/ipc.ts b/apps/x/apps/main/src/ipc.ts index ed7a4e4c..4ae5b8c4 100644 --- a/apps/x/apps/main/src/ipc.ts +++ b/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 | 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): 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): 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): void { + broadcastToWindows('runs:events', event); +} + function emitServiceEvent(event: z.infer): 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 { @@ -667,12 +656,7 @@ export async function startCodeSessionStatusWatcher(): Promise { const tracker = container.resolve('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 { // 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('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'); 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); }); } diff --git a/apps/x/apps/main/src/main.ts b/apps/x/apps/main/src/main.ts index 08299ce8..44358b45 100644 --- a/apps/x/apps/main/src/main.ts +++ b/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 diff --git a/apps/x/apps/renderer/src/components/sub-agent-block.tsx b/apps/x/apps/renderer/src/components/sub-agent-block.tsx index 71a81a41..2d295f03 100644 --- a/apps/x/apps/renderer/src/components/sub-agent-block.tsx +++ b/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(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 ( diff --git a/apps/x/apps/renderer/src/hooks/use-turn.ts b/apps/x/apps/renderer/src/hooks/use-turn.ts new file mode 100644 index 00000000..e619fcc2 --- /dev/null +++ b/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(null) + const [error, setError] = useState(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 } +} diff --git a/apps/x/apps/renderer/src/lib/agent-transcript.ts b/apps/x/apps/renderer/src/lib/agent-transcript.ts index 71c4f75e..5ec5b2d0 100644 --- a/apps/x/apps/renderer/src/lib/agent-transcript.ts +++ b/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>, ): 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, diff --git a/apps/x/apps/renderer/src/lib/broadcast-feed.ts b/apps/x/apps/renderer/src/lib/broadcast-feed.ts new file mode 100644 index 00000000..0d6e59d6 --- /dev/null +++ b/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 = (event: T) => void +export type FeedSource = (listener: FeedListener) => () => void + +export interface BroadcastFeed { + subscribe(listener: FeedListener): () => void +} + +export function createBroadcastFeed(source: FeedSource): BroadcastFeed { + const listeners = new Set>() + 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): () => void { + ensureStarted() + listeners.add(listener) + return () => { + listeners.delete(listener) + } + }, + } +} diff --git a/apps/x/apps/renderer/src/lib/session-chat/feed.ts b/apps/x/apps/renderer/src/lib/session-chat/feed.ts index bac8a735..1a633042 100644 --- a/apps/x/apps/renderer/src/lib/session-chat/feed.ts +++ b/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 +export type SessionFeedSource = FeedSource + export function createSessionFeed(source: SessionFeedSource) { - const listeners = new Set() - 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(source) } const appFeed = createSessionFeed((listener) => window.ipc.on('sessions:events', listener)) diff --git a/apps/x/apps/renderer/src/lib/turn-feed.ts b/apps/x/apps/renderer/src/lib/turn-feed.ts new file mode 100644 index 00000000..56e7bbec --- /dev/null +++ b/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((listener) => + window.ipc.on('turns:events', listener), +) + +export function subscribeTurnFeed(listener: FeedListener): () => void { + return appFeed.subscribe(listener) +} diff --git a/apps/x/apps/renderer/src/lib/turn-follower.test.ts b/apps/x/apps/renderer/src/lib/turn-follower.test.ts new file mode 100644 index 00000000..b17adbdd --- /dev/null +++ b/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 { + return new Promise((resolve) => setTimeout(resolve, 0)) +} + +function harness(fetches: Array) { + 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((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) + }) +}) diff --git a/apps/x/apps/renderer/src/lib/turn-follower.ts b/apps/x/apps/renderer/src/lib/turn-follower.ts new file mode 100644 index 00000000..7c289263 --- /dev/null +++ b/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> }> + 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> | null = null + let pending: TurnBusEvent[] = [] + let fetching = false + let retries = 0 + let retryTimer: ReturnType | 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() + } +} diff --git a/apps/x/packages/core/docs/turn-runtime-design.md b/apps/x/packages/core/docs/turn-runtime-design.md index f4169b64..5a4139c1 100644 --- a/apps/x/packages/core/docs/turn-runtime-design.md +++ b/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: diff --git a/apps/x/packages/core/src/di/container.ts b/apps/x/packages/core/src/di/container.ts index 2ba9ec84..2e926812 100644 --- a/apps/x/packages/core/src/di/container.ts +++ b/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(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(TurnEventHub).singleton(), usageReporter: asClass(RealUsageReporter).singleton(), agentResolver: asFunction( () => diff --git a/apps/x/packages/core/src/turns/event-hub.test.ts b/apps/x/packages/core/src/turns/event-hub.test.ts new file mode 100644 index 00000000..8e054c0c --- /dev/null +++ b/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); + }); +}); diff --git a/apps/x/packages/core/src/turns/event-hub.ts b/apps/x/packages/core/src/turns/event-hub.ts new file mode 100644 index 00000000..ba689c5a --- /dev/null +++ b/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(); + private readonly byTurn = new Map>(); + + 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); + } + }; + } +} diff --git a/apps/x/packages/core/src/turns/index.ts b/apps/x/packages/core/src/turns/index.ts index 77153361..ff15613e 100644 --- a/apps/x/packages/core/src/turns/index.ts +++ b/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"; diff --git a/apps/x/packages/core/src/turns/runtime.test.ts b/apps/x/packages/core/src/turns/runtime.test.ts index cec4a2c4..5b1c8d41 100644 --- a/apps/x/packages/core/src/turns/runtime.test.ts +++ b/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); + }); +}); diff --git a/apps/x/packages/core/src/turns/runtime.ts b/apps/x/packages/core/src/turns/runtime.ts index 2b1b7065..521a8f19 100644 --- a/apps/x/packages/core/src/turns/runtime.ts +++ b/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): 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 { 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, diff --git a/apps/x/packages/shared/src/ipc.ts b/apps/x/packages/shared/src/ipc.ts index f09d0a32..cf6dd880 100644 --- a/apps/x/packages/shared/src/ipc.ts +++ b/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(), 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(), + res: z.null(), + }, 'services:events': { req: ServiceEvent, res: z.null(), diff --git a/apps/x/packages/shared/src/turns.ts b/apps/x/packages/shared/src/turns.ts index c290607c..192e77ff 100644 --- a/apps/x/packages/shared/src/turns.ts +++ b/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 { + return event.type !== "text_delta" && event.type !== "reasoning_delta"; +} + // --------------------------------------------------------------------------- // Derived turn state // ---------------------------------------------------------------------------