From 3bbed557351219b31d4b1af72909b3c1cf0b40ac Mon Sep 17 00:00:00 2001 From: Ramnique Singh <30795890+ramnique@users.noreply.github.com> Date: Mon, 6 Jul 2026 14:04:33 +0530 Subject: [PATCH] feat(x): ephemeral CodeRunFeed for codex streaming + settle-time durable batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The runs->turns migration broke codex live streaming in copilot chat: the turns bridge's publish shim forwarded only tool-output-stream and silently dropped code-run-event / code-run-permission-request (the latter would deadlock a turn under policy 'ask'). An interim fix persisted every stream event as durable tool_progress, but that wrote each text chunk to the turn file — unsustainable. Final architecture — live and durable paths split: - Live (ephemeral, bypasses the turn runtime): code_agent_run broadcasts each ACP event on a new CodeRunFeed (core DI singleton), forwarded by main over a dedicated codeRun:events channel, buffered per toolCallId in a module-level renderer store and rendered by CodingRunBlock. The buffer survives session switches; nothing is persisted. - Durable (one line per run): when the run settles (success, error, or cancel), code_agent_run publishes a single code-run-events-batch with the whole ordered timeline, consecutive same-role message chunks coalesced (display-lossless — the timeline concatenates them anyway). The turns bridge maps it to tool_progress {kind:'code-run-events'}; turn-view derives the replay timeline from it, so reloads keep history. - Permissions stay durable per-ask (request + resolved marker): the renderer overlay resets on session switch, so an ephemeral-only ask would strand a blocked turn with no card to answer. Pending = requests minus resolutions (handles concurrent asks), cleared on tool result. The legacy code-section path (runs bus per-event) is untouched; its per-event ctx.publish remains and is a no-op under the turns shim. Also: repaired the two code_agent_run tests broken by the earlier cwd existence check (they used a nonexistent /repo), and added coverage for feed broadcast, batch coalescing, partial-batch-on-failure, bridge durability routing, and pending-permission derivation. Co-Authored-By: Claude Fable 5 --- apps/x/apps/main/src/ipc.ts | 20 ++++ apps/x/apps/main/src/main.ts | 2 + .../renderer/src/components/coding-run.tsx | 15 ++- apps/x/apps/renderer/src/lib/code-run-feed.ts | 52 +++++++++ .../src/lib/session-chat/turn-view.test.ts | 82 +++++++++++++ .../src/lib/session-chat/turn-view.ts | 37 ++++++ .../src/application/lib/builtin-tools.test.ts | 110 +++++++++++++++++- .../core/src/application/lib/builtin-tools.ts | 50 +++++++- apps/x/packages/core/src/code-mode/feed.ts | 32 +++++ apps/x/packages/core/src/di/container.ts | 4 + .../turns/bridges/real-tool-registry.test.ts | 53 +++++++++ .../src/turns/bridges/real-tool-registry.ts | 29 +++++ apps/x/packages/shared/src/code-mode.ts | 8 ++ apps/x/packages/shared/src/ipc.ts | 10 +- apps/x/packages/shared/src/runs.ts | 12 ++ 15 files changed, 506 insertions(+), 10 deletions(-) create mode 100644 apps/x/apps/renderer/src/lib/code-run-feed.ts create mode 100644 apps/x/packages/core/src/code-mode/feed.ts diff --git a/apps/x/apps/main/src/ipc.ts b/apps/x/apps/main/src/ipc.ts index a558f4ae..270b1e8f 100644 --- a/apps/x/apps/main/src/ipc.ts +++ b/apps/x/apps/main/src/ipc.ts @@ -38,6 +38,7 @@ import type { IOAuthRepo } from '@x/core/dist/auth/repo.js'; import { IGranolaConfigRepo } from '@x/core/dist/knowledge/granola/repo.js'; import { ICodeModeConfigRepo } from '@x/core/dist/code-mode/repo.js'; import { CodePermissionRegistry } from '@x/core/dist/code-mode/acp/permission-registry.js'; +import type { CodeRunFeed } from '@x/core/dist/code-mode/feed.js'; import { checkCodeModeAgentStatus } from '@x/core/dist/code-mode/status.js'; import { ensureEngine } from '@x/core/dist/code-mode/acp/engine-provisioner.js'; import type { ICodeProjectsRepo } from '@x/core/dist/code-mode/projects/repo.js'; @@ -703,6 +704,25 @@ export function startSessionsWatcher(): void { sessionsWatcher = sessionBus.subscribe((event) => emitSessionEvent(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. +let codeRunFeedWatcher: (() => void) | null = null; +export function startCodeRunFeedWatcher(): void { + if (codeRunFeedWatcher) { + return; + } + 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); + } + } + }); +} + // The renderer window is created before the session-index startup scan // finishes, so an early sessions:list could observe a partially built index // (the scan runs oldest-first — exactly the newest chats would be missing). diff --git a/apps/x/apps/main/src/main.ts b/apps/x/apps/main/src/main.ts index d8184cf1..4e1d42e0 100644 --- a/apps/x/apps/main/src/main.ts +++ b/apps/x/apps/main/src/main.ts @@ -3,6 +3,7 @@ import path from "node:path"; import { setupIpcHandlers, startRunsWatcher, startSessionsWatcher, markSessionsIndexReady, + startCodeRunFeedWatcher, startChannelsWatcher, startCodeSessionStatusWatcher, startServicesWatcher, @@ -422,6 +423,7 @@ app.whenReady().then(async () => { markSessionsIndexReady(); } startSessionsWatcher(); + startCodeRunFeedWatcher(); // Mobile channels (WhatsApp/Telegram bridge): needs the session index, so // start after initialize(). Failures must never block boot. diff --git a/apps/x/apps/renderer/src/components/coding-run.tsx b/apps/x/apps/renderer/src/components/coding-run.tsx index 2625b65a..62b1043d 100644 --- a/apps/x/apps/renderer/src/components/coding-run.tsx +++ b/apps/x/apps/renderer/src/components/coding-run.tsx @@ -1,4 +1,4 @@ -import { useMemo, useState } from 'react' +import { useEffect, useMemo, useState } from 'react' import { AlertCircle, CheckCircle2, @@ -19,6 +19,7 @@ import type { CodeRunEvent, PermissionAsk, PermissionDecision } from '@x/shared/ import { cn } from '@/lib/utils' import { Tool, ToolContent, ToolHeader } from '@/components/ai-elements/tool' import { getToolErrorText, toToolState, type ToolCall } from '@/lib/chat-conversation' +import { clearCodeRunBuffer, useCodeRunFeed } from '@/lib/code-run-feed' // ── Timeline reduction ────────────────────────────────────────────── // The raw ACP stream is a flat list of events; collapse it into ordered rows, @@ -283,12 +284,22 @@ export function CodingRunBlock({ (item.input as { agent?: string } | undefined)?.agent const title = AGENT_LABEL[agent ?? ''] ?? 'Coding agent' const error = getToolErrorText(item) + // Timeline source: the durable record (item.codeRunEvents — the settle-time + // batch, or the legacy path's inline accumulation) wins; while it's absent + // the live CodeRunFeed buffer streams the run in real time. + const liveEvents = useCodeRunFeed(item.id) + const durableEvents = item.codeRunEvents + const events = durableEvents?.length ? durableEvents : liveEvents + // Once the durable batch has landed the buffer is redundant — drop it. + useEffect(() => { + if (durableEvents?.length) clearCodeRunBuffer(item.id) + }, [durableEvents?.length, item.id]) return ( <> - + {item.pendingCodePermission && ( diff --git a/apps/x/apps/renderer/src/lib/code-run-feed.ts b/apps/x/apps/renderer/src/lib/code-run-feed.ts new file mode 100644 index 00000000..4ba08894 --- /dev/null +++ b/apps/x/apps/renderer/src/lib/code-run-feed.ts @@ -0,0 +1,52 @@ +import { useSyncExternalStore } from 'react' +import type { CodeRunEvent, CodeRunFeedEvent } from '@x/shared/src/code-mode.js' + +// Renderer half of the ephemeral CodeRunFeed side-channel: buffers the live +// `codeRun:events` broadcast per toolCallId so tool cards can render a +// code_agent_run's activity while it streams. Module-level (not tied to any +// session store) so the buffer survives session switches mid-run. Nothing here +// is persisted — on settle the durable code-run-events-batch in the turn +// record supersedes the buffer, which is then dropped. +const buffers = new Map() +const listeners = new Set<() => void>() +const EMPTY: CodeRunEvent[] = [] +// Backstop so abandoned runs can't grow the map forever (a run's buffer is +// normally dropped explicitly once its durable batch lands). +const MAX_TRACKED_RUNS = 32 + +let attached = false +function ensureAttached(): void { + if (attached) return + attached = true + window.ipc.on('codeRun:events', ((raw: unknown) => { + const { toolCallId, event } = raw as CodeRunFeedEvent + if (!toolCallId || !event) return + if (!buffers.has(toolCallId) && buffers.size >= MAX_TRACKED_RUNS) { + const oldest = buffers.keys().next().value + if (oldest !== undefined) buffers.delete(oldest) + } + // Immutable append: useSyncExternalStore consumers compare by reference. + buffers.set(toolCallId, [...(buffers.get(toolCallId) ?? EMPTY), event]) + for (const listener of [...listeners]) listener() + }) as never) +} + +function subscribe(onChange: () => void): () => void { + ensureAttached() + listeners.add(onChange) + return () => { + listeners.delete(onChange) + } +} + +export function clearCodeRunBuffer(toolCallId: string): void { + if (buffers.delete(toolCallId)) { + for (const listener of [...listeners]) listener() + } +} + +// Live events for one code_agent_run tool call, empty once the durable batch +// takes over (or if the buffer never existed — e.g. after an app reload). +export function useCodeRunFeed(toolCallId: string): CodeRunEvent[] { + return useSyncExternalStore(subscribe, () => buffers.get(toolCallId) ?? EMPTY) +} diff --git a/apps/x/apps/renderer/src/lib/session-chat/turn-view.test.ts b/apps/x/apps/renderer/src/lib/session-chat/turn-view.test.ts index 34df50ba..10671d9e 100644 --- a/apps/x/apps/renderer/src/lib/session-chat/turn-view.test.ts +++ b/apps/x/apps/renderer/src/lib/session-chat/turn-view.test.ts @@ -219,6 +219,88 @@ describe('buildTurnConversation', () => { expect(items.map((i) => i.status)).toEqual(['error', 'completed']) }) + it('derives the code-run timeline from the settle-time batch and asks from request events', () => { + const codeProgress = (progress: unknown): TEvent => ({ + type: 'tool_progress', + turnId: T1, + ts: TS, + toolCallId: 'cr1', + source: 'sync', + progress: progress as never, + }) + const state = reduceTurn([ + created(T1, S1), + requested(T1, 0), + completed(T1, 0, assistantCalls(toolCallPart('cr1', 'code_agent_run', { agent: 'codex' }))), + invocation(T1, 'cr1', 'code_agent_run'), + codeProgress({ + kind: 'code-run-permission-request', + requestId: 'cpr-1', + ask: { toolCallId: 'x', title: 'write file', options: [] }, + }), + codeProgress({ + kind: 'code-run-events', + events: [ + { type: 'message', role: 'agent', text: 'hi' }, + { type: 'tool_call', id: 'x', title: 'write file' }, + ], + }), + ]) + const tool = buildTurnConversation(state).filter(isToolCall)[0] + expect(tool.status).toBe('running') + expect(tool.codeRunEvents?.map((e) => e.type)).toEqual(['message', 'tool_call']) + expect(tool.pendingCodePermission?.requestId).toBe('cpr-1') + }) + + it('clears the pending code permission on the resolved marker and on tool result', () => { + const codeProgress = (toolCallId: string, progress: unknown): TEvent => ({ + type: 'tool_progress', + turnId: T1, + ts: TS, + toolCallId, + source: 'sync', + progress: progress as never, + }) + const ask = { toolCallId: 'x', title: 'write file', options: [] } + // resolved: the durable marker pairs off the request mid-run + const resolvedState = reduceTurn([ + created(T1, S1), + requested(T1, 0), + completed(T1, 0, assistantCalls(toolCallPart('cr1', 'code_agent_run'))), + invocation(T1, 'cr1', 'code_agent_run'), + codeProgress('cr1', { kind: 'code-run-permission-request', requestId: 'cpr-1', ask }), + codeProgress('cr1', { kind: 'code-run-permission-resolved' }), + ]) + const resolved = buildTurnConversation(resolvedState).filter(isToolCall)[0] + expect(resolved.pendingCodePermission).toBeUndefined() + + // a second ask after the first resolution is pending again + const secondAskState = reduceTurn([ + created(T1, S1), + requested(T1, 0), + completed(T1, 0, assistantCalls(toolCallPart('cr1', 'code_agent_run'))), + invocation(T1, 'cr1', 'code_agent_run'), + codeProgress('cr1', { kind: 'code-run-permission-request', requestId: 'cpr-1', ask }), + codeProgress('cr1', { kind: 'code-run-permission-resolved' }), + codeProgress('cr1', { kind: 'code-run-permission-request', requestId: 'cpr-2', ask }), + ]) + const secondAsk = buildTurnConversation(secondAskState).filter(isToolCall)[0] + expect(secondAsk.pendingCodePermission?.requestId).toBe('cpr-2') + + // settled: an unanswered ask must not survive the tool's terminal result + const settledState = reduceTurn([ + created(T1, S1), + requested(T1, 0), + completed(T1, 0, assistantCalls(toolCallPart('cr2', 'code_agent_run'))), + invocation(T1, 'cr2', 'code_agent_run'), + codeProgress('cr2', { kind: 'code-run-permission-request', requestId: 'cpr-2', ask }), + toolResult(T1, 'cr2', 'code_agent_run', { success: false, stopReason: 'cancelled' }), + ]) + const settled = buildTurnConversation(settledState).filter(isToolCall)[0] + expect(settled.pendingCodePermission).toBeUndefined() + expect(settled.status).toBe('completed') + }) + it('renders user attachments and a failed turn as an error item', () => { const input = { role: 'user' as const, diff --git a/apps/x/apps/renderer/src/lib/session-chat/turn-view.ts b/apps/x/apps/renderer/src/lib/session-chat/turn-view.ts index 2e8df218..cc1fc6da 100644 --- a/apps/x/apps/renderer/src/lib/session-chat/turn-view.ts +++ b/apps/x/apps/renderer/src/lib/session-chat/turn-view.ts @@ -13,6 +13,7 @@ import { type TurnState, type TurnStreamEvent, } from '@x/shared/src/turns.js' +import type { CodeRunEvent, PermissionAsk } from '@x/shared/src/code-mode.js' import type { ChatMessage, ConversationItem, @@ -205,6 +206,41 @@ function toolStatus(tc: ToolCallState): ToolCall['status'] { return 'running' } +// code_agent_run's durable trail in tool_progress (see the publish bridge in +// real-tool-registry.ts): ONE settle-time 'code-run-events' batch carrying the +// whole timeline (the live per-event stream travels over the ephemeral +// CodeRunFeed and never reaches turn state), plus per-ask +// 'code-run-permission-request' / 'code-run-permission-resolved' pairs. An ask +// is pending while requests outnumber resolutions and the tool hasn't settled. +function codeRunViewOf( + tc: ToolCallState, +): Pick { + let events: CodeRunEvent[] | undefined + let pending: { requestId: string; ask: PermissionAsk } | null = null + let unresolved = 0 + for (const p of tc.progress) { + const entry = p.progress + if (!entry || typeof entry !== 'object' || Array.isArray(entry)) continue + const kind = (entry as { kind?: unknown }).kind + if (kind === 'code-run-events') { + const batch = (entry as { events?: unknown }).events + if (Array.isArray(batch)) events = batch as CodeRunEvent[] + } else if (kind === 'code-run-permission-request') { + const { requestId, ask } = entry as { requestId?: unknown; ask?: unknown } + if (typeof requestId === 'string' && ask) { + pending = { requestId, ask: ask as PermissionAsk } + unresolved += 1 + } + } else if (kind === 'code-run-permission-resolved') { + unresolved -= 1 + } + } + return { + ...(events && events.length > 0 ? { codeRunEvents: events } : {}), + ...(pending && unresolved > 0 && !tc.result ? { pendingCodePermission: pending } : {}), + } +} + // One turn's contribution to the conversation: the user input, then per // completed model call its text and tool calls (with live status/results). export function buildTurnConversation(state: TurnState): ConversationItem[] { @@ -258,6 +294,7 @@ export function buildTurnConversation(state: TurnState): ConversationItem[] { ...(tc?.result ? { result: tc.result.result.output as ToolCall['result'] } : {}), status: tc ? toolStatus(tc) : 'running', timestamp: ts(), + ...(tc ? codeRunViewOf(tc) : {}), } satisfies ToolCall) } } diff --git a/apps/x/packages/core/src/application/lib/builtin-tools.test.ts b/apps/x/packages/core/src/application/lib/builtin-tools.test.ts index 1842024c..12ff2fe1 100644 --- a/apps/x/packages/core/src/application/lib/builtin-tools.test.ts +++ b/apps/x/packages/core/src/application/lib/builtin-tools.test.ts @@ -1,28 +1,42 @@ +import * as os from "os"; import { afterEach, describe, expect, it, vi } from "vitest"; +import type { CodeRunEvent } from "@x/shared/dist/code-mode.js"; import container from "../../di/container.js"; import { InMemoryAbortRegistry } from "../../runs/abort-registry.js"; -import { BuiltinTools } from "./builtin-tools.js"; +import { BuiltinTools, coalesceCodeRunEvents } from "./builtin-tools.js"; import type { ToolContext } from "./exec-tool.js"; -function context(signal: AbortSignal): ToolContext { +// A real directory: code_agent_run validates the cwd exists before spawning. +const CWD = os.tmpdir(); + +function context(signal: AbortSignal, published: unknown[] = []): ToolContext { return { runId: "turn-1", toolCallId: "tool-1", signal, abortRegistry: new InMemoryAbortRegistry(), - publish: async () => {}, + publish: async (event) => { + published.push(event); + }, codePolicy: "ask", }; } -function mockCodeServices(runPrompt: () => Promise): void { +function mockCodeServices( + runPrompt: (opts: { onEvent: (event: CodeRunEvent) => void }) => Promise, +): { feedEvents: unknown[] } { + const feedEvents: unknown[] = []; vi.spyOn(container, "resolve").mockImplementation(((name: string) => { if (name === "codeModeManager") return { runPrompt }; if (name === "codePermissionRegistry") { return { cancelRun: vi.fn(), request: vi.fn() }; } + if (name === "codeRunFeed") { + return { broadcast: (event: unknown) => feedEvents.push(event) }; + } throw new Error(`Unexpected dependency: ${name}`); }) as typeof container.resolve); + return { feedEvents }; } describe("code_agent_run", () => { @@ -34,11 +48,22 @@ describe("code_agent_run", () => { }); await expect(BuiltinTools.code_agent_run.execute( - { agent: "codex", cwd: "/repo", prompt: "Fix it" }, + { agent: "codex", cwd: CWD, prompt: "Fix it" }, context(new AbortController().signal), )).rejects.toThrow("Coding agent failed: spawn Electron ENOENT"); }); + it("rejects a working directory that does not exist with a clear error", async () => { + mockCodeServices(async () => { + throw new Error("unreachable"); + }); + + await expect(BuiltinTools.code_agent_run.execute( + { agent: "codex", cwd: "/nonexistent-dir-for-test", prompt: "Fix it" }, + context(new AbortController().signal), + )).rejects.toThrow("working directory does not exist"); + }); + it("returns an ordinary cancellation result when the turn was aborted", async () => { const controller = new AbortController(); controller.abort(); @@ -47,8 +72,81 @@ describe("code_agent_run", () => { }); await expect(BuiltinTools.code_agent_run.execute( - { agent: "codex", cwd: "/repo", prompt: "Fix it" }, + { agent: "codex", cwd: CWD, prompt: "Fix it" }, context(controller.signal), )).resolves.toMatchObject({ success: false, stopReason: "cancelled" }); }); + + it("broadcasts events on the feed live and publishes ONE coalesced durable batch", async () => { + const { feedEvents } = mockCodeServices(async ({ onEvent }) => { + onEvent({ type: "message", role: "agent", text: "hel" }); + onEvent({ type: "message", role: "agent", text: "lo" }); + onEvent({ type: "tool_call", id: "x", title: "write file" }); + return { stopReason: "end_turn", sessionId: "s1" }; + }); + const published: unknown[] = []; + + const result = await BuiltinTools.code_agent_run.execute( + { agent: "codex", cwd: CWD, prompt: "Fix it" }, + context(new AbortController().signal, published), + ); + + expect(result).toMatchObject({ success: true, summary: "hello" }); + // Live side-channel: every event, verbatim, keyed by the tool call. + expect(feedEvents).toHaveLength(3); + expect(feedEvents[0]).toMatchObject({ + toolCallId: "tool-1", + event: { type: "message", text: "hel" }, + }); + // Durable: per-event publishes for the legacy bus + exactly one batch, + // with consecutive same-role message chunks coalesced. + const batches = published.filter( + (e) => (e as { type?: string }).type === "code-run-events-batch", + ); + expect(batches).toHaveLength(1); + expect((batches[0] as { events: CodeRunEvent[] }).events).toEqual([ + { type: "message", role: "agent", text: "hello" }, + { type: "tool_call", id: "x", title: "write file" }, + ]); + }); + + it("publishes the partial batch even when the run fails", async () => { + mockCodeServices(async ({ onEvent }) => { + onEvent({ type: "message", role: "agent", text: "started..." }); + throw new Error("engine crashed"); + }); + const published: unknown[] = []; + + await expect(BuiltinTools.code_agent_run.execute( + { agent: "codex", cwd: CWD, prompt: "Fix it" }, + context(new AbortController().signal, published), + )).rejects.toThrow("Coding agent failed"); + const batches = published.filter( + (e) => (e as { type?: string }).type === "code-run-events-batch", + ); + expect(batches).toHaveLength(1); + }); +}); + +describe("coalesceCodeRunEvents", () => { + it("merges consecutive same-role message chunks and keeps everything else in order", () => { + const events: CodeRunEvent[] = [ + { type: "message", role: "agent", text: "a" }, + { type: "message", role: "agent", text: "b" }, + { type: "tool_call", id: "t1", title: "run" }, + { type: "message", role: "agent", text: "c" }, + { type: "message", role: "user", text: "d" }, + { type: "message", role: "user", text: "e" }, + ]; + expect(coalesceCodeRunEvents(events)).toEqual([ + { type: "message", role: "agent", text: "ab" }, + { type: "tool_call", id: "t1", title: "run" }, + { type: "message", role: "agent", text: "c" }, + { type: "message", role: "user", text: "de" }, + ]); + }); + + it("returns an empty list unchanged", () => { + expect(coalesceCodeRunEvents([])).toEqual([]); + }); }); diff --git a/apps/x/packages/core/src/application/lib/builtin-tools.ts b/apps/x/packages/core/src/application/lib/builtin-tools.ts index bf40ab1b..a7921852 100644 --- a/apps/x/packages/core/src/application/lib/builtin-tools.ts +++ b/apps/x/packages/core/src/application/lib/builtin-tools.ts @@ -20,7 +20,8 @@ import { BackgroundTaskSchema, TriggersSchema } from "@x/shared/dist/background- import type { CodeModeManager } from "../../code-mode/acp/manager.js"; import type { CodePermissionRegistry } from "../../code-mode/acp/permission-registry.js"; import { ICodeModeConfigRepo } from "../../code-mode/repo.js"; -import type { ApprovalPolicy } from "@x/shared/dist/code-mode.js"; +import type { ApprovalPolicy, CodeRunEvent as CodeRunEventType } from "@x/shared/dist/code-mode.js"; +import type { CodeRunFeed } from "../../code-mode/feed.js"; import type { ICodeProjectsRepo } from "../../code-mode/projects/repo.js"; import * as gitService from "../../code-mode/git/service.js"; import { listImportantThreads, searchThreads } from "../../knowledge/sync_gmail.js"; @@ -68,6 +69,27 @@ function expandHome(p: string): string { return t; } +// Shrink a code-run timeline for durable storage: consecutive same-role message +// chunks merge into one event. Display-lossless — the timeline renderer +// concatenates consecutive messages anyway (CodingRunTimeline) — and typically +// collapses the ~90% of a run's events that are per-token text deltas. +// Everything else (tool calls/updates, plans, permissions) is kept verbatim in +// order: updates are id-keyed transitions and must not be merged. +export function coalesceCodeRunEvents(events: CodeRunEventType[]): CodeRunEventType[] { + const out: CodeRunEventType[] = []; + for (const event of events) { + const last = out[out.length - 1]; + if ( + event.type === 'message' && last?.type === 'message' && last.role === event.role + ) { + out[out.length - 1] = { ...last, text: last.text + event.text }; + } else { + out.push(event); + } + } + return out; +} + async function resolveCodeProject(dirPath: string): Promise< { ok: true; projectId: string; path: string; warning?: string } | { ok: false; error: string } > { @@ -916,6 +938,10 @@ export const BuiltinTools: z.infer = { let finalText = ''; const changedFiles = new Set(); + // The full ordered timeline, published ONCE as a durable batch when the + // run settles (see finally). The per-event copies below are ephemeral. + const collected: CodeRunEventType[] = []; + const feed = container.resolve('codeRunFeed'); try { const result = await manager.runPrompt({ runId: ctx.runId, @@ -927,6 +953,11 @@ export const BuiltinTools: z.infer = { onEvent: (event) => { if (event.type === 'message' && event.role === 'agent') finalText += event.text; if (event.type === 'tool_call_update') for (const f of event.diffs) changedFiles.add(f); + collected.push(event); + // Live rendering, two transports: the CodeRunFeed side-channel + // (turns-runtime chats — the runtime never sees this traffic) + // and the legacy runs bus (code-section tabs). Both ephemeral. + feed.broadcast({ toolCallId: ctx.toolCallId, event }); void ctx.publish({ runId: ctx.runId, type: 'code-run-event', @@ -969,6 +1000,23 @@ export const BuiltinTools: z.infer = { throw new Error(`Coding agent failed: ${error instanceof Error ? error.message : String(error)}`); } finally { ctx.signal.removeEventListener('abort', onAbort); + // Durable record for replay-on-reload — one event with the whole + // (coalesced) timeline, on every settle path including errors and + // cancellation, so partial runs keep their history too. + if (collected.length > 0) { + await ctx.publish({ + runId: ctx.runId, + type: 'code-run-events-batch', + toolCallId: ctx.toolCallId, + events: coalesceCodeRunEvents(collected), + subflow: [], + }).catch((e: unknown) => { + // History is best-effort (rethrowing here would mask the run's + // real outcome) — but a lost timeline must leave a trail, since + // this batch is the only durable record of the run's activity. + console.warn(`[code_agent_run] failed to persist code-run timeline: ${e instanceof Error ? e.message : String(e)}`); + }); + } } }, }, diff --git a/apps/x/packages/core/src/code-mode/feed.ts b/apps/x/packages/core/src/code-mode/feed.ts new file mode 100644 index 00000000..f63b05b8 --- /dev/null +++ b/apps/x/packages/core/src/code-mode/feed.ts @@ -0,0 +1,32 @@ +import type { CodeRunFeedEvent } from '@x/shared/dist/code-mode.js'; + +// Ephemeral side-channel for code_agent_run's live ACP stream — a direct +// tool-implementation → renderer contract that deliberately bypasses the turn +// runtime. The stream is chatty (per-chunk agent messages, tool status +// updates) and only ever renders inside one tool card, so persisting each +// event as durable turn progress bloats the turn log for no benefit. Instead: +// - live: the tool broadcasts here; main forwards over `codeRun:events` +// (see apps/main ipc.ts) and the renderer buffers per toolCallId. +// - durable: ONE code-run-events-batch is published when the run settles, +// so reloads replay the full timeline from the turn record. +// Fire-and-forget: no subscribers ⇒ events vanish, which is the point. +export class CodeRunFeed { + private readonly listeners = new Set<(event: CodeRunFeedEvent) => void>(); + + broadcast(event: CodeRunFeedEvent): void { + for (const listener of [...this.listeners]) { + try { + listener(event); + } catch { + // A broken subscriber must not stall the coding turn. + } + } + } + + subscribe(listener: (event: CodeRunFeedEvent) => void): () => void { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + } +} diff --git a/apps/x/packages/core/src/di/container.ts b/apps/x/packages/core/src/di/container.ts index eefdaa5f..77bda6bb 100644 --- a/apps/x/packages/core/src/di/container.ts +++ b/apps/x/packages/core/src/di/container.ts @@ -21,6 +21,7 @@ import { FSSlackConfigRepo, ISlackConfigRepo } from "../slack/repo.js"; import { FSChannelsConfigRepo, IChannelsConfigRepo } from "../channels/repo.js"; import { CodeModeManager } from "../code-mode/acp/manager.js"; import { CodePermissionRegistry } from "../code-mode/acp/permission-registry.js"; +import { CodeRunFeed } from "../code-mode/feed.js"; import { FSCodeProjectsRepo, ICodeProjectsRepo } from "../code-mode/projects/repo.js"; import { FSCodeSessionsRepo, ICodeSessionsRepo } from "../code-mode/sessions/repo.js"; import { CodeSessionService } from "../code-mode/sessions/service.js"; @@ -98,6 +99,9 @@ container.register({ // session/load); the registry brokers mid-run approvals. codeModeManager: asClass(CodeModeManager).singleton(), codePermissionRegistry: asClass(CodePermissionRegistry).singleton(), + // Ephemeral live stream for code_agent_run (renderer side-channel; the + // durable record is the settle-time code-run-events-batch). + codeRunFeed: asClass(CodeRunFeed).singleton(), // Code section: project registry, session metadata, the direct-drive // session service, and the live status tracker. diff --git a/apps/x/packages/core/src/turns/bridges/real-tool-registry.test.ts b/apps/x/packages/core/src/turns/bridges/real-tool-registry.test.ts index 4fb6113e..4ccb15a6 100644 --- a/apps/x/packages/core/src/turns/bridges/real-tool-registry.test.ts +++ b/apps/x/packages/core/src/turns/bridges/real-tool-registry.test.ts @@ -125,6 +125,59 @@ describe("RealToolRegistry", () => { expect(ctx.progress).toEqual([{ kind: "tool-output", chunk: "chunk-1" }]); }); + it("keeps code-run durability to permission asks/resolutions and the settle batch", async () => { + const chunk = { type: "message", role: "agent", text: "hi" } as const; + const resolution = { + type: "permission", + ask: { toolCallId: "x", title: "write file", options: [] }, + decision: "allow_once", + auto: false, + } as const; + const ask = { toolCallId: "x", title: "write file", options: [] }; + const { registry } = makeRegistry(async ({ ctx }) => { + // Chatty stream events are ephemeral (CodeRunFeed) — NOT progress. + await ctx.publish({ + runId: "turn-1", + type: "code-run-event", + toolCallId: "tc-1", + event: chunk, + subflow: [], + }); + await ctx.publish({ + runId: "turn-1", + type: "code-run-permission-request", + toolCallId: "tc-1", + requestId: "cpr-1", + ask, + subflow: [], + }); + // A permission resolution in the stream leaves a durable marker. + await ctx.publish({ + runId: "turn-1", + type: "code-run-event", + toolCallId: "tc-1", + event: resolution, + subflow: [], + }); + await ctx.publish({ + runId: "turn-1", + type: "code-run-events-batch", + toolCallId: "tc-1", + events: [chunk, resolution], + subflow: [], + }); + return "done"; + }); + const tool = (await registry.resolve(descriptor())) as SyncRuntimeTool; + const ctx = makeCtx(); + await tool.execute({}, ctx); + expect(ctx.progress).toEqual([ + { kind: "code-run-permission-request", requestId: "cpr-1", ask }, + { kind: "code-run-permission-resolved" }, + { kind: "code-run-events", events: [chunk, resolution] }, + ]); + }); + it("wires the abort signal to the registry's force-kill path", async () => { const controller = new AbortController(); const { registry, abortRegistry } = makeRegistry(async () => { diff --git a/apps/x/packages/core/src/turns/bridges/real-tool-registry.ts b/apps/x/packages/core/src/turns/bridges/real-tool-registry.ts index 697978ab..e891ccb9 100644 --- a/apps/x/packages/core/src/turns/bridges/real-tool-registry.ts +++ b/apps/x/packages/core/src/turns/bridges/real-tool-registry.ts @@ -108,6 +108,35 @@ export class RealToolRegistry implements IToolRegistry { kind: "tool-output", chunk: event.output, }); + } else if (event.type === "code-run-event") { + // The live per-event stream travels over the + // ephemeral CodeRunFeed (never persisted) — but a + // permission RESOLUTION is durably marked so an + // answered ask never resurrects as a pending card + // after a reload or session switch. + if (event.event.type === "permission") { + await ctx.reportProgress({ + kind: "code-run-permission-resolved", + }); + } + } else if (event.type === "code-run-permission-request") { + // Durable (not feed-ephemeral): the coding turn is + // BLOCKED until the user answers via + // codeRun:resolvePermission, so the ask must survive + // session switches — dropping it would hang the turn + // under policy 'ask' with no card to answer. + await ctx.reportProgress({ + kind: "code-run-permission-request", + requestId: event.requestId, + ask: toJsonValue(event.ask), + }); + } else if (event.type === "code-run-events-batch") { + // Settle-time durable record of the whole timeline — + // what reloads replay instead of the live feed. + await ctx.reportProgress({ + kind: "code-run-events", + events: toJsonValue(event.events), + }); } }, }, diff --git a/apps/x/packages/shared/src/code-mode.ts b/apps/x/packages/shared/src/code-mode.ts index 8a85f62f..4a8429e8 100644 --- a/apps/x/packages/shared/src/code-mode.ts +++ b/apps/x/packages/shared/src/code-mode.ts @@ -73,3 +73,11 @@ export const RunPromptResult = z.object({ sessionId: z.string(), }); export type RunPromptResult = z.infer; + +// One item on the ephemeral CodeRunFeed (`codeRun:events` broadcast): a live +// code-run event tagged with the tool call it belongs to. Fire-and-forget — +// the durable record is the code-run-events-batch written when the run settles. +export type CodeRunFeedEvent = { + toolCallId: string; + event: CodeRunEvent; +}; diff --git a/apps/x/packages/shared/src/ipc.ts b/apps/x/packages/shared/src/ipc.ts index 04651dc8..4958a03a 100644 --- a/apps/x/packages/shared/src/ipc.ts +++ b/apps/x/packages/shared/src/ipc.ts @@ -21,7 +21,7 @@ import { ZListToolkitsResponse } from './composio.js'; import { BrowserStateSchema, HttpAuthRequestSchema } from './browser-control.js'; import { BillingInfoSchema } from './billing.js'; import { EmailBlockSchema, GmailThreadSchema } from './blocks.js'; -import { PermissionDecision, ApprovalPolicy, CodingAgent } from './code-mode.js'; +import { PermissionDecision, ApprovalPolicy, CodingAgent, type CodeRunFeedEvent } from './code-mode.js'; import { NotificationSettingsSchema } from './notification-settings.js'; import { CodeProject, CodeSession, CodeSessionMode, CodeSessionStatus, GitRepoInfo, GitStatusFile, CodeAgentModelOptions } from './code-sessions.js'; import { ChannelsConfig, ChannelsStatus } from './channels.js'; @@ -450,6 +450,14 @@ const ipcSchemas = { req: z.null(), res: z.null(), }, + // Ephemeral code-run stream (CodeRunFeed): per-event broadcast of a + // code_agent_run's live ACP activity, keyed by toolCallId. Never persisted — + // the durable record is the code-run-events-batch tool progress written when + // the run settles. Typed via z.custom like the other broadcast feeds. + 'codeRun:events': { + req: z.custom(), + res: z.null(), + }, // ── 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. diff --git a/apps/x/packages/shared/src/runs.ts b/apps/x/packages/shared/src/runs.ts index 571bf95a..e5771c2e 100644 --- a/apps/x/packages/shared/src/runs.ts +++ b/apps/x/packages/shared/src/runs.ts @@ -130,6 +130,17 @@ export const CodeRunPermissionRequestEvent = BaseRunEvent.extend({ ask: PermissionAsk, }); +// The complete, ordered code-run timeline, published ONCE when the coding turn +// settles (consecutive agent message chunks coalesced — display-lossless, the +// timeline concatenates them anyway). This is the durable record; the live +// per-event stream travels over the ephemeral CodeRunFeed (`codeRun:events`) +// and is never persisted. +export const CodeRunEventsBatchEvent = BaseRunEvent.extend({ + type: z.literal("code-run-events-batch"), + toolCallId: z.string(), + events: z.array(CodeRunEventSchema), +}); + export const ToolPermissionAutoDecisionEvent = BaseRunEvent.extend({ type: z.literal("tool-permission-auto-decision"), toolCallId: z.string(), @@ -165,6 +176,7 @@ export const RunEvent = z.union([ ToolPermissionResponseEvent, CodeRunStreamEvent, CodeRunPermissionRequestEvent, + CodeRunEventsBatchEvent, ToolPermissionAutoDecisionEvent, RunErrorEvent, RunStoppedEvent,