diff --git a/apps/x/apps/main/src/ipc.ts b/apps/x/apps/main/src/ipc.ts index 4ae5b8c4..e47d7593 100644 --- a/apps/x/apps/main/src/ipc.ts +++ b/apps/x/apps/main/src/ipc.ts @@ -697,9 +697,32 @@ export function startSessionsWatcher(): void { // runtime executes (session chat, headless background/knowledge runners, // spawned sub-agents), tagged with sessionId and the event's file offset so // consumers can join a live turn against a sessions:getTurn snapshot without -// gaps or duplicates. Deltas are deliberately not broadcast here in v1 — -// session chat already streams them via sessions:events; scoping deltas to -// subscribed turns is a follow-up. +// gaps or duplicates. Durable events are broadcast to every window; +// text/reasoning deltas are high-volume and ephemeral, so they are sent only +// to windows that subscribed to that turn via turns:subscribe. +const turnDeltaSubs = new Map>(); + +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('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('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('sessions'); diff --git a/apps/x/apps/renderer/src/components/bg-tasks-view.tsx b/apps/x/apps/renderer/src/components/bg-tasks-view.tsx index 45e23169..c0c3610f 100644 --- a/apps/x/apps/renderer/src/components/bg-tasks-view.tsx +++ b/apps/x/apps/renderer/src/components/bg-tasks-view.tsx @@ -15,7 +15,8 @@ import { useBackgroundTaskAgentStatus } from '@/hooks/use-bg-task-agent-status' import { formatRelativeTime } from '@/lib/relative-time' import { toast } from '@/lib/toast' import type { ConversationItem } from '@/lib/chat-conversation' -import { fetchAgentRunTranscript, type AgentRunTranscript } from '@/lib/agent-transcript' +import { fetchAgentRunTranscript } from '@/lib/agent-transcript' +import { useAgentRunTranscript } from '@/hooks/use-agent-run-transcript' import { CompactConversation } from '@/components/compact-conversation' import { RichMarkdownViewer } from '@/components/rich-markdown-viewer' import { HtmlFileViewer } from '@/components/html-file-viewer' @@ -1116,29 +1117,9 @@ function RunTranscriptView({ isInFlight: boolean onBack: () => void }) { - const [transcript, setTranscript] = useState(null) - const [loading, setLoading] = useState(true) - const [error, setError] = useState(null) - - useEffect(() => { - let cancelled = false - setLoading(true) - setError(null) - void (async () => { - try { - const t = await fetchAgentRunTranscript(runId) - if (cancelled) return - setTranscript(t) - } catch (err) { - if (cancelled) return - setError(err instanceof Error ? err.message : String(err)) - setTranscript(null) - } finally { - if (!cancelled) setLoading(false) - } - })() - return () => { cancelled = true } - }, [runId]) + // Live via the turns:events spine: an in-flight run's transcript streams + // in as the agent works; settled runs render from one snapshot fetch. + const { transcript, loading, error } = useAgentRunTranscript(runId) const summary = transcript ?? undefined const items: ConversationItem[] = transcript?.items ?? [] diff --git a/apps/x/apps/renderer/src/components/live-note-sidebar.tsx b/apps/x/apps/renderer/src/components/live-note-sidebar.tsx index 18654729..f863e6cb 100644 --- a/apps/x/apps/renderer/src/components/live-note-sidebar.tsx +++ b/apps/x/apps/renderer/src/components/live-note-sidebar.tsx @@ -13,7 +13,7 @@ import { import { LiveNoteSchema, type LiveNote, type Triggers } from '@x/shared/dist/live-note.js' import { useLiveNoteAgentStatus } from '@/hooks/use-live-note-agent-status' import { formatRelativeTime } from '@/lib/relative-time' -import { fetchAgentRunTranscript, type AgentRunTranscript } from '@/lib/agent-transcript' +import { useAgentRunTranscript } from '@/hooks/use-agent-run-transcript' import { CompactConversation } from '@/components/compact-conversation' export type OpenLiveNotePanelDetail = { @@ -659,37 +659,14 @@ function SectionRegion({ label, children }: { label?: string; children: React.Re } function LastRunTab({ live }: { live: LiveNote }) { - const [transcript, setTranscript] = useState(null) - const [loadingRun, setLoadingRun] = useState(false) - const [fetchError, setFetchError] = useState(null) - const runId = live.lastRunId ?? null - - useEffect(() => { - if (!runId) { - setTranscript(null) - setFetchError(null) - setLoadingRun(false) - return - } - let cancelled = false - setLoadingRun(true) - setFetchError(null) - void (async () => { - try { - const t = await fetchAgentRunTranscript(runId) - if (cancelled) return - setTranscript(t) - } catch (err) { - if (cancelled) return - setFetchError(err instanceof Error ? err.message : String(err)) - setTranscript(null) - } finally { - if (!cancelled) setLoadingRun(false) - } - })() - return () => { cancelled = true } - }, [runId]) + // Live via the turns:events spine: an in-flight run's transcript streams + // in as the agent works; settled runs render from one snapshot fetch. + const { + transcript, + loading: loadingRun, + error: fetchError, + } = useAgentRunTranscript(runId) if (!runId) { return ( diff --git a/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.test.tsx b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.test.tsx new file mode 100644 index 00000000..eda45598 --- /dev/null +++ b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.test.tsx @@ -0,0 +1,77 @@ +import { renderHook, waitFor } from '@testing-library/react' +import { beforeEach, describe, expect, it } from 'vitest' +import { completedTurnLog } from '@/lib/session-chat/test-fixtures' +import { useAgentRunTranscript } from './use-agent-run-transcript' + +// Same preload stub as use-turn.test.tsx: the hook rides useTurn (real +// turn-feed singleton) plus a runs:fetch legacy fallback. +let handlers: Record Promise> = {} + +;(window as unknown as { ipc: unknown }).ipc = { + on: () => () => undefined, + invoke: (channel: string, args: unknown) => { + const handler = handlers[channel] + return handler ? handler(args) : Promise.resolve({ success: true }) + }, +} + +beforeEach(() => { + handlers = {} +}) + +describe('useAgentRunTranscript', () => { + it('renders a turn-backed transcript', async () => { + handlers['sessions:getTurn'] = async () => ({ + turnId: 'run-1', + events: completedTurnLog('run-1', 's1', 'do the task', 'task done'), + }) + const { result } = renderHook(() => useAgentRunTranscript('run-1')) + + await waitFor(() => expect(result.current.transcript).not.toBeNull()) + expect(result.current.transcript?.id).toBe('run-1') + expect(result.current.transcript?.summary).toBe('task done') + expect(result.current.loading).toBe(false) + expect(result.current.error).toBeNull() + }) + + it('falls back to legacy runs:fetch when the id is not a turn', async () => { + handlers['sessions:getTurn'] = async () => { + throw new Error('turn not found: legacy id') + } + handlers['runs:fetch'] = async (args) => ({ + id: (args as { runId: string }).runId, + createdAt: '2026-07-01T00:00:00Z', + subUseCase: 'cron', + log: [], + }) + const { result } = renderHook(() => useAgentRunTranscript('legacy-run')) + + await waitFor(() => expect(result.current.transcript).not.toBeNull()) + expect(result.current.transcript?.id).toBe('legacy-run') + expect(result.current.transcript?.trigger).toBe('cron') + expect(result.current.loading).toBe(false) + expect(result.current.error).toBeNull() + }) + + it('surfaces an error when both the turn and legacy paths fail', async () => { + handlers['sessions:getTurn'] = async () => { + throw new Error('turn not found') + } + handlers['runs:fetch'] = async () => { + throw new Error('run not found either') + } + const { result } = renderHook(() => useAgentRunTranscript('ghost-run')) + + await waitFor(() => expect(result.current.error).not.toBeNull()) + expect(result.current.error).toMatch(/run not found either/) + expect(result.current.transcript).toBeNull() + expect(result.current.loading).toBe(false) + }) + + it('is idle without a run id', () => { + const { result } = renderHook(() => useAgentRunTranscript(null)) + expect(result.current.transcript).toBeNull() + expect(result.current.loading).toBe(false) + expect(result.current.error).toBeNull() + }) +}) diff --git a/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts new file mode 100644 index 00000000..eaa2b6ca --- /dev/null +++ b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts @@ -0,0 +1,61 @@ +import { useEffect, useMemo, useState } from 'react' +import { + runToTranscript, + turnStateToTranscript, + type AgentRunTranscript, +} from '@/lib/agent-transcript' +import { useTurn } from '@/hooks/use-turn' + +export interface UseAgentRunTranscriptResult { + transcript: AgentRunTranscript | null + loading: boolean + error: string | null +} + +// Live transcript of a headless agent run (background task, live note) by +// run id. Run ids are turn ids since the runs→turns migration, so useTurn +// keeps the transcript live via the turns:events spine; ids whose turn +// snapshot definitively fails (pre-migration legacy files) fall back to one +// runs:fetch. Callers open this for runs that already exist, so a failed +// snapshot means "not a turn", not "not created yet" — maxRetries 0 makes +// the legacy fallback immediate. +export function useAgentRunTranscript( + runId: string | undefined | null, +): UseAgentRunTranscriptResult { + const id = runId ?? undefined + const { state, error: turnError, snapshotFailed } = useTurn(id, { maxRetries: 0 }) + const [legacy, setLegacy] = useState(null) + const [legacyError, setLegacyError] = useState(null) + + useEffect(() => { + setLegacy(null) + setLegacyError(null) + if (!id || !snapshotFailed) { + return + } + let alive = true + window.ipc + .invoke('runs:fetch', { runId: id }) + .then((run) => { + if (alive) setLegacy(runToTranscript(run)) + }) + .catch((err: unknown) => { + if (alive) setLegacyError(err instanceof Error ? err.message : String(err)) + }) + return () => { + alive = false + } + }, [id, snapshotFailed]) + + const transcript = useMemo(() => { + if (id && state) { + return turnStateToTranscript(id, state) + } + return legacy + }, [id, state, legacy]) + + const error = transcript ? null : (turnError ?? legacyError) + const loading = !!id && !transcript && !error + + return { transcript, loading, error } +} diff --git a/apps/x/apps/renderer/src/hooks/use-turn.test.tsx b/apps/x/apps/renderer/src/hooks/use-turn.test.tsx new file mode 100644 index 00000000..60e58f98 --- /dev/null +++ b/apps/x/apps/renderer/src/hooks/use-turn.test.tsx @@ -0,0 +1,148 @@ +import { act, renderHook, waitFor } from '@testing-library/react' +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type { TurnBusEvent } from '@x/shared/src/turns.js' +import { + assistantText, + completed, + completedTurnLog, + created, + requested, + turnCompleted, + user, + type TEvent, +} from '@/lib/session-chat/test-fixtures' +import { useTurn } from './use-turn' + +// The hook wires the real turn-feed singleton and window.ipc, so the tests +// stub the preload surface: `on` captures the feed listener (the feed +// attaches lazily on first subscribe), `invoke` routes by channel through a +// per-test handler map. +let feedListener: ((event: TurnBusEvent) => void) | null = null +let handlers: Record Promise> = {} + +;(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): 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((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) + }) +}) diff --git a/apps/x/apps/renderer/src/hooks/use-turn.ts b/apps/x/apps/renderer/src/hooks/use-turn.ts index e619fcc2..da1c5ed4 100644 --- a/apps/x/apps/renderer/src/hooks/use-turn.ts +++ b/apps/x/apps/renderer/src/hooks/use-turn.ts @@ -1,4 +1,4 @@ -import { useEffect, useState } from 'react' +import { useEffect, useRef, useState } from 'react' import type { TurnState } from '@x/shared/src/turns.js' import { subscribeTurnFeed } from '@/lib/turn-feed' import { followTurn } from '@/lib/turn-follower' @@ -6,6 +6,10 @@ import { followTurn } from '@/lib/turn-follower' export interface UseTurnResult { state: TurnState | null error: string | null + // Snapshot retries exhausted with no state — the id may not be a turn at + // all (e.g. a pre-migration legacy run id). Clears if a later feed event + // recovers the turn. + snapshotFailed: boolean } // Live view of one turn by id: snapshot via sessions:getTurn, then durable @@ -14,20 +18,25 @@ export interface UseTurnResult { // sub-agents. export function useTurn( turnId: string | undefined, - opts?: { enabled?: boolean }, + opts?: { enabled?: boolean; maxRetries?: number }, ): UseTurnResult { const enabled = opts?.enabled ?? true + const maxRetries = opts?.maxRetries const [state, setState] = useState(null) const [error, setError] = useState(null) + const [snapshotFailed, setSnapshotFailed] = useState(false) + const lastTurnId = useRef(undefined) useEffect(() => { - if (!turnId) { + // A different turn means the previous turn's state is stale; a mere + // enabled toggle keeps the last rendered state while hidden. + if (turnId !== lastTurnId.current) { + lastTurnId.current = turnId setState(null) setError(null) - return + setSnapshotFailed(false) } - if (!enabled) { - // Keep the last rendered state while hidden; re-enabling refetches. + if (!turnId || !enabled) { return } return followTurn(turnId, { @@ -36,10 +45,13 @@ export function useTurn( onState: (next) => { setState(next) setError(null) + setSnapshotFailed(false) }, onError: (message) => setError(message), + onSnapshotFailed: () => setSnapshotFailed(true), + ...(maxRetries === undefined ? {} : { maxRetries }), }) - }, [turnId, enabled]) + }, [turnId, enabled, maxRetries]) - return { state, error } + return { state, error, snapshotFailed } } diff --git a/apps/x/apps/renderer/src/hooks/useSessionChat.test.tsx b/apps/x/apps/renderer/src/hooks/useSessionChat.test.tsx index b1b57124..44b84881 100644 --- a/apps/x/apps/renderer/src/hooks/useSessionChat.test.tsx +++ b/apps/x/apps/renderer/src/hooks/useSessionChat.test.tsx @@ -1,9 +1,8 @@ import { StrictMode } from 'react' import { act, renderHook, waitFor } from '@testing-library/react' import { describe, expect, it } from 'vitest' -import type { SessionBusEvent } from '@x/shared/src/sessions.js' +import type { TurnBusEvent } from '@x/shared/src/turns.js' import type { SessionsClient } from '@/lib/session-chat/client' -import type { SessionFeedListener } from '@/lib/session-chat/feed' import { assistantText, completed, @@ -21,8 +20,9 @@ const S1 = 'sess-1' function makeDeps() { const calls: Array<{ method: string; args: unknown[] }> = [] - let emit: SessionFeedListener = () => undefined + let emit: (event: TurnBusEvent) => void = () => undefined let unsubscribed = 0 + const deltaSubs: string[] = [] const sessions = new Map([[S1, sessionState(S1, ['turn-1'])]]) const turns = new Map([['turn-1', completedTurnLog('turn-1', S1, 'q1', 'a1')]]) const client: SessionsClient = { @@ -54,22 +54,46 @@ function makeDeps() { return { deps: { client, - subscribeFeed: (listener: SessionFeedListener) => { + subscribeTurnFeed: (listener: (event: TurnBusEvent) => void) => { emit = listener return () => { unsubscribed += 1 } }, + subscribeDeltas: (turnId: string) => { + deltaSubs.push(turnId) + return () => { + const i = deltaSubs.indexOf(turnId) + if (i >= 0) deltaSubs.splice(i, 1) + } + }, }, calls, - emit: (event: SessionBusEvent) => emit(event), + emit: (event: TurnBusEvent) => emit(event), + deltaSubs, getUnsubscribed: () => unsubscribed, } } +function durable( + turnId: string, + event: TurnBusEvent['event'], + offset: number, +): TurnBusEvent { + return { turnId, sessionId: S1, event, offset } +} + +function delta(turnId: string, text: string): TurnBusEvent { + return { + turnId, + sessionId: S1, + event: { type: 'text_delta', turnId, modelCallIndex: 0, delta: text }, + } +} + describe('useSessionChat', () => { it('seeds from the session, follows live events, and routes actions', async () => { - const { deps, calls, emit } = makeDeps() + const { deps, calls, emit, deltaSubs } = makeDeps() const { result } = renderHook(() => useSessionChat(S1, deps), { wrapper: StrictMode }) await waitFor(() => { @@ -78,25 +102,23 @@ describe('useSessionChat', () => { expect( result.current.chatState?.conversation.filter(isChatMessage).map((m) => m.content), ).toEqual(['q1', 'a1']) + // The window subscribed to the latest turn's deltas. + expect(deltaSubs).toEqual(['turn-1']) - // A new turn streams in over the feed. + // A new turn streams in over the turns:events spine. act(() => { - emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: created('turn-2', S1, user('q2')) }) - emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: requested('turn-2', 0) }) - emit({ - kind: 'turn-event', - sessionId: S1, - turnId: 'turn-2', - event: { type: 'text_delta', turnId: 'turn-2', modelCallIndex: 0, delta: 'a2…' }, - }) + emit(durable('turn-2', created('turn-2', S1, user('q2')), 1)) + emit(durable('turn-2', requested('turn-2', 0), 2)) + emit(delta('turn-2', 'a2…')) }) expect(result.current.latestTurnId).toBe('turn-2') expect(result.current.chatState?.currentAssistantMessage).toBe('a2…') expect(result.current.chatState?.isProcessing).toBe(true) + expect(deltaSubs).toEqual(['turn-2']) act(() => { - emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: completed('turn-2', 0, assistantText('a2')) }) - emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: turnCompleted('turn-2', 'a2') }) + emit(durable('turn-2', completed('turn-2', 0, assistantText('a2')), 3)) + emit(durable('turn-2', turnCompleted('turn-2', 'a2'), 4)) }) expect(result.current.chatState?.isProcessing).toBe(false) @@ -111,13 +133,14 @@ describe('useSessionChat', () => { }) it('unsubscribes from the feed on unmount (StrictMode double-mounts included)', async () => { - const { deps, getUnsubscribed } = makeDeps() + const { deps, getUnsubscribed, deltaSubs } = makeDeps() const { unmount } = renderHook(() => useSessionChat(S1, deps), { wrapper: StrictMode, }) unmount() // StrictMode's simulated cleanup plus the real unmount: every subscribe - // was matched by an unsubscribe. + // was matched by an unsubscribe, and no delta subscription leaks. expect(getUnsubscribed()).toBe(2) + expect(deltaSubs).toEqual([]) }) }) diff --git a/apps/x/apps/renderer/src/hooks/useSessionChat.ts b/apps/x/apps/renderer/src/hooks/useSessionChat.ts index 6f8ee5ce..363c7300 100644 --- a/apps/x/apps/renderer/src/hooks/useSessionChat.ts +++ b/apps/x/apps/renderer/src/hooks/useSessionChat.ts @@ -1,11 +1,22 @@ import { useEffect, useMemo, useState, useSyncExternalStore } from 'react' import { ipcSessionsClient } from '@/lib/session-chat/client' -import { subscribeSessionFeed } from '@/lib/session-chat/feed' +import { subscribeTurnFeed } from '@/lib/turn-feed' import { SessionChatStore, type SessionChatStoreDeps } from '@/lib/session-chat/store' +// Declare "this window is watching turn X" so main forwards its deltas. +// Fire-and-forget on both edges: a lost subscribe only degrades streaming +// granularity (durable events still arrive), never correctness. +function subscribeDeltas(turnId: string): () => void { + void window.ipc.invoke('turns:subscribe', { turnId }).catch(() => undefined) + return () => { + void window.ipc.invoke('turns:unsubscribe', { turnId }).catch(() => undefined) + } +} + const defaultDeps: SessionChatStoreDeps = { client: ipcSessionsClient, - subscribeFeed: subscribeSessionFeed, + subscribeTurnFeed, + subscribeDeltas, } // Thin subscription over SessionChatStore — all logic (seeding, feed events, diff --git a/apps/x/apps/renderer/src/hooks/useSessions.ts b/apps/x/apps/renderer/src/hooks/useSessions.ts index 9c4a7213..6c1bad1f 100644 --- a/apps/x/apps/renderer/src/hooks/useSessions.ts +++ b/apps/x/apps/renderer/src/hooks/useSessions.ts @@ -1,16 +1,16 @@ import { useEffect, useMemo, useState, useSyncExternalStore } from 'react' import { ipcSessionsClient } from '@/lib/session-chat/client' import { subscribeSessionFeed } from '@/lib/session-chat/feed' -import { SessionListStore, type SessionChatStoreDeps } from '@/lib/session-chat/store' +import { SessionListStore, type SessionListStoreDeps } from '@/lib/session-chat/store' -const defaultDeps: SessionChatStoreDeps = { +const defaultDeps: SessionListStoreDeps = { client: ipcSessionsClient, subscribeFeed: subscribeSessionFeed, } // The session list (chat history sidebar): seeded from sessions:list, kept // current by index-changed feed events. Logic lives in SessionListStore. -export function useSessions(deps: SessionChatStoreDeps = defaultDeps) { +export function useSessions(deps: SessionListStoreDeps = defaultDeps) { const [store] = useState(() => new SessionListStore(deps)) useEffect(() => { const disconnect = store.connect() diff --git a/apps/x/apps/renderer/src/lib/session-chat/store.test.ts b/apps/x/apps/renderer/src/lib/session-chat/store.test.ts index 10955cea..1224faa2 100644 --- a/apps/x/apps/renderer/src/lib/session-chat/store.test.ts +++ b/apps/x/apps/renderer/src/lib/session-chat/store.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from 'vitest' -import type { SessionBusEvent, SessionState } from '@x/shared/src/sessions.js' +import type { SessionState } from '@x/shared/src/sessions.js' +import { isDurableTurnEvent, type TurnBusEvent } from '@x/shared/src/turns.js' import { isChatMessage } from '@/lib/chat-conversation' import type { SessionsClient } from './client' import type { SessionFeedListener } from './feed' @@ -86,12 +87,16 @@ class FakeClient implements SessionsClient { function makeStore() { const client = new FakeClient() - let emit: SessionFeedListener = () => undefined + let emit: (event: TurnBusEvent) => void = () => undefined let subscribed = 0 let unsubscribed = 0 + // Live delta subscriptions by turn id (multiset semantics not needed: the + // store holds at most one). + const deltaSubs: string[] = [] + offsetByTurn.clear() const store = new SessionChatStore({ client, - subscribeFeed: (listener) => { + subscribeTurnFeed: (listener) => { subscribed += 1 emit = listener return () => { @@ -99,26 +104,44 @@ function makeStore() { emit = () => undefined } }, + subscribeDeltas: (turnId) => { + deltaSubs.push(turnId) + return () => { + const i = deltaSubs.indexOf(turnId) + if (i >= 0) deltaSubs.splice(i, 1) + } + }, }) const disconnect = store.connect() return { client, store, disconnect, - emit: (event: SessionBusEvent) => emit(event), + emit: (event: TurnBusEvent) => emit(event), + deltaSubs, getSubscribed: () => subscribed, getUnsubscribed: () => unsubscribed, } } +// Tags an event for the turn feed; durable events get their 1-based per-turn +// offset assigned automatically (mirroring the file line index the runtime +// stamps on the bus envelope). +const offsetByTurn = new Map() + function turnEvent( sessionId: string, turnId: string, event: | TEvent | { type: 'text_delta' | 'reasoning_delta'; turnId: string; modelCallIndex: number; delta: string }, -): SessionBusEvent { - return { kind: 'turn-event', sessionId, turnId, event } +): TurnBusEvent { + if (!isDurableTurnEvent(event)) { + return { turnId, sessionId, event } + } + const offset = (offsetByTurn.get(turnId) ?? 0) + 1 + offsetByTurn.set(turnId, offset) + return { turnId, sessionId, event, offset } } describe('SessionChatStore', () => { @@ -203,6 +226,35 @@ describe('SessionChatStore', () => { ).toEqual(['q1', 'a1', 'q2']) }) + it('keeps exactly one delta subscription, following the latest turn', async () => { + const { client, store, emit, deltaSubs, disconnect } = makeStore() + client.sessions.set(S1, sessionState(S1, ['turn-1'])) + client.turns.set('turn-1', completedTurnLog('turn-1', S1, 'q1', 'a1')) + await store.setSession(S1) + expect(deltaSubs).toEqual(['turn-1']) + + emit(turnEvent(S1, 'turn-2', created('turn-2', S1, user('q2')))) + expect(deltaSubs).toEqual(['turn-2']) + + disconnect() + expect(deltaSubs).toEqual([]) + }) + + it('drops duplicate feed events already covered by the snapshot', async () => { + const { client, store, emit } = makeStore() + client.sessions.set(S1, sessionState(S1, [])) + await store.setSession(S1) + emit(turnEvent(S1, 'turn-1', created('turn-1', S1, user('go')))) + emit(turnEvent(S1, 'turn-1', requested('turn-1', 0))) + // A replay of line 2 (e.g. snapshot/live race) must not double-apply — + // the reducer would throw on the impossible history. + emit({ turnId: 'turn-1', sessionId: S1, event: requested('turn-1', 0), offset: 2 }) + emit(turnEvent(S1, 'turn-1', completed('turn-1', 0, assistantText('done')))) + emit(turnEvent(S1, 'turn-1', turnCompleted('turn-1'))) + expect(store.getSnapshot().error).toBeNull() + expect(store.getSnapshot().chatState?.isProcessing).toBe(false) + }) + it('ignores events for other sessions', async () => { const { client, store, emit } = makeStore() client.sessions.set(S1, sessionState(S1, [])) diff --git a/apps/x/apps/renderer/src/lib/session-chat/store.ts b/apps/x/apps/renderer/src/lib/session-chat/store.ts index 2bd0b4b4..5b94b0da 100644 --- a/apps/x/apps/renderer/src/lib/session-chat/store.ts +++ b/apps/x/apps/renderer/src/lib/session-chat/store.ts @@ -2,11 +2,12 @@ import type { z } from 'zod' import type { UserMessage } from '@x/shared/src/message.js' import type { SessionBusEvent, SessionIndexEntry } from '@x/shared/src/sessions.js' import { + isDurableTurnEvent, reduceTurn, type JsonValue, + type TurnBusEvent, type TurnEvent, type TurnState, - type TurnStreamEvent, } from '@x/shared/src/turns.js' import type { SendMessageConfig, SessionsClient } from './client' import type { SessionFeedListener } from './feed' @@ -30,7 +31,12 @@ export interface SessionChatSnapshot { export interface SessionChatStoreDeps { client: SessionsClient - subscribeFeed: (listener: SessionFeedListener) => () => void + // The turns:events spine (durable events with offsets, plus deltas for + // turns this window subscribed to). + subscribeTurnFeed: (listener: (event: TurnBusEvent) => void) => () => void + // Declares "this window is watching turn X" so main forwards its deltas; + // returns the unsubscribe. + subscribeDeltas: (turnId: string) => () => void } // Framework-agnostic controller for one active session's chat. Owns all the @@ -39,7 +45,8 @@ export interface SessionChatStoreDeps { // hook is a thin useSyncExternalStore subscription over it. export class SessionChatStore { private readonly client: SessionsClient - private readonly subscribeFeed: (listener: SessionFeedListener) => () => void + private readonly subscribeTurnFeed: (listener: (event: TurnBusEvent) => void) => () => void + private readonly subscribeDeltas: (turnId: string) => () => void private feedDisconnect: (() => void) | null = null private readonly listeners = new Set<() => void>() @@ -53,6 +60,9 @@ export class SessionChatStore { private error: string | null = null // Guards stale async loads after a session switch. private generation = 0 + // The turn whose deltas this window currently receives. + private deltaTurnId: string | null = null + private deltaUnsub: (() => void) | null = null private snapshot: SessionChatSnapshot = { sessionId: null, @@ -64,7 +74,8 @@ export class SessionChatStore { constructor(deps: SessionChatStoreDeps) { this.client = deps.client - this.subscribeFeed = deps.subscribeFeed + this.subscribeTurnFeed = deps.subscribeTurnFeed + this.subscribeDeltas = deps.subscribeDeltas } // Feed attachment is effect-managed and idempotent so React StrictMode's @@ -72,11 +83,28 @@ export class SessionChatStore { // subscription would be torn down by the first cleanup and never restored). connect(): () => void { if (!this.feedDisconnect) { - this.feedDisconnect = this.subscribeFeed(this.onFeedEvent) + this.feedDisconnect = this.subscribeTurnFeed(this.onTurnEvent) + this.syncDeltas() } return () => { this.feedDisconnect?.() this.feedDisconnect = null + this.syncDeltas() + } + } + + // Keep exactly one delta subscription: the latest turn, while connected. + private syncDeltas(): void { + const want = + this.feedDisconnect && this.latestEvents + ? this.latestEvents[0].turnId + : null + if (want === this.deltaTurnId) return + this.deltaUnsub?.() + this.deltaUnsub = null + this.deltaTurnId = want + if (want) { + this.deltaUnsub = this.subscribeDeltas(want) } } @@ -99,6 +127,7 @@ export class SessionChatStore { this.overlay = emptyOverlay() this.error = null this.loading = sessionId !== null + this.syncDeltas() this.emit() if (sessionId === null) return @@ -112,6 +141,7 @@ export class SessionChatStore { this.priorTurns = reduced.slice(0, -1) this.latestEvents = turns.length > 0 ? turns[turns.length - 1].events : null this.loading = false + this.syncDeltas() this.emit() } catch (error) { if (generation !== this.generation) return @@ -122,23 +152,37 @@ export class SessionChatStore { } } - private onFeedEvent: SessionFeedListener = (event: SessionBusEvent) => { - if (event.kind !== 'turn-event' || event.sessionId !== this.sessionId) return + private onTurnEvent = (event: TurnBusEvent): void => { + if (this.sessionId === null || event.sessionId !== this.sessionId) return const turnEvent = event.event - if (isDurable(turnEvent)) { + if (isDurableTurnEvent(turnEvent) && event.offset !== undefined) { if (turnEvent.type === 'turn_created') { // A new turn started for this session: freeze the previous latest. this.freezeLatest() this.latestEvents = [turnEvent] this.overlay = emptyOverlay() - } else if (this.latestEvents && this.latestEvents[0].turnId === turnEvent.turnId) { - this.latestEvents.push(turnEvent) + this.syncDeltas() + } else if (this.latestEvents && this.latestEvents[0].turnId === event.turnId) { + // Offset join against the local log: drop already-known events, + // append the contiguous next line, refetch on a gap. + if (event.offset <= this.latestEvents.length) { + return + } + if (event.offset === this.latestEvents.length + 1) { + this.latestEvents.push(turnEvent) + } else { + void this.reloadTurn(event.turnId) + return + } } else { // An event for a turn we haven't seen (missed turn_created, e.g. the // feed attached mid-turn): reconcile by refetching that turn. void this.reloadTurn(event.turnId) return } + } else if (!this.latestEvents || this.latestEvents[0].turnId !== event.turnId) { + // Deltas are ephemeral: only the latest turn's deltas paint the overlay. + return } this.overlay = applyOverlay(this.overlay, turnEvent) this.emit() @@ -164,6 +208,7 @@ export class SessionChatStore { this.freezeLatest() } this.latestEvents = turn.events + this.syncDeltas() this.emit() } catch (error) { // The next snapshot-worthy event will retry. @@ -237,10 +282,6 @@ export class SessionChatStore { } } -function isDurable(event: TurnStreamEvent): event is TEvent { - return event.type !== 'text_delta' && event.type !== 'reasoning_delta' -} - // --------------------------------------------------------------------------- // Session list store // --------------------------------------------------------------------------- @@ -250,6 +291,12 @@ export interface SessionListSnapshot { loading: boolean } +export interface SessionListStoreDeps { + client: SessionsClient + // sessions:events — index-changed entries for the session list. + subscribeFeed: (listener: SessionFeedListener) => () => void +} + export class SessionListStore { private readonly client: SessionsClient private readonly subscribeFeed: (listener: SessionFeedListener) => () => void @@ -259,7 +306,7 @@ export class SessionListStore { private loading = true private snapshot: SessionListSnapshot = { sessions: [], loading: true } - constructor(deps: SessionChatStoreDeps) { + constructor(deps: SessionListStoreDeps) { this.client = deps.client this.subscribeFeed = deps.subscribeFeed } diff --git a/apps/x/apps/renderer/src/lib/turn-follower.test.ts b/apps/x/apps/renderer/src/lib/turn-follower.test.ts index b17adbdd..b8dae0dd 100644 --- a/apps/x/apps/renderer/src/lib/turn-follower.test.ts +++ b/apps/x/apps/renderer/src/lib/turn-follower.test.ts @@ -42,6 +42,7 @@ function harness(fetches: Array) { 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) { }, 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) { fetchTurn, states, errors, + snapshotFailures, emit: (event: TurnBusEvent) => listener?.(event), get unsubscribed() { return unsubscribed @@ -155,6 +158,18 @@ describe('followTurn', () => { detach() }) + it('reports snapshot failure once retries are exhausted', async () => { + const h = harness([new Error('turn not found: legacy run id')]) + const detach = followTurn(TURN, h.deps) + await flush() + + // maxRetries 0: the first failure is definitive (legacy-run fallback + // hinges on this signal firing). + expect(h.snapshotFailures).toEqual(['turn not found: legacy run id']) + expect(h.states).toHaveLength(0) + detach() + }) + it('stops delivering after detach and unsubscribes from the feed', async () => { const full = log() const h = harness([full.slice(0, 2)]) diff --git a/apps/x/apps/renderer/src/lib/turn-follower.ts b/apps/x/apps/renderer/src/lib/turn-follower.ts index 7c289263..e0a2878c 100644 --- a/apps/x/apps/renderer/src/lib/turn-follower.ts +++ b/apps/x/apps/renderer/src/lib/turn-follower.ts @@ -23,6 +23,10 @@ export interface TurnFollowerDeps { subscribe: (listener: (event: TurnBusEvent) => void) => () => void onState: (state: TurnState) => void onError: (message: string) => void + // Fired when snapshot retries are exhausted (the id may not be a turn at + // all — e.g. a pre-migration legacy run). A later feed event still retries + // and can recover. + onSnapshotFailed?: (message: string) => void retryDelayMs?: number maxRetries?: number } @@ -89,7 +93,7 @@ export function followTurn(turnId: string, deps: TurnFollowerDeps): () => void { } } publish() - } catch { + } catch (err) { if (!alive) return // Turn file may not exist yet (subscription raced creation) or the // fetch failed transiently. @@ -97,6 +101,8 @@ export function followTurn(turnId: string, deps: TurnFollowerDeps): () => void { if (retries < maxRetries) { retries += 1 retryTimer = setTimeout(() => void fetchSnapshot(), retryDelayMs) + } else { + deps.onSnapshotFailed?.(err instanceof Error ? err.message : String(err)) } } finally { fetching = false diff --git a/apps/x/packages/core/docs/session-design.md b/apps/x/packages/core/docs/session-design.md index 2694f9e9..ab4f122d 100644 --- a/apps/x/packages/core/docs/session-design.md +++ b/apps/x/packages/core/docs/session-design.md @@ -357,18 +357,23 @@ which governs mutation of live logs, not their removal. ## 10. Event forwarding and live UI -1. For every `advanceTurn` it initiates, the session layer consumes - `TurnExecution.events` and forwards each `TurnStreamEvent` — tagged with - `sessionId` — through the application bus to renderer windows over one - IPC channel (`sessions:events`). +1. Live turn delivery is not the session layer's job: the turn runtime + publishes every turn's events to the process-wide turn event bus + (turn-runtime-design.md §17.1), which the app layer bridges to renderer + windows over one IPC channel (`turns:events` — durable events broadcast + with their file offsets; deltas only to windows subscribed to that turn). + The session layer drains each `TurnExecution.events` it initiates so an + unconsumed stream never buffers, and `sessions:events` carries only + `session-index-changed` entries. 2. When `outcome` settles, the session layer updates the index entry and publishes `session-index-changed`. 3. The renderer follows the turn spec's historical/live pattern: fetch turns via `getTurn`, run the shared `reduceTurn` per turn, compose the session timeline turn-by-turn (each turn renders its input and its own activity; the referenced prefix is never re-rendered from context), - append live durable events and re-reduce, and keep text/reasoning deltas - in an ephemeral overlay cleared by canonical responses. + join live durable events by file offset (drop covered, append + contiguous, refetch on gap) and re-reduce, and keep text/reasoning + deltas in an ephemeral overlay cleared by canonical responses. 4. Pending approvals and ask-human prompts render from the suspended turn's reduced state, so they survive restarts without any session-layer bookkeeping. @@ -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: { diff --git a/apps/x/packages/core/docs/turn-runtime-design.md b/apps/x/packages/core/docs/turn-runtime-design.md index 5a4139c1..c3ef6be9 100644 --- a/apps/x/packages/core/docs/turn-runtime-design.md +++ b/apps/x/packages/core/docs/turn-runtime-design.md @@ -1435,9 +1435,10 @@ stream for the initiating caller. events with `offset <= snapshot.length` — no gaps, no duplicates, no sequence numbers in the durable schema. - Text/reasoning deltas are published without an offset (they are not - durable). The app-layer IPC bridge (`turns:events`) forwards durable - events only in v1; deltas still reach session chat via the session-layer - forwarding. Scoping delta delivery to subscribed turns is future work. + durable). The app-layer IPC bridge (`turns:events`) broadcasts durable + events to every window; deltas are forwarded only to windows that + declared they are watching that turn (`turns:subscribe` / + `turns:unsubscribe`, a per-webContents registry in the app layer). - The bus is ephemeral and observational, like the lifecycle bus: listener errors are swallowed, nothing durable depends on delivery, and a crash losing listeners accurately reflects that no execution is known active. diff --git a/apps/x/packages/core/src/agents/headless.ts b/apps/x/packages/core/src/agents/headless.ts index 0f17eae4..0372d977 100644 --- a/apps/x/packages/core/src/agents/headless.ts +++ b/apps/x/packages/core/src/agents/headless.ts @@ -163,6 +163,18 @@ export class HeadlessAgentRunner implements IHeadlessAgentRunner { const execution = this.turnRuntime.advanceTurn(turnId, undefined, { signal: options.signal, }); + // Drain the execution stream: live delivery rides the turn event bus, + // and an unconsumed HotStream would buffer every durable event in + // memory until the turn settles. + void (async () => { + try { + for await (const event of execution.events) { + void event; + } + } catch { + // Infrastructure failures surface through the outcome. + } + })(); const done = execution.outcome.then(async (outcome) => { const state = reduceTurn( (await this.turnRuntime.getTurn(turnId)).events, diff --git a/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts b/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts deleted file mode 100644 index d3d0e6e1..00000000 --- a/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts +++ /dev/null @@ -1,52 +0,0 @@ -export const skill = String.raw` -# Web Search Skill - -You have access to two search tools for finding information on the internet. Choose the right one based on the user's intent. - -## Tools - -### web-search (Brave Search) -Quick, general-purpose web search. Returns titles, URLs, and short descriptions. - -**Best for:** -- Quick lookups for things that change ("current price of Bitcoin", "weather in SF") -- Current events and breaking news -- Finding a specific website or page -- Simple questions with direct answers -- Checking a fact or date - -### research-search (Exa Search) -Deep, research-oriented search. Returns full article text, highlights, and metadata (author, published date). - -**Best for:** -- Exploring a topic in depth ("what are the latest advances in CRISPR") -- Finding articles, blog posts, papers, and quality sources -- Discovering companies, people, or organizations -- Research where you need rich context, not just links -- When the user says "research", "find articles about", "look into", "deep dive" - -**Category filter:** Use the category parameter when the user's intent clearly maps to one: company, research paper, news, tweet, personal site, financial report, people. - -## How Many Searches to Do - -**CRITICAL: Always start with exactly ONE search call.** Pick the single best tool (\`web-search\` or \`research-search\`) and make one request. Wait for the result before deciding if more searches are needed. - -**NEVER call multiple search tools simultaneously.** No parallel web-search + research-search. No firing off two web-searches at once. Always sequential: one search at a time. - -Only make a follow-up search if: -- The first search returned truly uninformative or irrelevant results -- The query has clearly distinct sub-topics that the first search couldn't cover (e.g., "compare X and Y" after getting results for X only) -- The user explicitly asks you to dig deeper - -One good search is almost always enough. Default to one and stop. - -## Choosing Between the Two - -If both tools are attached, prefer: -- \`web-search\` when the user wants a quick answer or specific link -- \`research-search\` when the user wants to learn, explore, or gather sources - -If only one is attached, use whichever is available. -`; - -export default skill; diff --git a/apps/x/packages/core/src/channels/bridge.test.ts b/apps/x/packages/core/src/channels/bridge.test.ts index 2c78f90a..c2b3adb7 100644 --- a/apps/x/packages/core/src/channels/bridge.test.ts +++ b/apps/x/packages/core/src/channels/bridge.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it, vi } from "vitest"; import type { SessionIndexEntry } from "@x/shared/dist/sessions.js"; import type { TurnStreamEvent } from "@x/shared/dist/turns.js"; -import { EmitterSessionBus } from "../sessions/bus.js"; +import { TurnEventHub } from "../turns/event-hub.js"; import { TurnInputError } from "../turns/api.js"; import type { ISessions } from "../sessions/api.js"; import { ChannelBridge, type ModelChoice } from "./bridge.js"; @@ -50,7 +50,7 @@ function askEvent(turnId: string, question: string, options?: string[]): TurnStr interface Harness { bridge: ChannelBridge; - bus: EmitterSessionBus; + bus: TurnEventHub; replies: string[]; reply: (text: string) => Promise; sessions: { @@ -72,9 +72,9 @@ const MODELS: ModelChoice[] = [ ]; function harness(entries: SessionIndexEntry[] = []): Harness { - const bus = new EmitterSessionBus(); + const bus = new TurnEventHub(); const publish = (turnId: string, event: TurnStreamEvent) => - bus.publish({ kind: "turn-event", sessionId: "s1", turnId, event }); + bus.publish({ turnId, sessionId: "s1", event }); const sessions = { createSession: vi.fn(async () => "s1"), sendMessage: vi.fn(async () => ({ turnId: "t1" })), @@ -86,7 +86,7 @@ function harness(entries: SessionIndexEntry[] = []): Harness { const listModels = vi.fn(async () => MODELS); const bridge = new ChannelBridge({ sessions: sessions as unknown as ISessions, - sessionBus: bus, + turnEventBus: bus, listModels, }); const replies: string[] = []; diff --git a/apps/x/packages/core/src/channels/bridge.ts b/apps/x/packages/core/src/channels/bridge.ts index e53be4d8..69e84e9e 100644 --- a/apps/x/packages/core/src/channels/bridge.ts +++ b/apps/x/packages/core/src/channels/bridge.ts @@ -4,7 +4,7 @@ import { assistantText, lastAssistantText } from "../agents/headless.js"; import { TurnInputError } from "../turns/api.js"; import { ASK_HUMAN_TOOL } from "../turns/bridges/real-agent-resolver.js"; import { TurnNotSettledError, type ISessions } from "../sessions/api.js"; -import type { EmitterSessionBus } from "../sessions/bus.js"; +import type { TurnEventHub } from "../turns/event-hub.js"; // Transport-agnostic command layer: inbound texts from a messaging channel // are parsed into commands (list / resume / new / stop / status) or forwarded @@ -136,7 +136,7 @@ export class ChannelBridge { constructor( private readonly deps: { sessions: ISessions; - sessionBus: EmitterSessionBus; + turnEventBus: TurnEventHub; listModels: () => Promise; }, ) {} @@ -502,8 +502,10 @@ export class ChannelBridge { const buffered: Array<{ turnId: string; settled: Settled }> = []; let waiter: { turnId: string; resolve: (settled: Settled) => void } | null = null; let cancelTimer: (() => void) | null = null; - const unsubscribe = this.deps.sessionBus.subscribe((event) => { - if (event.kind !== "turn-event") return; + // The turn event spine carries every turn's events (deltas included + // for subscribed windows, but settleOf ignores those); subscribeAll + // because the turnId is unknown until sendMessage returns. + const unsubscribe = this.deps.turnEventBus.subscribeAll((event) => { const settled = settleOf(event.event); if (!settled) return; if (waiter) { diff --git a/apps/x/packages/core/src/channels/service.ts b/apps/x/packages/core/src/channels/service.ts index 097d415a..a0bed3f4 100644 --- a/apps/x/packages/core/src/channels/service.ts +++ b/apps/x/packages/core/src/channels/service.ts @@ -5,7 +5,7 @@ import type { ChannelsConfig, ChannelsStatus } from "@x/shared/dist/channels.js" import container from "../di/container.js"; import { WorkDir } from "../config/config.js"; import type { ISessions } from "../sessions/api.js"; -import type { EmitterSessionBus } from "../sessions/bus.js"; +import type { TurnEventHub } from "../turns/event-hub.js"; import { isSignedIn } from "../account/account.js"; import { listGatewayModels } from "../models/gateway.js"; import { listOnboardingModels } from "../models/models-dev.js"; @@ -98,7 +98,7 @@ function ensureBridge(): ChannelBridge { if (!bridge) { bridge = new ChannelBridge({ sessions: container.resolve("sessions"), - sessionBus: container.resolve("sessionBus"), + turnEventBus: container.resolve("turnEventBus"), listModels: listBridgeModels, }); } diff --git a/apps/x/packages/core/src/sessions/headless.ts b/apps/x/packages/core/src/sessions/headless.ts deleted file mode 100644 index 071d7fd4..00000000 --- a/apps/x/packages/core/src/sessions/headless.ts +++ /dev/null @@ -1,41 +0,0 @@ -import type { z } from "zod"; -import type { UserMessage } from "@x/shared/dist/message.js"; -import type { - ConversationMessage, - RequestedAgent, -} from "@x/shared/dist/turns.js"; -import type { ITurnRuntime, TurnOutcome } from "../turns/api.js"; - -// Standalone turns for non-session callers (background tasks, live notes, -// knowledge pipelines, scheduled agents): sessionId null, automatic -// permissions, no human. Never appears in the session index; callers keep -// the turnId if they need history. -export async function runHeadlessTurn( - turnRuntime: ITurnRuntime, - input: { - agent: z.infer; - context?: Array>; - input: z.infer; - maxModelCalls?: number; - signal?: AbortSignal; - }, -): Promise<{ turnId: string; outcome: TurnOutcome }> { - const turnId = await turnRuntime.createTurn({ - agent: input.agent, - sessionId: null, - context: input.context ?? [], - input: input.input, - config: { - autoPermission: true, - humanAvailable: false, - ...(input.maxModelCalls === undefined - ? {} - : { maxModelCalls: input.maxModelCalls }), - }, - }); - const execution = turnRuntime.advanceTurn(turnId, undefined, { - signal: input.signal, - }); - const outcome = await execution.outcome; - return { turnId, outcome }; -} diff --git a/apps/x/packages/core/src/sessions/index.ts b/apps/x/packages/core/src/sessions/index.ts index df906d01..7e7e994c 100644 --- a/apps/x/packages/core/src/sessions/index.ts +++ b/apps/x/packages/core/src/sessions/index.ts @@ -1,7 +1,6 @@ export * from "./api.js"; export * from "./bus.js"; export * from "./fs-repo.js"; -export * from "./headless.js"; export * from "./in-memory-session-repo.js"; export * from "./repo.js"; export * from "./session-index.js"; diff --git a/apps/x/packages/core/src/sessions/sessions.test.ts b/apps/x/packages/core/src/sessions/sessions.test.ts index 93fc1a40..e1252a43 100644 --- a/apps/x/packages/core/src/sessions/sessions.test.ts +++ b/apps/x/packages/core/src/sessions/sessions.test.ts @@ -17,7 +17,6 @@ import type { import { TurnInputError } from "../turns/api.js"; import { HotStream } from "../turns/stream.js"; import { TurnNotSettledError } from "./api.js"; -import { runHeadlessTurn } from "./headless.js"; import { InMemorySessionRepo } from "./in-memory-session-repo.js"; import type { ISessionRepo } from "./repo.js"; import { SessionsImpl } from "./sessions.js"; @@ -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", diff --git a/apps/x/packages/core/src/sessions/sessions.ts b/apps/x/packages/core/src/sessions/sessions.ts index acd9d7ef..d42640b1 100644 --- a/apps/x/packages/core/src/sessions/sessions.ts +++ b/apps/x/packages/core/src/sessions/sessions.ts @@ -343,9 +343,11 @@ export class SessionsImpl implements ISessions { await execution.outcome; } - // Every advance this layer initiates: forward its stream to the bus - // tagged with sessionId, keep the abort controller for stopTurn, and - // update the index entry when the outcome settles. + // Every advance this layer initiates: keep the abort controller for + // stopTurn and update the index entry when the outcome settles. Live + // event delivery is not this layer's job — the runtime publishes every + // event to the turn event bus; the execution stream is drained so an + // unconsumed HotStream never buffers events until settle. private startTrackedAdvance( sessionId: string | null, turnId: string, @@ -368,14 +370,7 @@ export class SessionsImpl implements ISessions { void (async () => { try { for await (const event of execution.events) { - if (sessionId !== null) { - this.sessionBus.publish({ - kind: "turn-event", - sessionId, - turnId, - event, - }); - } + void event; } } catch { // Infrastructure failures surface through the outcome. diff --git a/apps/x/packages/shared/src/ipc.ts b/apps/x/packages/shared/src/ipc.ts index cf6dd880..c00220bc 100644 --- a/apps/x/packages/shared/src/ipc.ts +++ b/apps/x/packages/shared/src/ipc.ts @@ -461,7 +461,7 @@ const ipcSchemas = { }, // ── New runtime: sessions + turns (session-design.md) ──────────────────── // Turn-mutating calls return quickly; the renderer follows progress through - // the sessions:events feed and the shared reduceTurn reducer. + // the turns:events feed and the shared reduceTurn reducer. 'sessions:create': { req: z.object({ title: z.string().optional() }), res: z.object({ sessionId: z.string() }), @@ -545,12 +545,24 @@ const ipcSchemas = { }, // Process-wide turn event spine: every turn's durable events (with file // offsets), regardless of who started the turn — session chat, headless - // background/knowledge runners, spawned sub-agents. Deltas are not - // broadcast here in v1; session chat streams them via sessions:events. + // background/knowledge runners, spawned sub-agents. Text/reasoning deltas + // ride the same channel but only reach windows that subscribed to that + // turn via turns:subscribe. 'turns:events': { req: z.custom(), res: z.null(), }, + // Per-window delta subscription: deltas are high-volume and ephemeral, so + // they cross IPC only for turns this window declared it is watching. + // Durable events are always broadcast regardless. + 'turns:subscribe': { + req: z.object({ turnId: z.string() }), + res: z.object({ success: z.literal(true) }), + }, + 'turns:unsubscribe': { + req: z.object({ turnId: z.string() }), + res: z.object({ success: z.literal(true) }), + }, 'services:events': { req: ServiceEvent, res: z.null(), diff --git a/apps/x/packages/shared/src/sessions.ts b/apps/x/packages/shared/src/sessions.ts index 1a52d38f..08435650 100644 --- a/apps/x/packages/shared/src/sessions.ts +++ b/apps/x/packages/shared/src/sessions.ts @@ -1,9 +1,5 @@ import { z } from "zod"; -import { - ModelDescriptor, - type TurnStatus, - type TurnStreamEvent, -} from "./turns.js"; +import { ModelDescriptor, type TurnStatus } from "./turns.js"; // Durable session contract for the session layer (see // packages/core/docs/session-design.md). A session is an append-only chain @@ -171,21 +167,14 @@ export interface SessionIndexEntry { error?: string; } -// What the renderer's single feed consumer receives over IPC: live turn -// stream events (durable events + ephemeral deltas) and index updates. -// entry: null signals deletion. -export type SessionBusEvent = - | { - kind: "turn-event"; - sessionId: string; - turnId: string; - event: TurnStreamEvent; - } - | { - kind: "index-changed"; - sessionId: string; - entry: SessionIndexEntry | null; - }; +// What the renderer's session-feed consumer receives over IPC: session index +// updates only. Turn events (durable + deltas) travel on the turns:events +// spine (TurnBusEvent in turns.ts). entry: null signals deletion. +export type SessionBusEvent = { + kind: "index-changed"; + sessionId: string; + entry: SessionIndexEntry | null; +}; export function sessionIndexEntry( state: SessionState,