diff --git a/cli/src/agy/headless/agyHeadlessDriver.test.ts b/cli/src/agy/headless/agyHeadlessDriver.test.ts index e8bbc287..6b860ed0 100644 --- a/cli/src/agy/headless/agyHeadlessDriver.test.ts +++ b/cli/src/agy/headless/agyHeadlessDriver.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi } from 'vitest'; import { EventEmitter } from 'node:events'; +import { PassThrough, Writable } from 'node:stream'; import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'; import { logger } from '@/ui/logger'; import { buildAgyHeadlessArgs, AgyHeadlessDriver } from './agyHeadlessDriver'; @@ -11,8 +12,10 @@ import type { AgyMode, PermissionMode } from '../types'; function fakeAgyProcess(lines: string[], exitCode = 0): ChildProcessWithoutNullStreams { const child = spawn(process.execPath, ['-e', ` const lines = ${JSON.stringify(lines)}; - for (const line of lines) process.stdout.write(line + '\\n'); - process.exit(${exitCode}); + require('node:readline').createInterface({ input: process.stdin }).once('line', () => { + for (const line of lines) process.stdout.write(line + '\\n'); + process.exit(${exitCode}); + }); `], { stdio: ['pipe', 'pipe', 'pipe'] }) as ChildProcessWithoutNullStreams; return child; } @@ -26,18 +29,30 @@ function scriptableFakeChild(chunks: string[], exitCode = 0): ChildProcessWithou run: () => void; feedStdout: (chunk: string) => void; } { - const stdout = new EventEmitter() as unknown as ChildProcessWithoutNullStreams['stdout']; - const stderr = new EventEmitter() as unknown as ChildProcessWithoutNullStreams['stderr']; - for (const stream of [stdout, stderr]) { - (stream as unknown as { setEncoding: () => void }).setEncoding = () => {}; - } + const stdout = new PassThrough(); + const stderr = new PassThrough(); let killed = false; + let closed = false; const child = new EventEmitter() as unknown as ChildProcessWithoutNullStreams; + const close = (code: number | null, signal: string | null = null) => { + if (!closed) child.emit('close', code, signal); + }; + const stdin = new Writable({ + write(_chunk, _encoding, callback) { callback(); }, + final(callback) { callback(); queueMicrotask(() => close(0)); }, + }); Object.assign(child, { - stdout, - stderr, + stdin, stdout, stderr, exitCode: null, signalCode: null, get killed() { return killed; }, - kill: () => { killed = true; }, + kill: (signal = 'SIGTERM') => { + killed = true; + queueMicrotask(() => close(null, signal)); + return true; + }, + }); + child.on('close', (code, signal) => { + closed = true; + Object.assign(child, { exitCode: code, signalCode: signal }); }); let chunkIndex = 0; const run = () => { @@ -91,13 +106,12 @@ const FULL_TURN = [ ]; describe('buildAgyHeadlessArgs', () => { - it('builds the per-turn print-mode args', () => { + it('builds persistent stream args without placing the prompt on the command line', () => { const args = buildAgyHeadlessArgs({ - prompt: 'hello', permissionMode: 'request-review', }); expect(args).toEqual([ - '-p', 'hello', + '--input-format', 'stream-json', '--output-format', 'stream-json', '--print-timeout', '30m', ]); @@ -105,7 +119,6 @@ describe('buildAgyHeadlessArgs', () => { it('adds resume/model/mode/effort/skip-permissions flags', () => { const args = buildAgyHeadlessArgs({ - prompt: 'hi', conversationId: 'conv-1', model: 'gemini-3.5-flash', permissionMode: 'always-proceed', @@ -124,19 +137,19 @@ describe('buildAgyHeadlessArgs', () => { }); it('omits --dangerously-skip-permissions for request-review', () => { - const args = buildAgyHeadlessArgs({ prompt: 'hi', permissionMode: 'request-review' }); + const args = buildAgyHeadlessArgs({ permissionMode: 'request-review' }); expect(args).not.toContain('--dangerously-skip-permissions'); }); it('rejects a conflicting unnormalized variant instead of spawning an invalid selection', () => { expect(() => buildAgyHeadlessArgs({ - prompt: 'hi', permissionMode: 'request-review', model: 'gemini-4-argon-medium', effort: 'high', + permissionMode: 'request-review', model: 'gemini-4-argon-medium', effort: 'high', })).toThrow('conflicts with effort high'); }); it('avoids redundant effort flags for an exact variant', () => { const args = buildAgyHeadlessArgs({ - prompt: 'hi', permissionMode: 'request-review', model: 'gemini-4-argon-medium', effort: 'medium', + permissionMode: 'request-review', model: 'gemini-4-argon-medium', effort: 'medium', }); expect(args).toContain('gemini-4-argon-medium'); expect(args).not.toContain('--effort'); @@ -144,7 +157,7 @@ describe('buildAgyHeadlessArgs', () => { it.each(['xhigh', 'max'] as const)('passes native %s effort', (effort) => { const args = buildAgyHeadlessArgs({ - prompt: 'hi', permissionMode: 'request-review', model: 'gemini-4-argon', effort, + permissionMode: 'request-review', model: 'gemini-4-argon', effort, }); expect(args[args.indexOf('--effort') + 1]).toBe(effort); }); @@ -297,7 +310,7 @@ describe('AgyHeadlessDriver', () => { await new Promise((r) => setTimeout(r, 20)); child.run(); await new Promise((r) => setTimeout(r, 20)); - child.run(); // fourth run: no chunks left → emits close (the turn boundary) + child.run(); // fourth run: no chunks left → closes the process after its result await new Promise((r) => setTimeout(r, 200)); queue.close(); session.stopKeepAlive(); @@ -810,7 +823,7 @@ describe('AgyHeadlessDriver', () => { await new Promise((r) => setTimeout(r, 30)); child.run(); // one chunk await new Promise((r) => setTimeout(r, 20)); - child.run(); // no chunks left → close (the turn boundary) + child.run(); // no chunks left → close after the result await new Promise((r) => setTimeout(r, 500)); queue.close(); session.stopKeepAlive(); @@ -1485,9 +1498,9 @@ describe('AgyHeadlessDriver', () => { session.stopKeepAlive(); await launchPromise; - // The prompt must never be written to logs — only markers. + // The prompt travels over stdin and must never be written to debug logs. const allLogs = debugSpy.mock.calls.map((c) => c.join(' ')).join('\n'); - expect(allLogs).toContain(''); + expect(allLogs).toContain('--input-format stream-json'); expect(allLogs).not.toContain('super-secret-prompt-text'); debugSpy.mockRestore(); }); diff --git a/cli/src/agy/headless/agyHeadlessDriver.ts b/cli/src/agy/headless/agyHeadlessDriver.ts index 68ae3b96..aa4510af 100644 --- a/cli/src/agy/headless/agyHeadlessDriver.ts +++ b/cli/src/agy/headless/agyHeadlessDriver.ts @@ -14,31 +14,19 @@ import { import { createNativeSessionTitleMetadataSync } from '@/agent/nativeSessionTitle'; import { readAgyConversationTitle } from '../utils/agySessionTitle'; import { resolveAgyTurnModels } from '../utils/agyConversationModel'; -import { killProcessByChildProcess } from '@/utils/process'; +import { AgyStreamProcess } from './agyStreamProcess'; import { AGY_EFFORT_LEVELS, AGY_MODEL_LABELS, splitAgyModelSelection, type AgyEffortLevel } from '@hapi/protocol'; import { getAgentLaunchCommand } from '@/agent/agentLaunchCommand'; const AGY_PRINT_TIMEOUT = '30m'; // Bound on consecutive turns agy fails to accept (spawn error, crash, exit -// before any transcript output). Beyond this the session ends with the messages -// still queued rather than spinning forever. +// before any transcript output). Beyond this the prompt is restored to the +// composer rather than retried indefinitely. const MAX_UNACCEPTED_RETRIES = 3; -/** - * Per-turn spawn args for agy's headless print mode: - * - * agy -p --conversation --output-format stream-json \ - * --model --mode accept-edits|plan --effort \ - * --print-timeout 30m [--dangerously-skip-permissions] - * - * One spawn per user turn (the only official multi-turn headless channel is - * `--conversation` resume; the TUI is not used at all). The conversation id is - * learned from the `init` event on the first turn and persisted to session - * metadata, so every later turn resumes the same brain conversation. - */ +/** Build the persistent process arguments. Prompts travel only over stdin. */ export function buildAgyHeadlessArgs(opts: { - prompt: string; conversationId?: string; model?: string; permissionMode: PermissionMode; @@ -49,7 +37,7 @@ export function buildAgyHeadlessArgs(opts: { if (selection.effort && opts.effort && selection.effort !== opts.effort) { throw new Error(`Antigravity model ${opts.model} conflicts with effort ${opts.effort}`); } - const args = ['-p', opts.prompt, '--output-format', 'stream-json']; + const args = ['--input-format', 'stream-json', '--output-format', 'stream-json']; if (opts.conversationId) { args.push('--conversation', opts.conversationId); } @@ -70,7 +58,7 @@ export function buildAgyHeadlessArgs(opts: { } /** - * Build the spawn env for the per-turn agy process. Mirrors the model-probe + * Build the spawn env for the persistent agy process. Mirrors the model-probe * hardening (agyModels.ts): GEMINI_FORCE_FILE_STORAGE makes agy read the saved * OAuth file token directly instead of the keyring. * @@ -95,28 +83,18 @@ export type AgyHeadlessDriverOptions = { spawnAgy?: (args: string[], cwd: string) => ChildProcessWithoutNullStreams; }; -/** - * Headless print-mode transport for agy (replaces the PTY/TUI wrapper). - * - * Turn loop: - * 1. wait for a queued user message (MessageQueue2) - * 2. spawn `agy -p --conversation --output-format stream-json …` - * 3. stream NDJSON stdout through parseAgyNdjsonLine, mapping events onto the - * existing transcript-entry channel (session.client.sendAgySessionMessage) - * 4. kill the child on abort; the conversation id from the `init` envelope is - * adopted into session metadata (onSessionFound) so resume works - */ +/** Persistent NDJSON transport. One process serves sequential turns until its + * queued launch config changes, it fails, or the user interrupts it. */ export class AgyHeadlessDriver extends RemoteLauncherBase { private readonly session: AgySession; private readonly spawnAgy: (args: string[], cwd: string) => ChildProcessWithoutNullStreams; private readonly syncNativeTitle: (title: unknown) => void; /** Aborted on exit/switch/kill so an idle queue wait resolves immediately. */ private readonly loopAbortController = new AbortController(); - private child: ChildProcessWithoutNullStreams | null = null; + private process: AgyStreamProcess | null = null; private turnAbortController: AbortController | null = null; /** Brain conversation UUID — from resume seed or the first turn's init event. */ private conversationId: string | null; - private stderrTail = ''; constructor(opts: AgyHeadlessDriverOptions) { super(process.env.DEBUG ? opts.session.logPath : undefined); @@ -129,7 +107,7 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { }) as ChildProcessWithoutNullStreams; return child; }); - this.syncNativeTitle = createNativeSessionTitleMetadataSync(this.session.client); + this.syncNativeTitle = createNativeSessionTitleMetadataSync(this.session.client); this.conversationId = this.session.sessionId; this.session.setKillHandler(() => this.abort()); this.session.cancelRetryDelivery = this.cancelRetryItem; @@ -140,8 +118,8 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { } /** - * A requeued (unaccepted) delivery waiting out the retry backoff. finishTurn - * clears activeWebPrompt/activeLocalIds when the turn closes, so whole-session + * A requeued (unaccepted) delivery waiting out the retry backoff. Finalization + * clears activeWebPrompt/activeLocalIds when a turn finishes, so whole-session * shutdown during the backoff would otherwise find nothing to restore and the * session-end sweep would stamp the unacknowledged row invoked. Items keep * per-message localIds so a Cancel during the backoff can remove individual @@ -216,7 +194,7 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { /** * Snapshot the in-flight delivery and restore it to the composer (consuming * the hub row so it is neither replayed nor swept). Must run BEFORE any kill: - * the child close handler clears both fields via finishTurn. + * turn finalization clears both fields via resetTurn. */ private snapshotAndRestoreDelivery(): void { const prompt = this.activeWebPrompt; @@ -230,8 +208,8 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { this.session.client.sendSessionEvent({ type: 'abort-restore', text: prompt }); } - /** True while the close handler is finalizing a completed turn (parsing the - * final record, awaiting model resolution / transcript sends). During this + /** True while an accepted turn is finalizing its record and awaiting + * model resolution / transcript sends. During this * window the delivery is already accepted, so Stop/kill must not restore or * abort it. */ private turnFinalizing = false; @@ -251,7 +229,7 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { private async abort(): Promise { this.loopAbortController.abort(); if (this.turnCompleted || this.turnFinalizing) { - // The turn is accepted (result parsed) or finalizing (close handler + // The turn is accepted (result parsed) or finalizing (turn finalizer // awaiting model resolution / transcript sends): wait for it so the // final response and acks land before the API session closes. No // restore/abort — the prompt completed. Mark the interrupt on the @@ -267,7 +245,7 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { this.restoreRetryDelivery(); this.turnAbortController?.abort(); await this.terminateChild(); - // Let the close handler finish (its restore/ack work is already decided + // Let turn finalization finish (its restore/ack work is already decided // by the snapshot above; the turn promise must settle before teardown // closes the API session). await this.activeTurn; @@ -293,15 +271,28 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { * TERM-to-KILL escalation so session shutdown cannot exit before the tree is * actually gone. */ - private async terminateChild(): Promise { - const child = this.child; - if (!child || child.exitCode !== null) return; - const terminated = await killProcessByChildProcess(child); - if (!terminated && child.exitCode === null) { - // Last resort if tree-kill reports failure; the close handler is the - // real exit authority either way. - if (!child.killed) child.kill('SIGKILL'); + private async terminateChild(graceful = false): Promise { + const process = this.process; + if (!process) return; + await process.retire(graceful); + if (this.process === process) this.process = null; + } + + private async getOrStartProcess(args: string[], signal: AbortSignal): Promise { + // Conversation ids are discovered after launch and are deliberately not + // part of the config key: adopting init must not restart the warm process. + const configKey = JSON.stringify(args); + if (this.process && (this.process.configKey !== configKey || !this.process.reusable)) { + await this.terminateChild(true); } + if (signal.aborted || this.loopAbortController.signal.aborted || this.exitReason) return null; + if (!this.process) { + const spawnArgs = this.conversationId ? [...args, '--conversation', this.conversationId] : args; + // Prompts are never included in args or debug logs. + logger.debug(`[agy-headless] spawn: agy ${spawnArgs.join(' ')}`); + this.process = new AgyStreamProcess(this.spawnAgy(spawnArgs, this.session.path), configKey); + } + return this.process; } private async handleAbortRequest(): Promise { @@ -310,16 +301,16 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { // accepted its delivery — do not restore or abort it (resending could // repeat tool side effects). if (this.turnCompleted || this.turnFinalizing) { - // Mark the interrupt on the controller so the close handler reports + // Mark the interrupt on the controller so turn finalization reports // 'Turn aborted' (or nothing when the result already landed) instead // of a process failure. Do NOT restore — the prompt completed. this.turnAbortController?.abort(); await this.terminateChild(); return; } - // Snapshot BEFORE the kill: the child close handler (triggered by the + // Snapshot BEFORE the kill: turn finalization (triggered by process // termination) resolves the turn and clears activeWebPrompt/activeLocalIds - // via finishTurn — reading them after the wait would find nothing. + // via resetTurn — reading them after the wait would find nothing. this.snapshotAndRestoreDelivery(); this.turnAbortController?.abort(); @@ -342,8 +333,8 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { }); } - private describeFailure(code: number | null): string { - const stderr = this.stderrTail.trim(); + private describeFailure(code: number | null, stderrTail: string): string { + const stderr = stderrTail.trim(); if (/authentication|not signed in|keyring/i.test(stderr)) { return 'agy failed to authenticate. Ensure the login keyring is unlocked.'; } @@ -361,43 +352,19 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { /** Local ids of the in-flight delivery (for abort-time acknowledgement). */ private activeLocalIds: string[] = []; - /** - * True when the last turn's stderr carried a soft-deny notice (agy rejects a - * tool call under request-review because no allow-rule matches). Surfaced as - * a chat hint so the user can approve and resend. - */ - private lastTurnSoftDenied = false; - - /** - * Run a single turn: spawn agy with the queued prompt, stream NDJSON - * events onto the transcript-entry channel, wait for the `result` envelope - * or process exit. Returns whether agy accepted the prompt (user_input - * step seen or result envelope received). Aborting the returned - * controller's signal interrupts only this turn (the session loop stays - * alive for the next queued message). - */ + /** Serve one queued delivery; only native activity/result proves acceptance. */ private async runTurn( prompt: string, localIds: string[], - // The spawn config snapshot captured when the batch was queued — the - // session's live mode may have changed before dequeue, and a batch queued - // under request-review must never gain --dangerously-skip-permissions, - // nor must a queued prompt run on a model selected after it was sent. mode: AgyMode, signal: AbortSignal, ): Promise { - const onAbort = () => { - void this.terminateChild(); - }; + const onAbort = () => { void this.terminateChild(); }; signal.addEventListener('abort', onAbort, { once: true }); - const sessionModel = mode.model ?? undefined; const sessionEffort = mode.effort ?? undefined; - // Snapshot the model at turn start: a mid-response switch must not change - // attribution within one turn. const variantModel = sessionModel && sessionEffort && !splitAgyModelSelection(sessionModel).effort - ? `${sessionModel}-${sessionEffort}` - : sessionModel; + ? `${sessionModel}-${sessionEffort}` : sessionModel; const variantLabel = variantModel ? AGY_MODEL_LABELS[variantModel as keyof typeof AGY_MODEL_LABELS] : undefined; const turnModel = typeof sessionModel === 'string' && sessionModel ? variantLabel ?? AGY_MODEL_LABELS[sessionModel as keyof typeof AGY_MODEL_LABELS] @@ -405,55 +372,57 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { : undefined; let args: string[]; try { + if (!prompt.trim()) throw new Error('Antigravity requires a non-empty prompt'); args = buildAgyHeadlessArgs({ - prompt, - conversationId: this.conversationId ?? undefined, model: sessionModel, permissionMode: mode.permissionMode, effort: (AGY_EFFORT_LEVELS as readonly unknown[]).includes(sessionEffort) - ? sessionEffort as AgyEffortLevel - : undefined, + ? sessionEffort as AgyEffortLevel : undefined, }); } catch (error) { - // Invalid legacy config must not lose a dequeued delivery or retry - // the same deterministic failure. Restore it for the user to resend. if (localIds.length > 0) this.session.client.emitMessagesConsumed(localIds); this.session.client.sendSessionEvent({ type: 'error', message: error instanceof Error ? error.message : String(error) }); this.session.client.sendSessionEvent({ type: 'abort-restore', text: prompt }); signal.removeEventListener('abort', onAbort); return true; } - // Never persist the user's prompt to the log (HAPI debug logs can be - // forwarded under remote-debug configuration); keep the spawn shape. - const safeArgs = args.map((arg, index) => (args[index - 1] === '-p' ? '' : arg)); - logger.debug(`[agy-headless] spawn: agy ${safeArgs.map((a) => (a.length > 80 ? `${a.slice(0, 80)}…` : a)).join(' ')}`); this.session.onThinkingChange(true); - this.stderrTail = ''; this.activeWebPrompt = prompt; this.activeLocalIds = localIds; - this.lastTurnSoftDenied = false; this.turnFinalizing = false; this.turnCompleted = false; + const resetTurn = () => { + this.session.onThinkingChange(false); + this.activeWebPrompt = null; + this.activeLocalIds = []; + this.turnFinalizing = false; + this.turnCompleted = false; + signal.removeEventListener('abort', onAbort); + }; + let stream: AgyStreamProcess; + try { + const process = await this.getOrStartProcess(args, signal); + if (!process) { + this.snapshotAndRestoreDelivery(); + resetTurn(); + return false; + } + stream = process; + } catch (error) { + this.session.client.sendSessionEvent({ type: 'error', message: `Failed to start agy: ${error instanceof Error ? error.message : String(error)}` }); + resetTurn(); + return false; + } - // Per-turn planner: partial text must never leak into a later turn's - // response if this turn is aborted or crashes mid-stream. + stream.stderrTail = ''; const planner = new AgyPlannerAccumulator(); - - const child = this.spawnAgy(args, this.session.path); - this.child = child; - child.stdout.setEncoding('utf8'); - child.stderr.setEncoding('utf8'); - - let turnDone = false; let deliveryAcked = false; - let stdoutBuffer = ''; let accepted = false; let resultFailure: string | null = null; let sawResult = false; // Flush every pending planner step (result envelope AND turn close, so // prose emitted before a pre-result crash is not lost). - let plannerResponseSent = false; let lastPlannerContent: string | null = null; // Best-effort model resolution for default-model turns: when no explicit // --model was selected, the actual generation model is read from agy's @@ -489,29 +458,26 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { // reorder entries (a step with no DB lookup resolves before an earlier one // that queries the conversation DB; a tool emit could overtake a pending // planner emit from the same stdout chunk). The chain preserves the event - // order on the wire, and the close handler awaits it before resolving. + // order on the wire, and turn finalization awaits it before resolving. let sendChain: Promise = Promise.resolve(); const enqueueTranscript = (send: () => void | Promise): void => { sendChain = sendChain.then(send); }; const sendPlanner = (entry: AgyTranscriptEntry) => { - plannerResponseSent = true; if (typeof entry.content === 'string') lastPlannerContent = entry.content; enqueueTranscript(async () => { // Carry the turn's model (display label) on every planner entry so // the web's per-turn attribution survives model switches and // default-model turns still show the actual generation model. const model = await resolveModel(entry); - this.session.client.sendAgySessionMessage( + await this.session.client.sendAgySessionMessage( model ? { ...entry, model } : entry, this.conversationId ?? undefined, ); }); }; const sendTool = (entry: AgyTranscriptEntry, toolCall: AgyToolCall) => { - enqueueTranscript(() => { - this.session.client.sendAgySessionMessage(entry, this.conversationId ?? undefined, toolCall); - }); + enqueueTranscript(() => this.session.client.sendAgySessionMessage(entry, this.conversationId ?? undefined, toolCall)); }; const flushPlanner = async () => { for (const entry of planner.flushAll()) { @@ -520,280 +486,206 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { } await sendChain; }; - const finishTurn = () => { - if (turnDone) return; - turnDone = true; - this.session.onThinkingChange(false); - this.child = null; - this.activeWebPrompt = null; - this.activeLocalIds = []; - this.turnFinalizing = false; - this.turnCompleted = false; - signal.removeEventListener('abort', onAbort); + let softDenyReported = false; + const reportSoftDeny = () => { + if (softDenyReported || !/auto-denied|permissions\.allow|allow-rule/i.test(stream.stderrTail)) return; + softDenyReported = true; + const match = /(?:run_command|\S+)\s+auto-denied[^\n]*/i.exec(stream.stderrTail); + const detail = match?.[0] ? `: ${match[0].trim()}` : ''; + this.session.client.sendSessionEvent({ + type: 'message', + message: `A tool call was auto-denied (no allow-rule)${detail}. Approve it in agy's settings.json or switch to always-proceed, then resend.`, + }); }; - - try { - await new Promise((resolve) => { - const resolveTurn = () => resolve(); - - // Parse a complete NDJSON line. Extracted so both the streaming - // data handler and the close handler (EOF without trailing newline) - // can parse buffered records. - const handleLine = (line: string): void => { - if (!line.trim()) return; - const event = parseAgyNdjsonLine(line); - // Resume can silently fail: agy may create a REPLACEMENT - // conversation when the seeded --conversation id is stale. - // The streamed id is authoritative then — adopt it so later - // turns stop passing the stale value (the old PTY recovery - // handled this explicitly). - const adoptStreamConversationId = (id: string): void => { - if (!id || id === this.conversationId) return; - this.conversationId = id; - this.session.onSessionFound(id); - // emitSessionReady signals transport/session readiness (hub - // row resolvable); the user-facing 'ready' chat event is - // sent AFTER the turn completes (see runMainLoop) — sending - // it here fires while thinking is still active and would - // never fire on later turns (same conversation id). - this.session.client.emitSessionReady(); - void this.syncTitleIfKnown(); - }; - switch (event.kind) { - case 'init': { - // The conversation id is known from the very first - // envelope — adopt it immediately (and replace a - // stale resume seed when agy started fresh). - if (event.conversationId) { - adoptStreamConversationId(event.conversationId); - } - break; - } - case 'user-input': { - // NDJSON's user_input step is the authoritative - // delivery confirmation (replaces PTY echo matching). - if (event.conversationId) { - adoptStreamConversationId(event.conversationId); - } - accepted = true; - if (!deliveryAcked && localIds.length > 0) { - deliveryAcked = true; - this.session.client.emitMessagesConsumed(localIds); - } - break; - } - case 'planner-delta': { - if (event.conversationId) { - adoptStreamConversationId(event.conversationId); - } - // Any emitted prose/tool activity proves the prompt - // was accepted and the turn is running — a later - // pre-result failure must NOT trigger a retry that - // re-executes destructive tool effects. - accepted = true; - const entry = planner.feedDelta(event.stepIndex, event.delta, event.isDone); - if (entry) { - void sendPlanner(entry); - } - break; - } - case 'tool': { - if (event.conversationId) { - adoptStreamConversationId(event.conversationId); - } - // Only the DONE state carries the result; the ACTIVE - // line is the invocation start (parameters only). - accepted = true; - if (event.isDone) { - sendTool(event.entry, event.toolCall); - } - break; - } - case 'checkpoint': - case 'ignored': - break; - case 'result': { - sawResult = true; - // Fallback conversation-id adoption: if the init - // line was absent/malformed, the result envelope is - // authoritative — without it the next turn would - // start a fresh agy conversation and lose context. - // (adoptStreamConversationId also replaces a stale - // resume seed here.) - if (event.conversationId) { - adoptStreamConversationId(event.conversationId); - } - // Flush any planner text that never reached a DONE - // line (stream ended / response truncated). A - // SUCCESS result carries the authoritative complete - // answer: when unfinished deltas were force-flushed, - // the last pending step is replaced by the result - // response instead of discarding it or leaving a - // partial ("hel" → "hello"). - const pending = planner.flushAll(); - if (event.status === 'SUCCESS' && event.response?.trim() && pending.length > 0) { - for (const entry of pending.slice(0, -1)) { - void sendPlanner(entry); - } - void sendPlanner({ ...pending[pending.length - 1]!, content: event.response }); - } else { - for (const entry of pending) { - void sendPlanner(entry); - } - } - accepted = true; - // The authoritative result arrived: seal the turn so - // a Stop/kill before child close cannot restore it. - this.turnCompleted = true; - // A FAILURE envelope proves the prompt ran but the - // turn failed; surface it (exit code may still be 0). - if (event.status !== 'SUCCESS') { - // The reason lives in `error`; `response` is the - // answer field and stays filled on a failed turn, - // so reading it here repeated the whole answer in - // the chat as if it were the failure text. - resultFailure = event.error?.trim() - || `agy turn failed: ${event.status}`; - } else if (event.response?.trim()) { - const response = event.response.trim(); - // Deliver the authoritative final answer unless - // it was already emitted verbatim (trimmed — the - // envelope often carries a trailing newline the - // delta stream does not): a tool turn with - // completed pre-tool narration but no final - // agent_response would otherwise lose it. - if (lastPlannerContent?.trim() !== response) { - void sendPlanner({ - step_index: -1, - source: 'MODEL', - type: 'PLANNER_RESPONSE', - status: 'DONE', - created_at: '', - content: response, - }); - } - } - void this.syncTitleIfKnown(); - // Do NOT resolve here: the child's close handler is - // the turn boundary. Resolving early would let the - // queue loop spawn the next turn before this child - // has closed, racing shared turnAbortController / - // stderrTail state and briefly running two processes - // against the same conversation. - break; - } - } - }; - - child.stdout.on('data', (chunk: string) => { - // stdout chunks are arbitrary and do not preserve line - // boundaries: buffer partial lines across chunks before - // parsing, or a JSON object split across chunks would be - // parsed as two malformed lines (losing init ids, delivery - // acks, tool output, or assistant text). - stdoutBuffer += chunk; - const lines = stdoutBuffer.split('\n'); - stdoutBuffer = lines.pop() ?? ''; - for (const line of lines) { - handleLine(line); + let finalization: Promise | null = null; + let resolveTurn!: () => void; + let rejectTurn!: (error: unknown) => void; + const completed = new Promise((resolve, reject) => { resolveTurn = resolve; rejectTurn = reject; }); + const finalize = (code?: number | null): void => { + if (finalization) return; + // Accepted close/result finalization is a barrier: Stop cannot + // restore it, and neither a later prompt nor API teardown may pass it. + this.turnFinalizing = accepted; + finalization = (async () => { + await flushPlanner(); + if (!deliveryAcked && accepted && localIds.length > 0) { + deliveryAcked = true; + this.session.client.emitMessagesConsumed(localIds); + } + if (signal.aborted) { + if (!sawResult) this.session.client.sendSessionEvent({ type: 'message', message: 'Turn aborted' }); + } else if (resultFailure) { + this.session.client.sendSessionEvent({ type: 'error', message: resultFailure }); + } else if (!sawResult && stream.failure) { + this.session.client.sendSessionEvent({ type: 'error', message: stream.failure }); + } else if (!sawResult && code !== undefined && code !== 0) { + this.session.client.sendSessionEvent({ type: 'error', message: this.describeFailure(code, stream.stderrTail) }); + } else if (accepted && !sawResult) { + this.session.client.sendSessionEvent({ type: 'error', message: 'agy exited before returning a result' }); + } else reportSoftDeny(); + // Failed results may be followed by the native process exiting. + // Retire before releasing the turn so the next write cannot race it. + if (stream.retiring) await stream.retire(true); + })(); + void finalization.then(resolveTurn, rejectTurn); + }; + const handleLine = (line: string): void => { + if (!line.trim() || sawResult || finalization) return; + const event = parseAgyNdjsonLine(line); + // Resume can silently fail: agy may create a REPLACEMENT + // conversation when the seeded --conversation id is stale. + // The streamed id is authoritative then — adopt it so later + // turns stop passing the stale value (the old PTY recovery + // handled this explicitly). + const adoptStreamConversationId = (id: string): void => { + if (!id || id === this.conversationId) return; + this.conversationId = id; + this.session.onSessionFound(id); + // emitSessionReady signals transport/session readiness (hub + // row resolvable); the user-facing 'ready' chat event is + // sent AFTER the turn completes (see runMainLoop) — sending + // it here fires while thinking is still active and would + // never fire on later turns (same conversation id). + this.session.client.emitSessionReady(); + void this.syncTitleIfKnown(); + }; + switch (event.kind) { + case 'init': { + // The conversation id is known from the very first + // envelope — adopt it immediately (and replace a + // stale resume seed when agy started fresh). + if (event.conversationId) { + adoptStreamConversationId(event.conversationId); } - }); - - child.stderr.on('data', (chunk: string) => { - this.stderrTail = `${this.stderrTail}${chunk}`.slice(-4096); - }); - // Swallow pipe errors (e.g. ECONNRESET when we kill the child); - // the close handler is the real exit authority. - child.stdout.on('error', () => {}); - child.stderr.on('error', () => {}); - - let spawnFailed = false; - child.on('error', (err) => { - spawnFailed = true; - logger.debug(`[agy-headless] spawn error: ${err.message}`); - this.session.client.sendSessionEvent({ type: 'error', message: `Failed to start agy: ${err.message}` }); - // Node's ChildProcess emits close AFTER error; do NOT resolve - // here — close remains the sole turn boundary, so a late close - // cannot mutate shared state belonging to the next retry. - }); - - child.on('close', async (code) => { - logger.debug(`[agy-headless] agy exited with code ${code}`); - // EOF without a trailing newline: the final buffered record - // (result envelope, replacement id, failure status) would - // otherwise be discarded. Parse it FIRST — acceptance may be - // established by that record. - if (stdoutBuffer.trim()) { - handleLine(stdoutBuffer); - stdoutBuffer = ''; + break; + } + case 'user-input': { + // NDJSON's user_input step is the authoritative + // delivery confirmation (replaces PTY echo matching). + if (event.conversationId) { + adoptStreamConversationId(event.conversationId); } - // Seal the delivery ONLY when the turn was accepted: a close - // with no acceptance signal (crash, spawn error) must remain - // restorable/retryable — a concurrent archive/kill would - // otherwise skip snapshotAndRestoreDelivery and lose the prompt. - this.turnFinalizing = accepted; - // A close before the result envelope (crash, kill) must still - // deliver any prose emitted so far — and ALL pending emits - // (planner + tool) must land before the turn resolves, or a - // later turn could overtake them. - await flushPlanner(); - await sendChain; - // Ack whenever agy accepted the prompt — the user_input step - // or a result envelope (SUCCESS or failure: either proves the - // prompt was received, so the hub row must not stay stale), - // regardless of the exit code. - if (!deliveryAcked && accepted && localIds.length > 0) { + accepted = true; + if (!deliveryAcked && localIds.length > 0) { deliveryAcked = true; this.session.client.emitMessagesConsumed(localIds); } - if (this.turnAbortController?.signal.aborted) { - // A deliberate Stop (the controller is marked aborted, and - // a killed child closes with code null) is NOT an agy - // crash: report it as an interrupt, unless the result - // already landed (then the turn completed normally). - if (!sawResult) { - this.session.client.sendSessionEvent({ type: 'message', message: 'Turn aborted' }); - } - } else if (code !== 0 && !spawnFailed) { - // spawnFailed already surfaced the real cause via the - // error handler; avoid a second generic exit error. - this.session.client.sendSessionEvent({ - type: 'error', - message: this.describeFailure(code), - }); - } else if (resultFailure) { - // Exit 0 with a FAILURE result envelope: the turn ran but - // failed (e.g. quota, model error). Ack the delivery (it - // was accepted) but surface the failure visibly. - this.session.client.sendSessionEvent({ - type: 'error', - message: resultFailure, - }); - } else if (accepted && !sawResult) { - // The prompt was accepted (user_input/planner/tool) but - // agy exited without a result envelope: the turn ended - // with no answer. Ack the delivery (it ran) but surface - // the truncation instead of silently consuming it. - this.session.client.sendSessionEvent({ - type: 'error', - message: 'agy exited before returning a result', - }); - } else if (/auto-denied|permissions\.allow|allow-rule/i.test(this.stderrTail)) { - // Exit 0 with a soft-deny notice: surface a chat hint - // (headless has no mid-turn approval dialog). - this.lastTurnSoftDenied = true; - const match = /(?:run_command|\S+)\s+auto-denied[^\n]*/i.exec(this.stderrTail); - const detail = match?.[0] ? `: ${match[0].trim()}` : ''; - this.session.client.sendSessionEvent({ - type: 'message', - message: `A tool call was auto-denied (no allow-rule)${detail}. Approve it in agy's settings.json or switch to always-proceed, then resend.`, - }); + break; + } + case 'planner-delta': { + if (event.conversationId) { + adoptStreamConversationId(event.conversationId); } - resolveTurn(); - }); - }); + // Any emitted prose/tool activity proves the prompt + // was accepted and the turn is running — a later + // pre-result failure must NOT trigger a retry that + // re-executes destructive tool effects. + accepted = true; + const entry = planner.feedDelta(event.stepIndex, event.delta, event.isDone); + if (entry) { + void sendPlanner(entry); + } + break; + } + case 'tool': { + if (event.conversationId) { + adoptStreamConversationId(event.conversationId); + } + // Only the DONE state carries the result; the ACTIVE + // line is the invocation start (parameters only). + accepted = true; + if (event.isDone) { + sendTool(event.entry, event.toolCall); + } + break; + } + case 'checkpoint': + case 'ignored': + break; + case 'result': { + sawResult = true; + // Fallback conversation-id adoption: if the init + // line was absent/malformed, the result envelope is + // authoritative — without it the next turn would + // start a fresh agy conversation and lose context. + // (adoptStreamConversationId also replaces a stale + // resume seed here.) + if (event.conversationId) { + adoptStreamConversationId(event.conversationId); + } + // Flush any planner text that never reached a DONE + // line (stream ended / response truncated). A + // SUCCESS result carries the authoritative complete + // answer: when unfinished deltas were force-flushed, + // the last pending step is replaced by the result + // response instead of discarding it or leaving a + // partial ("hel" → "hello"). + const pending = planner.flushAll(); + if (event.status === 'SUCCESS' && event.response?.trim() && pending.length > 0) { + for (const entry of pending.slice(0, -1)) { + void sendPlanner(entry); + } + void sendPlanner({ ...pending[pending.length - 1]!, content: event.response }); + } else { + for (const entry of pending) { + void sendPlanner(entry); + } + } + accepted = true; + // The authoritative result arrived: seal the turn so + // a Stop/kill during finalization cannot restore it. + this.turnCompleted = true; + // A FAILURE envelope proves the prompt ran but the + // turn failed; surface it (exit code may still be 0). + if (event.status !== 'SUCCESS') { + // The reason lives in `error`; `response` is the + // answer field and stays filled on a failed turn, + // so reading it here repeated the whole answer in + // the chat as if it were the failure text. + resultFailure = event.error?.trim() + || `agy turn failed: ${event.status}`; + } else if (event.response?.trim()) { + const response = event.response.trim(); + // Deliver the authoritative final answer unless + // it was already emitted verbatim (trimmed — the + // envelope often carries a trailing newline the + // delta stream does not): a tool turn with + // completed pre-tool narration but no final + // agent_response would otherwise lose it. + if (lastPlannerContent?.trim() !== response) { + void sendPlanner({ + step_index: -1, + source: 'MODEL', + type: 'PLANNER_RESPONSE', + status: 'DONE', + created_at: '', + content: response, + }); + } + } + void this.syncTitleIfKnown(); + if (event.status !== 'SUCCESS') stream.retiring = true; + finalize(); + break; + } + } + }; + + stream.onLine = handleLine; + // stdout/stderr are independent pipes. A warning can arrive just after + // result; keep this reporter while idle, replacing it on the next turn. + stream.onStderr = reportSoftDeny; + const handleClose = (code: number | null) => finalize(code); + stream.onClose = handleClose; + try { + if (stream.closed) finalize(stream.closeCode); + else if (signal.aborted || stream.retiring) await stream.retire(); + else stream.writePrompt(prompt); + await completed; } finally { - finishTurn(); + if (stream.onLine === handleLine) stream.onLine = null; + if (stream.onClose === handleClose) stream.onClose = null; + resetTurn(); } return accepted; } @@ -831,6 +723,7 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { this.turnAbortController = turnController; const onLoopAbort = () => turnController.abort(); this.loopAbortController.signal.addEventListener('abort', onLoopAbort, { once: true }); + if (this.loopAbortController.signal.aborted) turnController.abort(); try { const turn = this.runTurn(msg.message, localIds, msg.mode, turnController.signal); this.activeTurn = turn; @@ -942,12 +835,15 @@ export class AgyHeadlessDriver extends RemoteLauncherBase { } } finally { this.clearAbortHandlers(session.client.rpcHandlerManager); + this.restoreRetryDelivery(); + await this.terminateChild(true); } logger.debug('[agy-headless]: main loop ended'); } protected async cleanup(): Promise { this.clearAbortHandlers(this.session.client.rpcHandlerManager); + await this.terminateChild(true); logger.debug('[agy-headless]: cleanup done'); } } diff --git a/cli/src/agy/headless/agyPersistentDriver.test.ts b/cli/src/agy/headless/agyPersistentDriver.test.ts new file mode 100644 index 00000000..65fd5ee2 --- /dev/null +++ b/cli/src/agy/headless/agyPersistentDriver.test.ts @@ -0,0 +1,380 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { EventEmitter } from 'node:events'; +import { PassThrough, Writable } from 'node:stream'; +import { tmpdir } from 'node:os'; +import type { ChildProcessWithoutNullStreams } from 'node:child_process'; +import { AgyHeadlessDriver } from './agyHeadlessDriver'; +import { AgySession } from '../session'; +import type { AgyMode } from '../types'; +import { MessageQueue2 } from '@/utils/MessageQueue2'; +import { RPC_METHODS } from '@hapi/protocol/rpcMethods'; + +const modelLookup = vi.hoisted(() => vi.fn(async (_id: unknown, indexes: readonly number[]) => ( + new Map(indexes.map((index) => [index, null])) +))); +vi.mock('../utils/agyConversationModel', () => ({ resolveAgyTurnModels: modelLookup })); +vi.mock('../utils/agySessionTitle', () => ({ readAgyConversationTitle: async () => null })); + +const CONVERSATION = '12345678-1234-4234-8234-123456789abc'; +const MODE: AgyMode = { permissionMode: 'request-review', model: 'gemini-4-argon', effort: 'medium' }; +const cleanups: Array<() => Promise> = []; +const releases: Array<() => void> = []; + +/** Pipes stay open after a result. Closing stdin or killing the process emits close once. */ +class PersistentChild extends EventEmitter { + readonly stdout = new PassThrough(); + readonly stderr = new PassThrough(); + readonly stdin: Writable; + readonly prompts: string[] = []; + exitCode: number | null = null; + killed = false; + closeOnStdinEnd = true; + onPrompt?: (prompt: string, callback: (error?: Error | null) => void) => void; + + constructor() { + super(); + this.stdin = new Writable({ + write: (chunk, _encoding, callback) => { + const line = JSON.parse(String(chunk)) as { event: string; message: { content: string } }; + expect(line.event).toBe('user'); + expect(String(chunk).endsWith('\n')).toBe(true); + this.prompts.push(line.message.content); + if (this.onPrompt) this.onPrompt(line.message.content, callback); + else callback(); + }, + final: (callback) => { + callback(); + if (this.closeOnStdinEnd) queueMicrotask(() => this.close(0)); + }, + }); + } + + get process(): ChildProcessWithoutNullStreams { + return this as unknown as ChildProcessWithoutNullStreams; + } + + feed(event: unknown): void { + this.stdout.write(JSON.stringify(event) + '\n'); + } + + accept(stepIndex = 0): void { + this.feed({ event: 'step_update', step_update: { + conversation_id: CONVERSATION, step_index: stepIndex, state: 'DONE', step_type: 'user_input', + } }); + } + + result(response: string): void { + this.feed({ event: 'result', result: { + conversation_id: CONVERSATION, status: 'SUCCESS', response, + } }); + } + + close(code = 0): void { + if (this.exitCode !== null) return; + this.exitCode = code; + this.stdout.end(); + this.stderr.end(); + this.stdin.destroy(); + this.emit('close', code, null); + } + + kill(): boolean { + this.killed = true; + this.close(0); + return true; + } +} + +function createHarness(configureChild?: (child: PersistentChild, index: number) => void) { + const sent: unknown[][] = []; + const client = { + sendAgySessionMessage: (...args: unknown[]) => { sent.push(args); }, + emitMessagesConsumed: vi.fn(), + emitSessionReady: vi.fn(), + sendSessionEvent: vi.fn(), + updateMetadata: vi.fn(), + keepAlive: vi.fn(), + rpcHandlerManager: { registerHandler: vi.fn() }, + }; + const queue = new MessageQueue2((mode) => JSON.stringify(mode)); + const session = new AgySession({ + api: {} as never, client: client as never, path: tmpdir(), + logPath: `${tmpdir()}/agy-persistent.log`, sessionId: null, + messageQueue: queue, onModeChange: () => {}, startedBy: 'runner', + }); + const children: PersistentChild[] = []; + const spawnAgy = vi.fn((args: string[]) => { + const child = new PersistentChild(); + configureChild?.(child, children.length); + children.push(child); + return child.process; + }); + const driver = new AgyHeadlessDriver({ session, spawnAgy }); + const launch = driver.launch(); + const stop = async () => { + queue.close(); + session.stopKeepAlive(); + await session.kill(); + children.forEach((child) => child.close()); + await launch; + }; + cleanups.push(stop); + return { client, queue, session, children, spawnAgy, sent, launch }; +} + +function waitFor(check: () => void, timeout = 2500): Promise { + return vi.waitFor(check, { timeout, interval: 5 }); +} + +function readyCount(client: ReturnType['client']): number { + return client.sendSessionEvent.mock.calls.filter(([event]) => event.type === 'ready').length; +} + +async function flushTasks(): Promise { + for (let i = 0; i < 5; i++) await new Promise((resolve) => setImmediate(resolve)); +} + +afterEach(async () => { + releases.splice(0).forEach((release) => release()); + for (const cleanup of cleanups.splice(0)) await cleanup(); + modelLookup.mockReset(); + modelLookup.mockImplementation(async (_id, indexes) => new Map(indexes.map((index) => [index, null]))); +}); + +describe('Agy persistent stream lifecycle', () => { + it('runs consecutive turns on one living process, using result rather than close as the boundary', async () => { + const h = createHarness(); + h.queue.push('first', MODE, 'first-id'); + await waitFor(() => expect(h.children[0]?.prompts).toEqual(['first'])); + const child = h.children[0]!; + child.feed({ event: 'init', conversation_id: CONVERSATION, init: {} }); + child.result('first answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(1)); + expect(child.exitCode).toBeNull(); + expect(child.stdin.destroyed).toBe(false); + h.queue.push('second', MODE, 'second-id'); + await waitFor(() => expect(child.prompts).toEqual(['first', 'second'])); + child.result('second answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(2)); + expect(h.spawnAgy).toHaveBeenCalledTimes(1); + expect(h.client.emitMessagesConsumed.mock.calls).toEqual([[['first-id']], [['second-id']]]); + expect(h.sent.map(([entry]) => (entry as { content: string }).content)).toEqual(['first answer', 'second answer']); + }); + + it('finishes asynchronous transcript sends before writing the next queued turn', async () => { + let release!: (value: Map) => void; + const lookup = new Promise>((resolve) => { release = resolve; }); + releases.push(() => release(new Map([[2, 'Gemini 4 Argon']]))); + modelLookup.mockImplementationOnce(() => lookup); + const h = createHarness(); + const mode: AgyMode = { permissionMode: 'request-review' }; + h.queue.push('first', mode, 'first-id'); + await waitFor(() => expect(h.children[0]?.prompts).toHaveLength(1)); + const child = h.children[0]!; + child.feed({ event: 'init', conversation_id: CONVERSATION, init: {} }); + child.feed({ event: 'step_update', step_update: { + conversation_id: CONVERSATION, step_index: 2, state: 'DONE', + step_type: 'agent_response', text_delta: 'first answer', + } }); + h.queue.push('second', mode, 'second-id'); + child.result('first answer'); + await waitFor(() => expect(modelLookup).toHaveBeenCalled()); + await flushTasks(); + expect(child.prompts).toEqual(['first']); + expect(h.sent).toHaveLength(0); + release(new Map([[2, 'Gemini 4 Argon']])); + await waitFor(() => expect(child.prompts).toEqual(['first', 'second'])); + expect(h.sent).toHaveLength(1); + child.result('second answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(1)); + }); + + it.each([ + ['model', { ...MODE, model: 'gemini-3.8-flash' }, '--model', 'gemini-3.8-flash'], + ['effort', { ...MODE, effort: 'high' }, '--effort', 'high'], + ['permission', { ...MODE, permissionMode: 'always-proceed' }, '--dangerously-skip-permissions', undefined], + ] as const)('restarts on a queued %s change and resumes the adopted conversation', async (_name, nextMode, flag, value) => { + const h = createHarness(); + h.queue.push('first', MODE, 'first-id'); + await waitFor(() => expect(h.children[0]?.prompts).toHaveLength(1)); + const first = h.children[0]!; + first.result('first answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(1)); + h.queue.push('second', nextMode, 'second-id'); + await waitFor(() => expect(h.children[1]?.prompts).toEqual(['second'])); + expect(first.exitCode).not.toBeNull(); + const args = h.spawnAgy.mock.calls[1]![0]; + expect(args[args.indexOf('--conversation') + 1]).toBe(CONVERSATION); + expect(args).toContain(flag); + if (value !== undefined) expect(args[args.indexOf(flag) + 1]).toBe(value); + h.children[1]!.result('second answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(2)); + expect(h.spawnAgy).toHaveBeenCalledTimes(2); + }); + + it('restores the unsent prompt when Stop interrupts a config restart', async () => { + const h = createHarness(); + h.queue.push('first', MODE, 'first-id'); + await waitFor(() => expect(h.children[0]?.prompts).toHaveLength(1)); + const first = h.children[0]!; + first.result('first answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(1)); + first.closeOnStdinEnd = false; + h.queue.push('not yet sent', { ...MODE, effort: 'high' }, 'second-id'); + await waitFor(() => expect(first.stdin.writableEnded).toBe(true)); + expect(h.session.thinking).toBe(true); + const handler = h.client.rpcHandlerManager.registerHandler.mock.calls + .find(([method]) => method === RPC_METHODS.Abort)?.[1] as (() => Promise) | undefined; + expect(handler).toBeDefined(); + await handler!(); + await waitFor(() => expect(h.session.thinking).toBe(false)); + expect(h.spawnAgy).toHaveBeenCalledTimes(1); + expect(first.prompts).toEqual(['first']); + expect(h.client.sendSessionEvent).toHaveBeenCalledWith({ type: 'abort-restore', text: 'not yet sent' }); + expect(h.client.emitMessagesConsumed).toHaveBeenCalledWith(['second-id']); + }); + + it('recovers from an idle process crash without replaying the completed turn', async () => { + const h = createHarness(); + h.queue.push('first', MODE, 'first-id'); + await waitFor(() => expect(h.children[0]?.prompts).toHaveLength(1)); + h.children[0]!.result('first answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(1)); + h.children[0]!.close(1); + h.queue.push('second', MODE, 'second-id'); + await waitFor(() => expect(h.children[1]?.prompts).toEqual(['second'])); + expect(h.spawnAgy.mock.calls[1]![0]).toContain(CONVERSATION); + h.children[1]!.result('second answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(2)); + expect(h.children.map((child) => child.prompts)).toEqual([['first'], ['second']]); + }); + + it.each([false, true])('handles stdin EPIPE with accepted=%s without replaying accepted work', async (accepted) => { + const h = createHarness((child, index) => { + if (index !== 0) return; + child.onPrompt = (_prompt, callback) => { + if (accepted) child.accept(); + callback(Object.assign(new Error('broken pipe'), { code: 'EPIPE' })); + }; + }); + h.queue.push('first', MODE, 'first-id'); + await waitFor(() => expect(h.children[0]?.prompts).toHaveLength(1)); + if (accepted) { + await waitFor(() => expect(h.session.thinking).toBe(false)); + h.queue.push('second', MODE, 'second-id'); + } + await waitFor(() => expect(h.children[1]?.prompts).toEqual([accepted ? 'second' : 'first'])); + h.children[1]!.result('answer'); + await waitFor(() => expect(h.session.thinking).toBe(false)); + expect(h.spawnAgy).toHaveBeenCalledTimes(2); + expect(h.client.emitMessagesConsumed.mock.calls.filter(([ids]) => ids.includes('first-id'))).toHaveLength(1); + }); + + it('ignores an old process close while its replacement is running', async () => { + const h = createHarness(); + h.queue.push('first', MODE, 'first-id'); + await waitFor(() => expect(h.children[0]?.prompts).toHaveLength(1)); + const oldChild = h.children[0]!; + oldChild.result('first answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(1)); + h.queue.push('second', { ...MODE, effort: 'high' }, 'second-id'); + await waitFor(() => expect(h.children[1]?.prompts).toHaveLength(1)); + oldChild.emit('close', 1, null); + await flushTasks(); + expect(h.session.thinking).toBe(true); + h.children[1]!.result('second answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(2)); + expect(h.client.emitMessagesConsumed).toHaveBeenCalledWith(['second-id']); + }); + + it('kills an idle warm process and stops the queue loop', async () => { + const h = createHarness(); + h.queue.push('first', MODE, 'first-id'); + await waitFor(() => expect(h.children[0]?.prompts).toHaveLength(1)); + h.children[0]!.result('answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(1)); + await h.session.kill(); + await h.launch; + expect(h.children[0]!.exitCode).not.toBeNull(); + h.queue.push('never sent', MODE, 'later-id'); + await flushTasks(); + expect(h.children[0]!.prompts).toEqual(['first']); + expect(h.spawnAgy).toHaveBeenCalledTimes(1); + }); + + it.each([false, true])('preserves a delivery before kill returns when queue wakeup races shutdown (warm=%s)', async (warm) => { + const h = createHarness(); + if (warm) { + h.queue.push('first', MODE, 'first-id'); + await waitFor(() => expect(h.children[0]?.prompts).toHaveLength(1)); + h.children[0]!.result('answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(1)); + } + h.queue.push('racing delivery', MODE, 'second-id'); + await h.session.kill(); + // Runner lifecycle may close the API immediately after kill resolves. + const restored = h.client.sendSessionEvent.mock.calls.some(([event]) => ( + event.type === 'abort-restore' && event.text === 'racing delivery' + )); + expect(restored || h.queue.pendingLocalIds().includes('second-id')).toBe(true); + if (restored) expect(h.client.emitMessagesConsumed).toHaveBeenCalledWith(['second-id']); + await h.launch; + expect(h.children.flatMap((child) => child.prompts)).toEqual(warm ? ['first'] : []); + expect(h.spawnAgy).toHaveBeenCalledTimes(warm ? 1 : 0); + }); + + it.each(['', ' \n\t '])('rejects empty prompts without retiring the warm process (%j)', async (prompt) => { + const h = createHarness(); + h.queue.push('first', MODE, 'first-id'); + await waitFor(() => expect(h.children[0]?.prompts).toHaveLength(1)); + const child = h.children[0]!; + child.result('first answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(1)); + h.queue.push(prompt, MODE, 'invalid-id'); + await waitFor(() => expect(readyCount(h.client)).toBe(2)); + expect(h.client.emitMessagesConsumed).toHaveBeenCalledWith(['invalid-id']); + expect(h.client.sendSessionEvent).toHaveBeenCalledWith({ type: 'abort-restore', text: prompt }); + expect(h.client.sendSessionEvent).toHaveBeenCalledWith({ + type: 'error', message: expect.stringContaining('non-empty prompt'), + }); + expect(child.prompts).toEqual(['first']); + expect(child.exitCode).toBeNull(); + expect(child.killed).toBe(false); + h.queue.push('valid after rejection', MODE, 'third-id'); + await waitFor(() => expect(child.prompts).toEqual(['first', 'valid after rejection'])); + child.result('third answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(3)); + expect(h.spawnAgy).toHaveBeenCalledTimes(1); + }); + + it('retires and retries after a synchronous stdin.write throw before acceptance', async () => { + const h = createHarness((child, index) => { + if (index === 0) vi.spyOn(child.stdin, 'write').mockImplementationOnce(() => { + throw new Error('synchronous write failure'); + }); + }); + h.queue.push('retry me', MODE, 'first-id'); + await waitFor(() => expect(h.children[1]?.prompts).toEqual(['retry me'])); + expect(h.children[0]!.exitCode).not.toBeNull(); + expect(h.children[0]!.prompts).toEqual([]); + expect(h.client.emitMessagesConsumed).not.toHaveBeenCalled(); + h.children[1]!.result('answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(1)); + expect(h.client.emitMessagesConsumed.mock.calls).toEqual([[['first-id']]]); + expect(h.spawnAgy).toHaveBeenCalledTimes(2); + }); + + it('does not restore or abort a completed turn when a process error arrives afterward', async () => { + const h = createHarness(); + h.queue.push('first', MODE, 'first-id'); + await waitFor(() => expect(h.children[0]?.prompts).toHaveLength(1)); + h.children[0]!.result('answer'); + await waitFor(() => expect(readyCount(h.client)).toBe(1)); + h.children[0]!.emit('error', new Error('late process failure')); + h.children[0]!.close(1); + await flushTasks(); + expect(h.client.sendSessionEvent.mock.calls.filter(([event]) => event.type === 'abort-restore')).toEqual([]); + expect(h.client.sendSessionEvent.mock.calls.filter(([event]) => event.message === 'Turn aborted')).toEqual([]); + expect(h.client.emitMessagesConsumed.mock.calls).toEqual([[['first-id']]]); + }); +}); diff --git a/cli/src/agy/headless/agyStreamProcess.ts b/cli/src/agy/headless/agyStreamProcess.ts new file mode 100644 index 00000000..0380be20 --- /dev/null +++ b/cli/src/agy/headless/agyStreamProcess.ts @@ -0,0 +1,106 @@ +import type { ChildProcessWithoutNullStreams } from 'node:child_process'; +import { killProcessByChildProcess } from '@/utils/process'; + +/** Process-owned pipes survive individual turns. Each listener closes over this + * instance so late output/close from a retired process cannot affect its replacement. */ +export class AgyStreamProcess { + onLine: ((line: string) => void) | null = null; + onClose: ((code: number | null) => void) | null = null; + onStderr: (() => void) | null = null; + stderrTail = ''; + failure: string | null = null; + closed = false; + closeCode: number | null = null; + retiring = false; + private stdoutBuffer = ''; + private readonly closedPromise: Promise; + private retirement: Promise | null = null; + private endGrace: (() => void) | null = null; + private forceRetirement = false; + + constructor(readonly child: ChildProcessWithoutNullStreams, readonly configKey: string) { + let resolveClosed!: () => void; + this.closedPromise = new Promise((resolve) => { resolveClosed = resolve; }); + child.stdout.setEncoding('utf8'); + child.stderr.setEncoding('utf8'); + child.stdout.on('data', (chunk: string) => { + if (this.closed) return; + this.stdoutBuffer += chunk; + const lines = this.stdoutBuffer.split('\n'); + this.stdoutBuffer = lines.pop() ?? ''; + for (const line of lines) this.onLine?.(line); + }); + child.stderr.on('data', (chunk: string) => { + if (this.closed) return; + this.stderrTail = `${this.stderrTail}${chunk}`.slice(-4096); + this.onStderr?.(); + }); + child.stdout.on('error', () => {}); + child.stderr.on('error', () => {}); + child.stdin.on('error', (error: Error) => this.failWrite(error)); + child.on('error', (error: Error) => { + this.failure ??= `Failed to start agy: ${error.message}`; + this.retiring = true; + // Node emits close after a spawn error. The turn stays attached until + // then, including when no process was created (ENOENT). + }); + child.on('close', (code: number | null) => { + if (this.closed) return; + // Preserve a valid final record even when EOF has no trailing newline. + if (this.stdoutBuffer.trim()) this.onLine?.(this.stdoutBuffer); + this.stdoutBuffer = ''; + this.closed = true; + this.closeCode = code; + this.retiring = true; + resolveClosed(); + this.onClose?.(code); + }); + } + + get reusable(): boolean { + return !this.closed && !this.retiring && this.child.exitCode === null && this.child.signalCode == null; + } + + writePrompt(prompt: string): void { + try { + this.child.stdin.write(JSON.stringify({ event: 'user', message: { content: prompt } }) + '\n', (error) => { + if (error) this.failWrite(error); + }); + } catch (error) { + this.failWrite(error instanceof Error ? error : new Error(String(error))); + } + } + + private failWrite(error: Error): void { + if (this.closed) return; + this.failure ??= `Failed to write to agy: ${error.message}`; + void this.retire(); + } + + retire(graceful = false): Promise { + this.retiring = true; + if (!graceful) { + this.forceRetirement = true; + this.endGrace?.(); + } + if (this.retirement) return this.retirement; + this.retirement = Promise.resolve().then(async () => { + if (this.closed) return; + if (graceful && !this.forceRetirement && !this.child.stdin.destroyed) { + try { this.child.stdin.end(); } catch { /* Fall through to tree kill. */ } + await new Promise((resolve) => { + const finish = () => { clearTimeout(timeout); this.endGrace = null; resolve(); }; + const timeout = setTimeout(finish, 500); + this.endGrace = finish; + void this.closedPromise.then(finish); + }); + } + if (!this.closed && this.child.exitCode === null && this.child.signalCode == null) { + const terminated = await killProcessByChildProcess(this.child); + if (!terminated && !this.closed) this.child.kill('SIGKILL'); + } + await this.closedPromise; + }); + return this.retirement; + } +} diff --git a/cli/src/agy/session.ts b/cli/src/agy/session.ts index 7e81309e..26218097 100644 --- a/cli/src/agy/session.ts +++ b/cli/src/agy/session.ts @@ -11,9 +11,9 @@ type LocalLaunchFailure = { }; /** - * Headless (print-mode) agy session. agy is spawned per user turn with - * `-p --conversation --output-format stream-json`; the driver - * maps NDJSON events onto the existing transcript-entry channel + * Headless agy session using a persistent stream-json stdin/stdout process. + * The driver reuses matching launch settings and resumes the conversation + * when the process restarts, mapping NDJSON onto the transcript-entry channel * (sendAgySessionMessage), so hub/web rendering is unchanged. */ export class AgySession extends AgentSessionBase { diff --git a/cli/src/agy/types.ts b/cli/src/agy/types.ts index e365a8e2..24e39a8e 100644 --- a/cli/src/agy/types.ts +++ b/cli/src/agy/types.ts @@ -4,9 +4,9 @@ import type { SessionEffort, SessionModel } from '@/api/types'; export type PermissionMode = AgyPermissionMode; /** - * Per-batch spawn config captured at enqueue time. MessageQueue2 batches + * Per-batch launch config captured at enqueue time. MessageQueue2 batches * messages by mode hash, so a prompt queued while the session runs on model A - * must keep spawning on A even if the user switches the session to B before + * must keep using A even if the user switches the session to B before * dequeue — otherwise the queued prompt silently runs on a different model * (and two prompts around the switch could be merged into one B turn). */ diff --git a/docs/guide/agents.md b/docs/guide/agents.md index c9a33c97..b01148a6 100644 --- a/docs/guide/agents.md +++ b/docs/guide/agents.md @@ -20,7 +20,7 @@ agent's integration; their supported syntax varies by agent. | Kimi | `hapi kimi` | ACP (`kimi acp`) | ✓ | ✓ | `default` `read-only` `safe-yolo` `yolo` | ✓ | | OpenCode | `hapi opencode` | ACP (`opencode acp`) | ✓ | ✓ | `default` `plan` `yolo` | ✓ | | DeepSeek Harness | `hapi dsh` | ACP (`dsh --profile acp` or configured server) | — | ✓ | Managed by DSH ACP composition | — | -| Antigravity (agy) | `hapi agy` | Headless print mode (per-turn `agy -p` + NDJSON) | — | ✓ | `request-review` `always-proceed` | ✓ | +| Antigravity (agy) | `hapi agy` | Persistent headless NDJSON over stdio | — | ✓ | `request-review` `always-proceed` | ✓ | | Pi | `hapi pi` | `pi --mode rpc` (JSON-line RPC over stdio) | — | ✓ | none (always auto-approve) | ✓ | | Gemini CLI | — | **Removed** — Google sunset the consumer Gemini CLI (2026-06-18) | — | — | — | — | @@ -308,7 +308,28 @@ the prompt is loaded by DSH through its own `skill` tool and catalog. - **GitHub Copilot** (`hapi copilot`) — Copilot CLI over ACP (`copilot --acp --stdio`). [GitHub Copilot](https://github.com/features/copilot) - **Kimi** (`hapi kimi`) — Moonshot AI's Kimi CLI over ACP (`kimi acp`). [MoonshotAI/kimi-cli](https://github.com/MoonshotAI/kimi-cli) - **OpenCode** (`hapi opencode`) — the open-source OpenCode agent over ACP (`opencode acp`). [opencode.ai](https://opencode.ai) -- **Antigravity** (`hapi agy`) — Google's Antigravity CLI (`agy`), driven headlessly via print mode: every turn spawns `agy -p --conversation --output-format stream-json`, and NDJSON events (init / step_update / result) are streamed into the chat. There is no PTY/TUI wrapper and no hook-based permission bridge: permission handling uses agy's own `settings.json` allow/deny rules (`request-review`) or `--dangerously-skip-permissions` (`always-proceed`). Tool calls that lack an allow-rule are auto-denied by agy and surfaced as a chat hint. MCP servers are configured the standard agy way — in the user's global `~/.gemini/config/mcp_config.json` or a workspace `.agents/mcp_config.json` — and are loaded natively by agy in headless mode (no HAPI injection). Remote-only — there is no local terminal input path. [Google Antigravity](https://antigravity.google) +- **Antigravity** (`hapi agy`) — Google's Antigravity CLI (`agy`), driven through + `agy --input-format stream-json --output-format stream-json`. HAPI starts the + process on the first prompt and reuses it for subsequent turns with the same + model, effort, and permission settings. Each prompt is sent through stdin; + the next turn starts after HAPI finishes handling the current `result` and + chat output. + Changing settings restarts the process with `--conversation `. Stop + terminates the process and its tools; the next send resumes the conversation. + A process failure also causes the next turn to start a fresh process, without + automatically replaying a prompt the agent already accepted. + + Requires an agy CLI with `--input-format stream-json` support. Native CLI + commands such as `/model` and `/usage` cannot run inside this stream; use + HAPI's model and effort controls to change settings. Remote-only — there is + no local terminal input path or PTY/TUI wrapper. + + Permission handling uses agy's `settings.json` allow/deny rules + (`request-review`) or `--dangerously-skip-permissions` (`always-proceed`), + without a hook-based approval bridge. Tools without an allow-rule are + auto-denied and shown as a chat hint. MCP servers are loaded natively from + `~/.gemini/config/mcp_config.json` or workspace `.agents/mcp_config.json`; + HAPI does not inject its own MCP server. [Google Antigravity](https://antigravity.google) In the web model picker, models with multiple effort variants appear once, with a separate effort selector showing the levels in the machine's catalog.