From 12a71184451157497709e38663e2e3290b4bfe38 Mon Sep 17 00:00:00 2001 From: Ramnique Singh <30795890+ramnique@users.noreply.github.com> Date: Thu, 9 Jul 2026 11:42:22 +0530 Subject: [PATCH 01/91] =?UTF-8?q?feat(x):=20turn=20event=20spine=20?= =?UTF-8?q?=E2=80=94=20one=20bus=20for=20every=20turn's=20events?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TurnRuntime now publishes every turn's events (durable ones tagged with their 1-based file offset, deltas without) to an injected TurnEventHub, regardless of who started the turn. Main forwards durable events to all windows on one turns:events channel; the renderer's new useTurn(turnId) hook joins live turns gap/duplicate-free via the offset protocol. The sub-agent card drops its 1s polling for push updates, and the nine per-channel window fan-out loops collapse into one broadcastToWindows. Co-Authored-By: Claude Fable 5 --- apps/x/apps/main/src/ipc.ts | 104 +++++------ apps/x/apps/main/src/main.ts | 5 +- .../src/components/sub-agent-block.tsx | 42 ++--- apps/x/apps/renderer/src/hooks/use-turn.ts | 45 +++++ .../apps/renderer/src/lib/agent-transcript.ts | 12 +- .../x/apps/renderer/src/lib/broadcast-feed.ts | 39 ++++ .../renderer/src/lib/session-chat/feed.ts | 42 ++--- apps/x/apps/renderer/src/lib/turn-feed.ts | 17 ++ .../renderer/src/lib/turn-follower.test.ts | 168 ++++++++++++++++++ apps/x/apps/renderer/src/lib/turn-follower.ts | 128 +++++++++++++ .../packages/core/docs/turn-runtime-design.md | 27 +++ apps/x/packages/core/src/di/container.ts | 4 + .../packages/core/src/turns/event-hub.test.ts | 69 +++++++ apps/x/packages/core/src/turns/event-hub.ts | 63 +++++++ apps/x/packages/core/src/turns/index.ts | 1 + .../x/packages/core/src/turns/runtime.test.ts | 84 ++++++++- apps/x/packages/core/src/turns/runtime.ts | 42 ++++- apps/x/packages/shared/src/ipc.ts | 10 +- apps/x/packages/shared/src/turns.ts | 20 +++ 19 files changed, 790 insertions(+), 132 deletions(-) create mode 100644 apps/x/apps/renderer/src/hooks/use-turn.ts create mode 100644 apps/x/apps/renderer/src/lib/broadcast-feed.ts create mode 100644 apps/x/apps/renderer/src/lib/turn-feed.ts create mode 100644 apps/x/apps/renderer/src/lib/turn-follower.test.ts create mode 100644 apps/x/apps/renderer/src/lib/turn-follower.ts create mode 100644 apps/x/packages/core/src/turns/event-hub.test.ts create mode 100644 apps/x/packages/core/src/turns/event-hub.ts 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 // --------------------------------------------------------------------------- From f916cb1d70ca3408083ed15dcbf26a23a4ff6c03 Mon Sep 17 00:00:00 2001 From: Ramnique Singh <30795890+ramnique@users.noreply.github.com> Date: Thu, 9 Jul 2026 14:59:22 +0530 Subject: [PATCH 02/91] refactor(x): finish turn-spine adoption for headless surfaces MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Background-task and live-note transcript views move from fetch-once to the shared useAgentRunTranscript hook (useTurn over the turns:events spine, with a one-shot legacy runs:fetch fallback for pre-migration run ids), so an in-flight run's transcript now streams live. The headless runner drains its execution stream — delivery rides the turn event bus, and an unconsumed HotStream buffered every durable event until settle. Deletes the caller-less runHeadlessTurn duplicate (HeadlessAgentRunner is the implementation; session-design.md now says so) and the orphaned web-search skill file that was never registered. Co-Authored-By: Claude Fable 5 --- .../renderer/src/components/bg-tasks-view.tsx | 29 ++------- .../src/components/live-note-sidebar.tsx | 39 +++--------- .../src/hooks/use-agent-run-transcript.ts | 61 +++++++++++++++++++ apps/x/apps/renderer/src/hooks/use-turn.ts | 28 ++++++--- .../renderer/src/lib/turn-follower.test.ts | 15 +++++ apps/x/apps/renderer/src/lib/turn-follower.ts | 8 ++- apps/x/packages/core/docs/session-design.md | 5 +- apps/x/packages/core/src/agents/headless.ts | 12 ++++ .../assistant/skills/web-search/skill.ts | 52 ---------------- apps/x/packages/core/src/sessions/headless.ts | 41 ------------- apps/x/packages/core/src/sessions/index.ts | 1 - .../core/src/sessions/sessions.test.ts | 21 ------- 12 files changed, 132 insertions(+), 180 deletions(-) create mode 100644 apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts delete mode 100644 apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts delete mode 100644 apps/x/packages/core/src/sessions/headless.ts diff --git a/apps/x/apps/renderer/src/components/bg-tasks-view.tsx b/apps/x/apps/renderer/src/components/bg-tasks-view.tsx index 45e23169..c0c3610f 100644 --- a/apps/x/apps/renderer/src/components/bg-tasks-view.tsx +++ b/apps/x/apps/renderer/src/components/bg-tasks-view.tsx @@ -15,7 +15,8 @@ import { useBackgroundTaskAgentStatus } from '@/hooks/use-bg-task-agent-status' import { formatRelativeTime } from '@/lib/relative-time' import { toast } from '@/lib/toast' import type { ConversationItem } from '@/lib/chat-conversation' -import { fetchAgentRunTranscript, type AgentRunTranscript } from '@/lib/agent-transcript' +import { fetchAgentRunTranscript } from '@/lib/agent-transcript' +import { useAgentRunTranscript } from '@/hooks/use-agent-run-transcript' import { CompactConversation } from '@/components/compact-conversation' import { RichMarkdownViewer } from '@/components/rich-markdown-viewer' import { HtmlFileViewer } from '@/components/html-file-viewer' @@ -1116,29 +1117,9 @@ function RunTranscriptView({ isInFlight: boolean onBack: () => void }) { - const [transcript, setTranscript] = useState(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.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.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/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..f2ba6ccc 100644 --- a/apps/x/packages/core/docs/session-design.md +++ b/apps/x/packages/core/docs/session-design.md @@ -376,7 +376,10 @@ which governs mutation of live logs, not their removal. ## 11. Headless standalone turns A helper covers the non-session callers (background tasks, live notes, -knowledge pipelines, scheduled agents): +knowledge pipelines, scheduled agents). Implemented as +`HeadlessAgentRunner` in `agents/headless.ts` (start/run handle with +turn id, reduced state, and final assistant text); the shape below is +the contract it fulfils: ```ts function runHeadlessTurn(input: { diff --git a/apps/x/packages/core/src/agents/headless.ts b/apps/x/packages/core/src/agents/headless.ts index 0f17eae4..0372d977 100644 --- a/apps/x/packages/core/src/agents/headless.ts +++ b/apps/x/packages/core/src/agents/headless.ts @@ -163,6 +163,18 @@ export class HeadlessAgentRunner implements IHeadlessAgentRunner { const execution = this.turnRuntime.advanceTurn(turnId, undefined, { signal: options.signal, }); + // Drain the execution stream: live delivery rides the turn event bus, + // and an unconsumed HotStream would buffer every durable event in + // memory until the turn settles. + void (async () => { + try { + for await (const event of execution.events) { + void event; + } + } catch { + // Infrastructure failures surface through the outcome. + } + })(); const done = execution.outcome.then(async (outcome) => { const state = reduceTurn( (await this.turnRuntime.getTurn(turnId)).events, diff --git a/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts b/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts deleted file mode 100644 index d3d0e6e1..00000000 --- a/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts +++ /dev/null @@ -1,52 +0,0 @@ -export const skill = String.raw` -# Web Search Skill - -You have access to two search tools for finding information on the internet. Choose the right one based on the user's intent. - -## Tools - -### web-search (Brave Search) -Quick, general-purpose web search. Returns titles, URLs, and short descriptions. - -**Best for:** -- Quick lookups for things that change ("current price of Bitcoin", "weather in SF") -- Current events and breaking news -- Finding a specific website or page -- Simple questions with direct answers -- Checking a fact or date - -### research-search (Exa Search) -Deep, research-oriented search. Returns full article text, highlights, and metadata (author, published date). - -**Best for:** -- Exploring a topic in depth ("what are the latest advances in CRISPR") -- Finding articles, blog posts, papers, and quality sources -- Discovering companies, people, or organizations -- Research where you need rich context, not just links -- When the user says "research", "find articles about", "look into", "deep dive" - -**Category filter:** Use the category parameter when the user's intent clearly maps to one: company, research paper, news, tweet, personal site, financial report, people. - -## How Many Searches to Do - -**CRITICAL: Always start with exactly ONE search call.** Pick the single best tool (\`web-search\` or \`research-search\`) and make one request. Wait for the result before deciding if more searches are needed. - -**NEVER call multiple search tools simultaneously.** No parallel web-search + research-search. No firing off two web-searches at once. Always sequential: one search at a time. - -Only make a follow-up search if: -- The first search returned truly uninformative or irrelevant results -- The query has clearly distinct sub-topics that the first search couldn't cover (e.g., "compare X and Y" after getting results for X only) -- The user explicitly asks you to dig deeper - -One good search is almost always enough. Default to one and stop. - -## Choosing Between the Two - -If both tools are attached, prefer: -- \`web-search\` when the user wants a quick answer or specific link -- \`research-search\` when the user wants to learn, explore, or gather sources - -If only one is attached, use whichever is available. -`; - -export default skill; diff --git a/apps/x/packages/core/src/sessions/headless.ts b/apps/x/packages/core/src/sessions/headless.ts deleted file mode 100644 index 071d7fd4..00000000 --- a/apps/x/packages/core/src/sessions/headless.ts +++ /dev/null @@ -1,41 +0,0 @@ -import type { z } from "zod"; -import type { UserMessage } from "@x/shared/dist/message.js"; -import type { - ConversationMessage, - RequestedAgent, -} from "@x/shared/dist/turns.js"; -import type { ITurnRuntime, TurnOutcome } from "../turns/api.js"; - -// Standalone turns for non-session callers (background tasks, live notes, -// knowledge pipelines, scheduled agents): sessionId null, automatic -// permissions, no human. Never appears in the session index; callers keep -// the turnId if they need history. -export async function runHeadlessTurn( - turnRuntime: ITurnRuntime, - input: { - agent: z.infer; - 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..9bca5360 100644 --- a/apps/x/packages/core/src/sessions/sessions.test.ts +++ b/apps/x/packages/core/src/sessions/sessions.test.ts @@ -17,7 +17,6 @@ import type { import { TurnInputError } from "../turns/api.js"; import { HotStream } from "../turns/stream.js"; import { TurnNotSettledError } from "./api.js"; -import { runHeadlessTurn } from "./headless.js"; import { InMemorySessionRepo } from "./in-memory-session-repo.js"; import type { ISessionRepo } from "./repo.js"; import { SessionsImpl } from "./sessions.js"; @@ -823,26 +822,6 @@ describe("startup scan (13.6)", () => { }); }); -describe("headless standalone turns (13.8)", () => { - it("runs a standalone turn with sessionId null, auto permission, no human", async () => { - const fake = new FakeTurnRuntime(); - const { turnId, outcome } = await runHeadlessTurn(fake, { - agent: { agentId: "background-task-agent" }, - input: user("summarize"), - maxModelCalls: 3, - }); - expect(fake.createTurnInputs[0]).toEqual({ - agent: { agentId: "background-task-agent" }, - sessionId: null, - context: [], - input: user("summarize"), - config: { autoPermission: true, humanAvailable: false, maxModelCalls: 3 }, - }); - expect(outcome.status).toBe("completed"); - expect(fake.advanceCalls).toEqual([{ turnId, input: undefined }]); - }); -}); - describe("active-skill carry-forward", () => { const skillTool = { toolId: "builtin:file-writeText", From 876bc35e9e2494e8ad376321f4c61f91ab2bbb86 Mon Sep 17 00:00:00 2001 From: Ramnique Singh <30795890+ramnique@users.noreply.github.com> Date: Thu, 9 Jul 2026 15:47:21 +0530 Subject: [PATCH 03/91] =?UTF-8?q?refactor(x):=20one=20delivery=20path=20fo?= =?UTF-8?q?r=20turn=20events=20=E2=80=94=20chat,=20channels,=20deltas?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the turn-spine unification. Chat's SessionChatStore now consumes turns:events like every other surface, joining live turns by file offset (drop covered, append contiguous, refetch on gap) instead of blind-appending session-bus events. Text/reasoning deltas cross IPC only for turns a window subscribed to (turns:subscribe/unsubscribe, a per-webContents registry that also survives window teardown), so headless pipeline chatter never reaches windows that aren't watching. The channels bridge settles turns off the turn event bus instead of filtering the session broadcast. With no turn-event consumers left, SessionsImpl stops forwarding entirely (it drains its execution streams and keeps outcome/index handling) and sessions:events shrinks to index-changed entries — one channel for turn events, one for session metadata. Co-Authored-By: Claude Fable 5 --- apps/x/apps/main/src/ipc.ts | 46 +++++++++-- .../src/hooks/useSessionChat.test.tsx | 61 ++++++++++----- .../apps/renderer/src/hooks/useSessionChat.ts | 15 +++- apps/x/apps/renderer/src/hooks/useSessions.ts | 6 +- .../src/lib/session-chat/store.test.ts | 64 +++++++++++++-- .../renderer/src/lib/session-chat/store.ts | 77 +++++++++++++++---- apps/x/packages/core/docs/session-design.md | 17 ++-- .../packages/core/docs/turn-runtime-design.md | 7 +- .../packages/core/src/channels/bridge.test.ts | 10 +-- apps/x/packages/core/src/channels/bridge.ts | 10 ++- apps/x/packages/core/src/channels/service.ts | 4 +- .../core/src/sessions/sessions.test.ts | 18 ++--- apps/x/packages/core/src/sessions/sessions.ts | 17 ++-- apps/x/packages/shared/src/ipc.ts | 18 ++++- apps/x/packages/shared/src/sessions.ts | 29 +++---- 15 files changed, 284 insertions(+), 115 deletions(-) diff --git a/apps/x/apps/main/src/ipc.ts b/apps/x/apps/main/src/ipc.ts index 4ae5b8c4..e47d7593 100644 --- a/apps/x/apps/main/src/ipc.ts +++ b/apps/x/apps/main/src/ipc.ts @@ -697,9 +697,32 @@ export function startSessionsWatcher(): void { // runtime executes (session chat, headless background/knowledge runners, // spawned sub-agents), tagged with sessionId and the event's file offset so // consumers can join a live turn against a sessions:getTurn snapshot without -// gaps or duplicates. Deltas are deliberately not broadcast here in v1 — -// session chat already streams them via sessions:events; scoping deltas to -// subscribed turns is a follow-up. +// gaps or duplicates. Durable events are broadcast to every window; +// text/reasoning deltas are high-volume and ephemeral, so they are sent only +// to windows that subscribed to that turn via turns:subscribe. +const turnDeltaSubs = new Map>(); + +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/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/packages/core/docs/session-design.md b/apps/x/packages/core/docs/session-design.md index f2ba6ccc..ab4f122d 100644 --- a/apps/x/packages/core/docs/session-design.md +++ b/apps/x/packages/core/docs/session-design.md @@ -357,18 +357,23 @@ which governs mutation of live logs, not their removal. ## 10. Event forwarding and live UI -1. For every `advanceTurn` it initiates, the session layer consumes - `TurnExecution.events` and forwards each `TurnStreamEvent` — tagged with - `sessionId` — through the application bus to renderer windows over one - IPC channel (`sessions:events`). +1. Live turn delivery is not the session layer's job: the turn runtime + publishes every turn's events to the process-wide turn event bus + (turn-runtime-design.md §17.1), which the app layer bridges to renderer + windows over one IPC channel (`turns:events` — durable events broadcast + with their file offsets; deltas only to windows subscribed to that turn). + The session layer drains each `TurnExecution.events` it initiates so an + unconsumed stream never buffers, and `sessions:events` carries only + `session-index-changed` entries. 2. When `outcome` settles, the session layer updates the index entry and publishes `session-index-changed`. 3. The renderer follows the turn spec's historical/live pattern: fetch turns via `getTurn`, run the shared `reduceTurn` per turn, compose the session timeline turn-by-turn (each turn renders its input and its own activity; the referenced prefix is never re-rendered from context), - append live durable events and re-reduce, and keep text/reasoning deltas - in an ephemeral overlay cleared by canonical responses. + join live durable events by file offset (drop covered, append + contiguous, refetch on gap) and re-reduce, and keep text/reasoning + deltas in an ephemeral overlay cleared by canonical responses. 4. Pending approvals and ask-human prompts render from the suspended turn's reduced state, so they survive restarts without any session-layer bookkeeping. diff --git a/apps/x/packages/core/docs/turn-runtime-design.md b/apps/x/packages/core/docs/turn-runtime-design.md index 5a4139c1..c3ef6be9 100644 --- a/apps/x/packages/core/docs/turn-runtime-design.md +++ b/apps/x/packages/core/docs/turn-runtime-design.md @@ -1435,9 +1435,10 @@ stream for the initiating caller. events with `offset <= snapshot.length` — no gaps, no duplicates, no sequence numbers in the durable schema. - Text/reasoning deltas are published without an offset (they are not - durable). The app-layer IPC bridge (`turns:events`) forwards durable - events only in v1; deltas still reach session chat via the session-layer - forwarding. Scoping delta delivery to subscribed turns is future work. + durable). The app-layer IPC bridge (`turns:events`) broadcasts durable + events to every window; deltas are forwarded only to windows that + declared they are watching that turn (`turns:subscribe` / + `turns:unsubscribe`, a per-webContents registry in the app layer). - The bus is ephemeral and observational, like the lifecycle bus: listener errors are swallowed, nothing durable depends on delivery, and a crash losing listeners accurately reflects that no execution is known active. diff --git a/apps/x/packages/core/src/channels/bridge.test.ts b/apps/x/packages/core/src/channels/bridge.test.ts index 2c78f90a..c2b3adb7 100644 --- a/apps/x/packages/core/src/channels/bridge.test.ts +++ b/apps/x/packages/core/src/channels/bridge.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it, vi } from "vitest"; import type { SessionIndexEntry } from "@x/shared/dist/sessions.js"; import type { TurnStreamEvent } from "@x/shared/dist/turns.js"; -import { EmitterSessionBus } from "../sessions/bus.js"; +import { TurnEventHub } from "../turns/event-hub.js"; import { TurnInputError } from "../turns/api.js"; import type { ISessions } from "../sessions/api.js"; import { ChannelBridge, type ModelChoice } from "./bridge.js"; @@ -50,7 +50,7 @@ function askEvent(turnId: string, question: string, options?: string[]): TurnStr interface Harness { bridge: ChannelBridge; - bus: EmitterSessionBus; + bus: TurnEventHub; replies: string[]; reply: (text: string) => Promise; 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/sessions.test.ts b/apps/x/packages/core/src/sessions/sessions.test.ts index 9bca5360..e1252a43 100644 --- a/apps/x/packages/core/src/sessions/sessions.test.ts +++ b/apps/x/packages/core/src/sessions/sessions.test.ts @@ -644,7 +644,10 @@ describe("stopTurn and resumeTurn", () => { }); describe("event forwarding and index maintenance (13.6, 13.7)", () => { - it("forwards stream events to the bus tagged with sessionId in order", async () => { + it("drains the execution stream and publishes only index updates", async () => { + // Live turn delivery is the turn event bus's job (published by the + // runtime itself); the session bus carries index changes only. The + // stream must still be consumed so it never buffers until settle. const { sessions, fake, bus } = makeSessions(); const streamed: TurnStreamEvent[] = [ { type: "text_delta", turnId: "x", modelCallIndex: 0, delta: "he" }, @@ -652,19 +655,12 @@ describe("event forwarding and index maintenance (13.6, 13.7)", () => { ]; fake.script = () => ({ events: streamed, outcome: completedOutcome() }); const sessionId = await sessions.createSession(); - const { turnId } = await sessions.sendMessage(sessionId, user("go"), { + await sessions.sendMessage(sessionId, user("go"), { agent: { agentId: "copilot" }, }); await flush(); - const turnEvents = bus.events.filter((e) => e.kind === "turn-event"); - expect(turnEvents).toEqual( - streamed.map((event) => ({ - kind: "turn-event", - sessionId, - turnId, - event, - })), - ); + expect(bus.events.length).toBeGreaterThan(0); + expect(bus.events.every((e) => e.kind === "index-changed")).toBe(true); }); it("outcome settlement updates the index entry's latest turn status", async () => { diff --git a/apps/x/packages/core/src/sessions/sessions.ts b/apps/x/packages/core/src/sessions/sessions.ts index acd9d7ef..d42640b1 100644 --- a/apps/x/packages/core/src/sessions/sessions.ts +++ b/apps/x/packages/core/src/sessions/sessions.ts @@ -343,9 +343,11 @@ export class SessionsImpl implements ISessions { await execution.outcome; } - // Every advance this layer initiates: forward its stream to the bus - // tagged with sessionId, keep the abort controller for stopTurn, and - // update the index entry when the outcome settles. + // Every advance this layer initiates: keep the abort controller for + // stopTurn and update the index entry when the outcome settles. Live + // event delivery is not this layer's job — the runtime publishes every + // event to the turn event bus; the execution stream is drained so an + // unconsumed HotStream never buffers events until settle. private startTrackedAdvance( sessionId: string | null, turnId: string, @@ -368,14 +370,7 @@ export class SessionsImpl implements ISessions { void (async () => { try { for await (const event of execution.events) { - if (sessionId !== null) { - this.sessionBus.publish({ - kind: "turn-event", - sessionId, - turnId, - event, - }); - } + void event; } } catch { // Infrastructure failures surface through the outcome. diff --git a/apps/x/packages/shared/src/ipc.ts b/apps/x/packages/shared/src/ipc.ts index cf6dd880..c00220bc 100644 --- a/apps/x/packages/shared/src/ipc.ts +++ b/apps/x/packages/shared/src/ipc.ts @@ -461,7 +461,7 @@ const ipcSchemas = { }, // ── New runtime: sessions + turns (session-design.md) ──────────────────── // Turn-mutating calls return quickly; the renderer follows progress through - // the sessions:events feed and the shared reduceTurn reducer. + // the turns:events feed and the shared reduceTurn reducer. 'sessions:create': { req: z.object({ title: z.string().optional() }), res: z.object({ sessionId: z.string() }), @@ -545,12 +545,24 @@ const ipcSchemas = { }, // Process-wide turn event spine: every turn's durable events (with file // offsets), regardless of who started the turn — session chat, headless - // background/knowledge runners, spawned sub-agents. Deltas are not - // broadcast here in v1; session chat streams them via sessions:events. + // background/knowledge runners, spawned sub-agents. Text/reasoning deltas + // ride the same channel but only reach windows that subscribed to that + // turn via turns:subscribe. 'turns:events': { req: z.custom(), res: z.null(), }, + // Per-window delta subscription: deltas are high-volume and ephemeral, so + // they cross IPC only for turns this window declared it is watching. + // Durable events are always broadcast regardless. + 'turns:subscribe': { + req: z.object({ turnId: z.string() }), + res: z.object({ success: z.literal(true) }), + }, + 'turns:unsubscribe': { + req: z.object({ turnId: z.string() }), + res: z.object({ success: z.literal(true) }), + }, 'services:events': { req: ServiceEvent, res: z.null(), diff --git a/apps/x/packages/shared/src/sessions.ts b/apps/x/packages/shared/src/sessions.ts index 1a52d38f..08435650 100644 --- a/apps/x/packages/shared/src/sessions.ts +++ b/apps/x/packages/shared/src/sessions.ts @@ -1,9 +1,5 @@ import { z } from "zod"; -import { - ModelDescriptor, - type TurnStatus, - type TurnStreamEvent, -} from "./turns.js"; +import { ModelDescriptor, type TurnStatus } from "./turns.js"; // Durable session contract for the session layer (see // packages/core/docs/session-design.md). A session is an append-only chain @@ -171,21 +167,14 @@ export interface SessionIndexEntry { error?: string; } -// What the renderer's single feed consumer receives over IPC: live turn -// stream events (durable events + ephemeral deltas) and index updates. -// entry: null signals deletion. -export type SessionBusEvent = - | { - kind: "turn-event"; - sessionId: string; - turnId: string; - event: TurnStreamEvent; - } - | { - kind: "index-changed"; - sessionId: string; - entry: SessionIndexEntry | null; - }; +// What the renderer's session-feed consumer receives over IPC: session index +// updates only. Turn events (durable + deltas) travel on the turns:events +// spine (TurnBusEvent in turns.ts). entry: null signals deletion. +export type SessionBusEvent = { + kind: "index-changed"; + sessionId: string; + entry: SessionIndexEntry | null; +}; export function sessionIndexEntry( state: SessionState, From 51b11101f040a6041decee87bb63c468b77e96de Mon Sep 17 00:00:00 2001 From: Ramnique Singh <30795890+ramnique@users.noreply.github.com> Date: Thu, 9 Jul 2026 15:56:45 +0530 Subject: [PATCH 04/91] test(x): cover useTurn and useAgentRunTranscript hook behavior The follower's join protocol was already unit-tested, but the hooks grew stateful behavior of their own: useTurn's reset-on-turnId-change vs keep-while-disabled, the snapshotFailed signal and feed-event recovery, and useAgentRunTranscript's legacy runs:fetch fallback and loading/error derivation. Tests drive the real turn-feed singleton through a stubbed window.ipc preload surface. Co-Authored-By: Claude Fable 5 --- .../hooks/use-agent-run-transcript.test.tsx | 77 +++++++++ .../apps/renderer/src/hooks/use-turn.test.tsx | 148 ++++++++++++++++++ 2 files changed, 225 insertions(+) create mode 100644 apps/x/apps/renderer/src/hooks/use-agent-run-transcript.test.tsx create mode 100644 apps/x/apps/renderer/src/hooks/use-turn.test.tsx diff --git a/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.test.tsx b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.test.tsx new file mode 100644 index 00000000..eda45598 --- /dev/null +++ b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.test.tsx @@ -0,0 +1,77 @@ +import { renderHook, waitFor } from '@testing-library/react' +import { beforeEach, describe, expect, it } from 'vitest' +import { completedTurnLog } from '@/lib/session-chat/test-fixtures' +import { useAgentRunTranscript } from './use-agent-run-transcript' + +// Same preload stub as use-turn.test.tsx: the hook rides useTurn (real +// turn-feed singleton) plus a runs:fetch legacy fallback. +let handlers: Record 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-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) + }) +}) From df3f0ba0289cb73226f254c111412ca31f4bf2f8 Mon Sep 17 00:00:00 2001 From: Ramnique Singh <30795890+ramnique@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:35:41 +0530 Subject: [PATCH 05/91] remove irrelevant gh workflows --- .github/workflows/rowboat-build.yml | 47 ----------------------------- .github/workflows/x-publish.yml | 43 -------------------------- 2 files changed, 90 deletions(-) delete mode 100644 .github/workflows/rowboat-build.yml delete mode 100644 .github/workflows/x-publish.yml diff --git a/.github/workflows/rowboat-build.yml b/.github/workflows/rowboat-build.yml deleted file mode 100644 index 270e6263..00000000 --- a/.github/workflows/rowboat-build.yml +++ /dev/null @@ -1,47 +0,0 @@ -name: Rowboat Build - -on: - pull_request: - -jobs: - build-rowboat-nextjs: - runs-on: ubuntu-latest - - steps: - - uses: actions/checkout@v6 - - - name: Setup Node.js - uses: actions/setup-node@v6 - with: - cache-dependency-path: 'apps/rowboat/package-lock.json' - node-version: '20' - cache: 'npm' - - - name: Install dependencies - run: npm ci - working-directory: apps/rowboat - - - name: Build Rowboat - run: npm run build - working-directory: apps/rowboat - - build-rowboatx: - runs-on: ubuntu-latest - - steps: - - uses: actions/checkout@v6 - - - name: Setup Node.js - uses: actions/setup-node@v6 - with: - cache-dependency-path: 'apps/rowboat/package-lock.json' - node-version: '24' - cache: 'npm' - - - name: Install dependencies - run: npm ci - working-directory: apps/cli - - - name: Build Rowboat - run: npm run build - working-directory: apps/cli \ No newline at end of file diff --git a/.github/workflows/x-publish.yml b/.github/workflows/x-publish.yml deleted file mode 100644 index c411ab68..00000000 --- a/.github/workflows/x-publish.yml +++ /dev/null @@ -1,43 +0,0 @@ -name: Publish to npm - -on: workflow_dispatch - -permissions: - id-token: write # Required for OIDC - contents: read - -jobs: - publish: - runs-on: ubuntu-latest - - steps: - - name: Checkout repo - uses: actions/checkout@v6 - - - name: Set up Node - uses: actions/setup-node@v6 - with: - node-version: 24 - registry-url: https://registry.npmjs.org/ - - - name: Update npm - run: npm install -g npm@latest - - - name: Install deps - run: npm ci - working-directory: apps/cli - - # optional: run tests - # - run: npm test - - - name: Build - run: npm run build - working-directory: apps/cli - - - name: Pack - run: npm pack - working-directory: apps/cli - - - name: Publish to npm - run: npm publish --access public - working-directory: apps/cli \ No newline at end of file From 602889e340b1bceafd6f3f4e6f32966badabdc5d Mon Sep 17 00:00:00 2001 From: Gagan Date: Thu, 9 Jul 2026 18:11:51 +0530 Subject: [PATCH 06/91] style(x): unify tab headers and clean up Brain/Home UI - Shared header treatment across Apps, Background tasks, Workspace, Meetings, Brain: 24px/650 heading, no leading icons, no divider line, centered 1120px column with fixed 34px/30px padding - Apps-style page background on BG tasks, Workspace, Meetings - Brain files view: monochrome then icon-free rows, inline note-count meta line instead of pill chips, semibold item names, uniform row heights, breadcrumb-only navigation, ghost quick actions - Bases toolbar flush with table edge - Meetings: full-width Coming up section - Home: drop Connections/Ask-anything icon tiles, real Slack logo - Sidebar: Apps above Background agents --- .../src/components/apps/apps-view.tsx | 4 +- .../renderer/src/components/bases-view.tsx | 2 +- .../renderer/src/components/bg-tasks-view.tsx | 15 +- .../renderer/src/components/home-view.tsx | 6 +- .../src/components/knowledge-view.tsx | 131 ++++++------------ .../renderer/src/components/meetings-view.tsx | 19 ++- .../src/components/sidebar-content.tsx | 20 +-- .../src/components/tool-connections-card.tsx | 5 +- .../src/components/workspace-view.tsx | 16 +-- 9 files changed, 86 insertions(+), 132 deletions(-) diff --git a/apps/x/apps/renderer/src/components/apps/apps-view.tsx b/apps/x/apps/renderer/src/components/apps/apps-view.tsx index 9fdb89de..04a01bce 100644 --- a/apps/x/apps/renderer/src/components/apps/apps-view.tsx +++ b/apps/x/apps/renderer/src/components/apps/apps-view.tsx @@ -58,8 +58,8 @@ const CARD_CSS = ` --ma-pat-opacity:0.05; --ma-glow-opacity:0.10; --ma-glow-hover-opacity:0.16; --ma-badge-mix:15%; --ma-pill-mix:13%; --ma-tint:20%; --ma-tint-hover:26%; } -.ma-inner { max-width:1120px; margin:0 auto; padding:clamp(20px,3.5cqw,34px) clamp(16px,3cqw,30px) 48px; } -.ma-h1 { font-size:clamp(19px,2.6cqw,24px); font-weight:650; letter-spacing:-0.02em; color:var(--ma-h1); margin:0 0 4px; } +.ma-inner { max-width:1120px; margin:0 auto; padding:34px 30px 48px; } +.ma-h1 { font-size:24px; font-weight:650; letter-spacing:-0.02em; color:var(--ma-h1); margin:0 0 4px; } .ma-sub { font-size:clamp(13px,1.5cqw,14px); color:var(--ma-sub); margin:0 0 clamp(14px,2cqw,20px); } .ma-tabs { display:flex; gap:6px; margin-bottom:clamp(14px,2cqw,22px); } .ma-tab { border:1px solid var(--ma-border); background:transparent; color:var(--ma-sub); border-radius:999px; padding:5px 14px; font-size:13px; font-weight:600; cursor:pointer; } diff --git a/apps/x/apps/renderer/src/components/bases-view.tsx b/apps/x/apps/renderer/src/components/bases-view.tsx index b1f5413c..f41a7a82 100644 --- a/apps/x/apps/renderer/src/components/bases-view.tsx +++ b/apps/x/apps/renderer/src/components/bases-view.tsx @@ -466,7 +466,7 @@ export function BasesView({ tree, onSelectNote, config, onConfigChange, isDefaul return (
{/* Toolbar */} -
+
-

+

Persistent agents that fire on a schedule or in response to events. Toggle a task inactive to pause it.

-
+
+
{loading ? (
@@ -1885,6 +1883,7 @@ export function BgTasksView({ onCreateWithCopilot, onEditWithCopilot, initialSlu
)} +
- + Slack Latest messages @@ -548,9 +549,6 @@ export function HomeView({ onClick={onOpenChat} className="flex items-center gap-3.5 rounded-xl border border-border bg-card p-4 text-left transition-colors hover:bg-accent" > -
- -
Ask anything — create presentations, do research, collaborate on docs. diff --git a/apps/x/apps/renderer/src/components/knowledge-view.tsx b/apps/x/apps/renderer/src/components/knowledge-view.tsx index 1d5b1397..254dc0e6 100644 --- a/apps/x/apps/renderer/src/components/knowledge-view.tsx +++ b/apps/x/apps/renderer/src/components/knowledge-view.tsx @@ -1,6 +1,5 @@ import { useCallback, useEffect, useMemo, useRef, useState, type ReactNode } from 'react' import { - ArrowLeft, ChevronRight, Copy, ExternalLink, @@ -70,20 +69,6 @@ const HIDDEN_PATHS = new Set(['knowledge/Meetings', 'knowledge/Workspace']) // Theme-aware accent palette for folder avatars — colored letter on a faint // tint of the same hue. Mirrors the design's six-colour rotation. -const AVATAR_PALETTE = [ - 'bg-indigo-500/10 text-indigo-600 dark:text-indigo-400', - 'bg-violet-500/10 text-violet-600 dark:text-violet-400', - 'bg-amber-500/10 text-amber-600 dark:text-amber-400', - 'bg-rose-500/10 text-rose-600 dark:text-rose-400', - 'bg-emerald-500/10 text-emerald-600 dark:text-emerald-400', - 'bg-sky-500/10 text-sky-600 dark:text-sky-400', -] as const - -function avatarClass(name: string): string { - let hash = 0 - for (let i = 0; i < name.length; i++) hash = (hash * 31 + name.charCodeAt(i)) >>> 0 - return AVATAR_PALETTE[hash % AVATAR_PALETTE.length] -} function isMarkdown(node: TreeNode): boolean { return node.kind === 'file' && node.name.toLowerCase().endsWith('.md') @@ -204,10 +189,10 @@ export function KnowledgeView({ return (
-
+
-

Brain

-

+

Brain

+

{totalNotes} {totalNotes === 1 ? 'note' : 'notes'} across {folders.length}{' '} {folders.length === 1 ? 'folder' : 'folders'}

@@ -242,12 +227,12 @@ export function KnowledgeView({ {graphContent}
) : mode === 'basis' ? ( -
+
{basisContent}
) : (
-
+
{currentFolder ? ( ) : ( <> - + {folders.length === 0 ? ( ) : ( @@ -398,9 +383,9 @@ function QuickAction({ ) @@ -425,20 +410,6 @@ function EmptyState({ text }: { text: string }) { ) } -function FolderAvatar({ name, className }: { name: string; className?: string }) { - return ( -
- {name.charAt(0).toUpperCase() || '?'} -
- ) -} - function FolderCard({ node, actions, @@ -472,9 +443,8 @@ function FolderCard({ onOpenFolder(node.path) } }} - className="group flex w-full cursor-pointer items-start gap-3 px-4 py-3 text-left transition-colors hover:bg-accent/50" + className="group flex w-full cursor-pointer items-center gap-3 px-4 py-3 text-left transition-colors hover:bg-accent/50" > -
{renameActive ? ( )} -
- {count} {count === 1 ? 'note' : 'notes'} +
+ + {count} {count === 1 ? 'note' : 'notes'} + + {peek.length > 0 && ( + + {peek.map((n) => ( + + {' · '} + + + ))} + + )}
- {peek.length > 0 && ( -
- {peek.map((n) => ( - - ))} -
- )}
-
+
{modified} @@ -565,17 +539,6 @@ function FolderDetail({ return ( <>
-
-

+

Upcoming events and meeting notes.

+
-
+
{loading ? (
@@ -1329,6 +1327,7 @@ export function MeetingsView({ onOpenNote, onTakeMeetingNotes, meetingState, mee
)}
+
) diff --git a/apps/x/apps/renderer/src/components/sidebar-content.tsx b/apps/x/apps/renderer/src/components/sidebar-content.tsx index b6b42c9e..f192df1c 100644 --- a/apps/x/apps/renderer/src/components/sidebar-content.tsx +++ b/apps/x/apps/renderer/src/components/sidebar-content.tsx @@ -890,6 +890,16 @@ export function SidebarContentPanel({
+ + + + Apps + + - - - - Apps - -
-
- -
Bring context from and take action in the apps you already use. diff --git a/apps/x/apps/renderer/src/components/workspace-view.tsx b/apps/x/apps/renderer/src/components/workspace-view.tsx index d5738e9b..9d475e13 100644 --- a/apps/x/apps/renderer/src/components/workspace-view.tsx +++ b/apps/x/apps/renderer/src/components/workspace-view.tsx @@ -8,7 +8,6 @@ import { Folder as FolderIcon, FolderOpen, FolderPlus, - Home, Loader2, MessageSquare, Pencil, @@ -365,19 +364,18 @@ export function WorkspaceView({ tree, initialPath, actions, onNavigate, onOpenNo }, [newName, onCreateWorkspace, resetAddDialog]) return ( -
-
+
+
{breadcrumbs.map((crumb, idx) => { const isLast = idx === breadcrumbs.length - 1 @@ -482,12 +480,13 @@ export function WorkspaceView({ tree, initialPath, actions, onNavigate, onOpenNo
+
{items.length === 0 ? (
@@ -594,6 +593,7 @@ export function WorkspaceView({ tree, initialPath, actions, onNavigate, onOpenNo })}
)} +
{dropEnabled && isDraggingOver && (
From 6def1bf7d3e131cadcb68561175a1c3563ef5750 Mon Sep 17 00:00:00 2001 From: Gagan Date: Thu, 9 Jul 2026 18:14:21 +0530 Subject: [PATCH 07/91] style(x): icon-free up-next hero on home tab --- apps/x/apps/renderer/src/components/home-view.tsx | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/apps/x/apps/renderer/src/components/home-view.tsx b/apps/x/apps/renderer/src/components/home-view.tsx index d5e1e3a4..cc834004 100644 --- a/apps/x/apps/renderer/src/components/home-view.tsx +++ b/apps/x/apps/renderer/src/components/home-view.tsx @@ -332,15 +332,12 @@ export function HomeView({ {/* Up-next hero */} {nextEvent && ( -
-
- -
+
-
+
Up next · {nextEvent.isAllDay ? 'today' : relativeFromNow(nextEvent.start)}
-
{nextEvent.summary}
+
{nextEvent.summary}
{nextEvent.isAllDay ? 'All day' : `${timeOfDay(nextEvent.start)}${nextEvent.end ? ` – ${timeOfDay(nextEvent.end)}` : ''}`} {nextEvent.location ? ` · ${nextEvent.location}` : ''} From 992c1883a7d0f4c95622c5868e318eeece4c5ec9 Mon Sep 17 00:00:00 2001 From: Arjun <6592213+arkml@users.noreply.github.com> Date: Thu, 9 Jul 2026 18:15:13 +0530 Subject: [PATCH 08/91] fix(x): collapse quoted replies in Outlook-authored email MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Quoted-chain detection only matched Gmail/Apple markup (.gmail_quote, .gmail_attr, blockquote[type=cite]). Outlook emits none of those, so hasQuote stayed false, no "•••" toggle rendered, and the whole quoted history was drawn inline — each message in a long thread re-rendering every message before it. Outlook desktop marks the boundary with a border-top:solid #E1E1E1 div that holds only the From:/Sent:/Subject: header; the quoted body trails it as siblings. Hiding just the matched element is therefore not enough, so mark the divider and everything after it, climbing ancestors. Require the header block before treating a border-top div as a boundary — signatures and marketing footers draw the same rule, and wrongly hiding a signature is worse than wrongly showing a quote. Measured against a 558-message local corpus: 7/7 real quotes caught, 0 false positives. Also fixes a pre-existing bug: a forward is quoted top to bottom, so hiding its quote left a blank body behind a toggle nobody would press. When nothing visible remains, show everything and drop the toggle. The plain-text path had the same Gmail-only hole, and the fix already existed in the file — isReplyQuoteBoundary handled Outlook headers but was wired only to the composer. It now backs display too. Quote tagging moves into the srcDoc build, so quotes are hidden on first paint and the measured iframe height is the collapsed height. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../renderer/src/components/email-view.tsx | 82 +------- .../renderer/src/lib/email-quotes.test.ts | 138 ++++++++++++ apps/x/apps/renderer/src/lib/email-quotes.ts | 198 ++++++++++++++++++ 3 files changed, 345 insertions(+), 73 deletions(-) create mode 100644 apps/x/apps/renderer/src/lib/email-quotes.test.ts create mode 100644 apps/x/apps/renderer/src/lib/email-quotes.ts diff --git a/apps/x/apps/renderer/src/components/email-view.tsx b/apps/x/apps/renderer/src/components/email-view.tsx index f1b9ff17..bd720692 100644 --- a/apps/x/apps/renderer/src/components/email-view.tsx +++ b/apps/x/apps/renderer/src/components/email-view.tsx @@ -7,6 +7,7 @@ import Placeholder from '@tiptap/extension-placeholder' import type { blocks } from '@x/shared' import { cn } from '@/lib/utils' import { toast } from '@/lib/toast' +import { prepareEmailHtml, splitPlainTextQuote, stripQuotedReplyText, QUOTED_CLASS } from '@/lib/email-quotes' import { useTheme } from '@/contexts/theme-context' import { SettingsDialog } from '@/components/settings-dialog' import { Button } from '@/components/ui/button' @@ -72,31 +73,6 @@ function snippet(text?: string): string { return (text || '').replace(/\s+/g, ' ').trim().slice(0, 180) } -function isReplyQuoteBoundary(lines: string[], index: number): boolean { - const line = lines[index]?.trim() || '' - if (/^On\b.+\bwrote:\s*$/i.test(line)) return true - if (/^-{2,}\s*(Original Message|Forwarded message)\s*-{2,}$/i.test(line)) return true - if (/^From:\s+\S/i.test(line)) { - const next = lines.slice(index + 1, index + 6).map((value) => value.trim()) - return next.some((value) => /^(Sent|Date):\s+\S/i.test(value)) - && next.some((value) => /^To:\s+\S/i.test(value)) - && next.some((value) => /^Subject:\s+\S/i.test(value)) - } - return false -} - -function stripQuotedReplyText(text: string): string { - const lines = text.replace(/\r\n/g, '\n').replace(/\r/g, '\n').split('\n') - const boundary = lines.findIndex((line, index) => { - if (isReplyQuoteBoundary(lines, index)) return true - return index > 0 - && line.trim().startsWith('>') - && (lines[index - 1]?.trim() === '' || lines[index - 1]?.trim().startsWith('>')) - }) - const visible = boundary >= 0 ? lines.slice(0, boundary) : lines - return visible.join('\n').replace(/[ \t]+\n/g, '\n').replace(/\n{3,}/g, '\n\n').trim() -} - function getInitial(from?: string): string { return (extractName(from)[0] || '?').toUpperCase() } @@ -307,43 +283,6 @@ function plainTextToHtml(text: string): string { .join('') } -function splitPlainTextQuote(text: string): { visible: string; quoted: string | null } { - const re = /(?:^|\n)On\s+.+?\swrote:\s*(?:\n|$)/ - const match = re.exec(text) - if (!match) return { visible: text, quoted: null } - const start = match.index === 0 ? 0 : match.index + 1 - const visible = text.slice(0, start).trimEnd() - const quoted = text.slice(start) - if (!quoted.trim()) return { visible: text, quoted: null } - return { visible, quoted } -} - -// True if the HTML — after stripping quoted/hidden content — defines its -// own visual layout (real images, tables, explicit backgrounds). Unstyled -// HTML (Gmail replies, Outlook one-liners wrapped in MsoNormal boilerplate, -// outreach emails with only a tracking pixel, reply HTML whose only image -// lives inside the inline-quoted thread) gets an iframe that adapts to the -// app theme; styled HTML keeps the white "paper" look so newsletters / -// branded designs render as their senders intended. -function isStyledHtml(html: string): boolean { - const doc = new DOMParser().parseFromString(html, 'text/html') - doc.querySelectorAll('.gmail_quote, .gmail_attr, blockquote[type="cite"]').forEach((n) => n.remove()) - if (doc.querySelector('table')) return true - for (const img of Array.from(doc.querySelectorAll('img'))) { - const w = parseInt(img.getAttribute('width') || '0', 10) - const h = parseInt(img.getAttribute('height') || '0', 10) - if (w === 1 && h === 1) continue - const style = img.getAttribute('style') || '' - if (/display\s*:\s*none/i.test(style)) continue - if (/visibility\s*:\s*hidden/i.test(style)) continue - return true - } - const visible = doc.body?.innerHTML || '' - if (/bgcolor\s*=/i.test(visible)) return true - if (/background-(color|image)\s*:/i.test(visible)) return true - return false -} - function buildEmailDocument( html: string, opts: { theme: 'light' | 'dark'; adaptToTheme: boolean }, @@ -384,12 +323,8 @@ function buildEmailDocument( border-left: 2px solid ${quoteBorder}; color: ${quoteColor}; } - .gmail_quote, - .gmail_attr, - blockquote[type="cite"] { display: none; } - [data-show-quotes="true"] .gmail_quote, - [data-show-quotes="true"] .gmail_attr, - [data-show-quotes="true"] blockquote[type="cite"] { display: block; } + .${QUOTED_CLASS} { display: none; } + [data-show-quotes="true"] .${QUOTED_CLASS} { display: revert; } ${html}` } @@ -436,24 +371,25 @@ function HtmlMessageBody({ message, threadId }: { message: GmailThreadMessage; t const saveTimerRef = useRef | null>(null) const lastSavedHeightRef = useRef(message.bodyHeight ?? 0) const [height, setHeight] = useState(message.bodyHeight ?? 80) - const [hasQuote, setHasQuote] = useState(false) const [showQuotes, setShowQuotes] = useState(false) // Read by handleLoad so a reload (theme switch rebuilds srcDoc) restores the // expanded-quotes state on the fresh document. const showQuotesRef = useRef(showQuotes) useEffect(() => { showQuotesRef.current = showQuotes }, [showQuotes]) - const adaptToTheme = useMemo(() => !isStyledHtml(message.bodyHtml!), [message.bodyHtml]) + // Tag the quotes before the iframe ever paints, so quoted history is hidden + // on the first frame and the height we measure is the collapsed height. + const { html, hasQuote, styled } = useMemo(() => prepareEmailHtml(message.bodyHtml!), [message.bodyHtml]) + const adaptToTheme = !styled const srcDoc = useMemo( - () => buildEmailDocument(message.bodyHtml!, { theme: resolvedTheme, adaptToTheme }), - [message.bodyHtml, resolvedTheme, adaptToTheme], + () => buildEmailDocument(html, { theme: resolvedTheme, adaptToTheme }), + [html, resolvedTheme, adaptToTheme], ) const handleLoad = useCallback(() => { const iframe = iframeRef.current const doc = iframe?.contentDocument if (!doc?.body) return - setHasQuote(!!doc.querySelector('.gmail_quote, .gmail_attr, blockquote[type="cite"]')) if (showQuotesRef.current) doc.documentElement.dataset.showQuotes = 'true' // Clicking into the email body focuses the iframe document, which would // otherwise swallow every list/thread shortcut (the parent's document diff --git a/apps/x/apps/renderer/src/lib/email-quotes.test.ts b/apps/x/apps/renderer/src/lib/email-quotes.test.ts new file mode 100644 index 00000000..0caf540b --- /dev/null +++ b/apps/x/apps/renderer/src/lib/email-quotes.test.ts @@ -0,0 +1,138 @@ +import { describe, expect, it } from 'vitest' +import { prepareEmailHtml, QUOTED_CLASS, splitPlainTextQuote, stripQuotedReplyText } from './email-quotes' + +/** Visible text once the stylesheet hides the tagged nodes. */ +function visible(html: string): string { + const doc = new DOMParser().parseFromString(html, 'text/html') + doc.querySelectorAll(`.${QUOTED_CLASS}`).forEach((n) => n.remove()) + return (doc.body?.textContent || '').replace(/\s+/g, ' ').trim() +} + +// The divider Outlook desktop emits: the header lives inside it, the quoted +// chain trails it as siblings. +const OUTLOOK_QUOTE = ` +
+

From: Counsel <c@firm.com>
+ Sent: Tuesday, July 7, 2026 9:14 AM
+ To: Me <me@co.com>
+ Subject: Re: Series A docs

+
+

The indemnity clause needs revisiting.

` + +describe('prepareEmailHtml — quote detection', () => { + it('hides an Outlook quoted chain, including the siblings after the divider', () => { + const { html, hasQuote } = prepareEmailHtml( + `

Agreed, see below.

${OUTLOOK_QUOTE}`, + ) + expect(hasQuote).toBe(true) + expect(visible(html)).toBe('Agreed, see below.') + }) + + it('hides an Outlook quote nested beside a signature in an outer wrapper', () => { + const { html, hasQuote } = prepareEmailHtml( + `

Sounds good.

-- Jane, Partner

${OUTLOOK_QUOTE}
`, + ) + expect(hasQuote).toBe(true) + expect(visible(html)).toContain('Sounds good.') + expect(visible(html)).toContain('-- Jane, Partner') + expect(visible(html)).not.toContain('indemnity') + }) + + it('hides Outlook-on-the-web quoted chains via divRplyFwdMsg', () => { + const { html, hasQuote } = prepareEmailHtml( + `
Short reply.
From: C
Older content.
`, + ) + expect(hasQuote).toBe(true) + expect(visible(html)).toBe('Short reply.') + }) + + it('still hides Gmail wrapper quotes', () => { + const { html, hasQuote } = prepareEmailHtml( + `
Thanks!
On Tue, X wrote:
old
`, + ) + expect(hasQuote).toBe(true) + expect(visible(html)).toBe('Thanks!') + }) + + it('leaves a border-top divider alone when it is a signature rule, not a quote header', () => { + const html = `

Hi there.

Jane Doe | Acme | jane@acme.com

` + const prepared = prepareEmailHtml(html) + expect(prepared.hasQuote).toBe(false) + expect(visible(prepared.html)).toContain('Jane Doe') + }) + + it('shows a forward in full rather than collapsing the whole body', () => { + // Everything is quoted, so hiding would leave a blank message. + const prepared = prepareEmailHtml( + `
---------- Forwarded message ---------
The actual content.
`, + ) + expect(prepared.hasQuote).toBe(false) + expect(visible(prepared.html)).toContain('The actual content.') + }) + + it('reports no quote for a plain message', () => { + expect(prepareEmailHtml('

Just a note.

').hasQuote).toBe(false) + }) + + it('keeps a

Body

') + expect(html).toContain('.a{color:red}') + }) + + it('ignores quoted content when deciding whether the email is styled', () => { + // The table lives only in the quote, so the body should still adapt to theme. + const { styled } = prepareEmailHtml( + `

ok

old
`, + ) + expect(styled).toBe(false) + }) +}) + +describe('splitPlainTextQuote', () => { + it('splits on an Outlook From:/Sent:/To:/Subject: header block', () => { + const body = [ + 'Agreed.', + '', + 'From: Counsel ', + 'Sent: Tuesday, July 7, 2026 9:14 AM', + 'To: Me ', + 'Subject: Re: Series A docs', + '', + 'The indemnity clause needs revisiting.', + ].join('\n') + const { visible: head, quoted } = splitPlainTextQuote(body) + expect(head).toBe('Agreed.') + expect(quoted).toContain('The indemnity clause') + }) + + it('splits on "On … wrote:"', () => { + const { visible: head, quoted } = splitPlainTextQuote('Thanks!\n\nOn Tue, Jul 7, X wrote:\n> old') + expect(head).toBe('Thanks!') + expect(quoted).toContain('> old') + }) + + it('splits on a forwarded-message separator', () => { + const { visible: head, quoted } = splitPlainTextQuote('FYI\n\n---------- Forwarded message ---------\nbody') + expect(head).toBe('FYI') + expect(quoted).toContain('body') + }) + + it('returns no quote when there is no boundary', () => { + expect(splitPlainTextQuote('Just a note.').quoted).toBeNull() + }) + + it('does not treat a bare "From:" line as a boundary', () => { + expect(splitPlainTextQuote('From: the desk of Jane\n\nHello.').quoted).toBeNull() + }) +}) + +describe('stripQuotedReplyText', () => { + it('drops an Outlook quoted tail', () => { + const body = 'Agreed.\n\nFrom: C \nSent: Tue\nTo: Me\nSubject: Re: x\n\nold' + expect(stripQuotedReplyText(body)).toBe('Agreed.') + }) + + it('drops a ">" quoted tail', () => { + expect(stripQuotedReplyText('Reply.\n\n> old line')).toBe('Reply.') + }) +}) diff --git a/apps/x/apps/renderer/src/lib/email-quotes.ts b/apps/x/apps/renderer/src/lib/email-quotes.ts new file mode 100644 index 00000000..37fdd553 --- /dev/null +++ b/apps/x/apps/renderer/src/lib/email-quotes.ts @@ -0,0 +1,198 @@ +/** + * Quoted-reply detection for rendered email bodies. + * + * Quoted chains come in two shapes, and the difference decides how they hide: + * + * - Wrapper quotes (Gmail, Apple Mail, Yahoo, Proton): a single element + * contains the whole quoted chain, so hiding that element is enough. + * - Boundary quotes (Outlook, Thunderbird): a divider element holds only the + * "From:/Sent:/Subject:" header and the quoted chain trails it as siblings, + * so everything from the divider onward has to be hidden. + * + * Treating the second kind as the first is why Outlook threads used to render + * their entire history inline with no toggle to collapse it. + */ + +/** Marks nodes the iframe stylesheet hides until quotes are toggled on. */ +export const QUOTED_CLASS = 'rb-quoted' + +const WRAPPER_QUOTE_SELECTOR = [ + '.gmail_quote', + '.gmail_attr', + 'blockquote[type="cite"]', + '.yahoo_quoted', + '.protonmail_quote', +].join(', ') + +const BOUNDARY_QUOTE_SELECTOR = [ + '#divRplyFwdMsg', // Outlook on the web / "new Outlook" + '#appendonsend', // Outlook, an empty marker sitting before the quoted chain + '.moz-cite-prefix', // Thunderbird +].join(', ') + +// Outlook desktop draws the divider as an inline-styled
. The colour is +// the only stable part of the rule — padding units vary (0in vs 0cm), casing +// varies (#E1E1E1 vs #e1e1e1), and `border:none` may or may not precede it. +const OUTLOOK_DIVIDER_STYLE = /border-top\s*:\s*solid\s*#?e1e1e1/i + +const HEADER_SCAN_CHARS = 800 + +function hasQuotedHeaderBlock(text: string): boolean { + const head = text.slice(0, HEADER_SCAN_CHARS) + return /\bFrom:/i.test(head) && /\b(Sent|Date):/i.test(head) && /\bSubject:/i.test(head) +} + +// A bare border-top div is also how plenty of signatures and marketing footers +// draw a horizontal rule, so require the header block before calling it a +// quote boundary. Wrongly hiding someone's signature is worse than wrongly +// showing a quote. +function isOutlookDivider(el: Element): boolean { + if (!OUTLOOK_DIVIDER_STYLE.test(el.getAttribute('style') || '')) return false + return hasQuotedHeaderBlock(el.textContent || '') +} + +// Hide the divider plus everything after it. The quoted chain can sit outside +// the divider's parent, so climb to the body marking later siblings at each +// level. Quoted history always terminates the message, so nothing below the +// boundary is content worth keeping. +function markFromBoundary(el: Element, body: Element): void { + el.classList.add(QUOTED_CLASS) + let node: Element | null = el + while (node && node !== body) { + for (let sib = node.nextElementSibling; sib; sib = sib.nextElementSibling) { + sib.classList.add(QUOTED_CLASS) + } + node = node.parentElement + } +} + +/** Tags quoted nodes with {@link QUOTED_CLASS}. Returns true if any were found. */ +export function markQuotedNodes(doc: Document): boolean { + const body = doc.body + if (!body) return false + let found = false + + for (const el of Array.from(body.querySelectorAll(WRAPPER_QUOTE_SELECTOR))) { + el.classList.add(QUOTED_CLASS) + found = true + } + + const boundaries = Array.from(body.querySelectorAll(BOUNDARY_QUOTE_SELECTOR)) + for (const el of Array.from(body.querySelectorAll('div[style]'))) { + if (isOutlookDivider(el)) boundaries.push(el) + } + for (const el of boundaries) { + // Already inside a quote we hid, or reached by an earlier boundary's sweep. + if (el.closest(`.${QUOTED_CLASS}`)) continue + markFromBoundary(el, body) + found = true + } + + return found +} + +// True if the HTML — ignoring quoted content — defines its own visual layout +// (real images, tables, explicit backgrounds). Unstyled HTML (Gmail replies, +// Outlook one-liners wrapped in MsoNormal boilerplate, outreach emails with +// only a tracking pixel) gets an iframe that adapts to the app theme; styled +// HTML keeps the white "paper" look so newsletters render as sent. +function isStyledDocument(doc: Document): boolean { + const clone = doc.cloneNode(true) as Document + clone.querySelectorAll(`.${QUOTED_CLASS}`).forEach((n) => n.remove()) + if (clone.querySelector('table')) return true + for (const img of Array.from(clone.querySelectorAll('img'))) { + const w = parseInt(img.getAttribute('width') || '0', 10) + const h = parseInt(img.getAttribute('height') || '0', 10) + if (w === 1 && h === 1) continue + const style = img.getAttribute('style') || '' + if (/display\s*:\s*none/i.test(style)) continue + if (/visibility\s*:\s*hidden/i.test(style)) continue + return true + } + const visible = clone.body?.innerHTML || '' + if (/bgcolor\s*=/i.test(visible)) return true + if (/background-(color|image)\s*:/i.test(visible)) return true + return false +} + +export interface PreparedEmail { + /** Body HTML with quoted nodes tagged, ready to embed in the iframe. */ + html: string + hasQuote: boolean + styled: boolean +} + +// A forward is quoted content all the way down: hiding it leaves a blank body +// behind a "•••" nobody knows to press. When nothing survives, show everything +// and drop the toggle. Real replies clear this easily — the shortest in a +// 558-message sample left 13 visible characters, forwards left zero. +function unmarkIfNothingRemains(doc: Document): boolean { + const clone = doc.cloneNode(true) as Document + clone.querySelectorAll(`.${QUOTED_CLASS}`).forEach((n) => n.remove()) + const visible = (clone.body?.textContent || '').replace(/[\s\u00a0]+/g, '') + if (visible.length > 0) return true + doc.querySelectorAll(`.${QUOTED_CLASS}`).forEach((n) => n.classList.remove(QUOTED_CLASS)) + return false +} + +/** + * Parse once, tag the quotes, and decide the styling — so the iframe hides + * quoted history on its first paint and reports the right height immediately. + */ +export function prepareEmailHtml(rawHtml: string): PreparedEmail { + const doc = new DOMParser().parseFromString(rawHtml, 'text/html') + const hasQuote = markQuotedNodes(doc) && unmarkIfNothingRemains(doc) + const styled = isStyledDocument(doc) + // Emails ship