wushenghua/hapi
1
0
Fork 0
mirror of https://github.com/wu736139669/hapi.git synced 2026-11-08 01:43:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(agy): reuse a persistent headless stream process (#2248)

Browse Source
This commit is contained in:
SSU-WEI HUANG
2026-10-30 19:11:26 +08:00
committed by GitHub
parent 013ace8e88
commit f8e55fb930
7 changed files with 819 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/agy/headless/agyHeadlessDriver.test.ts
+35 -22
View File
@@ -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 <redacted> 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('<redacted>');
expect(allLogs).toContain('--input-format stream-json');
expect(allLogs).not.toContain('super-secret-prompt-text');
debugSpy.mockRestore();
});
cli/src/agy/headless/agyHeadlessDriver.ts
+270 -374
View File
@@ -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 <msg> --conversation <uuid> --output-format stream-json \
* --model <slug> --mode accept-edits|plan --effort <level> \
* --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 <msg> --conversation <uuid> --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<void> {
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<void> {
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<void> {
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<AgyStreamProcess | null> {
// 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<void> {
@@ -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<boolean> {
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' ? '<redacted>' : 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<void> = Promise.resolve();
const enqueueTranscript = (send: () => void | Promise<void>): 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<void>((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<void> | null = null;
let resolveTurn!: () => void;
let rejectTurn!: (error: unknown) => void;
const completed = new Promise<void>((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<void> {
this.clearAbortHandlers(this.session.client.rpcHandlerManager);
await this.terminateChild(true);
logger.debug('[agy-headless]: cleanup done');
}
}
cli/src/agy/headless/agyPersistentDriver.test.ts
+380
View File
@@ -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<number, string | null>(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<void>> = [];
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<AgyMode>((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<void> {
return vi.waitFor(check, { timeout, interval: 5 });
}
function readyCount(client: ReturnType<typeof createHarness>['client']): number {
return client.sendSessionEvent.mock.calls.filter(([event]) => event.type === 'ready').length;
}
async function flushTasks(): Promise<void> {
for (let i = 0; i < 5; i++) await new Promise<void>((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<number, string | null>) => void;
const lookup = new Promise<Map<number, string | null>>((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<void>) | 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']]]);
});
});
cli/src/agy/headless/agyStreamProcess.ts
+106
View File
@@ -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<void>;
private retirement: Promise<void> | 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<void> {
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<void>((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;
}
}
cli/src/agy/session.ts
+3 -3
View File
@@ -11,9 +11,9 @@ type LocalLaunchFailure = {
};
/**
* Headless (print-mode) agy session. agy is spawned per user turn with
* `-p <msg> --conversation <uuid> --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<AgyMode> {
cli/src/agy/types.ts
+2 -2
View File
@@ -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).
*/