diff --git a/desktop/electron/resources/agent_families.generated.json b/desktop/electron/resources/agent_families.generated.json index 6d4d987d..315fe727 100644 --- a/desktop/electron/resources/agent_families.generated.json +++ b/desktop/electron/resources/agent_families.generated.json @@ -712,7 +712,7 @@ }, "prompt_pdf": { "M1": true, - "M2": true, + "M2": false, "M4": false } }, diff --git a/desktop/electron/src/localagent/claudechild.test.ts b/desktop/electron/src/localagent/claudechild.test.ts index 4b1f833e..e1c4a0b3 100644 --- a/desktop/electron/src/localagent/claudechild.test.ts +++ b/desktop/electron/src/localagent/claudechild.test.ts @@ -13,7 +13,8 @@ import assert from 'node:assert/strict'; import { EventEmitter } from 'node:events'; import { PassThrough } from 'node:stream'; -import { ClaudeChild, MAX_FRAME_BYTES, type DriverEvent, type SpawnedChild } from './claudechild.ts'; +import { ClaudeChild, MAX_FRAME_BYTES, type SpawnedChild } from './claudechild.ts'; +import type { DriverEvent } from './driver.ts'; import type { Family } from './families.ts'; class FakeChild extends EventEmitter { diff --git a/desktop/electron/src/localagent/claudechild.ts b/desktop/electron/src/localagent/claudechild.ts index d971c0ea..7ed0a879 100644 --- a/desktop/electron/src/localagent/claudechild.ts +++ b/desktop/electron/src/localagent/claudechild.ts @@ -18,17 +18,16 @@ import { spawn as nodeSpawn } from 'node:child_process'; import { applyProfile } from '../frameprofile/translate.ts'; import type { EmittedEvent } from '../frameprofile/types.ts'; -import type { Family } from './families.ts'; import { - buildInputFrame, - buildLaunchArgs, - ContextWindows, DEFAULT_TOOL_POSTURE, - TurnClock, + type DriverEvent, type InputKind, type InputPayload, + type LocalDriver, type ToolPosture, -} from './claudewire.ts'; +} from './driver.ts'; +import type { Family } from './families.ts'; +import { buildInputFrame, buildLaunchArgs, ContextWindows, TurnClock } from './claudewire.ts'; /// The child's streams, narrowed to what this module uses. Structural so a /// test can supply plain Node streams. @@ -46,14 +45,6 @@ export type SpawnFn = ( opts: { cwd: string; env: NodeJS.ProcessEnv }, ) => SpawnedChild; -/// An event on its way to the session log — kind/producer/payload, before the -/// log assigns it a `seq`. -export interface DriverEvent { - kind: string; - producer: string; - payload: Record; -} - export interface ClaudeChildOptions { family: Family; cwd: string; @@ -83,7 +74,7 @@ export interface ClaudeChildOptions { /// growing main's heap without bound. export const MAX_FRAME_BYTES = 1 << 20; -export class ClaudeChild { +export class ClaudeChild implements LocalDriver { readonly #opts: ClaudeChildOptions; readonly #turns = new TurnClock(); readonly #windows = new ContextWindows(); diff --git a/desktop/electron/src/localagent/claudewire.test.ts b/desktop/electron/src/localagent/claudewire.test.ts index d3580281..74a96130 100644 --- a/desktop/electron/src/localagent/claudewire.test.ts +++ b/desktop/electron/src/localagent/claudewire.test.ts @@ -7,12 +7,11 @@ import { buildLaunchArgs, ContextWindows, CONFIG_HOME_ENV, - DEFAULT_TOOL_POSTURE, - isToolPosture, resolveConfigHome, toolArgs, TurnClock, } from './claudewire.ts'; +import { DEFAULT_TOOL_POSTURE, isToolPosture } from './driver.ts'; import type { Family } from './families.ts'; const FAMILY: Family = { diff --git a/desktop/electron/src/localagent/claudewire.ts b/desktop/electron/src/localagent/claudewire.ts index 5856f7cf..c74c3720 100644 --- a/desktop/electron/src/localagent/claudewire.ts +++ b/desktop/electron/src/localagent/claudewire.ts @@ -10,6 +10,7 @@ /// state (a turn counter, a per-model window learned from an earlier frame). /// L2's note says they "belong to whatever owns the session", and this is it. +import type { InputKind, InputPayload, ToolPosture } from './driver.ts'; import type { Family } from './families.ts'; // ── Config root ────────────────────────────────────────────────────────────── @@ -43,7 +44,7 @@ export function resolveConfigHome( // ── Tool posture ───────────────────────────────────────────────────────────── -/// Which built-in tools a local child is launched with. +/// How claude keeps the posture contract (`driver.ts` `ToolPosture`). /// /// **This is not the hub's `permission mode`, and the difference is not /// cosmetic.** Permission mode gates whether a tool call is *approved*, and it @@ -65,17 +66,12 @@ export function resolveConfigHome( /// entirely. The model reports them as unavailable and calls nothing. /// /// So the only lever that gates a non-interactive child is the tool list, and -/// this type is that list behind three named postures. An allowlist, not a +/// this table is that list behind the three named postures. An allowlist, not a /// denylist: a denylist fails open for every tool claude adds after this file /// was written, and "the engine grew a capability" must not silently widen what -/// a local session can do to the director's machine. -export type ToolPosture = 'converse' | 'read_local' | 'unrestricted'; - -/// The default. Reading the workdir is what makes a co-working Companion -/// useful; writing, executing and reaching the network are what make an -/// unattended one dangerous, and none of the three are here. -export const DEFAULT_TOOL_POSTURE: ToolPosture = 'read_local'; - +/// a local session can do to the director's machine. (codex keeps the same +/// contract by a different mechanism — an OS sandbox — see `codexPosture`.) +/// /// Tool names are claude's own, read off a live `system/init` frame's `tools` /// array rather than from documentation. const POSTURE_TOOLS: Record = { @@ -90,10 +86,6 @@ const POSTURE_TOOLS: Record = { unrestricted: null, }; -export function isToolPosture(v: unknown): v is ToolPosture { - return v === 'converse' || v === 'read_local' || v === 'unrestricted'; -} - /// The `--tools` argv for a posture, or [] when the posture adds no flag. export function toolArgs(posture: ToolPosture): string[] { const tools = POSTURE_TOOLS[posture]; @@ -159,30 +151,6 @@ export function buildLaunchArgs(family: Family, opts: LaunchOptions): string[] { // ── Input frames ───────────────────────────────────────────────────────────── -/// A binary attachment lowered onto an Anthropic content block. -export interface AttachmentInput { - mime: string; - /// base64, without a data: prefix. - data: string; - filename?: string; -} - -export interface InputPayload { - body?: string; - images?: AttachmentInput[]; - pdfs?: AttachmentInput[]; - request_id?: string; - decision?: string; - note?: string; - reason?: string; -} - -/// The input kinds the local claude driver accepts. A deliberate subset of the -/// Go driver's: `attention_reply` and `attach` are hub concepts (an attention -/// table, a document entity) that a local session has none of, so they are -/// absent rather than stubbed (D-4). -export type InputKind = 'text' | 'approval' | 'answer' | 'cancel'; - /// Build the stream-json line for one user-side input — the port of /// `buildStreamJSONInputFrame` (driver_stdio.go:707). /// diff --git a/desktop/electron/src/localagent/codexdriver.e2e.test.ts b/desktop/electron/src/localagent/codexdriver.e2e.test.ts new file mode 100644 index 00000000..4b327b0e --- /dev/null +++ b/desktop/electron/src/localagent/codexdriver.e2e.test.ts @@ -0,0 +1,224 @@ +/// End-to-end against a REAL codex app-server (vision-parity L4c). +/// +/// The unit tests fake the channel, which proves the driver's bookkeeping and +/// proves nothing about the PROTOCOL: whether `thread/start` takes the params +/// we send it, whether `turn/start.input` accepts the blocks we build, whether +/// `thread/resume` finds the thread again. Every one of those is a claim about +/// a program we did not write, and this file is where they are measured. +/// +/// It has already earned itself. The shapes the hub's Go driver has been +/// sending — `{type:"input_image", image_url}` and `{type:"input_file", +/// file_data}` — are answered by codex-cli 0.147.0 with `-32600 unknown +/// variant`, which fails the whole `turn/start`. A fake server accepts anything +/// and had pinned the wrong shape for as long as it shipped. +/// +/// **Opt-in, because this one costs tokens** — it runs real model turns against +/// the operator's codex account: +/// +/// TERMIPOD_CODEX_DRIVER_E2E=1 npm test +/// +/// Skipped without the variable, and skipped with a clear reason when codex is +/// absent or unauthenticated (neither is a failure of this code). +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; + +import { codexBinary, codexHome } from './codexattach.ts'; +import { CodexDriver } from './codexdriver.ts'; +import type { DriverEvent } from './driver.ts'; +import { parseFamilies, familyByName, type Family } from './families.ts'; + +const ENABLED = process.env['TERMIPOD_CODEX_DRIVER_E2E'] === '1'; +const HOME = os.homedir(); + +function codexPresent(): boolean { + const bin = codexBinary(HOME, process.env); + return path.isAbsolute(bin) && fs.existsSync(bin); +} + +function authPresent(): boolean { + return fs.existsSync(path.join(codexHome(HOME, process.env), 'auth.json')); +} + +/// The SHIPPED registry, not a fixture — the whole point is that the profile +/// the Companion translates codex frames with is the hub's own. +function codexFamily(): Family { + const artifact = fileURLToPath(new URL('../../resources/agent_families.generated.json', import.meta.url)); + const fam = familyByName(parseFamilies(fs.readFileSync(artifact, 'utf-8')), 'codex'); + assert.ok(fam !== undefined, 'the generated registry has no codex family'); + return fam; +} + +interface Live { + driver: CodexDriver; + events: DriverEvent[]; + /// Resolve when an event satisfying `pred` arrives, or reject on timeout. + wait: (pred: (ev: DriverEvent) => boolean, what: string, ms?: number) => Promise; +} + +function live(cwd: string, resumeThreadId?: string): Live { + const events: DriverEvent[] = []; + const waiters: Array<(ev: DriverEvent) => void> = []; + const driver = new CodexDriver({ + family: codexFamily(), + cwd, + posture: 'read_local', + env: process.env, + homeDir: HOME, + ...(resumeThreadId !== undefined ? { resumeThreadId } : {}), + onEvent: (ev) => { + events.push(ev); + for (const w of [...waiters]) w(ev); + }, + }); + return { + driver, + events, + wait: (pred, what, ms = 90_000) => + new Promise((resolve, reject) => { + const hit = events.find(pred); + if (hit !== undefined) { + resolve(hit); + return; + } + const timer = setTimeout(() => reject(new Error(`timed out waiting for ${what}`)), ms); + waiters.push((ev) => { + if (!pred(ev)) return; + clearTimeout(timer); + resolve(ev); + }); + }), + }; +} + +const skip = (): string | false => { + if (!ENABLED) return 'set TERMIPOD_CODEX_DRIVER_E2E=1 to run (spends tokens on a real turn)'; + if (!codexPresent()) return 'no codex binary on this host'; + if (!authPresent()) return 'codex is not authenticated on this host'; + return false; +}; + +test('a real turn round-trips through the shipped frame profile', async (t) => { + const why = skip(); + if (why !== false) { + t.skip(why); + return; + } + const cwd = fs.mkdtempSync(path.join(os.tmpdir(), 'termipod-l4c-')); + const l = live(cwd); + try { + l.driver.start(); + // `session.init` comes from the profile's `thread/started` rule — so this + // assertion covers the handshake AND the translation in one. + const init = await l.wait((e) => e.kind === 'session.init', 'session.init'); + const threadId = String(init.payload.session_id); + assert.notEqual(threadId, ''); + + // Long enough to outlive the 200 ms flush window. A short reply + // legitimately produces NO partial — item/completed cancels the pending + // flush, which is the designed behaviour, so asserting a partial on a + // one-token answer would be asserting a race. + l.driver.input('text', { + body: 'Write four sentences about why streaming output matters, then end with exactly: L4C-OK', + }); + const text = await l.wait( + (e) => e.kind === 'text' && e.payload.partial !== true && String(e.payload.text).includes('L4C-OK'), + 'the final text event', + ); + assert.match(String(text.payload.text), /L4C-OK/); + + // Deltas arrived and were throttled into partials, which is E3's whole + // point: a unit test can prove the buffer, only this can prove codex + // actually sends them. + assert.ok( + l.events.some((e) => e.kind === 'text' && e.payload.partial === true), + 'expected at least one streamed partial before the final', + ); + + // ── The measurement that matters most: resume finds the thread again ── + l.driver.stop(); + const again = live(cwd, threadId); + try { + again.driver.start(); + const reinit = await again.wait((e) => e.kind === 'session.init', 'session.init after resume'); + assert.equal(reinit.payload.session_id, threadId); + assert.equal(reinit.payload.resumed, true); + + again.driver.input('text', { body: 'What exact token did I ask you to reply with earlier? Answer with just that token.' }); + const memory = await again.wait( + (e) => e.kind === 'text' && e.payload.partial !== true && String(e.payload.text).includes('L4C-OK'), + 'the resumed turn to remember the token', + ); + // The engine remembers across a fresh app-server process — the same + // property claude's `--resume` has, reached by a completely different + // mechanism. + assert.match(String(memory.payload.text), /L4C-OK/); + } finally { + again.driver.stop(); + } + } finally { + l.driver.stop(); + fs.rmSync(cwd, { recursive: true, force: true }); + } +}); + +test('an image reaches the model in the variant this build accepts', async (t) => { + const why = skip(); + if (why !== false) { + t.skip(why); + return; + } + const cwd = fs.mkdtempSync(path.join(os.tmpdir(), 'termipod-l4c-img-')); + const l = live(cwd); + try { + l.driver.start(); + await l.wait((e) => e.kind === 'session.init', 'session.init'); + // A 1×1 png of one OPAQUE pure-green pixel (colour type 2, no alpha + // channel). Both halves of that are deliberate. Opaque, because the + // obvious sample to reach for — the widely-copied 1×1 "red dot" — is + // RGBA(255,0,0,127), and a half-transparent pixel gets described + // differently depending on what the reader composites it against: the + // same bytes drew "Light blue." from one turn and "red" from another. + // Green, because red and blue are what a model guesses when it cannot see + // the image at all, so an answer of "green" is evidence the bytes + // ARRIVED rather than evidence the model is agreeable. + l.driver.input('text', { + body: 'Name the single colour of this 1x1 image in one word.', + images: [ + { + mime: 'image/png', + data: 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAADElEQVR4nGNg+M8AAAICAQB7CYF4AAAAAElFTkSuQmCC', + }, + ], + }); + // Wait for a completed text that NAMES the colour, not merely the first + // completed text: a turn is free to open with a preamble message ("i'll + // inspect the image…") before the answer, and asserting on whichever text + // lands first races that choice — the same class as the two assertions + // this e2e already had to unlearn. If no text ever says green, the wait + // times out and fails with every collected text in hand. + const answered = l.wait( + (e) => e.kind === 'text' && e.payload.partial !== true && /green/.test(String(e.payload.text).toLowerCase()), + 'an answer naming the colour green', + ); + await answered.catch((err: unknown) => { + const texts = l.events + .filter((e) => e.kind === 'text' && e.payload.partial !== true) + .map((e) => JSON.stringify(e.payload.text)); + throw new Error(`${err instanceof Error ? err.message : String(err)} — completed texts: ${texts.join(' | ')}`); + }); + // If the block shape were wrong, `turn/start` would have failed with + // `-32600 unknown variant` and this would be an `error` row instead. + assert.equal( + l.events.some((e) => e.kind === 'error' && /unknown variant/.test(String(e.payload.text))), + false, + 'turn/start rejected the image block', + ); + } finally { + l.driver.stop(); + fs.rmSync(cwd, { recursive: true, force: true }); + } +}); diff --git a/desktop/electron/src/localagent/codexdriver.test.ts b/desktop/electron/src/localagent/codexdriver.test.ts new file mode 100644 index 00000000..d32a1d93 --- /dev/null +++ b/desktop/electron/src/localagent/codexdriver.test.ts @@ -0,0 +1,492 @@ +/// The codex app-server driver (vision-parity L4c). Run with `node --test`. +/// +/// Driven against a fake channel rather than a real app-server: the byte path +/// is L4b's and already has its own live e2e, so what is asserted here is the +/// PROTOCOL — which call opens a thread, what a director's click writes back, +/// and what the transcript says while all of it happens. +import { test } from 'node:test'; +import assert from 'node:assert/strict'; + +import { CodexDriver } from './codexdriver.ts'; +import type { CodexChannel, CodexFrame, CodexChannelHandlers } from './codexchannel.ts'; +import type { DriverEvent } from './driver.ts'; +import type { Family } from './families.ts'; + +const FAMILY: Family = { + family: 'codex', + bin: 'codex', + supports: ['M2'], + launch: { M2: { mode_args: ['app-server', '--listen', 'stdio://'] } }, + frame_profile: { + rules: [ + { + match: { method: 'thread/started' }, + emit: { + kind: 'session.init', + producer: 'agent', + payload: { session_id: '$.params.thread.id' }, + }, + }, + { + match: { method: 'item/completed', 'params.item.type': 'agentMessage' }, + emit: { + kind: 'text', + producer: 'agent', + payload: { text: '$.params.item.text', message_id: '$.params.item.id' }, + }, + }, + ], + }, +}; + +interface Rig { + driver: CodexDriver; + events: DriverEvent[]; + /// Frames the driver wrote to the channel. + sent: CodexFrame[]; + /// Push a frame at the driver, as the engine would. + recv: (frame: CodexFrame) => void; + /// Drop the channel, as a dead engine would. + hangup: (code: number | null, reason: string) => void; + closed: () => number; + exits: (number | null)[]; +} + +function rig(opts: { + resumeThreadId?: string; + posture?: 'converse' | 'read_local' | 'unrestricted'; + flushIntervalMs?: number; + mode?: 'daemon' | 'spawn'; + reason?: string; +} = {}): Rig { + const events: DriverEvent[] = []; + const sent: CodexFrame[] = []; + const exits: (number | null)[] = []; + let handlers: CodexChannelHandlers | null = null; + let closes = 0; + + const driver = new CodexDriver({ + family: FAMILY, + cwd: '/w', + env: { PATH: '/bin' }, + homeDir: '/home/u', + ...(opts.posture !== undefined ? { posture: opts.posture } : {}), + ...(opts.resumeThreadId !== undefined ? { resumeThreadId: opts.resumeThreadId } : {}), + ...(opts.flushIntervalMs !== undefined ? { flushIntervalMs: opts.flushIntervalMs } : {}), + plan: { mode: 'spawn', argv: ['codex', 'app-server'], reason: 'test rung' }, + openChannel: (plan, h) => { + handlers = h; + const channel: CodexChannel = { + mode: opts.mode ?? 'spawn', + reason: opts.reason ?? plan.reason, + send: (frame) => { + sent.push(frame); + }, + close: () => { + closes += 1; + }, + }; + return Promise.resolve(channel); + }, + onEvent: (ev) => events.push(ev), + onExit: (code) => exits.push(code), + }); + + return { + driver, + events, + sent, + recv: (frame) => handlers?.onFrame(frame), + hangup: (code, reason) => handlers?.onClose({ code, reason }), + closed: () => closes, + exits, + }; +} + +const tick = (): Promise => new Promise((r) => setImmediate(r)); +const after = (ms: number): Promise => new Promise((r) => setTimeout(r, ms)); + +/// Answer the handshake the way a real app-server does, and return the thread +/// id it reported. +async function handshake(r: Rig, threadId = 'th-1'): Promise { + r.driver.start(); + await tick(); + // initialize + const init = r.sent.find((f) => f.method === 'initialize'); + assert.ok(init !== undefined, 'expected an initialize call'); + r.recv({ jsonrpc: '2.0', id: init.id, result: { userAgent: 'codex/0.147.0' } }); + await tick(); + const open = r.sent.find((f) => f.method === 'thread/start' || f.method === 'thread/resume'); + assert.ok(open !== undefined, 'expected thread/start or thread/resume'); + r.recv({ + jsonrpc: '2.0', + id: open.id, + result: { thread: { id: threadId }, model: 'gpt-5.6', cwd: '/w' }, + }); + await tick(); + await tick(); + return threadId; +} + +const kinds = (events: DriverEvent[]): string[] => events.map((e) => e.kind); +const find = (events: DriverEvent[], kind: string): DriverEvent | undefined => + events.find((e) => e.kind === kind); + +// ── Opening ────────────────────────────────────────────────────────────────── + +test('the lifecycle row states the boundary, not just the posture name', async () => { + const r = rig({ posture: 'read_local' }); + await handshake(r); + const life = find(r.events, 'lifecycle'); + assert.ok(life !== undefined); + assert.equal(life.payload.tool_posture, 'read_local'); + // A reader should not have to know our mapping table to know whether this + // agent can write to their disk. + assert.equal(life.payload.sandbox, 'read-only'); + assert.equal(life.payload.approval_policy, 'never'); + assert.equal(life.payload.resumed, false); +}); + +test('converse records the difference it could not keep', async () => { + const r = rig({ posture: 'converse' }); + await handshake(r); + assert.match(String(find(r.events, 'lifecycle')?.payload.posture_note ?? ''), /no tool-disable switch/); +}); + +test('which rung opened is a transcript row, because the two make different promises', async () => { + const r = rig({ mode: 'daemon', reason: 'attached to the shared codex app-server daemon' }); + await handshake(r); + const row = r.events.find((e) => e.kind === 'system' && e.payload.kind === 'codex_channel'); + assert.ok(row !== undefined, 'expected a codex_channel row'); + assert.equal(row.payload.channel, 'daemon'); + assert.match(String(row.payload.reason), /shared codex app-server daemon/); +}); + +test('the handshake is initialize, then the initialized notification, then thread/start', async () => { + const r = rig(); + await handshake(r); + const methods = r.sent.map((f) => f.method); + assert.deepEqual(methods, ['initialize', 'initialized', 'thread/start']); + // The notification carries no id — a request would block waiting for a + // response app-server never sends. + assert.equal(r.sent[1].id, undefined); + assert.equal(r.driver.threadId, 'th-1'); +}); + +test('a resume calls thread/resume and mints the session.init the engine will not', async () => { + const r = rig({ resumeThreadId: 'th-old' }); + await handshake(r, 'th-old'); + const resume = r.sent.find((f) => f.method === 'thread/resume'); + assert.ok(resume !== undefined); + assert.equal((resume.params as Record).threadId, 'th-old'); + // Measured: thread/resume emits no `thread/started`, so the profile's + // session.init rule never fires and the row has to come from here — without + // it a reattached session has no init row at all. + const init = find(r.events, 'session.init'); + assert.ok(init !== undefined, 'expected a synthesized session.init on resume'); + assert.equal(init.payload.session_id, 'th-old'); + assert.equal(init.payload.resumed, true); + assert.equal(find(r.events, 'lifecycle')?.payload.resumed, true); +}); + +test('a fresh start does NOT synthesize session.init — the profile owns that row', async () => { + const r = rig(); + await handshake(r); + assert.equal(find(r.events, 'session.init'), undefined); + // It arrives when the engine says so, through the shared frame profile. + r.recv({ jsonrpc: '2.0', method: 'thread/started', params: { thread: { id: 'th-1' } } }); + assert.equal(find(r.events, 'session.init')?.payload.session_id, 'th-1'); +}); + +test('a failed handshake reports an error row and stops, rather than throwing into nowhere', async () => { + const r = rig(); + r.driver.start(); + await tick(); + const init = r.sent.find((f) => f.method === 'initialize'); + r.recv({ jsonrpc: '2.0', id: init?.id, error: { code: -32600, message: 'bad client' } }); + await tick(); + await tick(); + assert.match(String(find(r.events, 'error')?.payload.text ?? ''), /bad client/); + const stopped = r.events.filter((e) => e.kind === 'lifecycle' && e.payload.phase === 'stopped'); + assert.equal(stopped.length, 1); + assert.equal(stopped[0].payload.expected, false); + assert.equal(r.driver.running, false); +}); + +// ── Turns ──────────────────────────────────────────────────────────────────── + +test('text before the thread exists is QUEUED, not lost', async () => { + const r = rig(); + r.driver.start(); + await tick(); + r.driver.input('text', { body: 'early' }); + assert.equal(r.sent.some((f) => f.method === 'turn/start'), false); + + await handshake(r); + const turn = r.sent.find((f) => f.method === 'turn/start'); + assert.ok(turn !== undefined, 'the queued turn should be sent once the thread opens'); + assert.deepEqual((turn.params as Record).input, [{ type: 'text', text: 'early' }]); +}); + +test('a payload codex cannot carry throws at the call site, even while queued', async () => { + const r = rig(); + r.driver.start(); + await tick(); + // Built before the queue, so the caller learns immediately rather than + // having the failure surface later against a turn they think is running. + assert.throws( + () => r.driver.input('text', { body: 'x', pdfs: [{ mime: 'application/pdf', data: 'JVBER' }] }), + /no file attachments/, + ); +}); + +test('cancel interrupts the tracked turn with BOTH ids codex requires', async () => { + const r = rig(); + await handshake(r); + r.driver.input('text', { body: 'go' }); + await tick(); + const turn = r.sent.find((f) => f.method === 'turn/start'); + r.recv({ jsonrpc: '2.0', id: turn?.id, result: { turn: { id: 'tu-1' } } }); + await tick(); + + r.driver.input('cancel', { reason: 'stop' }); + await tick(); + const interrupt = r.sent.find((f) => f.method === 'turn/interrupt'); + assert.ok(interrupt !== undefined); + // Without either id codex answers -32600 "missing field". + assert.deepEqual(interrupt.params, { threadId: 'th-1', turnId: 'tu-1' }); +}); + +// ── Approvals ──────────────────────────────────────────────────────────────── + +test('an approval is parked as an R1 card and the click becomes the JSON-RPC response', async () => { + const r = rig(); + await handshake(r); + r.recv({ + jsonrpc: '2.0', + id: 7, + method: 'item/commandExecution/requestApproval', + params: { command: 'ls', availableDecisions: ['accept', 'decline'] }, + }); + + const card = find(r.events, 'approval_request'); + assert.ok(card !== undefined, 'expected an approval_request event'); + const requestId = String(card.payload.request_id); + assert.match(requestId, /^codex-[0-9a-f]{8}-7$/); + + r.driver.input('approval', { request_id: requestId, decision: 'accept' }); + const answer = r.sent.find((f) => f.id === 7); + assert.ok(answer !== undefined, 'expected a response on the parked id'); + assert.deepEqual(answer.result, { decision: 'accept' }); +}); + +test('request ids never collide across driver instances, so a stale card cannot answer a new request', async () => { + // The server's JSON-RPC id counter restarts with every connection, and a + // rebind is a NEW driver instance — so any per-instance counter in the + // request id collides exactly there: the card left over from the instance + // that died would answer whatever request happened to reach the same number + // on the next connection. That click can APPROVE a command the director + // never saw. The prefix must therefore differ per instance. + const ask = { + jsonrpc: '2.0', + id: 7, + method: 'item/commandExecution/requestApproval', + params: { command: 'ls' }, + }; + const a = rig(); + await handshake(a); + a.recv(ask); + const b = rig(); + await handshake(b); + b.recv(ask); + const idA = String(find(a.events, 'approval_request')?.payload.request_id); + const idB = String(find(b.events, 'approval_request')?.payload.request_id); + assert.notEqual(idA, idB, 'same jsonrpc id on two instances must mint two request ids'); + + // The stale click lands as unmatched — reported, not misdelivered. + b.driver.input('approval', { request_id: idA, decision: 'accept' }); + assert.equal(b.sent.some((f) => f.id === 7 && f.result !== undefined), false, 'the old card must not answer the new request'); + const unmatched = b.events.find((e) => e.kind === 'system' && e.payload.kind === 'codex_answer_unmatched'); + assert.ok(unmatched !== undefined, 'the stale click must be reported as unmatched'); +}); + +test('a question is answered with the option label, on the question-id map', async () => { + const r = rig(); + await handshake(r); + r.recv({ + jsonrpc: '2.0', + id: 9, + method: 'item/tool/requestUserInput', + params: { questions: [{ id: 'q1', question: 'which?', options: [{ label: 'staging' }] }] }, + }); + const card = find(r.events, 'approval_request'); + assert.equal(card?.payload.dialog_type, 'user_question'); + + r.driver.input('answer', { request_id: String(card?.payload.tool_use_id), body: 'staging' }); + const answer = r.sent.find((f) => f.id === 9); + assert.deepEqual(answer?.result, { answers: { q1: { answers: ['staging'] } } }); +}); + +test('a request we cannot present is refused AT ONCE, not parked forever', async () => { + const r = rig(); + await handshake(r); + r.recv({ + jsonrpc: '2.0', + id: 11, + method: 'mcpServer/elicitation/request', + params: { serverName: 'docs', mode: 'form', message: 'branch?', requestedSchema: { properties: { b: {} } } }, + }); + // No card — a card nobody can answer holds the engine open forever. + assert.equal(find(r.events, 'approval_request'), undefined); + const answer = r.sent.find((f) => f.id === 11); + assert.deepEqual(answer?.result, { action: 'decline', content: null, _meta: null }); + const row = r.events.find((e) => e.kind === 'system' && e.payload.kind === 'codex_request_refused'); + assert.match(String(row?.payload.reason ?? ''), /structured fields/); +}); + +test('cancel drains parked gates, so the next turn is not stuck behind one', async () => { + const r = rig(); + await handshake(r); + r.recv({ jsonrpc: '2.0', id: 5, method: 'item/fileChange/requestApproval', params: {} }); + assert.ok(find(r.events, 'approval_request') !== undefined); + + r.driver.input('cancel', { reason: 'stop' }); + await tick(); + // turn/interrupt aborts in-flight tool calls, but a parked JSON-RPC id stays + // open until we write a response. + assert.deepEqual(r.sent.find((f) => f.id === 5)?.result, { decision: 'decline' }); +}); + +test('stop() answers parked gates before closing the socket', async () => { + const r = rig({ mode: 'daemon' }); + await handshake(r); + r.recv({ jsonrpc: '2.0', id: 3, method: 'item/commandExecution/requestApproval', params: { command: 'ls' } }); + + r.driver.stop(); + assert.deepEqual(r.sent.find((f) => f.id === 3)?.result, { decision: 'decline' }); + assert.equal(r.closed(), 1); + assert.equal(r.driver.running, false); + const stopped = r.events.filter((e) => e.kind === 'lifecycle' && e.payload.phase === 'stopped'); + assert.equal(stopped[0].payload.expected, true); +}); + +test('answering a card from a previous connection is reported, not misrouted', async () => { + const r = rig(); + await handshake(r); + // `codex-0-7` names epoch 0; this connection is epoch 1. The id counter + // restarts with every connection, so without the epoch a stale card could + // answer a DIFFERENT request that reached the same number. + r.driver.input('approval', { request_id: 'codex-0-7', decision: 'accept' }); + const row = r.events.find((e) => e.kind === 'system' && e.payload.kind === 'codex_answer_unmatched'); + assert.ok(row !== undefined, 'expected an unmatched-answer row'); + assert.equal(r.sent.some((f) => f.result !== undefined), false); +}); + +// ── Streaming ──────────────────────────────────────────────────────────────── + +test('agentMessage deltas are throttled into one growing partial', async () => { + const r = rig({ flushIntervalMs: 5 }); + await handshake(r); + for (const delta of ['Hel', 'lo ', 'there']) { + r.recv({ jsonrpc: '2.0', method: 'item/agentMessage/delta', params: { itemId: 'msg-1', delta } }); + } + // A throttle, not a debounce: one flush covers all three. + await after(25); + const partials = r.events.filter((e) => e.kind === 'text' && e.payload.partial === true); + assert.equal(partials.length, 1); + assert.equal(partials[0].payload.text, 'Hello there'); + assert.equal(partials[0].payload.message_id, 'msg-1'); +}); + +test('command output streams as a tool_call_update with no status', async () => { + const r = rig({ flushIntervalMs: 5 }); + await handshake(r); + r.recv({ + jsonrpc: '2.0', + method: 'item/commandExecution/outputDelta', + params: { itemId: 'exec-1', delta: 'line one\n' }, + }); + await after(25); + const update = find(r.events, 'tool_call_update'); + assert.ok(update !== undefined); + // `toolCallId` is the key BOTH clients fold a running tool card on — not the + // `tool_use_id` a tool_result carries. + assert.equal(update.payload.toolCallId, 'exec-1'); + // No `status`: the latest update's status wins over the tool_result's, so a + // trailing "in_progress" would pin the card at running forever. + assert.equal('status' in update.payload, false); +}); + +test('item/completed cancels the pending flush so no partial lands after the final', async () => { + const r = rig({ flushIntervalMs: 20 }); + await handshake(r); + r.recv({ jsonrpc: '2.0', method: 'item/agentMessage/delta', params: { itemId: 'msg-1', delta: 'Hel' } }); + r.recv({ + jsonrpc: '2.0', + method: 'item/completed', + params: { item: { id: 'msg-1', type: 'agentMessage', text: 'Hello there' } }, + }); + await after(50); + const partials = r.events.filter((e) => e.kind === 'text' && e.payload.partial === true); + assert.equal(partials.length, 0, 'a stale partial would supersede the authoritative final'); + assert.equal(find(r.events, 'text')?.payload.text, 'Hello there'); +}); + +test('streaming can be turned off entirely', async () => { + const r = rig({ flushIntervalMs: -1 }); + await handshake(r); + r.recv({ jsonrpc: '2.0', method: 'item/agentMessage/delta', params: { itemId: 'msg-1', delta: 'x' } }); + await after(20); + assert.equal(r.events.some((e) => e.payload.partial === true), false); +}); + +test('reasoning deltas stay dropped — they are not in the vocabulary', async () => { + const r = rig({ flushIntervalMs: 5 }); + await handshake(r); + r.recv({ jsonrpc: '2.0', method: 'item/reasoning/textDelta', params: { itemId: 'rs-1', delta: 'hmm' } }); + await after(20); + assert.equal(r.events.some((e) => e.payload.partial === true), false); +}); + +// ── Losing the channel ─────────────────────────────────────────────────────── + +test('a hangup ends the session as UNexpected and fails anything in flight', async () => { + const r = rig(); + await handshake(r); + r.driver.input('text', { body: 'go' }); + await tick(); + + r.hangup(1, 'codex app-server exited (code 1)'); + await tick(); + const stopped = r.events.filter((e) => e.kind === 'lifecycle' && e.payload.phase === 'stopped'); + assert.equal(stopped.length, 1); + assert.equal(stopped[0].payload.expected, false); + assert.equal(stopped[0].payload.exit_code, 1); + // The turn that was waiting reports rather than hanging until the app quits. + assert.match(String(find(r.events, 'error')?.payload.text ?? ''), /exited \(code 1\)/); + assert.deepEqual(r.exits, [1]); +}); + +test('stop() after a hangup does not emit a second lifecycle row', async () => { + const r = rig(); + await handshake(r); + r.hangup(null, 'socket closed'); + await tick(); + r.driver.stop(); + const stopped = r.events.filter((e) => e.kind === 'lifecycle' && e.payload.phase === 'stopped'); + assert.equal(stopped.length, 1); +}); + +test('notifications translate through the SHARED frame profile', async () => { + const r = rig(); + await handshake(r); + r.recv({ + jsonrpc: '2.0', + method: 'item/completed', + params: { item: { id: 'msg-2', type: 'agentMessage', text: 'done' } }, + }); + // No second vocabulary: this is the hub's own codex profile, shipped in + // agent_families.generated.json. + assert.deepEqual(kinds(r.events).filter((k) => k === 'text'), ['text']); + assert.equal(find(r.events, 'text')?.payload.message_id, 'msg-2'); +}); diff --git a/desktop/electron/src/localagent/codexdriver.ts b/desktop/electron/src/localagent/codexdriver.ts new file mode 100644 index 00000000..f9f8a958 --- /dev/null +++ b/desktop/electron/src/localagent/codexdriver.ts @@ -0,0 +1,639 @@ +/// One codex session over the app-server protocol (vision-parity **L4c**). +/// +/// L4a chose the rung, L4b opened the byte path, and this is what finally +/// speaks: a JSON-RPC client that turns `turn/start` into a transcript and a +/// director's click into a JSON-RPC response. It is the codex counterpart of +/// `claudechild.ts`, and `service.ts` holds either behind `LocalDriver` without +/// knowing which. +/// +/// ## What is NOT here, deliberately +/// +/// - **A second frame vocabulary.** Notifications go through the L2 +/// interpreter with the hub's own `codex` frame profile, which ships to the +/// desktop in `agent_families.generated.json`. `thread/started` → +/// `session.init`, `item/completed` → `text` / `tool_result`, and so on are +/// *data*, already written and already drift-tested against the hub. +/// - **An attention table.** A local session has no hub, so a parked approval +/// surfaces as an `approval_request` event and is answered by R1's inline +/// cards — which is precisely what R1's own comment said would happen. +/// - **A written `.codex/config.toml`.** `thread/start` takes a `config` map, +/// so a session's overrides never become a file in the director's project. +/// +/// ## Two things the transcript has to be honest about +/// +/// 1. **Which rung opened.** "shared with your codex TUI" and "dies with this +/// window" are different promises, so the channel's own sentence is +/// recorded as a `system` row rather than being inferred from silence. +/// 2. **What the agent may actually do.** The lifecycle row carries the +/// posture AND the sandbox/approval policy it lowered to, because a name +/// is not a boundary — see `codexPosture` for the probe that measured it. +/// +/// ## Async start behind a synchronous door +/// +/// Opening the channel and running `initialize → thread/start` is a multi-step +/// async handshake, but `LocalDriver.start()` returns void: the service must be +/// able to hand back a session descriptor immediately. So input sent before the +/// thread exists is QUEUED and flushed when it opens, and a handshake that +/// fails reports an `error` row plus `lifecycle{phase:stopped}` — the same +/// place a failed claude launch already reports, rather than an exception +/// thrown somewhere that cannot say which session it was about. + +import { randomUUID } from 'node:crypto'; +import fs from 'node:fs'; +import { applyProfile } from '../frameprofile/translate.ts'; +import type { EmittedEvent } from '../frameprofile/types.ts'; +import { + codexBinary, + managedCodexPath, + codexHome, + planCodexAttach, + type CodexAttachPlan, +} from './codexattach.ts'; +import { + openCodexChannel, + type ChannelDeps, + type CodexChannel, + type CodexFrame, +} from './codexchannel.ts'; +import { + AGENT_MESSAGE_DELTA, + COMMAND_OUTPUT_DELTA, + CODEX_CLIENT_INFO, + MAX_STREAMED_OUTPUT_BYTES, + askEventPayload, + askResult, + buildTurnInput, + codexPosture, + isDeltaMethod, + parseServerAsk, + refuseResult, + threadResumeParams, + threadStartParams, + trimToTail, + type CodexAsk, +} from './codexwire.ts'; +import { + DEFAULT_TOOL_POSTURE, + type DriverEvent, + type InputKind, + type InputPayload, + type LocalDriver, + type ToolPosture, +} from './driver.ts'; +import type { Family } from './families.ts'; + +export interface CodexDriverOptions { + family: Family; + cwd: string; + posture?: ToolPosture; + model?: string; + /// Base environment. Supplied by the caller rather than read from the global + /// so a test can drive a hermetic one. + env: NodeJS.ProcessEnv; + homeDir: string; + /// An existing codex thread to reattach to. Absent for a fresh session. + /// + /// This is a THREAD id, not an argv token. N1's table says so itself: the + /// codex family's mechanism is `appserver_thread_resume`, and its note + /// records that the `codex resume ` CLI row is "the spawn-per-session + /// fallback rung … not what the hub drives today". So the plan's line about + /// taking resume argv from the N1 table does not describe this rung — the + /// table it points at is the one that says to use an RPC instead. + resumeThreadId?: string; + /// Per-thread codex config overrides, merged over `config.toml` by codex. + config?: Record; + /// Pre-resolved rung. Injected by tests; production computes it. + plan?: CodexAttachPlan; + channelDeps?: ChannelDeps; + /// Injected so a test can open a fake channel. + openChannel?: typeof openCodexChannel; + exists?: (p: string) => boolean; + now?: () => Date; + /// Throttle for streaming deltas. Negative disables streaming entirely. + flushIntervalMs?: number; + onEvent: (ev: DriverEvent) => void; + onExit?: (code: number | null) => void; +} + +const DEFAULT_FLUSH_MS = 200; + +interface Parked { + jsonrpcId: number | string; + ask: CodexAsk; +} + +interface StreamBuffer { + text: string; + timer: ReturnType | null; +} + +export class CodexDriver implements LocalDriver { + readonly #opts: CodexDriverOptions; + readonly #pending = new Map void>(); + readonly #parked = new Map(); + readonly #messageBuffers = new Map(); + readonly #outputBuffers = new Map(); + /// Text inputs that arrived before the thread was open. + readonly #queue: InputPayload[] = []; + #channel: CodexChannel | null = null; + #nextId = 0; + /// Distinguishes THIS driver instance's server requests from every other's — + /// the instance a rebind replaced, and one from a previous app run whose + /// unanswered card a restored transcript still shows. The server's JSON-RPC + /// id counter restarts with every connection, so a per-instance counter + /// would collide exactly where it matters: each rebind is a new instance + /// counting from the same start. Random, because transcripts outlive the + /// process; a stale card's click then lands as `codex_answer_unmatched` + /// instead of answering a request the director never saw. + readonly #instance = randomUUID().slice(0, 8); + #threadId = ''; + #turnId = ''; + #started = false; + #stopped = false; + #ready = false; + + constructor(opts: CodexDriverOptions) { + this.#opts = opts; + } + + get running(): boolean { + return this.#started && !this.#stopped; + } + + get posture(): ToolPosture { + return this.#opts.posture ?? DEFAULT_TOOL_POSTURE; + } + + /// The codex thread this session is bound to — the handle `thread/resume` + /// takes. Empty until the handshake completes. + get threadId(): string { + return this.#threadId; + } + + start(): void { + if (this.#started) return; + this.#started = true; + + const posture = codexPosture(this.posture); + this.#emit('lifecycle', 'system', { + phase: 'started', + mode: 'M2', + source: 'local', + engine: 'codex-app-server', + tool_posture: this.posture, + // What the posture actually LOWERED to. A reader should not have to know + // our mapping table to know whether this agent can write to their disk. + sandbox: posture.sandbox, + approval_policy: posture.approvalPolicy, + ...(posture.note !== undefined ? { posture_note: posture.note } : {}), + cwd: this.#opts.cwd, + resumed: this.#opts.resumeThreadId !== undefined && this.#opts.resumeThreadId !== '', + }); + + void this.#open().catch((err: unknown) => { + this.#fail(err instanceof Error ? err.message : String(err)); + }); + } + + input(kind: InputKind, payload: InputPayload): void { + if (!this.#started) throw new Error('local codex: session not started'); + if (this.#stopped) throw new Error('local codex: session has stopped'); + + switch (kind) { + case 'text': { + // Built before anything is queued or sent so a payload codex cannot + // carry (a PDF) throws at the call site instead of failing later, out + // of sight, against a turn the director thinks is running. + const input = buildTurnInput(payload); + if (!this.#ready) { + this.#queue.push(payload); + return; + } + void this.#startTurn(input); + return; + } + case 'approval': + case 'answer': { + const requestId = payload.request_id ?? ''; + const choice = kind === 'approval' ? payload.decision ?? '' : payload.body ?? ''; + if (requestId === '' || choice === '') { + throw new Error(`local codex: ${kind} missing request_id/${kind === 'approval' ? 'decision' : 'body'}`); + } + this.#answerParked(requestId, choice); + return; + } + case 'cancel': { + void this.#cancel(payload.reason ?? ''); + return; + } + } + } + + stop(): void { + if (!this.#started || this.#stopped) return; + this.#stopped = true; + this.#clearTimers(); + // Refuse anything still parked BEFORE the socket goes: an unanswered + // JSON-RPC request leaves the thread waiting on a client that no longer + // exists, and on the daemon rung that thread outlives us. + this.#refuseAllParked('the Companion closed this session'); + try { + this.#channel?.close(); + } catch { + // An already-closed channel is not a failure to report. + } + this.#channel = null; + this.#emit('lifecycle', 'system', { + phase: 'stopped', + mode: 'M2', + source: 'local', + engine: 'codex-app-server', + expected: true, + }); + this.#opts.onExit?.(null); + } + + // ── Opening ──────────────────────────────────────────────────────────────── + + async #open(): Promise { + const plan = this.#opts.plan ?? this.#planRung(); + const open = this.#opts.openChannel ?? openCodexChannel; + const channel = await open( + plan, + { + onFrame: (frame) => this.#onFrame(frame), + onClose: (info) => this.#onClose(info), + onJunk: (line) => this.#emit('error', 'system', { text: line, stream: 'stderr' }), + }, + { cwd: this.#opts.cwd, env: this.#opts.env }, + this.#opts.channelDeps, + ); + if (this.#stopped) { + channel.close(); + return; + } + this.#channel = channel; + // The rung is a fact about what this session IS, so it is a row, not a log + // line: on `daemon` the thread survives the app and shows up in the codex + // TUI; on `spawn` it dies with this window. + this.#emit('system', 'system', { + kind: 'codex_channel', + channel: channel.mode, + reason: channel.reason, + }); + await this.#handshake(); + } + + #planRung(): CodexAttachPlan { + const exists = this.#opts.exists ?? fs.existsSync; + const home = this.#opts.homeDir; + return planCodexAttach( + home, + { + managedInstallPresent: exists(managedCodexPath(codexHome(home, this.#opts.env))), + bin: codexBinary(home, this.#opts.env, exists), + }, + this.#opts.env, + ); + } + + async #handshake(): Promise { + await this.#call('initialize', { + clientInfo: { ...CODEX_CLIENT_INFO }, + capabilities: { experimentalApi: false }, + }); + // The protocol's one client notification. Send-and-forget: app-server + // treats it as confirmation and answers nothing. + this.#notify('initialized', null); + + const threadOpts = { + cwd: this.#opts.cwd, + posture: this.posture, + ...(this.#opts.model !== undefined ? { model: this.#opts.model } : {}), + ...(this.#opts.config !== undefined ? { config: this.#opts.config } : {}), + }; + const resume = this.#opts.resumeThreadId ?? ''; + const result = resume !== '' + ? await this.#call('thread/resume', threadResumeParams(resume, threadOpts)) + : await this.#call('thread/start', threadStartParams(threadOpts)); + + const thread = asRecord(result.thread); + const id = thread === undefined ? '' : asString(thread.id); + if (id === '') { + throw new Error('codex app-server returned no thread id'); + } + this.#threadId = id; + + if (resume !== '') { + // A fresh start gets its `session.init` from the profile's + // `thread/started` rule. A RESUME emits no such notification (measured: + // the only notifications after `thread/resume` are configWarning, + // remoteControl/status, thread/status, tokenUsage and goal/cleared), so + // without this a reattached session would have no init row and the + // service would have nothing to re-confirm its handle from. + this.#emit('session.init', 'agent', { + session_id: id, + engine: 'codex', + model: asString(result.model), + cwd: asString(result.cwd) || this.#opts.cwd, + resumed: true, + }); + } + + this.#ready = true; + const queued = this.#queue.splice(0, this.#queue.length); + for (const payload of queued) { + await this.#startTurn(buildTurnInput(payload)); + } + } + + #fail(message: string): void { + if (this.#stopped) return; + this.#stopped = true; + this.#clearTimers(); + this.#emit('error', 'system', { text: message, stream: 'handshake' }); + this.#emit('lifecycle', 'system', { + phase: 'stopped', + mode: 'M2', + source: 'local', + engine: 'codex-app-server', + expected: false, + }); + this.#opts.onExit?.(null); + } + + // ── Turns ────────────────────────────────────────────────────────────────── + + async #startTurn(input: Record[]): Promise { + try { + const result = await this.#call('turn/start', { threadId: this.#threadId, input }); + const turn = asRecord(result.turn); + if (turn !== undefined) this.#turnId = asString(turn.id); + } catch (err) { + this.#emit('error', 'system', { + text: err instanceof Error ? err.message : String(err), + stream: 'turn', + }); + } + } + + async #cancel(reason: string): Promise { + // Unblock anything parked first. `turn/interrupt` aborts in-flight tool + // calls, but a parked JSON-RPC id stays open until we write a response — + // so an interrupt with a gate still held leaves the next turn stuck behind + // it. + this.#refuseAllParked(reason !== '' ? reason : 'the director cancelled this turn'); + if (this.#threadId === '' || this.#turnId === '') return; + try { + // codex requires BOTH ids; without either it answers -32600 "missing + // field". + await this.#call('turn/interrupt', { threadId: this.#threadId, turnId: this.#turnId }); + } catch (err) { + this.#emit('error', 'system', { + text: err instanceof Error ? err.message : String(err), + stream: 'cancel', + }); + } + } + + // ── Parked requests ──────────────────────────────────────────────────────── + + #park(jsonrpcId: number | string, method: string, params: unknown): void { + const requestId = `codex-${this.#instance}-${String(jsonrpcId)}`; + const ask = parseServerAsk(method, params, requestId); + + if (ask.form === 'unsupported') { + // Answered immediately, never parked: a card the director cannot answer + // would hold the engine open forever, which is a stub wearing a card's + // clothes (D-4). + this.#respond(jsonrpcId, refuseResult(method)); + this.#emit('system', 'system', { + kind: 'codex_request_refused', + method, + reason: ask.note ?? 'unbridged', + summary: ask.summary, + }); + return; + } + + this.#parked.set(requestId, { jsonrpcId, ask }); + this.#emit('approval_request', 'agent', askEventPayload(ask)); + } + + #answerParked(requestId: string, choice: string): void { + const parked = this.#parked.get(requestId); + if (parked === undefined) { + // The connection restarted, or the request was already answered. Report + // it rather than throwing: the card is gone either way, and an exception + // here would surface on a surface that cannot say which session it was. + this.#emit('system', 'system', { + kind: 'codex_answer_unmatched', + request_id: requestId, + choice, + }); + return; + } + this.#parked.delete(requestId); + this.#respond(parked.jsonrpcId, askResult(parked.ask, choice)); + } + + #refuseAllParked(reason: string): void { + for (const [requestId, parked] of [...this.#parked]) { + this.#parked.delete(requestId); + this.#respond(parked.jsonrpcId, refuseResult(parked.ask.method)); + this.#emit('system', 'system', { + kind: 'codex_request_refused', + method: parked.ask.method, + reason, + }); + } + } + + // ── Frames ───────────────────────────────────────────────────────────────── + + #onFrame(frame: CodexFrame): void { + const id = frame.id; + const method = asString(frame.method); + + if (id !== undefined && method === '') { + const waiter = typeof id === 'number' ? this.#pending.get(id) : undefined; + if (waiter !== undefined) { + this.#pending.delete(id as number); + waiter(frame); + } + return; + } + if (id !== undefined && method !== '') { + this.#park(id as number | string, method, frame.params); + return; + } + if (method === '') return; + this.#onNotification(method, frame); + } + + #onNotification(method: string, frame: CodexFrame): void { + if (isDeltaMethod(method)) { + if (method === AGENT_MESSAGE_DELTA) this.#onDelta(this.#messageBuffers, frame, false); + else if (method === COMMAND_OUTPUT_DELTA) this.#onDelta(this.#outputBuffers, frame, true); + // Every other delta stays dropped: reasoning text and raw reasoning + // content are internal monologue the vocabulary does not surface. + return; + } + + const params = asRecord(frame.params); + if (method === 'item/completed' && params !== undefined) { + const item = asRecord(params.item); + if (item !== undefined) { + const itemId = asString(item.id); + const type = asString(item.type); + // The profile is about to post the authoritative row for this item, so + // stop streaming partials for it. Without this the last partial can + // land AFTER the final and pin the card at a stale value. + if (itemId !== '') { + if (type === 'agentMessage') this.#finalize(this.#messageBuffers, itemId); + else if (type === 'commandExecution') this.#finalize(this.#outputBuffers, itemId); + } + } + } + if (params !== undefined) { + if (method === 'turn/started') { + const turn = asRecord(params.turn); + if (turn !== undefined) this.#turnId = asString(turn.id) || this.#turnId; + } else if (method === 'turn/completed' || method === 'turn/failed') { + this.#turnId = ''; + } + } + + const events: EmittedEvent[] = applyProfile(frame, this.#opts.family.frame_profile); + for (const ev of events) this.#emit(ev.kind, ev.producer, ev.payload ?? {}); + } + + #onClose(info: { code: number | null; reason: string }): void { + if (this.#stopped) return; + this.#stopped = true; + this.#clearTimers(); + this.#channel = null; + // Fail everything still waiting: a caller blocked on a response that can + // no longer arrive would hang until the process ended. + for (const [id, waiter] of [...this.#pending]) { + this.#pending.delete(id); + waiter({ id, error: { code: -1, message: info.reason } }); + } + this.#emit('lifecycle', 'system', { + phase: 'stopped', + mode: 'M2', + source: 'local', + engine: 'codex-app-server', + exit_code: info.code, + reason: info.reason, + // We did not ask for this — `stop()` sets `#stopped` before closing, so + // reaching here means the engine or the socket went on its own. + expected: false, + }); + this.#opts.onExit?.(info.code); + } + + // ── Streaming deltas ─────────────────────────────────────────────────────── + + #onDelta(buffers: Map, frame: CodexFrame, bounded: boolean): void { + const flushMs = this.#opts.flushIntervalMs ?? DEFAULT_FLUSH_MS; + if (flushMs < 0) return; + const params = asRecord(frame.params); + if (params === undefined) return; + const itemId = asString(params.itemId); + const delta = asString(params.delta); + if (itemId === '' || delta === '') return; + + let buf = buffers.get(itemId); + if (buf === undefined) { + buf = { text: '', timer: null }; + buffers.set(itemId, buf); + } + buf.text = bounded ? trimToTail(buf.text + delta, MAX_STREAMED_OUTPUT_BYTES) : buf.text + delta; + // A THROTTLE, not a debounce: the timer is only armed when there is none, + // so a fast stream flushes every `flushMs` instead of never. + if (buf.timer === null) { + buf.timer = setTimeout(() => this.#flush(buffers, itemId, bounded), flushMs); + } + } + + #flush(buffers: Map, itemId: string, bounded: boolean): void { + const buf = buffers.get(itemId); + if (buf === undefined) return; + buf.timer = null; + if (buf.text === '') return; + if (bounded) { + // `toolCallId` (not `tool_use_id`) is the key BOTH clients fold a running + // tool card on, and the payload carries NO `status`: the latest update's + // status wins over the one derived from the paired tool_result, so a + // trailing "in_progress" would pin the card at running forever. + this.#emit('tool_call_update', 'agent', { + toolCallId: itemId, + content: [{ type: 'content', content: { type: 'text', text: buf.text } }], + partial: true, + }); + } else { + this.#emit('text', 'agent', { text: buf.text, message_id: itemId, partial: true }); + } + } + + #finalize(buffers: Map, itemId: string): void { + const buf = buffers.get(itemId); + if (buf === undefined) return; + if (buf.timer !== null) clearTimeout(buf.timer); + buffers.delete(itemId); + } + + #clearTimers(): void { + for (const buffers of [this.#messageBuffers, this.#outputBuffers]) { + for (const buf of buffers.values()) { + if (buf.timer !== null) clearTimeout(buf.timer); + } + buffers.clear(); + } + } + + // ── JSON-RPC plumbing ────────────────────────────────────────────────────── + + #call(method: string, params: unknown): Promise> { + const channel = this.#channel; + if (channel === null) return Promise.reject(new Error(`local codex: channel is closed (${method})`)); + const id = ++this.#nextId; + return new Promise>((resolve, reject) => { + this.#pending.set(id, (frame) => { + const err = asRecord(frame.error); + if (err !== undefined) { + reject(new Error(`codex ${method}: ${asString(err.message) || JSON.stringify(err)}`)); + return; + } + resolve(asRecord(frame.result) ?? {}); + }); + try { + channel.send({ jsonrpc: '2.0', id, method, params }); + } catch (err) { + this.#pending.delete(id); + reject(err instanceof Error ? err : new Error(String(err))); + } + }); + } + + #notify(method: string, params: unknown): void { + this.#channel?.send({ jsonrpc: '2.0', method, params }); + } + + #respond(id: number | string, result: Record): void { + this.#channel?.send({ jsonrpc: '2.0', id, result }); + } + + #emit(kind: string, producer: string, payload: Record): void { + this.#opts.onEvent({ kind, producer, payload }); + } +} + +function asRecord(v: unknown): Record | undefined { + return v !== null && typeof v === 'object' && !Array.isArray(v) ? (v as Record) : undefined; +} + +function asString(v: unknown): string { + return typeof v === 'string' ? v : ''; +} diff --git a/desktop/electron/src/localagent/codexwire.test.ts b/desktop/electron/src/localagent/codexwire.test.ts new file mode 100644 index 00000000..b75aabfe --- /dev/null +++ b/desktop/electron/src/localagent/codexwire.test.ts @@ -0,0 +1,320 @@ +/// codex app-server wire shapes (vision-parity L4c). Run with `node --test`. +/// +/// Every shape asserted here was checked against a LIVE codex-cli 0.147.0 — +/// either by sending it and reading the reply, or by reading the vendor's own +/// `codex app-server generate-ts` output. The comments name which. +import { test } from 'node:test'; +import assert from 'node:assert/strict'; + +import { + askEventPayload, + askResult, + buildTurnInput, + codexPosture, + isDeltaMethod, + parseServerAsk, + refuseResult, + threadResumeParams, + threadStartParams, + trimToTail, +} from './codexwire.ts'; + +// ── Posture ────────────────────────────────────────────────────────────────── + +test('read_local lowers to the sandbox that was MEASURED to refuse a write', () => { + const p = codexPosture('read_local'); + assert.equal(p.sandbox, 'read-only'); + // `never` and not `untrusted`: with `untrusted` codex asks to retry outside + // the sandbox when a command fails (observed: `reason: "command failed; + // retry without sandbox?"`), which would make the boundary clickable. + assert.equal(p.approvalPolicy, 'never'); + assert.equal(p.note, undefined); +}); + +test('unrestricted is the only posture that leaves the sandbox', () => { + assert.equal(codexPosture('unrestricted').sandbox, 'danger-full-access'); + for (const posture of ['converse', 'read_local'] as const) { + assert.equal(codexPosture(posture).sandbox, 'read-only'); + } +}); + +test('converse says out loud that codex cannot disable its tools', () => { + const p = codexPosture('converse'); + // The posture cannot be kept exactly, so the transcript carries the + // difference rather than the name alone (D-4). + assert.match(p.note ?? '', /no tool-disable switch/); + assert.match(p.note ?? '', /can still read files/); +}); + +// ── thread/start + thread/resume ───────────────────────────────────────────── + +test('thread/start carries cwd and the lowered posture, and no config file', () => { + const params = threadStartParams({ cwd: '/w', posture: 'read_local', model: 'gpt-5.6' }); + assert.deepEqual(params, { + cwd: '/w', + sandbox: 'read-only', + approvalPolicy: 'never', + model: 'gpt-5.6', + }); +}); + +test('config overrides ride the RPC instead of a written .codex/config.toml', () => { + const params = threadStartParams({ + cwd: '/w', + posture: 'read_local', + config: { model_reasoning_effort: 'low' }, + }); + assert.deepEqual(params.config, { model_reasoning_effort: 'low' }); + // An empty map is not sent at all — an empty override object would be a + // claim that overrides were configured. + assert.equal(threadStartParams({ cwd: '/w', posture: 'read_local', config: {} }).config, undefined); +}); + +test('resume re-sends the overrides, because a resumed thread takes what it is given', () => { + const params = threadResumeParams('th-1', { cwd: '/w', posture: 'read_local' }); + assert.equal(params.threadId, 'th-1'); + assert.equal(params.sandbox, 'read-only'); + assert.equal(params.approvalPolicy, 'never'); +}); + +// ── turn/start input ───────────────────────────────────────────────────────── + +test('an image rides the variant codex actually accepts', () => { + const input = buildTurnInput({ body: 'what is this', images: [{ mime: 'image/png', data: 'AAA' }] }); + // Measured: `{type:"input_image", image_url}` — the shape the hub's Go driver + // sends — is answered `-32600 unknown variant 'input_image', expected one of + // 'text', 'image', 'localImage', 'audio', 'localAudio', 'skill', 'mention'`. + assert.deepEqual(input, [ + { type: 'image', url: 'data:image/png;base64,AAA' }, + { type: 'text', text: 'what is this' }, + ]); +}); + +test('text carries no text_elements, which the server fills itself', () => { + const input = buildTurnInput({ body: 'hi' }); + assert.deepEqual(input, [{ type: 'text', text: 'hi' }]); + // The generated `UserInput` marks `text_elements` required; the live server + // accepts its absence and echoes back `text_elements: []`. Sending one would + // be inventing UI spans we do not have. + assert.equal('text_elements' in (input[0] as object), false); +}); + +test('a PDF throws rather than being dropped', () => { + // codex 0.147.0 has no file variant at all — `input_file` is rejected with + // the same "unknown variant" error. A silent drop would send the agent a + // question about a document it never received. + assert.throws( + () => buildTurnInput({ body: 'summarise', pdfs: [{ mime: 'application/pdf', data: 'JVBER' }] }), + /no file attachments/, + ); +}); + +test('an empty message is refused', () => { + assert.throws(() => buildTurnInput({}), /no body and no attachments/); +}); + +// ── Server-initiated requests ──────────────────────────────────────────────── + +const EXEC_APPROVAL = 'item/commandExecution/requestApproval'; + +test('a command approval offers exactly the decisions codex advertised', () => { + // Verbatim from the wire: codex advertised `cancel`, NOT `decline`, plus an + // object-shaped amendment. + const ask = parseServerAsk( + EXEC_APPROVAL, + { + command: "/bin/bash -lc 'echo hi'", + reason: 'command failed; retry without sandbox?', + availableDecisions: ['accept', { acceptWithExecpolicyAmendment: { execpolicy_amendment: ['echo'] } }, 'cancel'], + }, + 'req-1', + ); + assert.equal(ask.form, 'approval'); + assert.deepEqual(ask.options.map((o) => o.id), ['accept', 'cancel']); + assert.match(ask.summary, /Run: \/bin\/bash -lc 'echo hi'/); + assert.match(ask.summary, /retry without sandbox/); +}); + +test('an object-shaped decision is never offered as a button', () => { + const ask = parseServerAsk( + EXEC_APPROVAL, + { command: 'rm -rf /', availableDecisions: [{ applyNetworkPolicyAmendment: {} }, 'accept', 'decline'] }, + 'req-1', + ); + // Both amendment variants mint a STANDING policy from one click — the same + // class R1's `inlineDecidable` already refuses to put on an inline card. + assert.deepEqual(ask.options.map((o) => o.id), ['accept', 'decline']); +}); + +test('no advertised decisions still yields an answerable card', () => { + const ask = parseServerAsk('item/fileChange/requestApproval', { grantRoot: '/w' }, 'req-2'); + assert.equal(ask.form, 'approval'); + assert.deepEqual(ask.options.map((o) => o.id), ['accept', 'decline']); + assert.match(ask.summary, /under \/w/); +}); + +test('an MCP tool-call gate is an approval; a real form fill is refused', () => { + const gate = parseServerAsk( + 'mcpServer/elicitation/request', + { serverName: 'docs', message: 'Run search?', mode: 'form', requestedSchema: { properties: {} } }, + 'req-3', + ); + assert.equal(gate.form, 'approval'); + assert.deepEqual(gate.options.map((o) => o.id), ['accept', 'decline']); + + const form = parseServerAsk( + 'mcpServer/elicitation/request', + { + serverName: 'docs', + message: 'Which branch?', + mode: 'form', + requestedSchema: { properties: { branch: { type: 'string' } } }, + }, + 'req-4', + ); + // A card with no free-text input could never answer this, and a card that + // can never be answered parks the engine forever. + assert.equal(form.form, 'unsupported'); + assert.match(form.note ?? '', /structured fields/); +}); + +test('a url-mode elicitation is refused rather than silently opening a browser', () => { + const ask = parseServerAsk( + 'mcpServer/elicitation/request', + { serverName: 'auth', mode: 'url', message: 'Sign in', url: 'https://example.test' }, + 'req-5', + ); + assert.equal(ask.form, 'unsupported'); + assert.match(ask.note ?? '', /opening a browser/); +}); + +test('requestUserInput becomes a question card keyed on the question id', () => { + const ask = parseServerAsk( + 'item/tool/requestUserInput', + { + questions: [ + { + id: 'q1', + header: 'Deploy', + question: 'Which environment?', + isSecret: false, + options: [ + { label: 'staging', description: 'safe' }, + { label: 'prod', description: 'not safe' }, + ], + }, + ], + isBlocking: true, + }, + 'req-6', + ); + assert.equal(ask.form, 'question'); + assert.equal(ask.questionId, 'q1'); + assert.deepEqual(ask.options.map((o) => o.label), ['staging', 'prod']); +}); + +test('a SECRET question is never rendered as a card', () => { + const ask = parseServerAsk( + 'item/tool/requestUserInput', + { questions: [{ id: 'q1', question: 'API key?', isSecret: true, options: [{ label: 'x' }] }] }, + 'req-7', + ); + // `service.input()` records every input into the durable on-disk transcript + // BEFORE sending it, so answering here would write the secret to disk. + assert.equal(ask.form, 'unsupported'); + assert.match(ask.note ?? '', /transcript on disk/); +}); + +test('an option-less question is refused, because the card answers with an option', () => { + const ask = parseServerAsk( + 'item/tool/requestUserInput', + { questions: [{ id: 'q1', question: 'Describe the bug', options: [] }] }, + 'req-8', + ); + assert.equal(ask.form, 'unsupported'); +}); + +test('a permissions request grants nothing rather than guessing a profile', () => { + const ask = parseServerAsk('item/permissions/requestApproval', { reason: 'needs network' }, 'req-9'); + assert.equal(ask.form, 'unsupported'); + assert.deepEqual(refuseResult('item/permissions/requestApproval'), { permissions: {}, scope: 'turn' }); +}); + +// ── Answering ──────────────────────────────────────────────────────────────── + +test('each method is answered in ITS OWN response shape', () => { + const exec = parseServerAsk(EXEC_APPROVAL, {}, 'r'); + assert.deepEqual(askResult(exec, 'accept'), { decision: 'accept' }); + + const elicit = parseServerAsk('mcpServer/elicitation/request', { requestedSchema: { properties: {} } }, 'r'); + assert.deepEqual(askResult(elicit, 'accept'), { action: 'accept', content: {}, _meta: null }); + assert.deepEqual(askResult(elicit, 'decline'), { action: 'decline', content: null, _meta: null }); + + const question = parseServerAsk( + 'item/tool/requestUserInput', + { questions: [{ id: 'q1', question: 'which?', options: [{ label: 'a' }] }] }, + 'r', + ); + assert.deepEqual(askResult(question, 'a'), { answers: { q1: { answers: ['a'] } } }); +}); + +test('the refusal shape differs per method — an empty object fails every one', () => { + assert.deepEqual(refuseResult('mcpServer/elicitation/request'), { + action: 'decline', + content: null, + _meta: null, + }); + assert.deepEqual(refuseResult(EXEC_APPROVAL), { decision: 'decline' }); + assert.deepEqual(refuseResult('item/tool/requestUserInput'), { answers: {} }); + // The legacy v1 `ReviewDecision` refusal is an OBJECT carrying its reason. + assert.deepEqual(refuseResult('execCommandApproval'), { + decision: { denied: { rejection: 'the Companion cannot present this request' } }, + }); +}); + +// ── The R1 card shapes ─────────────────────────────────────────────────────── + +test('an approval renders through R1s existing permission card', () => { + const ask = parseServerAsk(EXEC_APPROVAL, { command: 'ls', availableDecisions: ['accept', 'decline'] }, 'req-1'); + const payload = askEventPayload(ask); + // The ACP-permission shape `parseApprovalRequest` reads: a `request_id`, a + // `params.toolCall` naming what is gated, and `params.options[].optionId`. + assert.equal(payload.request_id, 'req-1'); + const params = payload.params as Record; + assert.equal((params.toolCall as Record).name, EXEC_APPROVAL); + assert.deepEqual((params.options as Record[]).map((o) => o.optionId), ['accept', 'decline']); +}); + +test('a question renders through R1s existing question card', () => { + const ask = parseServerAsk( + 'item/tool/requestUserInput', + { questions: [{ id: 'q1', question: 'Which env?', options: [{ label: 'staging' }] }] }, + 'req-6', + ); + const payload = askEventPayload(ask); + // `parseApprovalRequest` discriminates on `dialog_type` and keys the reply on + // `tool_use_id`; the card answers with the option's LABEL. + assert.equal(payload.dialog_type, 'user_question'); + assert.equal(payload.tool_use_id, 'req-6'); + const q = (payload.questions as Record[])[0]; + assert.equal(q.question, 'Which env?'); + assert.deepEqual((q.options as Record[]).map((o) => o.label), ['staging']); +}); + +// ── Deltas ─────────────────────────────────────────────────────────────────── + +test('both delta spellings observed on the wire are recognised', () => { + assert.equal(isDeltaMethod('item/agentMessage/delta'), true); + assert.equal(isDeltaMethod('item/commandExecution/outputDelta'), true); + assert.equal(isDeltaMethod('item/agentReasoningRawContentDelta'), true); + assert.equal(isDeltaMethod('item/completed'), false); + assert.equal(isDeltaMethod(''), false); +}); + +test('trimToTail keeps the end and never leaves half a line', () => { + assert.equal(trimToTail('abc', 10), 'abc'); + // Cut lands mid-line, so it moves forward to the next boundary. + assert.equal(trimToTail('aaaa\nbbbb\ncccc', 9), 'cccc'); + // No newline in the kept window: the raw tail is the honest answer. + assert.equal(trimToTail('abcdefghij', 4), 'ghij'); +}); diff --git a/desktop/electron/src/localagent/codexwire.ts b/desktop/electron/src/localagent/codexwire.ts new file mode 100644 index 00000000..b4063c89 --- /dev/null +++ b/desktop/electron/src/localagent/codexwire.ts @@ -0,0 +1,551 @@ +/// The codex app-server wire, as pure functions (vision-parity **L4c**). +/// +/// Everything here is shape: JSON-RPC params in, JSON-RPC results out, plus the +/// translation of a server-initiated request into the `approval_request` event +/// R1's inline cards already render. No socket, no process, no clock — so every +/// claim below is asserted in `codexwire.test.ts` rather than eyeballed on a +/// machine with a display. `codexdriver.ts` is what wires these to a channel. +/// +/// **Measured against codex-cli 0.147.0, not read off documentation.** Three of +/// the shapes the hub's Go driver sends are rejected by this build, so the +/// probe log is the authority for what is here: +/// +/// - `{type:"input_image", image_url}` → `-32600 unknown variant +/// 'input_image', expected one of 'text', 'image', 'localImage', 'audio', +/// 'localAudio', 'skill', 'mention'`. The accepted image form is +/// `{type:"image", url:"data:;base64,"}` — probed with a 1×1 png +/// the model then described, so it is carried end to end, not just parsed. +/// - `{type:"input_file", file_data}` → the same error. **There is no PDF +/// variant at all**, which is why `buildTurnInput` refuses one instead of +/// dropping it silently. +/// - `{type:"text", text}` with no `text_elements` IS accepted; the server +/// fills `text_elements: []` itself. The generated type marks the field +/// required, so this is the one place the wire is more forgiving than the +/// schema, and sending an empty array anyway would be inventing UI spans. +/// +/// The vendor generates its own protocol — `codex app-server generate-ts --out +/// DIR` — and that generated union is what every shape here was checked against. + +import type { AttachmentInput, InputPayload, ToolPosture } from './driver.ts'; + +/// Who we say we are on `initialize`. `title` is what codex shows a user when +/// it names the client that opened a thread. +export const CODEX_CLIENT_INFO = { + name: 'termipod-companion', + title: 'TermiPod', + version: '0', +} as const; + +// ── Posture ────────────────────────────────────────────────────────────────── + +/// codex's sandbox mode for a posture, and the approval policy that goes with +/// it. Returned together because they are one decision: a sandbox with a policy +/// that lets the agent ask its way out of it is not the boundary its name +/// claims. +export interface CodexPosture { + /// `SandboxMode` — `read-only` | `workspace-write` | `danger-full-access`. + sandbox: string; + /// `AskForApproval` — `untrusted` | `on-request` | `never` (or a granular + /// object we do not use). + approvalPolicy: string; + /// Present when the posture cannot be kept exactly. Recorded on the + /// lifecycle event so the transcript states the real boundary rather than + /// the name we asked for (D-4). + note?: string; +} + +/// **The proof, not the documentation.** Against codex-cli 0.147.0, a thread +/// started with `{sandbox:'read-only', approvalPolicy:'never'}` was asked to +/// create a file in its cwd. It tried, failed, said *"the environment is +/// read-only, so POSTURE-PROBE.txt could not be created"*, and **no file +/// existed on disk afterwards**. The response echoed the lowered policy as +/// `{"type":"readOnly","networkAccess":false}`, so the network half is refused +/// by the same setting. +/// +/// This is a different mechanism from claude's, for the same contract: claude +/// enforces `read_local` by which tools exist (`--tools Read,Glob,Grep`), codex +/// by an OS sandbox. One posture, two proofs — and each driver owns its own. +/// +/// `approvalPolicy` is `never` on every posture on purpose. The alternative, +/// `untrusted`, makes codex ask to retry outside the sandbox when a command +/// fails (measured — it emits `item/commandExecution/requestApproval` with +/// `reason: "command failed; retry without sandbox?"`), which would turn +/// `read_local` into "read local until the director clicks accept". A posture +/// whose boundary can be clicked away is not the boundary claude's posture of +/// the same name is. +export function codexPosture(posture: ToolPosture): CodexPosture { + switch (posture) { + case 'unrestricted': + return { sandbox: 'danger-full-access', approvalPolicy: 'never' }; + case 'converse': + // codex has no tool-disable switch — nothing in `ThreadStartParams` + // corresponds to claude's `--tools ""`. The closest true statement is + // the read-only sandbox, so that is what it gets, and the difference is + // RECORDED rather than smoothed over: a director who picked "converse" + // is entitled to know the agent can still read their files. + return { + sandbox: 'read-only', + approvalPolicy: 'never', + note: 'codex has no tool-disable switch, so `converse` runs with the read-only sandbox: the agent can still read files in the workspace, but cannot write, execute outside it, or reach the network', + }; + case 'read_local': + default: + return { sandbox: 'read-only', approvalPolicy: 'never' }; + } +} + +// ── Handshake + turn params ────────────────────────────────────────────────── + +export interface ThreadOptions { + cwd: string; + posture: ToolPosture; + model?: string; + /// Per-thread config overrides, merged over `config.toml` by codex itself. + config?: Record; +} + +/// `thread/start` params. +/// +/// Note what is NOT here: a written `.codex/config.toml`. `ThreadStartParams` +/// carries a `config` map of the same overrides, applied to this thread only, +/// so seeding a session's configuration never means writing a file into the +/// director's project — a file we would then own the lifecycle of, and which +/// would outlive the session that wanted it. +export function threadStartParams(opts: ThreadOptions): Record { + const posture = codexPosture(opts.posture); + const params: Record = { + cwd: opts.cwd, + sandbox: posture.sandbox, + approvalPolicy: posture.approvalPolicy, + }; + if (opts.model !== undefined && opts.model !== '') params.model = opts.model; + if (opts.config !== undefined && Object.keys(opts.config).length > 0) params.config = opts.config; + return params; +} + +/// `thread/resume` params. The overrides are re-sent because a resumed thread +/// takes the ones it is given, not the ones it had — a resume that omitted the +/// sandbox would silently reopen the session under codex's own default. +export function threadResumeParams(threadId: string, opts: ThreadOptions): Record { + return { threadId, ...threadStartParams(opts) }; +} + +/// One `UserInput` block for an image attachment. +function imageInput(att: AttachmentInput): Record { + return { type: 'image', url: `data:${att.mime};base64,${att.data}` }; +} + +/// Build `turn/start.input` from a director's message. +/// +/// Attachments lead and text follows — the order a caption reads in, and the +/// order the hub's driver already builds. +/// +/// A PDF **throws** rather than being dropped. codex 0.147.0's `UserInput` +/// union has no file variant, so there is no shape that could carry one; a +/// silent drop would send the agent a question about a document it never +/// received, which is the failure mode that is worst to debug. The composer +/// should never offer it in the first place — the family registry's +/// `prompt_pdf.M2` is what gates that, and this is the backstop for an +/// out-of-band caller. +export function buildTurnInput(payload: InputPayload): Record[] { + const body = payload.body ?? ''; + const images = payload.images ?? []; + const pdfs = payload.pdfs ?? []; + if (pdfs.length > 0) { + throw new Error( + 'local codex: this build of codex accepts no file attachments — ' + + "turn/start rejects `input_file` with \"unknown variant\", and its UserInput union is text|image|localImage|audio|localAudio|skill|mention", + ); + } + if (body === '' && images.length === 0) { + throw new Error('local codex: text input has no body and no attachments'); + } + const input: Record[] = images.map(imageInput); + if (body !== '') input.push({ type: 'text', text: body }); + return input; +} + +// ── Server-initiated requests ──────────────────────────────────────────────── + +/// How a parked request can be answered by the Companion. +/// +/// - `approval` — a yes/no gate. Renders as R1's PermissionCard, answered by +/// `input.approval`. +/// - `question` — the agent asking the director something, with options. +/// Renders as R1's QuestionCard, answered by `input.answer`. +/// - `unsupported` — we cannot honestly present it, so it is refused +/// immediately with the shape codex expects and a `system` row saying why. +/// The alternative is a card that can never be answered, which parks the +/// engine forever: a stub, which D-4 forbids. +export type CodexAskForm = 'approval' | 'question' | 'unsupported'; + +export interface CodexAskOption { + id: string; + label: string; + description?: string; +} + +export interface CodexAsk { + method: string; + form: CodexAskForm; + /// What the transcript event carries as `request_id`, and what comes back on + /// `input.approval` / `input.answer`. + requestId: string; + summary: string; + options: CodexAskOption[]; + /// `question` form: the id the answers map must be keyed on. + questionId?: string; + /// `unsupported` form: the sentence the `system` row carries. + note?: string; +} + +const APPROVAL_METHODS = new Set([ + 'item/commandExecution/requestApproval', + 'item/fileChange/requestApproval', +]); + +/// Decisions we will offer, in the order a card should show them, mapped to the +/// label the button carries. +/// +/// **Object-shaped decisions are deliberately omitted.** +/// `CommandExecutionApprovalDecision` also admits +/// `{acceptWithExecpolicyAmendment}` and `{applyNetworkPolicyAmendment}` — both +/// mint a STANDING policy from a single click, which is the same class of +/// decision R1's `inlineDecidable` already refuses to put on an inline card +/// (browser_action's session grant). `acceptForSession` is offered only when +/// codex itself advertises it, and reads as what it is. +const DECISION_LABELS: Record = { + accept: 'accept', + acceptForSession: 'accept for session', + decline: 'decline', + cancel: 'cancel', +}; +const ACCEPT_ORDER = ['accept', 'acceptForSession']; +const REFUSE_ORDER = ['decline', 'cancel']; + +function str(v: unknown): string { + return typeof v === 'string' ? v : ''; +} + +function record(v: unknown): Record | undefined { + return v !== null && typeof v === 'object' && !Array.isArray(v) ? (v as Record) : undefined; +} + +/// The string variants of `availableDecisions`, if the server sent it. +/// +/// It is on the wire (probed: `["accept", {"acceptWithExecpolicyAmendment": +/// …}, "cancel"]`) but **not** in the generated `CommandExecutionRequest +/// ApprovalParams`, so it is read defensively: an absent field means "offer +/// the defaults", never "offer nothing". Notice that probe — the advertised +/// refusal there was `cancel`, not `decline`, which is exactly why the +/// refusal button is chosen from what was advertised rather than hard-coded. +function advertisedDecisions(params: Record): string[] | undefined { + const raw = params.availableDecisions; + if (!Array.isArray(raw)) return undefined; + const out = raw.filter((d): d is string => typeof d === 'string'); + return out.length > 0 ? out : undefined; +} + +function approvalOptions(params: Record): CodexAskOption[] { + const advertised = advertisedDecisions(params); + const pick = (order: string[]): CodexAskOption[] => { + const ids = advertised === undefined ? order.slice(0, 1) : order.filter((id) => advertised.includes(id)); + // Nothing advertised that we understand: fall back to the first known id + // rather than rendering a card with a missing half. + const chosen = ids.length > 0 ? ids : order.slice(0, 1); + return chosen.map((id) => ({ id, label: DECISION_LABELS[id] ?? id })); + }; + return [...pick(ACCEPT_ORDER), ...pick(REFUSE_ORDER)]; +} + +function approvalSummary(method: string, params: Record): string { + const reason = str(params.reason); + if (method === 'item/commandExecution/requestApproval') { + const cmd = str(params.command); + const head = cmd !== '' ? `Run: ${cmd}` : 'Run a command'; + return reason !== '' ? `${head} — ${reason}` : head; + } + const head = 'Apply a file change'; + const root = str(params.grantRoot); + const withRoot = root !== '' ? `${head} under ${root}` : head; + return reason !== '' ? `${withRoot} — ${reason}` : withRoot; +} + +/// Whether an elicitation is codex asking permission for an MCP tool call +/// rather than forwarding a real form fill. A permission gate collects no +/// input, so it has an empty schema; a form fill describes its fields. +function isToolCallGate(params: Record): boolean { + const meta = record(params._meta); + if (meta !== undefined && str(meta.codex_approval_kind) === 'mcp_tool_call') return true; + const schema = record(params.requestedSchema); + if (schema === undefined) return true; + const props = record(schema.properties); + return props === undefined || Object.keys(props).length === 0; +} + +function elicitationAsk(requestId: string, params: Record): CodexAsk { + const server = str(params.serverName) || str(params.server); + const message = str(params.message); + const mode = str(params.mode); + const base = { method: 'mcpServer/elicitation/request', requestId }; + + if (mode === 'url') { + return { + ...base, + form: 'unsupported', + summary: message !== '' ? message : `${server || 'An MCP server'} asked the user to open a URL`, + options: [], + note: `declined a url-mode elicitation from ${server || 'an MCP server'}: opening a browser on the director's behalf is a decision this driver does not make on its own`, + }; + } + if (!isToolCallGate(params)) { + return { + ...base, + form: 'unsupported', + summary: message !== '' ? message : `${server || 'An MCP server'} asked for input`, + options: [], + note: `declined a form elicitation from ${server || 'an MCP server'}: it wants structured fields, and the Companion's inline cards answer with a chosen option, not a filled form`, + }; + } + return { + ...base, + form: 'approval', + summary: message !== '' ? message : `${server || 'An MCP server'} wants to run a tool`, + options: [ + { id: 'accept', label: DECISION_LABELS.accept }, + { id: 'decline', label: DECISION_LABELS.decline }, + ], + }; +} + +function userInputAsk(requestId: string, params: Record): CodexAsk { + const questions = Array.isArray(params.questions) ? params.questions : []; + const first = record(questions[0]); + if (first === undefined) { + return { + method: 'item/tool/requestUserInput', + form: 'unsupported', + requestId, + summary: 'The agent asked a question with no content', + options: [], + note: 'declined a requestUserInput carrying no questions', + }; + } + // A secret answer must never take this path. `service.input()` RECORDS every + // input into the durable transcript before sending it, so answering a secret + // question through the card would write the secret to disk in plain text — + // a place the director never chose to put it and cannot easily unwrite. + if (first.isSecret === true) { + return { + method: 'item/tool/requestUserInput', + form: 'unsupported', + requestId, + summary: str(first.question) || 'The agent asked for a secret', + options: [], + note: 'declined a secret question: an answer sent through the Companion is written to the session transcript on disk, which is not where a secret belongs', + }; + } + const rawOptions = Array.isArray(first.options) ? first.options : []; + const options: CodexAskOption[] = []; + for (const o of rawOptions) { + const m = record(o); + if (m === undefined) continue; + const label = str(m.label); + if (label === '') continue; + const description = str(m.description); + options.push({ id: label, label, ...(description !== '' ? { description } : {}) }); + } + if (options.length === 0) { + return { + method: 'item/tool/requestUserInput', + form: 'unsupported', + requestId, + summary: str(first.question) || 'The agent asked an open question', + options: [], + note: "declined an open question: the Companion's question card answers with one of the offered options, and this one offered none", + }; + } + return { + method: 'item/tool/requestUserInput', + form: 'question', + requestId, + summary: str(first.question), + options, + questionId: str(first.id), + ...(questions.length > 1 + ? { note: `${questions.length - 1} further question(s) in this request are not shown` } + : {}), + }; +} + +/// Classify one server-initiated request. +/// +/// `requestId` is minted by the caller rather than taken from the JSON-RPC id: +/// the id counter restarts with every connection, so a card left on screen +/// across a rebind could otherwise answer a DIFFERENT request that happens to +/// have reached the same number. +export function parseServerAsk( + method: string, + rawParams: unknown, + requestId: string, +): CodexAsk { + const params = record(rawParams) ?? {}; + if (APPROVAL_METHODS.has(method)) { + return { + method, + form: 'approval', + requestId, + summary: approvalSummary(method, params), + options: approvalOptions(params), + }; + } + if (method === 'mcpServer/elicitation/request') return elicitationAsk(requestId, params); + if (method === 'item/tool/requestUserInput') return userInputAsk(requestId, params); + if (method === 'item/permissions/requestApproval') { + return { + method, + form: 'unsupported', + requestId, + summary: str(params.reason) || 'The agent asked for additional permissions', + options: [], + // The response shape is `{permissions, scope}` — a granted profile, not a + // verdict. Synthesizing one from a yes/no button would be inventing the + // grant's contents, so this refuses by granting nothing. + note: 'granted no additional permissions: this request asks for a permission profile, which is a decision with contents an approve button cannot express', + }; + } + return { + method, + form: 'unsupported', + requestId, + summary: method, + options: [], + note: `declined an unbridged server request (${method})`, + }; +} + +/// The JSON-RPC result that ANSWERS an ask. `choice` is an option id for the +/// `approval` form and the chosen option's label for `question`. +/// +/// Every shape here is the vendor's generated response type for that method; +/// sending the wrong one trips a deserialization error on the codex side that +/// surfaces to the agent as a flat rejection with no explanation. +export function askResult(ask: CodexAsk, choice: string): Record { + if (ask.form === 'question') { + // `ToolRequestUserInputResponse` = `{answers: {[questionId]: {answers: + // [string]}}}` — a map keyed by question id, each holding a LIST, because + // a multi-select question returns more than one. + return { answers: { [ask.questionId ?? '']: { answers: [choice] } } }; + } + if (ask.method === 'mcpServer/elicitation/request') { + const accepted = choice === 'accept' || choice === 'acceptForSession'; + return accepted + ? { action: 'accept', content: {}, _meta: null } + : { action: 'decline', content: null, _meta: null }; + } + return { decision: choice }; +} + +/// The result that REFUSES a request we cannot present, per method. +/// +/// Not one shape: `mcpServer/elicitation/request` wants `{action}`, the v2 +/// approvals want `{decision}`, `item/permissions/requestApproval` wants a +/// granted profile, and the two legacy v1 methods take a `ReviewDecision` +/// whose refusal is an object. An empty `{}` — the hub's fallback for anything +/// unknown — fails to deserialize on every one of them. +export function refuseResult(method: string): Record { + switch (method) { + case 'mcpServer/elicitation/request': + return { action: 'decline', content: null, _meta: null }; + case 'item/commandExecution/requestApproval': + case 'item/fileChange/requestApproval': + return { decision: 'decline' }; + case 'item/permissions/requestApproval': + // Granting an empty profile IS the refusal: the response has no + // decision field, so "no additional permissions" is the only way to say + // no. `turn` scope keeps even that scoped to the turn that asked. + return { permissions: {}, scope: 'turn' }; + case 'item/tool/requestUserInput': + return { answers: {} }; + case 'applyPatchApproval': + case 'execCommandApproval': + // The v1 `ReviewDecision` refusal carries the reason with it. + return { decision: { denied: { rejection: 'the Companion cannot present this request' } } }; + default: + return {}; + } +} + +/// The `approval_request` payload an ask becomes. +/// +/// This is R1's ACP-permission shape (`approvalRequest.ts` `parseApprovalRequest`) +/// for the approval form, and its claude AskUserQuestion shape for the question +/// form — deliberately, so codex approvals render through the cards that +/// already ship rather than through a third parser. R1's own comment predicted +/// this: *"A local driver parks nothing: its approvals arrive as +/// `approval_request` events and are answered by the two cards above (plan +/// L4)."* +export function askEventPayload(ask: CodexAsk): Record { + if (ask.form === 'question') { + return { + dialog_type: 'user_question', + tool_use_id: ask.requestId, + questions: [ + { + header: 'codex', + question: ask.summary, + options: ask.options.map((o) => ({ + label: o.label, + ...(o.description !== undefined ? { description: o.description } : {}), + })), + }, + ], + }; + } + return { + request_id: ask.requestId, + params: { + toolCall: { name: ask.method, title: ask.summary }, + options: ask.options.map((o) => ({ + optionId: o.id, + name: o.label, + ...(o.description !== undefined ? { description: o.description } : {}), + })), + }, + }; +} + +// ── Notifications ──────────────────────────────────────────────────────────── + +/// codex's streaming-delta notifications, which do NOT traverse the frame +/// profile: they arrive at 50–200× the rate of the completed item they build +/// up, so they are buffered per item and throttle-flushed instead +/// (vision-parity E3). Method spellings observed on the wire include both +/// `.../delta` and camelCase `...Delta`. +export function isDeltaMethod(method: string): boolean { + return method !== '' && (method.endsWith('/delta') || method.endsWith('Delta')); +} + +/// The two deltas that reach the transcript. The rest — reasoning text, raw +/// reasoning content — stay dropped: internal monologue the `agent_events` +/// vocabulary does not surface. +export const AGENT_MESSAGE_DELTA = 'item/agentMessage/delta'; +export const COMMAND_OUTPUT_DELTA = 'item/commandExecution/outputDelta'; + +/// Cap on the cumulative output one running command may carry. Each flush posts +/// the WHOLE buffer — that is what makes a partial self-contained for a client +/// that joins late — so an uncapped buffer would cost O(n²) bytes over a chatty +/// command's life. Past the cap the TAIL is kept: for a running command the +/// newest output is the interesting end, and the authoritative full text +/// arrives with the tool_result anyway. +export const MAX_STREAMED_OUTPUT_BYTES = 32 * 1024; + +/// Bound `s` to `max` characters, keeping the end, and cut forward to the next +/// line boundary when one is inside the kept window so no client renders half a +/// line as if it were whole. +export function trimToTail(s: string, max: number): string { + if (max <= 0 || s.length <= max) return s; + const tail = s.slice(s.length - max); + const nl = tail.indexOf('\n'); + return nl >= 0 && nl + 1 < tail.length ? tail.slice(nl + 1) : tail; +} diff --git a/desktop/electron/src/localagent/driver.ts b/desktop/electron/src/localagent/driver.ts new file mode 100644 index 00000000..f14162cd --- /dev/null +++ b/desktop/electron/src/localagent/driver.ts @@ -0,0 +1,98 @@ +/// What every local engine driver has in common (vision-parity L4c). +/// +/// L3a had one driver, so its vocabulary lived in `claudewire.ts` — the posture, +/// the input kinds, the event shape. L4c adds a second engine, and a type named +/// after claude that codex has to import is a name that lies to the next reader. +/// So the engine-neutral half moves here and the two wires keep only what is +/// genuinely theirs: `claudewire.ts` builds stream-json frames and `--tools` +/// argv, `codexwire.ts` builds JSON-RPC params. +/// +/// Nothing here spawns, connects, or imports `electron` — `service.ts` owns one +/// of these per session and cannot tell which engine is behind it. + +// ── Tool posture ───────────────────────────────────────────────────────────── + +/// How much of the director's machine a local session may touch. +/// +/// **The name is the promise; each driver has to MEASURE that it keeps it.** +/// The two engines enforce it by completely different means — claude by which +/// tools exist (`--tools`), codex by an OS sandbox (`sandbox` + +/// `approvalPolicy`) — so the same posture is one contract with two proofs. +/// Both proofs are in the wire modules beside the mapping they justify, and +/// both were run against a live engine rather than read off documentation. +export type ToolPosture = 'converse' | 'read_local' | 'unrestricted'; + +/// The default. Reading the workdir is what makes a co-working Companion +/// useful; writing, executing and reaching the network are what make an +/// unattended one dangerous, and none of the three are here. +export const DEFAULT_TOOL_POSTURE: ToolPosture = 'read_local'; + +export function isToolPosture(v: unknown): v is ToolPosture { + return v === 'converse' || v === 'read_local' || v === 'unrestricted'; +} + +// ── Input ──────────────────────────────────────────────────────────────────── + +/// A binary attachment, already base64 and without a `data:` prefix. +export interface AttachmentInput { + mime: string; + data: string; + filename?: string; +} + +export interface InputPayload { + body?: string; + images?: AttachmentInput[]; + pdfs?: AttachmentInput[]; + request_id?: string; + decision?: string; + note?: string; + reason?: string; +} + +/// The input kinds a local driver accepts. A deliberate subset of the hub +/// driver's: `attention_reply` and `attach` are hub concepts (an attention +/// table, a document entity) that a local session has none of, so they are +/// absent rather than stubbed (D-4). +export type InputKind = 'text' | 'approval' | 'answer' | 'cancel'; + +// ── Output ─────────────────────────────────────────────────────────────────── + +/// An event on its way to the session log — kind/producer/payload, before the +/// log assigns it a `seq`. +export interface DriverEvent { + kind: string; + producer: string; + payload: Record; +} + +// ── The driver itself ──────────────────────────────────────────────────────── + +/// One engine session, as `service.ts` uses it. +/// +/// `start()` returns void on both engines even though codex's opening move is a +/// multi-step async handshake. That is deliberate: the service's job is to have +/// a session that exists immediately and a transcript that explains itself, and +/// an async `start` would either make `create()` async (changing the IPC +/// contract for every caller) or leave a window where a session exists with no +/// driver. The codex driver therefore queues input until its thread is open and +/// reports a failed handshake as an `error` row — the same place a failed +/// claude launch already reports. +export interface LocalDriver { + /// False once the engine is gone, whether we stopped it or it died. + readonly running: boolean; + /// The posture this session actually launched with, recorded on the + /// lifecycle event so a transcript states what the agent was allowed to do + /// rather than leaving a reader to infer it. + readonly posture: ToolPosture; + start(): void; + input(kind: InputKind, payload: InputPayload): void; + stop(): void; +} + +/// What a driver needs from the session that owns it. +export interface DriverHooks { + onEvent: (ev: DriverEvent) => void; + /// The engine is gone. `service.ts` marks the session stopped and flushes. + onExit?: (code: number | null) => void; +} diff --git a/desktop/electron/src/localagent/families.ts b/desktop/electron/src/localagent/families.ts index 7f9a8656..338734f3 100644 --- a/desktop/electron/src/localagent/families.ts +++ b/desktop/electron/src/localagent/families.ts @@ -74,15 +74,34 @@ export function familyByName(families: readonly Family[], name: string): Family return families.find((f) => f.family === name); } -/// Whether a family can be driven by the local service. +/// The engines the local service has a driver for. +/// +/// A closed union rather than a string: every branch that builds a driver has +/// to handle each member, so adding an engine is a compile error at each site +/// instead of a silent fallthrough to claude's. +export type LocalEngine = 'claude-code' | 'codex'; + +/// Which local driver a family gets, or null when it gets none. +/// +/// A family qualifies when we have a driver for it AND it declares the mode +/// that driver speaks — the second half matters because the mode set is data, +/// and a family that drops M2 should stop being offered without anyone editing +/// this file. /// -/// L3a ships the claude-code driver only; L4 adds codex through the vendor's -/// app-server. A family is "supported locally" when we have a driver for it AND -/// it declares the mode that driver speaks — the second half matters because -/// the mode set is data, and a family that drops M2 should stop being offered -/// without anyone editing this file. +/// Both drivers are M2, and both read `launch.M2.mode_args` from the same +/// registry, but they use it differently: claude's IS its argv, while codex's +/// (`app-server --listen stdio://`) states the launch contract the desktop's +/// spawn rung honours through `codexattach.ts`, which additionally has to +/// FIND the binary (the installer's PATH line lives in `.bashrc`, which a +/// GUI-launched app never sources). +export function localEngine(fam: Family | undefined): LocalEngine | null { + if (fam === undefined) return null; + if (fam.family !== 'claude-code' && fam.family !== 'codex') return null; + const usable = (fam.supports ?? []).includes('M2') && (fam.launch?.M2?.mode_args ?? []).length > 0; + return usable ? fam.family : null; +} + +/// Whether a family can be driven by the local service. export function supportsLocalDriving(fam: Family | undefined): boolean { - if (fam === undefined) return false; - if (fam.family !== 'claude-code') return false; - return (fam.supports ?? []).includes('M2') && (fam.launch?.M2?.mode_args ?? []).length > 0; + return localEngine(fam) !== null; } diff --git a/desktop/electron/src/localagent/hostargs.ts b/desktop/electron/src/localagent/hostargs.ts index 22c8bf9b..e24403d6 100644 --- a/desktop/electron/src/localagent/hostargs.ts +++ b/desktop/electron/src/localagent/hostargs.ts @@ -5,7 +5,7 @@ /// rest of `electron/src` uses. Everything here arrives from the renderer as /// `unknown`, so this is the boundary that gives it a type. -import { isToolPosture, type InputKind, type InputPayload, type ToolPosture } from './claudewire.ts'; +import { isToolPosture, type InputKind, type InputPayload, type ToolPosture } from './driver.ts'; export function requireString(args: Record, key: string): string { const v = args[key]; diff --git a/desktop/electron/src/localagent/service.test.ts b/desktop/electron/src/localagent/service.test.ts index 3d571fa8..8188d5a1 100644 --- a/desktop/electron/src/localagent/service.test.ts +++ b/desktop/electron/src/localagent/service.test.ts @@ -8,7 +8,8 @@ import { mkdtempSync, rmSync, existsSync } from 'node:fs'; import { tmpdir } from 'node:os'; import path from 'node:path'; -import { LocalAgentService } from './service.ts'; +import { LocalAgentService, type ServiceOptions } from './service.ts'; +import type { CodexChannelHandlers, CodexFrame } from './codexchannel.ts'; import type { SpawnedChild } from './claudechild.ts'; import type { Family } from './families.ts'; import type { LocalAgentEvent } from './log.ts'; @@ -49,6 +50,22 @@ const CLAUDE: Family = { }; // Declares M2 but ships no launch contract — offered by the hub, undrivable here. const CODEX: Family = { family: 'codex', bin: 'codex', supports: ['M2'] }; +/// The drivable codex row (vision-parity L4c), with the launch contract the +/// real registry carries. +const CODEX_LOCAL: Family = { + family: 'codex', + bin: 'codex', + supports: ['M2'], + launch: { M2: { mode_args: ['app-server', '--listen', 'stdio://'] } }, + frame_profile: { + rules: [ + { + match: { method: 'thread/started' }, + emit: { kind: 'session.init', producer: 'agent', payload: { session_id: '$.params.thread.id' } }, + }, + ], + }, +}; const settle = (): Promise => new Promise((r) => setImmediate(r)); @@ -94,7 +111,11 @@ process.on('exit', () => { for (const d of tempDirs) rmSync(d, { recursive: true, force: true }); }); -function service(families: Family[] = [CLAUDE, CODEX], dataDir?: string): Harness { +function service( + families: Family[] = [CLAUDE, CODEX], + dataDir?: string, + codex?: ServiceOptions['codex'], +): Harness { const dir = dataDir ?? mkdtempSync(path.join(tmpdir(), 'termipod-l3b-svc-')); if (dataDir === undefined) tempDirs.push(dir); const children: FakeChild[] = []; @@ -111,16 +132,65 @@ function service(families: Family[] = [CLAUDE, CODEX], dataDir?: string): Harnes children.push(c); return c as unknown as SpawnedChild; }, + ...(codex !== undefined ? { codex } : {}), }); return { svc, children, dataDir: dir, - restart: () => service(families, dir), + restart: () => service(families, dir, codex), cleanup: () => rmSync(dir, { recursive: true, force: true }), }; } +/// A codex service whose channel is a fake: the byte path is L4b's and has its +/// own live e2e, so what these tests exercise is the SERVICE's half — which +/// driver a family gets, where the handle comes from, and how a rebind carries +/// it. +interface CodexHarness extends Harness { + sent: CodexFrame[]; + recv: (frame: CodexFrame) => void; + restart: () => CodexHarness; +} + +function codexService(dataDir?: string): CodexHarness { + const sent: CodexFrame[] = []; + let handlers: CodexChannelHandlers | null = null; + const h = service([CLAUDE, CODEX_LOCAL], dataDir, { + plan: { mode: 'spawn', argv: ['codex', 'app-server'], reason: 'test rung' }, + openChannel: (plan, hs) => { + handlers = hs; + return Promise.resolve({ + mode: 'spawn' as const, + reason: plan.reason, + send: (frame: CodexFrame) => sent.push(frame), + close: () => undefined, + }); + }, + }); + return { + ...h, + sent, + recv: (frame) => handlers?.onFrame(frame), + restart: () => codexService(h.dataDir), + }; +} + +/// Answer the codex handshake and report the thread id, as a real app-server +/// would. +async function codexHandshake(h: CodexHarness, threadId: string): Promise { + await settle(); + const init = h.sent.find((f) => f.method === 'initialize'); + assert.ok(init !== undefined, 'expected an initialize call'); + h.recv({ jsonrpc: '2.0', id: init.id, result: {} }); + await settle(); + const open = h.sent.find((f) => f.method === 'thread/start' || f.method === 'thread/resume'); + assert.ok(open !== undefined, 'expected thread/start or thread/resume'); + h.recv({ jsonrpc: '2.0', id: open.id, result: { thread: { id: threadId } } }); + await settle(); + await settle(); +} + /// Capture the argv a spawn was called with, which is how the resume flags are /// observed — the FakeChild above is deliberately ignorant of them. function spawnRecorder(): { calls: Array<{ bin: string; args: string[] }>; children: FakeChild[]; fn: (bin: string, args: string[]) => SpawnedChild } { @@ -516,18 +586,123 @@ test('a session with no engine handle refuses to rebind rather than cold-startin } }); -test('a family that does not resume by argv refuses to rebind', () => { - const h = service([CLAUDE, { ...CODEX, launch: { M2: { mode_args: ['--x'] } } }]); +test('an argv-resume family whose recipe went missing refuses to rebind', () => { + // The guard that used to catch codex here. codex now rebinds by RPC (see the + // L4c tests below), so the only way to reach `not_argv_resume` is a family + // whose driver splices argv while its N1 row says it does not — a generated + // table drifting from the driver, which is exactly the case worth refusing + // rather than silently cold-starting. + const h = service([CLAUDE], undefined); try { const desc = h.svc.create({ cwd: '/w' }); h.svc.stop(desc.id); const meta = readSessionMeta(h.dataDir, desc.id); assert.ok(meta !== null); - writeSessionMeta(h.dataDir, { ...meta, family: 'codex' }); + writeSessionMeta(h.dataDir, { ...meta, family: 'ghost' }); const next = h.restart(); next.svc.reload(); - assert.throws(() => next.svc.rebind(desc.id), /not_argv_resume/); + // No driver at all for `ghost`, so the refusal lands one step earlier — + // which is the honest place for it. + assert.throws(() => next.svc.rebind(desc.id), /cannot drive family ghost/); + } finally { + h.cleanup(); + } +}); + +// ── codex (vision-parity L4c) ──────────────────────────────────────────────── + +test('a codex session gets NO pre-assigned handle, and learns one from the engine', async () => { + const h = codexService(); + try { + const desc = h.svc.create({ family: 'codex', cwd: '/w' }); + // claude honours `--session-id`, so its handle is assigned at create time. + // codex mints its own thread id and takes no such flag — writing a UUID + // here would record a thread id codex has never heard of, and the later + // `thread/resume` would fail against it. + assert.equal(desc.engine_session_id, undefined); + + await codexHandshake(h, 'th-42'); + h.recv({ jsonrpc: '2.0', method: 'thread/started', params: { thread: { id: 'th-42' } } }); + await settle(); + assert.equal(h.svc.get(desc.id)?.engine_session_id, 'th-42'); + // And it survives to disk, which is what makes the rebind below possible. + assert.equal(readSessionMeta(h.dataDir, desc.id)?.engine_session_id, 'th-42'); + } finally { + h.cleanup(); + } +}); + +test('a codex rebind is thread/resume, not argv — which is what N1s own row says', async () => { + const h = codexService(); + try { + const desc = h.svc.create({ family: 'codex', cwd: '/w' }); + await codexHandshake(h, 'th-42'); + h.recv({ jsonrpc: '2.0', method: 'thread/started', params: { thread: { id: 'th-42' } } }); + await settle(); + h.svc.stop(desc.id); + + const next = h.restart(); + next.svc.reload(); + next.svc.rebind(desc.id); + await settle(); + const init = next.sent.find((f) => f.method === 'initialize'); + assert.ok(init !== undefined); + next.recv({ jsonrpc: '2.0', id: init.id, result: {} }); + await settle(); + + const resume = next.sent.find((f) => f.method === 'thread/resume'); + assert.ok(resume !== undefined, 'expected a thread/resume, not a resume argv'); + assert.equal((resume.params as Record).threadId, 'th-42'); + // The table says so itself: the codex family's mechanism is + // `appserver_thread_resume`, and its `codex resume ` CLI row is a + // different rung. + assert.equal(next.sent.some((f) => f.method === 'thread/start'), false); + next.cleanup(); + } finally { + h.cleanup(); + } +}); + +test('a codex session with no learned handle refuses to rebind', async () => { + const h = codexService(); + try { + const desc = h.svc.create({ family: 'codex', cwd: '/w' }); + // Never handshook, so no thread id was ever reported. + h.svc.stop(desc.id); + const next = h.restart(); + next.svc.reload(); + assert.throws(() => next.svc.rebind(desc.id), /no engine session id/); + next.cleanup(); + } finally { + h.cleanup(); + } +}); + +test('a codex input is recorded in the transcript and reaches the engine', async () => { + const h = codexService(); + try { + const desc = h.svc.create({ family: 'codex', cwd: '/w' }); + await codexHandshake(h, 'th-42'); + h.svc.input(desc.id, 'text', { body: 'hello codex' }); + await settle(); + + const turn = h.sent.find((f) => f.method === 'turn/start'); + assert.ok(turn !== undefined); + assert.deepEqual((turn.params as Record).input, [{ type: 'text', text: 'hello codex' }]); + // The director's own turn is in the log — otherwise the transcript is the + // agent's half of a conversation (the L3b defect, one engine over). + const page = h.svc.history(desc.id); + assert.ok(page.events.some((e) => e.kind === 'input.text' && e.payload?.body === 'hello codex')); + } finally { + h.cleanup(); + } +}); + +test('both engines are offered when both declare a launch contract', () => { + const h = codexService(); + try { + assert.deepEqual(h.svc.localFamilies().map((f) => f.family), ['claude-code', 'codex']); } finally { h.cleanup(); } diff --git a/desktop/electron/src/localagent/service.ts b/desktop/electron/src/localagent/service.ts index e0311992..4ef3804f 100644 --- a/desktop/electron/src/localagent/service.ts +++ b/desktop/electron/src/localagent/service.ts @@ -29,10 +29,21 @@ /// into IPC handlers, and `service.test.ts` drives it directly. import { randomUUID } from 'node:crypto'; -import { ClaudeChild, type DriverEvent, type SpawnFn } from './claudechild.ts'; -import { DEFAULT_TOOL_POSTURE, resolveConfigHome, type InputKind, type InputPayload, type ToolPosture } from './claudewire.ts'; +import { ClaudeChild, type SpawnFn } from './claudechild.ts'; +import { resolveConfigHome } from './claudewire.ts'; +import type { CodexAttachPlan } from './codexattach.ts'; +import type { ChannelDeps, openCodexChannel } from './codexchannel.ts'; +import { CodexDriver } from './codexdriver.ts'; +import { + DEFAULT_TOOL_POSTURE, + type DriverEvent, + type InputKind, + type InputPayload, + type LocalDriver, + type ToolPosture, +} from './driver.ts'; import { DurableSessionLog } from './durablelog.ts'; -import { familyByName, supportsLocalDriving, type Family } from './families.ts'; +import { familyByName, localEngine, supportsLocalDriving, type Family } from './families.ts'; import { type LocalAgentEvent, type LogPage } from './log.ts'; import { newSessionID, resumeSplice, type ResumeTable } from './resumerecipes.ts'; import { @@ -52,12 +63,20 @@ export interface SessionDescriptor { model?: string; status: 'running' | 'stopped'; created_at: string; - /// The engine's own session id — the handle `--resume` takes. + /// The engine's own session id — the handle a reattach is keyed on. + /// + /// **How it arrives differs by engine, and that is a measured difference, + /// not a style choice.** claude honours `--session-id `, so since L3b + /// its handle is ASSIGNED at create time and exists from the moment the + /// session does — closing the window where a child that died before its + /// first frame left an unreattachable transcript. codex mints its own thread + /// id and takes no equivalent flag, so a codex session has no handle until + /// `thread/started` reports one. Assigning a UUID anyway would write down a + /// thread id codex has never heard of, and a later `thread/resume` would + /// fail against it. /// - /// Since L3b this is ASSIGNED at create time rather than learned from the - /// init frame, so it is present from the moment the session exists. The - /// engine's own report still wins if the two ever disagree: it is the one - /// that decides what `--resume` will find. + /// The engine's own report wins on both paths: it is what a reattach will + /// look up. engine_session_id?: string; /// True for a session read back off disk that has not been rebound yet. The /// transcript is readable; the engine is not attached. @@ -88,6 +107,18 @@ export interface ServiceOptions { spawnFn?: SpawnFn; now?: () => Date; logCapacity?: number; + /// Overrides for the codex rung, injected so a test can drive a codex + /// session with no app-server on the box. Production leaves them absent and + /// the driver resolves the rung itself. + codex?: { + plan?: CodexAttachPlan; + deps?: ChannelDeps; + openChannel?: typeof openCodexChannel; + flushIntervalMs?: number; + /// Per-thread codex config overrides (`thread/start.config`). + config?: Record; + exists?: (p: string) => boolean; + }; } /// A live event, already assigned its `seq` by the session's log. @@ -107,11 +138,12 @@ export interface ReloadReport { interface Session { desc: SessionDescriptor; log: DurableSessionLog; - child: ClaudeChild | null; + driver: LocalDriver | null; /// The resolved claude config root, kept so a rebind spawns against the same - /// one the session was created against. + /// one the session was created against. Unused by codex, which is configured + /// per thread through `thread/start.config` rather than by a config root. configHome: string; - /// Whether the CURRENT child has emitted its init frame yet. Reset on every + /// Whether the CURRENT driver has emitted its init frame yet. Reset on every /// spawn — see `#record` for the distinction this draws. initSeen: boolean; } @@ -168,7 +200,7 @@ export class LocalAgentService { restored: true, }, log, - child: null, + driver: null, configHome: meta.config_home ?? resolveConfigHome(undefined, this.#opts.env, this.#opts.homeDir), initSeen: false, }); @@ -189,7 +221,8 @@ export class LocalAgentService { create(opts: CreateOptions): SessionDescriptor { const familyName = opts.family ?? 'claude-code'; const family = familyByName(this.#opts.families, familyName); - if (!supportsLocalDriving(family)) { + const engine = localEngine(family); + if (engine === null) { throw new Error(`local agent service cannot drive family ${familyName}`); } if (opts.cwd.trim() === '') { @@ -197,11 +230,6 @@ export class LocalAgentService { } const id = `local-${randomUUID()}`; - // The engine's handle, assigned rather than awaited. claude honours - // `--session-id ` (probed on 2.1.220), so the session has something - // to resume against from the moment its directory exists — including if - // the very first launch dies before emitting a frame. - const engineSessionId = randomUUID(); const posture = opts.posture ?? DEFAULT_TOOL_POSTURE; const configHome = resolveConfigHome(opts.configHome, this.#opts.env, this.#opts.homeDir); const desc: SessionDescriptor = { @@ -212,17 +240,21 @@ export class LocalAgentService { model: opts.model, status: 'running', created_at: this.#nowIso(), - engine_session_id: engineSessionId, + // claude's handle is assigned rather than awaited (`--session-id`, + // probed on 2.1.220); codex mints its own thread id and takes no such + // flag, so its handle stays absent until `thread/started` reports one. + // See `SessionDescriptor.engine_session_id`. + ...(engine === 'claude-code' ? { engine_session_id: randomUUID() } : {}), }; const { dir } = sessionPaths(this.#opts.dataDir, id); const log = DurableSessionLog.create(dir, { capacity: this.#opts.logCapacity }); - // Written BEFORE the child starts: a descriptor that appeared only after a - // successful launch would leave a crashed first launch with a transcript + // Written BEFORE the driver starts: a descriptor that appeared only after + // a successful launch would leave a crashed first launch with a transcript // nobody can attribute. this.#persist(desc, configHome); - this.#sessions.set(id, { desc, log, child: null, configHome, initSeen: false }); + this.#sessions.set(id, { desc, log, driver: null, configHome, initSeen: false }); this.#spawn(id, undefined); return { ...desc }; } @@ -243,9 +275,9 @@ export class LocalAgentService { /// session that is not listening. input(id: string, kind: InputKind, payload: InputPayload): void { const s = this.#require(id); - if (s.child === null || !s.child.running) this.rebind(id); - const child = this.#require(id).child; - if (child === null) throw new Error(`local session ${id} is not running`); + if (s.driver === null || !s.driver.running) this.rebind(id); + const driver = this.#require(id).driver; + if (driver === null) throw new Error(`local session ${id} is not running`); // Record the director's own turn BEFORE sending it. Two reasons, and the // first is why this exists at all: // @@ -264,48 +296,48 @@ export class LocalAgentService { // Before rather than after, because the prompt is the cause of the turn the // child is about to open, and a reader sorting by seq should see it that way. this.#record(id, { kind: `input.${kind}`, producer: 'user', payload: inputEventPayload(payload) }); - child.input(kind, payload); + driver.input(kind, payload); } - /// Respawn a stopped session's engine child, reattached to its conversation. + /// Restart a stopped session's engine, reattached to its conversation. /// - /// Throws when there is no handle to resume against, or when the family does - /// not resume by argv. It does NOT throw when the engine rejects the handle: - /// claude exits 1 with `No conversation found with session ID`, which arrives - /// as an `error` row and a `lifecycle{expected:false}` in the transcript. That - /// is the honest place for it — the reader sees the session failed to - /// reattach, rather than an exception surfacing somewhere that cannot say - /// which session it was about. + /// Throws when there is no handle to resume against. It does NOT throw when + /// the engine rejects the handle: claude exits 1 with `No conversation found + /// with session ID`, which arrives as an `error` row and a + /// `lifecycle{expected:false}` in the transcript. That is the honest place + /// for it — the reader sees the session failed to reattach, rather than an + /// exception surfacing somewhere that cannot say which session it was about. + /// + /// **The mechanism is per family, and the N1 table is what says which.** + /// claude reattaches by argv (`--resume ` spliced from the recipe); + /// codex reattaches by RPC (`thread/resume`), which is exactly what that + /// table's codex row declares — `mechanism: appserver_thread_resume`, with a + /// note recording that the `codex resume ` CLI row is a different rung. + /// So the argv is built here only for the families whose mechanism IS argv, + /// and the rest hand their handle to a driver that knows what to do with it. rebind(id: string): SessionDescriptor { const s = this.#require(id); - if (s.child !== null && s.child.running) return { ...s.desc }; + if (s.driver !== null && s.driver.running) return { ...s.desc }; const handle = s.desc.engine_session_id; if (handle === undefined || handle === '') { throw new Error(`local session ${id} has no engine session id to resume`); } - const ref = newSessionID(handle); - if (ref === null) { + // Screened before it is used, on both paths: the handle arrives from the + // engine and is read back off disk, so neither its length nor its bytes + // are ours to assume. + if (newSessionID(handle) === null) { throw new Error(`local session ${id} has an unusable engine session id`); } - const splice = resumeSplice( - this.#opts.resumeTable, - s.desc.family, - ref, - this.#opts.platform ?? process.platform, - ); - if (!splice.ok) { - throw new Error(`local session ${id} cannot resume: ${splice.error}`); - } - this.#spawn(id, splice.tokens); + this.#spawn(id, handle); return { ...this.#require(id).desc }; } stop(id: string): void { const s = this.#sessions.get(id); if (s === undefined) return; - s.child?.stop(); + s.driver?.stop(); s.desc.status = 'stopped'; // Flush before the process can go away. The batching in `durablelog` is a // performance choice, not a durability one, and this is where that has to @@ -314,8 +346,9 @@ export class LocalAgentService { } /// Stop every session. Called on app quit, beside the other hosts' - /// `disposeAll` — a child left running after the window closes is a claude - /// process nobody can see and nobody will reap. + /// `disposeAll` — a child left running after the window closes is an engine + /// process nobody can see and nobody will reap. (On codex's daemon rung this + /// closes our socket only: that rung exists so the thread outlives us.) disposeAll(): void { for (const id of [...this.#sessions.keys()]) this.stop(id); } @@ -328,7 +361,7 @@ export class LocalAgentService { forget(id: string): boolean { const s = this.#sessions.get(id); if (s === undefined) return false; - if (s.desc.status === 'running' && s.child !== null && s.child.running) { + if (s.desc.status === 'running' && s.driver !== null && s.driver.running) { throw new Error(`session ${id} is still running; stop it first`); } s.log.close(); @@ -344,21 +377,63 @@ export class LocalAgentService { // ── internals ────────────────────────────────────────────────────────────── - /// Start (or restart) a session's engine child. + /// Start (or restart) a session's engine. /// - /// `resumeTokens` empty means a fresh conversation, in which case the engine - /// id is ASSIGNED; non-empty means a rebind, in which case it is already - /// spoken for and `--session-id` must not also be passed. - #spawn(id: string, resumeTokens: readonly string[] | undefined): void { + /// `handle` absent means a fresh conversation; present means a rebind + /// against the engine's own session id. + #spawn(id: string, handle: string | undefined): void { const s = this.#require(id); const family = familyByName(this.#opts.families, s.desc.family); - if (!supportsLocalDriving(family)) { + const engine = localEngine(family); + if (engine === null) { throw new Error(`local agent service cannot drive family ${s.desc.family}`); } - const resuming = resumeTokens !== undefined && resumeTokens.length > 0; + const onExit = (): void => { + const cur = this.#sessions.get(id); + if (cur !== undefined) { + cur.desc.status = 'stopped'; + cur.log.flush(); + } + }; + + const driver = + engine === 'codex' + ? this.#codexDriver(id, family as Family, s, handle, onExit) + : this.#claudeDriver(id, family as Family, s, handle, onExit); + + s.driver = driver; + s.desc.status = 'running'; + s.desc.restored = false; + s.initSeen = false; + // After the field is set: `start()` emits its lifecycle event + // synchronously, and #record has to find the session to log it into. + driver.start(); + } - const child = new ClaudeChild({ - family: family as Family, + #claudeDriver( + id: string, + family: Family, + s: Session, + handle: string | undefined, + onExit: () => void, + ): LocalDriver { + let resumeTokens: string[] | undefined; + if (handle !== undefined) { + const ref = newSessionID(handle); + // `rebind` already screened it; this narrows the type without a cast, + // and would catch an internal caller that skipped the check. + if (ref === null) throw new Error(`local session ${id} has an unusable engine session id`); + const splice = resumeSplice( + this.#opts.resumeTable, + s.desc.family, + ref, + this.#opts.platform ?? process.platform, + ); + if (!splice.ok) throw new Error(`local session ${id} cannot resume: ${splice.error}`); + resumeTokens = splice.tokens; + } + return new ClaudeChild({ + family, cwd: s.desc.cwd, posture: s.desc.posture, model: s.desc.model, @@ -366,25 +441,41 @@ export class LocalAgentService { env: this.#opts.env, spawnFn: this.#opts.spawnFn, now: this.#opts.now, - sessionId: resuming ? undefined : s.desc.engine_session_id, + // `--resume` names an existing conversation and `--session-id` names a + // new one, so exactly one of them is ever passed. + sessionId: handle === undefined ? s.desc.engine_session_id : undefined, resumeTokens, onEvent: (ev) => this.#record(id, ev), - onExit: () => { - const cur = this.#sessions.get(id); - if (cur !== undefined) { - cur.desc.status = 'stopped'; - cur.log.flush(); - } - }, + onExit, }); + } - s.child = child; - s.desc.status = 'running'; - s.desc.restored = false; - s.initSeen = false; - // After the field is set: `start()` emits its lifecycle event - // synchronously, and #record has to find the session to log it into. - child.start(); + #codexDriver( + id: string, + family: Family, + s: Session, + handle: string | undefined, + onExit: () => void, + ): LocalDriver { + const codex = this.#opts.codex ?? {}; + return new CodexDriver({ + family, + cwd: s.desc.cwd, + posture: s.desc.posture, + model: s.desc.model, + env: this.#opts.env, + homeDir: this.#opts.homeDir, + ...(handle !== undefined ? { resumeThreadId: handle } : {}), + ...(codex.config !== undefined ? { config: codex.config } : {}), + ...(codex.plan !== undefined ? { plan: codex.plan } : {}), + ...(codex.deps !== undefined ? { channelDeps: codex.deps } : {}), + ...(codex.openChannel !== undefined ? { openChannel: codex.openChannel } : {}), + ...(codex.flushIntervalMs !== undefined ? { flushIntervalMs: codex.flushIntervalMs } : {}), + ...(codex.exists !== undefined ? { exists: codex.exists } : {}), + now: this.#opts.now, + onEvent: (ev) => this.#record(id, ev), + onExit, + }); } #record(sessionId: string, ev: DriverEvent): void { diff --git a/desktop/electron/src/localagent/store.ts b/desktop/electron/src/localagent/store.ts index 65804467..4638d3fd 100644 --- a/desktop/electron/src/localagent/store.ts +++ b/desktop/electron/src/localagent/store.ts @@ -28,7 +28,7 @@ import { existsSync, mkdirSync, readdirSync, readFileSync, renameSync, rmSync, writeFileSync } from 'node:fs'; import path from 'node:path'; -import type { ToolPosture } from './claudewire.ts'; +import type { ToolPosture } from './driver.ts'; export const SESSIONS_DIRNAME = 'local-sessions'; export const META_FILENAME = 'meta.json'; diff --git a/docs/changelog-desktop.md b/docs/changelog-desktop.md index be00b8de..de8ed7fb 100644 --- a/docs/changelog-desktop.md +++ b/docs/changelog-desktop.md @@ -42,6 +42,34 @@ This complements: ## Unreleased ### Added +- **The Companion can now run a *codex* session on this machine too.** Pick + `codex` in the local picker and the dock drives it through the vendor's own + `app-server` — a JSON-RPC thread that streams as it writes, folds tool calls + the same way a claude session does, and comes back after an app restart with + its memory intact (`thread/resume`) and its transcript re-read from disk. It + reuses the hub's own codex frame profile, so a local codex transcript and a + hub-driven one are the same rows. Where a shared codex daemon is available + the session attaches to it and survives the app; otherwise it runs a + per-session app server, and the transcript **says which one it got** — those + are different promises. (vision-parity L4c) + + **What the agent may do is measured, not named.** A codex session lowers the + same three postures to codex's own sandbox: `read_local` (the default) opens + the thread with `sandbox: read-only` and `approvalPolicy: never`, which was + probed by asking codex to create a file — it tried, failed, said *"the + environment is read-only"*, and nothing appeared on disk. `converse` cannot + be kept exactly (codex has no way to turn its tools off), so the transcript + carries a line saying the agent can still read files, rather than letting the + name imply otherwise. + + **Approvals and questions arrive as cards in the feed**, answered inline — + a local session has no attention table, so this is where codex's gates live. + Four kinds are declined immediately instead, each with a line saying why: a + form-fill an inline card cannot type into, a "open this URL" request, a + permission-profile grant, and — deliberately — any question codex marks + **secret**, because an answer sent through the Companion is written to the + session transcript on disk, which is not where a secret belongs. + - **The Companion can run a claude session on this machine, with no hub.** A local agent service in Electron main owns the engine child and keeps the session's transcript in an append-only log with cursor semantics, so a diff --git a/docs/changelog.md b/docs/changelog.md index 220765fc..843a56f8 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -210,6 +210,26 @@ binding). Seed entries prior to that are in ### Fixed +- **Sending an image to a codex agent never reached the model, and sending a + PDF silently lost the whole message.** The app-server driver lowered + attachments to OpenAI *responses-API* content types — `input_image` and + `input_file` — which codex's `turn/start` does not accept. Measured against + codex-cli 0.147.0, it answers both with `-32600 Invalid request: unknown + variant`, and because that rejects the entire call, a PDF took the + director's text down with it. Images now ride the variant codex actually + documents in its own generated protocol (`{type:"image", url:"data:…"}`, + confirmed end to end — the model described the picture), and PDFs are + stripped with a `system` row saying so, the same strip-and-warn shape the + gemini exec-per-turn driver uses for what it cannot carry. The engine + registry agreed with the old shape and has been corrected too: + `codex.prompt_pdf.M2` is now `false` — codex 0.147.0 has no file input + variant at all — so the composer stops offering an attachment that cannot + arrive. `prompt_image.M2` stays `true`; only the shape was wrong. The + driver's test used a fake app-server that accepts any params, so it had + pinned the broken shape rather than the protocol; the replacement asserts + the stripped-but-still-sent turn, and the desktop's new codex driver ships + an opt-in live e2e against a real app-server. (vision-parity L4c) + - **Changing a steward's model or permission mode never worked.** Both the routing gate and the executor resolved the engine from `agents.kind`, which for a steward is the persona template (`steward.claude-m4`), not the family diff --git a/docs/plans/desktop-companion-vision-parity.md b/docs/plans/desktop-companion-vision-parity.md index 223e895c..ec881e2b 100644 --- a/docs/plans/desktop-companion-vision-parity.md +++ b/docs/plans/desktop-companion-vision-parity.md @@ -1,21 +1,10 @@ # Desktop Companion vision parity — kimi-web's bar on our data scheme > **Type:** plan -> **Status:** In flight (2026-08-05) — W1 landed (F1, F2, L1, E1, R1); -> **W2 complete** (L2, E2, R2, R3, F3). Principal review done. **W3 in -> flight**: L3 split into **L3a** (shipped 2026-08-10 — the local agent -> service, claude M2 child, renderer source), **L3b** (shipped -> 2026-08-14 — on-disk log + restart rebind) and **L3c** (session -> catalog + loopback WS, split out of L3b). **E3 + E4 shipped -> 2026-08-16** (streaming command output; relay result passthrough). -> **R4 shipped 2026-08-16** (live output + agent-produced media) — -> **W3 complete**; L3c is deferrable. **W4 started 2026-08-16**: -> **F4 shipped** (user-level MCP reseed for claude + codex), -> **L4a shipped** (codex attach-rung resolver) and **L4b shipped** -> (codex app-server transport — which corrected L4a: the daemon's -> control socket *is* a WebSocket, and L4a's `app-server proxy` data -> path was a dead channel). L4 split again: **L4c** is the driver. -> Left in W4: L4c, R5, R6 +> **Status:** In flight (2026-08-05) — **W1 + W2 + W3 complete** +> (F1 F2 L1 E1 R1 · L2 E2 R2 R3 F3 · L3a L3b E3 E4 R4; L3c deferrable). +> **W4 in flight (2026-08-16)**: F4, L4a, L4b and L4c shipped, so +> **lane L4 is complete** — codex is drivable locally. Left in W4: R5, R6. > **Audience:** principal · contributors · maintainers > **Last verified vs code:** 2026.730.1231-alpha (`cea267fa`) — every > anchor below re-verified against that tip by the authoring audit @@ -562,17 +551,112 @@ transport rung (lane T), never the renderer's ceiling. resolves through PATH plus the well-known dirs the way `kimiweb.ts` does for kimi, with `TERMIPOD_CODEX_BIN` as the override. - **L4c** (still to do) is the driver itself: the app-server JSON-RPC - client over L4b's channel, delta throttling, R1 approval cards, - `turn/interrupt`, config seeding, and resume argv from the N1 table. - The translation half is already done and shipped — the hub's `codex` - frame profile maps `thread/started` → `session.init`, - `item/completed` → `text`/`tool_result` and so on, and it rides to - the desktop in `agent_families.generated.json`, so L4c reads that - table rather than writing a second one. The vendor also generates the - whole protocol on demand (`codex app-server generate-ts --out DIR`, - and `generate-json-schema`), which is the authority to check request - shapes against. + **L4c shipped 2026-08-16** — the driver + (`localagent/codexdriver.ts` + `localagent/codexwire.ts`), which + makes the whole lane user-visible for the first time: a codex session + is now offered in the local picker, opens a thread, streams, takes + approvals, and reattaches after a restart. + + *What the plan got right.* The translation half really was already + done: the hub's `codex` frame profile ships to the desktop in + `agent_families.generated.json`, and the driver reads that table + rather than writing a second one. And the vendor really does generate + its own protocol (`codex app-server generate-ts --out DIR`, 93 files + at 0.147.0), which is what every request shape below was checked + against. + + ★★ *What the plan got wrong, and what it cost.* Two lines: + + 1. **"resume argv from the N1 table" describes a rung this driver + does not use** — and N1's own table says so. Its codex family row + reads `mechanism: appserver_thread_resume`, with a note recording + that the `codex resume ` CLI recipe is "the spawn-per-session + fallback rung … not what the hub drives today". Resume here is an + RPC (`thread/resume`), so `service.rebind` now routes by + *mechanism*: argv families get a splice, the rest hand their handle + to a driver that knows what to do with it. The plan line pointed at + a table that contradicted it. + 2. ★★ **Two of the input shapes we ship are REJECTED by codex, and + the fixtures could not see it.** Measured against codex-cli + 0.147.0: + + | sent | answer | + |---|---| + | `{type:"input_image", image_url:"data:…"}` | `-32600 Invalid request: unknown variant 'input_image', expected one of 'text', 'image', 'localImage', 'audio', 'localAudio', 'skill', 'mention'` | + | `{type:"input_file", file_data:"data:…"}` | the same error | + | `{type:"image", url:"data:…"}` | accepted — the model described the pixel | + + Those are OpenAI **responses-API** content types, not app-server + ones. This is not only the new driver's problem: the hub's + `AppServerDriver.startTurn` has been sending both since W4.3, so + **E4's image passthrough has never reached a codex M2 agent**, and + a PDF was worse than useless — an unknown variant fails the whole + `turn/start`, so the director's message was lost with it. Fixed in + the same commit, with the strip-and-warn shape the gemini + exec-per-turn driver already uses for what it cannot carry. + + The registry was asserting it too: `codex.prompt_pdf.M2` was + `true`, which is what the composer's F3 gate reads. It is now + `false`, measured — codex 0.147.0's `UserInput` union has no file + variant at all. `prompt_image.M2` stays `true`; only the shape was + wrong. + + **Why it survived a green suite:** the Go test's fake app-server + accepts any params, so it pinned the shipped shape rather than the + protocol — a fixture that cannot disprove the rule it is testing. + The desktop's equivalent would have had the same blind spot, which + is why L4c ships an opt-in live e2e + (`TERMIPOD_CODEX_DRIVER_E2E=1`) that runs real turns. + + *Two more measurements the driver is built on.* + + - **`thread/resume` restores the memory and emits no replay** — the + only notifications after it are `configWarning`, + `remoteControl/status/changed`, `thread/status/changed`, + `thread/tokenUsage/updated`, `thread/goal/cleared`. Same shape as + claude's `--resume`, so L3b's durable log is what restores the view + for codex too. (Its *response*, unlike claude's, does carry the + whole `thread.turns[]` history — a replay source we do not need + while the log exists, recorded because it would be the answer if + that ever changed.) + - **`sandbox: read-only` + `approvalPolicy: never` is a real write + barrier**, not a name: asked to create a file in its cwd, codex + tried, failed, said *"the environment is read-only"*, and nothing + appeared on disk. The response echoes the lowered policy as + `{"type":"readOnly","networkAccess":false}`. That is how the codex + driver keeps the `read_local` posture claude keeps with `--tools` — + one contract, two mechanisms, each with its own probe. `converse` + cannot be kept exactly (codex has no tool-disable switch) so the + lifecycle row carries a `posture_note` saying so rather than + letting the name imply a boundary that is not there. + + *Approvals, with no attention table.* Server-initiated requests + become `approval_request` events in R1's own shapes and are answered + by the cards that already ship — which is exactly what R1's comment + predicted ("A local driver parks nothing … answered by the two cards + above (plan L4)"). Four things are deliberately **refused at once + rather than parked**, because a card nobody can answer holds the + engine open forever: a form-mode elicitation (its schema wants typed + fields the card has no input for), a `url` elicitation, an + `item/permissions/requestApproval` (its response is a granted + *profile*, not a verdict), and — the one worth naming — a question + flagged `isSecret`, because `service.input()` records every input into + the durable on-disk transcript before sending it, so answering a + secret through a card would write it to disk. Each refusal uses that + method's own response shape; an empty `{}` fails to deserialize on + every one of them. + + Also carried over: `.codex/config.toml` seeding is **not** a written + file. `ThreadStartParams` takes a `config` map of the same overrides, + scoped to the thread — so configuring a session never leaves a file in + the director's project that outlives it. + + Still open in this lane, named rather than silently skipped: no + posture maps to `approvalPolicy: on-request`, so codex's own + command/file-change gates never fire from a Companion session today. + Adding one is a posture decision (it means "read local, but the agent + may ask to escape"), not a driver change — the approval path is built + and tested, and MCP elicitations already exercise it. ### Lane E — event-vocabulary gaps (hub; verified per-driver) diff --git a/hub/internal/agentfamilies/agent_families.yaml b/hub/internal/agentfamilies/agent_families.yaml index ce82d683..100fcc29 100644 --- a/hub/internal/agentfamilies/agent_families.yaml +++ b/hub/internal/agentfamilies/agent_families.yaml @@ -549,18 +549,29 @@ families: M1: respawn M2: respawn # ADR-021 D5 / W4.3 — codex's app-server JSON-RPC `turn/start.input` - # accepts content blocks including `{type:"input_image", - # image_url:"data:..."}`. Driver wire shape lands in W4.3. + # accepts an image content block. MEASURED against codex-cli + # 0.147.0 (vision-parity L4c): the accepted shape is + # `{type:"image", url:"data:;base64,"}`. The + # `{type:"input_image", image_url:...}` this comment previously + # claimed is REJECTED — `-32600 Invalid request: unknown variant + # 'input_image', expected one of 'text', 'image', 'localImage', + # 'audio', 'localAudio', 'skill', 'mention'`. prompt_image: M1: true M2: true M4: false - # artifact-type-registry W7.2 — codex accepts PDF as `file_data` - # content block; audio/video aren't supported on OpenAI chat - # input today. + # M2 is FALSE, measured, and this row used to say true. + # codex 0.147.0's `UserInput` union (the vendor's own + # `codex app-server generate-ts`) has no file variant at all, and + # `{type:"input_file", file_data:...}` is rejected with the same + # "unknown variant" error — so a PDF did not merely fail to reach + # the model, it failed the whole `turn/start` and lost the turn + # with it. M1 (the ACP adapter) is a different transport and is + # unaffected. Audio/video aren't supported on OpenAI chat input + # today. prompt_pdf: M1: true - M2: true + M2: false M4: false # ─── Launch contract (ADR-043) ──────────────────────────────── # codex M2 drives the app-server's JSON-RPC stdio protocol. The diff --git a/hub/internal/hostrunner/driver_appserver.go b/hub/internal/hostrunner/driver_appserver.go index 3668f80e..a504eba5 100644 --- a/hub/internal/hostrunner/driver_appserver.go +++ b/hub/internal/hostrunner/driver_appserver.go @@ -538,15 +538,37 @@ func (d *AppServerDriver) Input(ctx context.Context, kind string, payload map[st case "text": body, _ := payload["body"].(string) images := extractImageInputs(payload) - pdfs := extractAttachmentInputs(payload, "pdfs") - // audios/videos: silently dropped — OpenAI chat API doesn't - // accept audio/video input. The composer's family gate already - // hides the affordance; this is the backstop for A2A or other + // PDFs are STRIPPED here, not forwarded. Measured against + // codex-cli 0.147.0 (vision-parity L4c): `turn/start` answers + // `{type:"input_file", …}` with `-32600 Invalid request: + // unknown variant 'input_file', expected one of 'text', + // 'image', 'localImage', 'audio', 'localAudio', 'skill', + // 'mention'` — the vendor's `UserInput` union has no file + // variant at all. So forwarding one did not merely fail to + // reach the model: it failed the WHOLE turn/start and lost the + // director's message with it. Same strip-and-warn shape the + // gemini exec-per-turn driver uses for the modalities it + // cannot carry (driver_exec_resume.go). The family registry's + // `prompt_pdf.M2` is now `false`, so the composer no longer + // offers it; this is the backstop for A2A and other // out-of-band injection paths. - if body == "" && len(images) == 0 && len(pdfs) == 0 { + // + // audios/videos: dropped for the same reason, and always were + // — OpenAI chat input accepts neither. + dropped := len(extractAttachmentInputs(payload, "pdfs")) + + len(extractAttachmentInputs(payload, "audios")) + + len(extractAttachmentInputs(payload, "videos")) + if dropped > 0 { + _ = d.Poster.PostAgentEvent(ctx, d.AgentID, "system", "agent", map[string]any{ + "reason": "codex app-server accepts no file/audio/video attachments — its turn/start input union is text|image|localImage|audio|localAudio|skill|mention, and a file block fails the whole turn", + "engine": "codex-app-server", + "dropped": dropped, + }) + } + if body == "" && len(images) == 0 { return fmt.Errorf("appserver driver: text input missing body") } - return d.startTurn(ctx, body, images, pdfs) + return d.startTurn(ctx, body, images) case "attention_reply": // Three paths depending on attention kind: // - kind=permission_prompt → we have a parked codex JSON-RPC @@ -573,7 +595,7 @@ func (d *AppServerDriver) Input(ctx context.Context, kind string, payload map[st if body == "" { return fmt.Errorf("appserver driver: attention_reply produced no text") } - return d.startTurn(ctx, body, nil, nil) + return d.startTurn(ctx, body, nil) case "cancel": // codex requires both `threadId` and `turnId` on turn/interrupt. // Without either the server replies -32600 "missing field …". @@ -846,42 +868,41 @@ func elicitationContentFromBody(body string) map[string]any { // through notifications (item/*, turn/started, turn/completed) // translated by the frame profile. // -// ADR-021 W4.3 — image content blocks lower to OpenAI responses-API -// shape `{type:"input_image", image_url:"data:;base64,"}` -// and lead the input array; the text block (if any) comes last so -// the model sees the imagery before the question. Image-only inputs -// (no body text) are accepted at this layer. +// ADR-021 W4.3 — images lead the input array and the text block (if +// any) comes last, so the model sees the imagery before the question. +// Image-only inputs (no body text) are accepted at this layer. +// +// It builds the array from the vendor's own `UserInput` +// union — checked against `codex app-server generate-ts` @ codex-cli +// 0.147.0 and confirmed on the wire, not read off prose: +// +// text | image | localImage | audio | localAudio | skill | mention // -// artifact-type-registry W7.2 — PDFs lower to -// `{type:"input_file", filename, file_data:"data:application/pdf;base64,..."}` -// the OpenAI responses-API shape for inline document input. +// An image is `{type:"image", url:"data:;base64,"}`. The +// `input_image` / `input_file` spellings this function used to send are +// OpenAI *responses-API* content types, not app-server ones, and codex +// rejects them with `-32600 unknown variant`, failing the turn. +// +// `text` is sent without `text_elements` even though the generated type +// marks it required: the server fills `text_elements: []` itself +// (observed on the echoed userMessage item), and synthesizing UI spans +// we do not have would be inventing data. func (d *AppServerDriver) startTurn( ctx context.Context, text string, images []imageInput, - pdfs []attachmentInput, ) error { tid := d.ThreadID() if tid == "" { return fmt.Errorf("appserver driver: no active thread (handshake didn't complete?)") } - input := make([]map[string]any, 0, len(images)+len(pdfs)+1) + input := make([]map[string]any, 0, len(images)+1) for _, img := range images { input = append(input, map[string]any{ - "type": "input_image", - "image_url": "data:" + img.mime + ";base64," + img.data, + "type": "image", + "url": "data:" + img.mime + ";base64," + img.data, }) } - for _, p := range pdfs { - block := map[string]any{ - "type": "input_file", - "file_data": "data:" + p.mime + ";base64," + p.data, - } - if p.filename != "" { - block["filename"] = p.filename - } - input = append(input, block) - } if text != "" { input = append(input, map[string]any{"type": "text", "text": text}) } diff --git a/hub/internal/hostrunner/driver_appserver_test.go b/hub/internal/hostrunner/driver_appserver_test.go index 77c4665e..0bef87ae 100644 --- a/hub/internal/hostrunner/driver_appserver_test.go +++ b/hub/internal/hostrunner/driver_appserver_test.go @@ -305,12 +305,19 @@ func TestAppServerDriver_HandshakeAndTurn(t *testing.T) { } } -// TestAppServerDriver_TurnStart_ImageBlocks pins the W4.3 wire shape: -// when payload["images"] is set, turn/start.params.input leads with -// `{type:"input_image", image_url:"data:;base64,"}` blocks -// and follows with the `{type:"text", text:body}` block. Image-only -// inputs (no body) produce a single image block. Hub-side W4.1 -// validation is upstream; the driver trusts the payload shape. +// TestAppServerDriver_TurnStart_ImageBlocks pins the wire shape codex +// actually accepts: turn/start.params.input leads with +// `{type:"image", url:"data:;base64,"}` blocks and follows +// with `{type:"text", text:body}`. Hub-side W4.1 validation is +// upstream; the driver trusts the payload shape. +// +// It used to pin `{type:"input_image", image_url:…}` — an OpenAI +// responses-API content type, not an app-server one. The fake server +// below accepts any params, so the test passed for as long as the +// shipped shape was wrong; measured against codex-cli 0.147.0, the +// real server answers it `-32600 Invalid request: unknown variant +// 'input_image', expected one of 'text', 'image', 'localImage', +// 'audio', 'localAudio', 'skill', 'mention'` (vision-parity L4c). func TestAppServerDriver_TurnStart_ImageBlocks(t *testing.T) { pipes := newPipePair() t.Cleanup(pipes.closeFn) @@ -363,14 +370,14 @@ func TestAppServerDriver_TurnStart_ImageBlocks(t *testing.T) { t.Fatalf("input: want 3 blocks, got %d (%+v)", len(input), input) } first, _ := input[0].(map[string]any) - if first["type"] != "input_image" { - t.Errorf("input[0].type = %v, want input_image", first["type"]) + if first["type"] != "image" { + t.Errorf("input[0].type = %v, want image", first["type"]) } - if got := first["image_url"]; got != "data:image/png;base64,AAA=" { - t.Errorf("input[0].image_url = %v", got) + if got := first["url"]; got != "data:image/png;base64,AAA=" { + t.Errorf("input[0].url = %v", got) } second, _ := input[1].(map[string]any) - if second["type"] != "input_image" || second["image_url"] != "data:image/jpeg;base64,BBB=" { + if second["type"] != "image" || second["url"] != "data:image/jpeg;base64,BBB=" { t.Errorf("input[1] malformed: %+v", second) } third, _ := input[2].(map[string]any) @@ -379,6 +386,86 @@ func TestAppServerDriver_TurnStart_ImageBlocks(t *testing.T) { } } +// TestAppServerDriver_TurnStart_PdfsAreStrippedNotForwarded pins the +// other half of the same measurement. codex-cli 0.147.0's `UserInput` +// union has no file variant at all, so `{type:"input_file", …}` is +// answered `-32600 unknown variant 'input_file'` — which fails the +// WHOLE turn/start, not just the attachment. A PDF must therefore be +// stripped with a `system` row saying so, the same strip-and-warn +// shape driver_exec_resume.go uses for what gemini cannot carry. +// +// The separating input is a PDF *with body text*: the turn must still +// reach the engine carrying the text, which is what distinguishes +// "stripped" from "refused". +func TestAppServerDriver_TurnStart_PdfsAreStrippedNotForwarded(t *testing.T) { + pipes := newPipePair() + t.Cleanup(pipes.closeFn) + + server := newFakeAppServer(t, pipes.serverRead, pipes.serverWrite) + server.onCall("initialize", func(_ map[string]any) any { + return map[string]any{"protocolVersion": "1.0"} + }) + server.onCall("thread/start", func(_ map[string]any) any { + return map[string]any{"thread": map[string]any{"id": "thr_pdf"}} + }) + server.onCall("turn/start", func(_ map[string]any) any { + return map[string]any{"turn": map[string]any{"id": "turn_pdf"}} + }) + go server.run() + + poster := &fakePoster{} + drv := &AppServerDriver{ + AgentID: "agent-pdf", + Poster: poster, + Stdout: pipes.driverStdout, + Stdin: pipes.driverStdin, + FrameProfile: codexProfileForTest(t), + HandshakeTimeout: 2 * time.Second, + CallTimeout: 2 * time.Second, + Closer: pipes.closeFn, + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := drv.Start(ctx); err != nil { + t.Fatalf("Start: %v", err) + } + t.Cleanup(drv.Stop) + server.waitForMethod("thread/start", time.Second) + + if err := drv.Input(ctx, "text", map[string]any{ + "body": "summarise this", + "pdfs": []any{map[string]any{"mime_type": "application/pdf", "data": "JVBER"}}, + }); err != nil { + t.Fatalf("Input: %v", err) + } + + turnFrame := server.waitForMethod("turn/start", time.Second) + params, _ := turnFrame["params"].(map[string]any) + input, _ := params["input"].([]any) + if len(input) != 1 { + t.Fatalf("input: want only the text block, got %d (%+v)", len(input), input) + } + block, _ := input[0].(map[string]any) + if block["type"] != "text" || block["text"] != "summarise this" { + t.Errorf("input[0] = %+v, want the text block", block) + } + + // Stripped, and SAID SO — a silent drop would leave the agent + // answering a question about a document it never received. + var dropped bool + for _, ev := range poster.snapshot() { + if ev.Kind != "system" { + continue + } + if n, ok := ev.Payload["dropped"].(int); ok && n == 1 { + dropped = true + } + } + if !dropped { + t.Errorf("expected a system event reporting the dropped attachment, got %+v", poster.snapshot()) + } +} + // TestAppServerDriver_Cancel_IncludesThreadID pins the cancel→ // turn/interrupt translation: codex's app-server returns // -32600 "Invalid request: missing field `threadId`" when threadId