mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-07-12 21:02:17 +02:00
feat(x): ephemeral CodeRunFeed for codex streaming + settle-time durable batch
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 <noreply@anthropic.com>
This commit is contained in:
Ramnique Singh
parent
7c06af04a7
commit
3bbed55735
15 changed files with 506 additions and 10 deletions
|
|
@ -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>('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).
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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 (
|
||||
<>
|
||||
<Tool open={open} onOpenChange={onOpenChange}>
|
||||
<ToolHeader title={title} type="tool-code_agent_run" state={toToolState(item.status)} />
|
||||
<ToolContent>
|
||||
<CodingRunTimeline events={item.codeRunEvents ?? []} error={error} />
|
||||
<CodingRunTimeline events={events} error={error} />
|
||||
</ToolContent>
|
||||
</Tool>
|
||||
{item.pendingCodePermission && (
|
||||
|
|
|
|||
52
apps/x/apps/renderer/src/lib/code-run-feed.ts
Normal file
52
apps/x/apps/renderer/src/lib/code-run-feed.ts
Normal file
|
|
@ -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<string, CodeRunEvent[]>()
|
||||
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)
|
||||
}
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<ToolCall, 'codeRunEvents' | 'pendingCodePermission'> {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<never>): void {
|
||||
function mockCodeServices(
|
||||
runPrompt: (opts: { onEvent: (event: CodeRunEvent) => void }) => Promise<unknown>,
|
||||
): { 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([]);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<typeof BuiltinToolsSchema> = {
|
|||
|
||||
let finalText = '';
|
||||
const changedFiles = new Set<string>();
|
||||
// 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>('codeRunFeed');
|
||||
try {
|
||||
const result = await manager.runPrompt({
|
||||
runId: ctx.runId,
|
||||
|
|
@ -927,6 +953,11 @@ export const BuiltinTools: z.infer<typeof BuiltinToolsSchema> = {
|
|||
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<typeof BuiltinToolsSchema> = {
|
|||
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)}`);
|
||||
});
|
||||
}
|
||||
}
|
||||
},
|
||||
},
|
||||
|
|
|
|||
32
apps/x/packages/core/src/code-mode/feed.ts
Normal file
32
apps/x/packages/core/src/code-mode/feed.ts
Normal file
|
|
@ -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);
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
});
|
||||
}
|
||||
},
|
||||
},
|
||||
|
|
|
|||
|
|
@ -73,3 +73,11 @@ export const RunPromptResult = z.object({
|
|||
sessionId: z.string(),
|
||||
});
|
||||
export type RunPromptResult = z.infer<typeof RunPromptResult>;
|
||||
|
||||
// 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;
|
||||
};
|
||||
|
|
|
|||
|
|
@ -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<CodeRunFeedEvent>(),
|
||||
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.
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue