From 278496e900fb419476e407308f253fdfd05be68a Mon Sep 17 00:00:00 2001 From: tonyzaca7755963 <245165781+tonyzaca7755963@users.noreply.github.com> Date: Mon, 26 Oct 2026 13:31:24 +0800 Subject: [PATCH 01/11] feat(antigravity): reuse one stream-json driver across successful turns (#535) Each turn.start previously spawned agy and every #completeTurn closed its stdin and waited for exit, so continuing a conversation repaid the full CLI, Language Server, and customization startup cost. The official headless protocol keeps one stream-json process and reads the next user message from the same stdin. - A successful Turn hands its process, Question bridge, and log to the Session; the next Turn rebinds the bridge to its own turn id, ownership check, and duplicate guard - Reused processes ignore the previous Turn's replayed result and steps, keyed on num_turns and step index, so history receives each Turn exactly once - A different model/effort/permission selection, a dead or unwritable process, or a Turn that failed, was cancelled, or closed the Session disposes the driver with bounded SIGTERM/SIGKILL escalation - Test fixtures now answer each stdin line like the real CLI instead of exiting after one Turn, and a new test asserts one PID across three Turns and a new one after model.select Co-Authored-By: Claude Code --- .../antigravity/src/antigravity-adapter.ts | 313 +++++++++++++----- .../antigravity/src/question-bridge.ts | 23 +- .../test/adversarial-fork-rollback.test.ts | 25 +- .../test/adversarial-stress.test.ts | 27 +- .../test/antigravity-adapter.test.ts | 204 ++++++++++-- .../adapters/antigravity/test/fork.test.ts | 25 +- .../antigravity/test/rollback.test.ts | 25 +- 7 files changed, 491 insertions(+), 151 deletions(-) diff --git a/packages/adapters/antigravity/src/antigravity-adapter.ts b/packages/adapters/antigravity/src/antigravity-adapter.ts index 1c3549074..23cf0cc57 100644 --- a/packages/adapters/antigravity/src/antigravity-adapter.ts +++ b/packages/adapters/antigravity/src/antigravity-adapter.ts @@ -92,7 +92,7 @@ import { type AntigravityCodeAction, } from "./code-action-diff.js"; import { fetchAntigravityQuota, type AntigravityQuotaSnapshot } from "./quota.js"; -import { AntigravityQuestionBridge } from "./question-bridge.js"; +import { AntigravityQuestionBridge, type BridgeOptions } from "./question-bridge.js"; import { AntigravitySubagents } from "./subagents.js"; import { nativeSubagentIdSchema, readSubagentTranscript } from "./subagent-transcript.js"; import { createAntigravityUsageStatistics } from "./usage-statistics.js"; @@ -136,6 +136,10 @@ interface ActiveTurn { questions: AntigravityQuestionBridge; subagents: AntigravitySubagents; logPath: string; + /** The native driver this Turn runs on; a successful Turn hands it to the next one. */ + transport: AntigravityTransport; + /** Steps below this index belong to an earlier Turn on a reused driver. */ + firstStepIndex: number; agentItem: HostAgentMessageItem | null; agentText: string; tools: Map; @@ -168,6 +172,30 @@ interface ActiveTurn { httpsPort: number | null; } +/** + * One agy stream-json driver. A successful Turn keeps it (and its Question bridge) + * alive for the next Turn of the same model/effort/permission selection, so + * continued conversations skip CLI, Language Server, and customization startup. + * Any other outcome disposes it. + */ +interface AntigravityTransport { + process: ChildProcessByStdio; + exited: Promise; + questions: AntigravityQuestionBridge; + logPath: string; + /** Model, effort, and permission mode this process was started with. */ + key: string; + nativePermissionMode: string | null; + httpsPort: number | null; + codeActionCursor: number; + /** Highest `num_turns` whose result this process already reported. */ + lastResultTurn: number; + /** Fingerprint of the last successful result, to ignore a replayed one. */ + lastResultJSON: string | null; + /** Highest step index this process already reported. */ + lastStepIndex: number; +} + const antigravityHarnessId = harnessIdSchema.parse("antigravity"); const DEFAULT_INSPECT_TIMEOUT_MS = 20_000; const DEFAULT_PRINT_TIMEOUT = "30m"; @@ -549,6 +577,8 @@ class AntigravitySession implements HarnessSession { readonly #history: AntigravityHistory; readonly #subagentObservers = new Set(); #active: ActiveTurn | null = null; + /** The driver a successful Turn kept for this Session's next Turn. */ + #transport: AntigravityTransport | null = null; #preparingQuestions: Promise | null = null; #questionCleanup: Promise = Promise.resolve(); #closed = false; @@ -701,37 +731,58 @@ class AntigravitySession implements HarnessSession { let questions: AntigravityQuestionBridge; const hookDirectory = antigravityHookDirectory(this.#environment); - // The Thread's Hook directory outlives each Turn, while the previous Turn's agy may still be - // exiting and its bridge closing; the next bridge rewrites the directory after that cleanup, - // which is itself bounded. A private per-Turn directory is shared with no one. - const previousTurn = hookDirectory ? this.#questionCleanup : Promise.resolve(); - this.#preparingQuestions = previousTurn.then(() => - AntigravityQuestionBridge.create({ - ...(hookDirectory ? { directory: hookDirectory } : {}), - turnId: command.turnId, - nativeSessionId: () => - this.#active?.command === command && !this.#active.cancellationRequested - ? this.#nativeRef?.nativeSessionId - : undefined, - schedule: (action) => { - if (this.#active?.command === command) this.#enqueue(this.#active, action); - else action(); - }, - emit: (output) => { - const active = this.#active; - if (!active || active.command !== command) return; - if (output.kind === "event" && output.event.type === "item.started" && active.agentItem) { - this.#completeItem(active, active.agentItem, { status: "succeeded" }); - active.agentItem = null; - active.agentText = ""; - } - if (output.kind === "event" && output.event.type === "item.completed") { - active.completedItems.push(output.event.snapshot); - } - this.#channel.emit(output); - }, - }), - ); + // A successful Turn keeps its driver, bridge, and log alive for the next Turn, so a + // continued conversation skips CLI, Language Server, and customization startup. A + // different model/effort/permission selection, a dead or unwritable process, or any + // earlier cleanup still in flight falls back to a fresh driver. + const transportKey = JSON.stringify([ + this.#model?.id ?? null, + this.#thinkingOptionId ?? null, + this.#permissionMode, + ]); + await this.#questionCleanup.catch(() => undefined); + if (this.#closed) { + return { ok: false, error: invalidState("Antigravity Session is closed") }; + } + let reusable = this.#transport; + if ( + reusable && + (reusable.key !== transportKey || + reusable.process.exitCode !== null || + reusable.process.signalCode !== null || + !reusable.process.stdin.writable) + ) { + await this.#disposeTransport(reusable); + reusable = null; + } + const questionOptions: BridgeOptions = { + ...(hookDirectory ? { directory: hookDirectory } : {}), + turnId: command.turnId, + nativeSessionId: () => + this.#active?.command === command && !this.#active.cancellationRequested + ? this.#nativeRef?.nativeSessionId + : undefined, + schedule: (action) => { + if (this.#active?.command === command) this.#enqueue(this.#active, action); + else action(); + }, + emit: (output) => { + const active = this.#active; + if (!active || active.command !== command) return; + if (output.kind === "event" && output.event.type === "item.started" && active.agentItem) { + this.#completeItem(active, active.agentItem, { status: "succeeded" }); + active.agentItem = null; + active.agentText = ""; + } + if (output.kind === "event" && output.event.type === "item.completed") { + active.completedItems.push(output.event.snapshot); + } + this.#channel.emit(output); + }, + }; + this.#preparingQuestions = reusable + ? Promise.resolve(reusable.questions.rebind(questionOptions)) + : AntigravityQuestionBridge.create(questionOptions); try { questions = await this.#preparingQuestions; } catch (error) { @@ -747,11 +798,13 @@ class AntigravitySession implements HarnessSession { this.#preparingQuestions = null; } if (this.#closed) { - await questions.dispose(); + await this.#disposeTransport(reusable); + if (!reusable) await questions.dispose(); return { ok: false, error: invalidState("Antigravity Session is closed") }; } const environment = { ...this.#environment, ...questions.environment }; - const logPath = path.join(os.tmpdir(), `codexhost-antigravity-${randomUUID()}.log`); + const logPath = + reusable?.logPath ?? path.join(os.tmpdir(), `codexhost-antigravity-${randomUUID()}.log`); const arguments_ = [ "--input-format", "stream-json", @@ -769,13 +822,15 @@ class AntigravitySession implements HarnessSession { const invocation = commandInvocation(this.#executable, arguments_, environment); let child: ChildProcessByStdio; try { - child = spawn(invocation.command, invocation.arguments, { - cwd: this.#cwd, - env: environment, - windowsHide: true, - windowsVerbatimArguments: invocation.windowsVerbatimArguments, - stdio: ["pipe", "pipe", "pipe"], - }); + child = + reusable?.process ?? + spawn(invocation.command, invocation.arguments, { + cwd: this.#cwd, + env: environment, + windowsHide: true, + windowsVerbatimArguments: invocation.windowsVerbatimArguments, + stdio: ["pipe", "pipe", "pipe"], + }); } catch (error) { await questions.dispose(); return { @@ -783,11 +838,24 @@ class AntigravitySession implements HarnessSession { error: { code: "nativeFailure", message: errorMessage(error), retryable: true }, }; } + const transport: AntigravityTransport = reusable ?? { + process: child, + exited: new Promise((resolve) => child.once("close", () => resolve())), + questions, + logPath, + key: transportKey, + nativePermissionMode: null, + httpsPort: null, + codeActionCursor: 0, + lastResultTurn: -1, + lastResultJSON: null, + lastStepIndex: -1, + }; const active: ActiveTurn = { command, model: this.#model, process: child, - exited: new Promise((resolve) => child.once("close", () => resolve())), + exited: transport.exited, questions, subagents: new AntigravitySubagents({ turnId: command.turnId, @@ -814,6 +882,8 @@ class AntigravitySession implements HarnessSession { schedule: (work) => this.#enqueue(active, work), }), logPath, + transport, + firstStepIndex: reusable ? reusable.lastStepIndex + 1 : 0, agentItem: null, agentText: "", tools: new Map(), @@ -826,51 +896,87 @@ class AntigravitySession implements HarnessSession { latestUsage: null, contextUsagePromise: null, queue: Promise.resolve(), - codeActionCursor: 0, + codeActionCursor: reusable?.codeActionCursor ?? 0, fileChanges: new Map(), pendingSteps: new Map(), httpsPort: null, }; this.#active = active; + this.#transport = transport; this.#subagentObservers.add(active.subagents); - child.stdin.on("error", () => undefined); - child.stderr.setEncoding("utf8").on("data", (chunk: string) => { - active.stderr = (active.stderr + chunk).slice(-8_000); - }); - const lines = readline.createInterface({ input: child.stdout }); - lines.on("line", (line) => { - const event = parseAntigravityStreamLine(line); - if (event) this.#enqueue(active, () => this.#handleEvent(active, event)); - else if (line.trim()) active.stderr = (active.stderr + `\n${line}`).slice(-8_000); - }); - child.once("error", (error) => { - this.#enqueue(active, () => { - if (this.#active !== active) return; - this.#completeTurn(active, { - status: "failed", - error: { code: "nativeFailure", message: error.message, retryable: true }, - }); + // A reused process already has its stream listeners and its stdin error handler; + // attaching a second set would deliver every event twice. + if (!reusable) { + child.stdin.on("error", () => undefined); + child.stderr.setEncoding("utf8").on("data", (chunk: string) => { + const turn = this.#active; + if (turn?.process === child) turn.stderr = (turn.stderr + chunk).slice(-8_000); }); - }); - child.once("close", (code) => { - this.#enqueue(active, () => { - void unlink(active.logPath).catch(() => undefined); - if (active.subagents.running) void active.subagents.cancel(); - else active.subagents.stop(); - if (this.#active !== active || active.receivedResult) return; - if (active.cancellationRequested) { - this.#completeTurn(active, { status: "cancelled", reason: "Cancelled by user" }); - } else { - this.#completeTurn(active, { + const lines = readline.createInterface({ input: child.stdout }); + lines.on("line", (line) => { + const turn = this.#active; + if (!turn || turn.process !== child) return; + const event = parseAntigravityStreamLine(line); + if (!event) { + if (line.trim()) turn.stderr = (turn.stderr + `\n${line}`).slice(-8_000); + return; + } + // agy replays the previous successful result and old steps when it starts the + // next stdin message, so only genuinely new Turn output reaches the Host. + if (event.event === "result" && Number.isInteger(event.result.num_turns)) { + const fingerprint = JSON.stringify(event.result); + if ( + event.result.status === "SUCCESS" && + event.result.num_turns <= transport.lastResultTurn && + fingerprint === transport.lastResultJSON + ) { + return; + } + transport.lastResultTurn = Math.max(transport.lastResultTurn, event.result.num_turns); + transport.lastResultJSON = fingerprint; + } + if (event.event === "step_update") { + const index = event.step_update.step_index; + if (Number.isInteger(index)) { + if (index < turn.firstStepIndex) return; + transport.lastStepIndex = Math.max(transport.lastStepIndex, index); + } + } + this.#enqueue(turn, () => this.#handleEvent(turn, event)); + }); + child.once("error", (error) => { + const turn = this.#active; + if (turn?.process !== child) return; + this.#enqueue(turn, () => { + if (this.#active !== turn) return; + this.#completeTurn(turn, { status: "failed", - error: normalizedProcessError( - active.stderr, - `Antigravity CLI exited before a result event (code ${String(code)})`, - ), + error: { code: "nativeFailure", message: error.message, retryable: true }, }); - } + }); }); - }); + child.once("close", (code) => { + if (this.#transport === transport) this.#transport = null; + const turn = this.#active; + if (!turn || turn.process !== child) return; + this.#enqueue(turn, () => { + if (turn.subagents.running) void turn.subagents.cancel(); + else turn.subagents.stop(); + if (this.#active !== turn || turn.receivedResult) return; + if (turn.cancellationRequested) { + this.#completeTurn(turn, { status: "cancelled", reason: "Cancelled by user" }); + } else { + this.#completeTurn(turn, { + status: "failed", + error: normalizedProcessError( + turn.stderr, + `Antigravity CLI exited before a result event (code ${String(code)})`, + ), + }); + } + }); + }); + } try { const turnPrompt = formatAntigravityTurnPrompt(text, this.#cwd); if (child.stdin.writable) { @@ -905,6 +1011,9 @@ class AntigravitySession implements HarnessSession { await active.subagents.cancel(); active.process.kill(); this.#completeTurn(active, { status: "cancelled", reason: "Session closed" }); + } else if (this.#transport) { + // A retained driver from a successful Turn belongs to the Session; close releases it. + await this.#disposeTransport(this.#transport); } await Promise.all([...this.#subagentObservers].map((observer) => observer.cancel())); await this.#questionCleanup; @@ -1276,17 +1385,28 @@ class AntigravitySession implements HarnessSession { #completeTurn(active: ActiveTurn, outcome: TurnOutcome, nativeTurnRef?: NativeTurnRef): void { if (this.#active !== active) return; active.questions.stop(); + const transport = active.transport; + // A successful Turn keeps its driver for the next Turn of this Session; the bridge is + // rebound there. Anything else (failure, cancellation, close) releases the driver. + const keepTransport = + outcome.status === "succeeded" && + !this.#closed && + this.#transport === transport && + active.process.exitCode === null && + active.process.signalCode === null && + active.process.stdin.writable; + if (keepTransport) { + transport.httpsPort = active.httpsPort ?? transport.httpsPort; + transport.codeActionCursor = active.codeActionCursor; + transport.nativePermissionMode = active.nativePermissionMode; + } this.#questionCleanup = this.#questionCleanup .then(async () => { - if (outcome.status !== "succeeded") await active.subagents.cancel(); await active.subagents.settled; - if (active.process.stdin.writable) active.process.stdin.end(); - const timer = setTimeout(() => active.process.kill(), NATIVE_EXIT_GRACE_MS); - // A child that keeps agy's output open can delay `close` indefinitely, and later Turns' - // cleanup and close() queue behind this one. - await settledWithin(active.exited, NATIVE_EXIT_GRACE_MS + this.#turnCleanupWaitMs); - clearTimeout(timer); - await active.questions.dispose(); + if (!keepTransport) { + if (outcome.status !== "succeeded") await active.subagents.cancel(); + await this.#disposeTransport(transport); + } this.#subagentObservers.delete(active.subagents); }) .catch(() => undefined); @@ -1335,6 +1455,29 @@ class AntigravitySession implements HarnessSession { }); } + /** Releases a native driver: stop its bridge, end stdin, then kill if it lingers. */ + async #disposeTransport(transport: AntigravityTransport | null): Promise { + if (!transport) return; + if (this.#transport === transport) this.#transport = null; + transport.questions.stop(); + if (transport.process.stdin.writable) transport.process.stdin.end(); + const timer = setTimeout(() => transport.process.kill(), NATIVE_EXIT_GRACE_MS); + const hardTimer = setTimeout( + () => transport.process.kill("SIGKILL"), + NATIVE_EXIT_GRACE_MS + 5_000, + ); + try { + // A process that keeps agy's output open can delay `close` indefinitely, and later + // Turns' cleanup and close() queue behind this one. + await settledWithin(transport.exited, NATIVE_EXIT_GRACE_MS + this.#turnCleanupWaitMs); + } finally { + clearTimeout(timer); + clearTimeout(hardTimer); + await transport.questions.dispose(); + await unlink(transport.logPath).catch(() => undefined); + } + } + #completeItem(active: ActiveTurn, item: HostItem, outcome: HostItemOutcome): void { const snapshot = { item, outcome } satisfies HostItemSnapshot; active.completedItems.push(snapshot); diff --git a/packages/adapters/antigravity/src/question-bridge.ts b/packages/adapters/antigravity/src/question-bridge.ts index c995d795f..ff66420b3 100644 --- a/packages/adapters/antigravity/src/question-bridge.ts +++ b/packages/adapters/antigravity/src/question-bridge.ts @@ -49,7 +49,7 @@ const responseSchema = z cancelled: z.boolean().optional(), }) .strict(); -interface BridgeOptions { +export interface BridgeOptions { /** * A directory the caller keeps across Turns. The bridge writes its Hook files there and leaves * the directory on dispose; without one it uses a private temporary directory it removes. @@ -374,6 +374,27 @@ export class AntigravityQuestionBridge { }); } + /** + * Hands the bridge to the next Turn of a reused native driver. Ownership, the Turn + * identity used in emitted Items, and the duplicate-question guard all move to the new + * Turn, so a question from an earlier Turn can no longer be attributed or answered. + */ + rebind(options: BridgeOptions): AntigravityQuestionBridge { + if (this.#disposal) throw new Error("Cannot reuse a disposed Question bridge"); + if (this.#pending.size) throw new Error("Cannot reuse a busy Question bridge"); + if (options.directory !== this.#options.directory) { + throw new Error("A reused Question bridge keeps its directory"); + } + this.#options.turnId = options.turnId; + this.#options.nativeSessionId = options.nativeSessionId; + this.#options.schedule = options.schedule; + this.#options.emit = options.emit; + if (options.timeoutMs !== undefined) this.#options.timeoutMs = options.timeoutMs; + this.#stopped = false; + this.#seenSteps.clear(); + return this; + } + stop(): void { if (this.#stopped) return; this.#stopped = true; diff --git a/packages/adapters/antigravity/test/adversarial-fork-rollback.test.ts b/packages/adapters/antigravity/test/adversarial-fork-rollback.test.ts index 3db2202ff..1305498fb 100644 --- a/packages/adapters/antigravity/test/adversarial-fork-rollback.test.ts +++ b/packages/adapters/antigravity/test/adversarial-fork-rollback.test.ts @@ -60,7 +60,9 @@ if (process.argv.includes("models") || process.argv.some((arg) => arg.includes(" const runsDir = ${JSON.stringify(runsDir)}; fs.mkdirSync(runsDir, { recursive: true }); -const count = fs.readdirSync(runsDir).length; +// The script position is shared by every process, so a forked or resumed session picks +// up where the previous process left off rather than replaying an earlier Turn. +let count = fs.readdirSync(runsDir).length; fs.writeFileSync(path.join(runsDir, "run-" + count + ".txt"), JSON.stringify(process.argv) + "\\n"); const convIdx = process.argv.indexOf("--conversation"); @@ -71,6 +73,7 @@ const delay = ${delayMs}; function emit() { const rawLines = turnResponses[Math.min(count, turnResponses.length - 1)] || []; + count += 1; for (const rawLine of rawLines) { try { const parsed = JSON.parse(rawLine); @@ -90,17 +93,23 @@ function emit() { } } +let buffer = ""; +function emitNext() { + if (delay > 0) setTimeout(emit, delay); + else emit(); +} process.stdin.resume(); -process.stdin.on("data", () => {}); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let index; + while ((index = buffer.indexOf("\\n")) >= 0) { + buffer = buffer.slice(index + 1); + emitNext(); + } +}); process.stdin.on("end", () => { process.exit(0); }); - -if (delay > 0) { - setTimeout(emit, delay); -} else { - emit(); -} `; const jsPath = path.join(directory, "agy.cjs"); diff --git a/packages/adapters/antigravity/test/adversarial-stress.test.ts b/packages/adapters/antigravity/test/adversarial-stress.test.ts index 5ef7ea8c7..0865f9867 100644 --- a/packages/adapters/antigravity/test/adversarial-stress.test.ts +++ b/packages/adapters/antigravity/test/adversarial-stress.test.ts @@ -352,6 +352,8 @@ describe("Antigravity Adversarial & Stress Testing", () => { const runsDir = path.join(directory, "runs"); const jsPath = path.join(directory, "agy.cjs"); + // Like the real CLI, one process serves every Turn: each stdin line drives the + // next scripted response, so a reused driver never replays an earlier Turn. const scriptContent = ` const fs = require('fs'); const path = require('path'); @@ -363,16 +365,25 @@ if (process.argv.includes("models") || process.argv.some((arg) => arg.includes(" } const runsDir = ${JSON.stringify(runsDir)}; fs.mkdirSync(runsDir, { recursive: true }); -// Only a Turn consumes a scripted response. Quota probes also request -// stream-json output, but unlike Turns they do not use stream-json input. -const isTurn = process.argv.includes("--input-format"); -const counter = isTurn ? fs.readdirSync(runsDir).length : -1; -if (isTurn) fs.writeFileSync(path.join(runsDir, "run-" + counter + ".txt"), ""); const allResponses = JSON.parse(fs.readFileSync(${JSON.stringify(runsFile)}, 'utf8')); -const lines = isTurn ? allResponses[counter] || allResponses[0] || [] : []; -for (const line of lines) { - process.stdout.write(line + "\\n"); +let turn = 0; +let buffer = ""; +function emit() { + const lines = allResponses[turn] || allResponses[allResponses.length - 1] || []; + turn += 1; + for (const line of lines) process.stdout.write(line + "\\n"); } +process.stdin.resume(); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let index; + while ((index = buffer.indexOf("\\n")) >= 0) { + buffer = buffer.slice(index + 1); + fs.writeFileSync(path.join(runsDir, "run-" + turn + ".txt"), ""); + emit(); + } +}); +process.stdin.on("end", () => process.exit(0)); `; await writeFile(jsPath, scriptContent); if (process.platform === "win32") { diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index fd67c1503..eefb33e55 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -1427,21 +1427,23 @@ if (process.argv.includes("models")) { // The quota probe runs as --print=/usage; Turns carry --print-timeout instead. if (process.argv.some((value) => value.startsWith("--print="))) process.exit(0); const addDirs = process.argv.filter((_, index, all) => all[index - 1] === "--add-dir"); -let input = ""; +let buffer = ""; process.stdin.setEncoding("utf8"); process.stdin.on("data", (chunk) => { - if (input.includes("\\n")) return; - input += chunk; - const newline = input.indexOf("\\n"); - if (newline === -1) return; - const prompt = JSON.parse(input.slice(0, newline)).message.content; - fs.appendFileSync(${JSON.stringify(runs)}, JSON.stringify({ addDirs, prompt }) + "\\n"); - const turn = fs.readFileSync(${JSON.stringify(runs)}, "utf8").trim().split("\\n").length; - for (const event of [ - { event: "init", conversation_id: "conv-hooks", init: { permission_mode: "dangerously-skip-permissions" } }, - { event: "result", result: { conversation_id: "conv-hooks", status: "SUCCESS", num_turns: turn, response: "ok" } }, - ]) { - process.stdout.write(JSON.stringify(event) + "\\n"); + buffer += chunk; + let newline; + while ((newline = buffer.indexOf("\\n")) >= 0) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + const prompt = JSON.parse(line).message.content; + fs.appendFileSync(${JSON.stringify(runs)}, JSON.stringify({ addDirs, prompt }) + "\\n"); + const turn = fs.readFileSync(${JSON.stringify(runs)}, "utf8").trim().split("\\n").length; + for (const event of [ + { event: "init", conversation_id: "conv-hooks", init: { permission_mode: "dangerously-skip-permissions" } }, + { event: "result", result: { conversation_id: "conv-hooks", status: "SUCCESS", num_turns: turn, response: "ok" } }, + ]) { + process.stdout.write(JSON.stringify(event) + "\\n"); + } } }); process.stdin.on("end", () => process.exit(0)); @@ -1553,24 +1555,32 @@ if (process.argv.includes("models")) { process.exit(0); } if (process.argv.some((value) => value.startsWith("--print="))) process.exit(0); -fs.appendFileSync(${JSON.stringify(runs)}, "run\\n"); -const turn = fs.readFileSync(${JSON.stringify(runs)}, "utf8").trim().split("\\n").length; -if (turn === 1) { - // Detached so it outlives this process on Windows too, where a child otherwise ends with it. - const holder = spawn(process.execPath, ["-e", "setTimeout(() => {}, 60000)"], { - detached: true, - stdio: ["ignore", "inherit", "inherit"], - windowsHide: true, - }); - fs.writeFileSync(${JSON.stringify(holderPidFile)}, String(holder.pid)); -} -for (const event of [ - { event: "init", conversation_id: "conv-linger", init: { permission_mode: "dangerously-skip-permissions" } }, - { event: "result", result: { conversation_id: "conv-linger", status: "SUCCESS", num_turns: turn, response: "ok" } }, -]) { - process.stdout.write(JSON.stringify(event) + "\\n"); -} +let buffer = ""; process.stdin.resume(); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let newline; + while ((newline = buffer.indexOf("\\n")) >= 0) { + buffer = buffer.slice(newline + 1); + fs.appendFileSync(${JSON.stringify(runs)}, "run\\n"); + const turn = fs.readFileSync(${JSON.stringify(runs)}, "utf8").trim().split("\\n").length; + if (turn === 1) { + // Detached so it outlives this process on Windows too, where a child otherwise ends with it. + const holder = spawn(process.execPath, ["-e", "setTimeout(() => {}, 60000)"], { + detached: true, + stdio: ["ignore", "inherit", "inherit"], + windowsHide: true, + }); + fs.writeFileSync(${JSON.stringify(holderPidFile)}, String(holder.pid)); + } + for (const event of [ + { event: "init", conversation_id: "conv-linger", init: { permission_mode: "dangerously-skip-permissions" } }, + { event: "result", result: { conversation_id: "conv-linger", status: "SUCCESS", num_turns: turn, response: "ok" } }, + ]) { + process.stdout.write(JSON.stringify(event) + "\\n"); + } + } +}); process.stdin.on("end", () => process.exit(0)); `; const jsPath = path.join(directory, "agy.cjs"); @@ -1642,6 +1652,113 @@ process.stdin.on("end", () => process.exit(0)); 20_000, ); + it("reuses one native driver across successful Turns and releases it on model select", async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-reuse-")); + const cwd = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-reuse-cwd-")); + const pids = path.join(directory, "pids.jsonl"); + const cleanup = async (): Promise => { + for (const target of [directory, cwd]) { + await rm(target, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + } + }; + // One process answers every Turn it is given, and records its own PID per Turn. + const scriptContent = ` +const fs = require("node:fs"); +if (process.argv.includes("models")) { + process.stdout.write("gemini-3.7-flash-high\\tGemini 3.7 Flash High\\ngemini-3.1-pro-high\\tGemini 3.1 Pro High\\n"); + process.exit(0); +} +if (process.argv.some((value) => value.startsWith("--print="))) process.exit(0); +// The Conversation keeps counting across processes, as agy reports num_turns for the +// Conversation rather than for the process. +let turn = fs.existsSync(${JSON.stringify(pids)}) + ? fs.readFileSync(${JSON.stringify(pids)}, "utf8").trim().split("\\n").length + : 0; +let buffer = ""; +function answer() { + turn += 1; + fs.appendFileSync(${JSON.stringify(pids)}, JSON.stringify({ pid: process.pid, turn }) + "\\n"); + for (const event of [ + { event: "init", conversation_id: "conv-reuse", init: { permission_mode: "dangerously-skip-permissions" } }, + { event: "result", result: { conversation_id: "conv-reuse", status: "SUCCESS", num_turns: turn, response: "reply " + turn } }, + ]) { + process.stdout.write(JSON.stringify(event) + "\\n"); + } +} +process.stdin.resume(); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let newline; + while ((newline = buffer.indexOf("\\n")) >= 0) { + buffer = buffer.slice(newline + 1); + answer(); + } +}); +process.stdin.on("end", () => process.exit(0)); +`; + const jsPath = path.join(directory, "agy.cjs"); + await writeFile(jsPath, scriptContent); + let command: string; + if (process.platform === "win32") { + command = path.join(directory, "agy.cmd"); + await writeFile(command, `@node "${jsPath}" %*\r\n`); + } else { + command = path.join(directory, "agy"); + await writeFile(command, `#!/usr/bin/env node\n${scriptContent}`); + await chmod(command, 0o755); + } + + const adapter = new AntigravityAdapter({ command }); + try { + const opened = await adapter.open({ kind: "create", cwd }); + if (!opened.ok) throw new Error(opened.error.message); + const session = opened.value; + const iterator = session.outputs[Symbol.asyncIterator](); + const runTurn = async (turnId: string): Promise => { + const started = await session.execute({ + type: "turn.start", + turnId: hostTurnIdSchema.parse(turnId), + input: [{ type: "text", text: turnId }], + }); + expect(started.ok).toBe(true); + while ((await nextEvent(iterator)).type !== "turn.completed") { + // Remaining events are not under test. + } + }; + + await runTurn("turn-reuse-1"); + await runTurn("turn-reuse-2"); + await runTurn("turn-reuse-3"); + const observed = (await readFile(pids, "utf8")) + .trim() + .split("\n") + .map((line) => JSON.parse(line) as { pid: number; turn: number }); + expect(observed).toHaveLength(3); + expect(new Set(observed.map(({ pid }) => pid)).size).toBe(1); + + // A different model needs a different invocation, so the next Turn starts a new driver. + await session.execute({ + type: "model.select", + model: harnessModelRefSchema.parse({ id: "gemini-3.1-pro-high" }), + }); + await runTurn("turn-reuse-4"); + const afterSelect = (await readFile(pids, "utf8")) + .trim() + .split("\n") + .map((line) => JSON.parse(line) as { pid: number; turn: number }); + expect(afterSelect).toHaveLength(4); + expect(new Set(afterSelect.map(({ pid }) => pid)).size).toBe(2); + + const snapshot = await session.readSnapshot(); + expect(snapshot.ok).toBe(true); + if (snapshot.ok) expect(snapshot.value.turns).toHaveLength(4); + await session.close(); + } finally { + await adapter.close(); + await cleanup(); + } + }); + it("projects file steps as Tool Executions when the applied patch cannot be read", async () => { const streamLines = [ JSON.stringify({ @@ -1784,11 +1901,28 @@ if (process.argv.includes("models")) { if (!process.argv.includes("--input-format")) process.exit(0); const runsDir = ${JSON.stringify(runsDir)}; fs.mkdirSync(runsDir, { recursive: true }); -const count = fs.readdirSync(runsDir).length; -fs.writeFileSync(path.join(runsDir, "run-" + count + ".txt"), ""); -const result = ${JSON.stringify(results)}[count]; -process.stdout.write(JSON.stringify({ event: "init", conversation_id: "conv-sticky", init: { permission_mode: "default" } }) + "\\n"); -process.stdout.write(JSON.stringify({ event: "result", result: { conversation_id: "conv-sticky", ...result } }) + "\\n"); +const results = ${JSON.stringify(results)}; +// One process serves every Turn, as the real CLI does: each stdin line advances the script. +let count = fs.readdirSync(runsDir).length; +let buffer = ""; +function answer() { + fs.writeFileSync(path.join(runsDir, "run-" + count + ".txt"), ""); + const result = results[count]; + count += 1; + if (!result) return; + process.stdout.write(JSON.stringify({ event: "init", conversation_id: "conv-sticky", init: { permission_mode: "default" } }) + "\\n"); + process.stdout.write(JSON.stringify({ event: "result", result: { conversation_id: "conv-sticky", ...result } }) + "\\n"); +} +process.stdin.resume(); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let newline; + while ((newline = buffer.indexOf("\\n")) >= 0) { + buffer = buffer.slice(newline + 1); + answer(); + } +}); +process.stdin.on("end", () => process.exit(0)); `; const jsPath = path.join(directory, "agy.cjs"); await writeFile(jsPath, scriptContent); diff --git a/packages/adapters/antigravity/test/fork.test.ts b/packages/adapters/antigravity/test/fork.test.ts index 09987a1fe..c54aa62e2 100644 --- a/packages/adapters/antigravity/test/fork.test.ts +++ b/packages/adapters/antigravity/test/fork.test.ts @@ -81,15 +81,26 @@ if (process.argv.includes("models") || process.argv.some((arg) => arg.includes(" const runsDir = ${JSON.stringify(runsDir)}; fs.mkdirSync(runsDir, { recursive: true }); -const count = fs.readdirSync(runsDir).length; -fs.writeFileSync(path.join(runsDir, "run-" + count + ".txt"), JSON.stringify(process.argv) + "\\n"); - const turnStreamLines = ${JSON.stringify(turnStreamLines)}; -const lines = turnStreamLines[Math.min(count, turnStreamLines.length - 1)] || []; -for (const line of lines) { - process.stdout.write(line + "\\n"); +// One process serves every Turn, as the real CLI does: each stdin line advances the script. +let count = fs.readdirSync(runsDir).length; +function emit() { + fs.writeFileSync(path.join(runsDir, "run-" + count + ".txt"), JSON.stringify(process.argv) + "\\n"); + const lines = turnStreamLines[Math.min(count, turnStreamLines.length - 1)] || []; + count += 1; + for (const line of lines) process.stdout.write(line + "\\n"); } -setTimeout(() => { process.exit(0); }, 50); +let buffer = ""; +process.stdin.resume(); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let index; + while ((index = buffer.indexOf("\\n")) >= 0) { + buffer = buffer.slice(index + 1); + emit(); + } +}); +process.stdin.on("end", () => process.exit(0)); `; const jsPath = path.join(directory, "agy.cjs"); await writeFile(jsPath, scriptContent); diff --git a/packages/adapters/antigravity/test/rollback.test.ts b/packages/adapters/antigravity/test/rollback.test.ts index 75012bb07..db6e59d0d 100644 --- a/packages/adapters/antigravity/test/rollback.test.ts +++ b/packages/adapters/antigravity/test/rollback.test.ts @@ -44,15 +44,26 @@ if (process.argv.includes("models") || process.argv.some((arg) => arg.includes(" const runsDir = ${JSON.stringify(runsDir)}; fs.mkdirSync(runsDir, { recursive: true }); -const count = fs.readdirSync(runsDir).length; -fs.writeFileSync(path.join(runsDir, "run-" + count + ".txt"), JSON.stringify(process.argv) + "\\n"); - const turnStreamLines = ${JSON.stringify(turnStreamLines)}; -const lines = turnStreamLines[Math.min(count, turnStreamLines.length - 1)] || []; -for (const line of lines) { - process.stdout.write(line + "\\n"); +// One process serves every Turn, as the real CLI does: each stdin line advances the script. +let count = fs.readdirSync(runsDir).length; +function emit() { + fs.writeFileSync(path.join(runsDir, "run-" + count + ".txt"), JSON.stringify(process.argv) + "\\n"); + const lines = turnStreamLines[Math.min(count, turnStreamLines.length - 1)] || []; + count += 1; + for (const line of lines) process.stdout.write(line + "\\n"); } -setTimeout(() => { process.exit(0); }, 50); +let buffer = ""; +process.stdin.resume(); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let index; + while ((index = buffer.indexOf("\\n")) >= 0) { + buffer = buffer.slice(index + 1); + emit(); + } +}); +process.stdin.on("end", () => process.exit(0)); `; const jsPath = path.join(directory, "agy.cjs"); await writeFile(jsPath, scriptContent); From ff436f1f71aea637ab8b3d7357d18c0b4b7868e8 Mon Sep 17 00:00:00 2001 From: tonyzaca7755963 <245165781+tonyzaca7755963@users.noreply.github.com> Date: Mon, 26 Oct 2026 13:37:45 +0800 Subject: [PATCH 02/11] refactor(antigravity): drop the per-turn exit promise a reused driver made ambiguous (#535) Co-Authored-By: Claude Code --- packages/adapters/antigravity/src/antigravity-adapter.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/packages/adapters/antigravity/src/antigravity-adapter.ts b/packages/adapters/antigravity/src/antigravity-adapter.ts index 23cf0cc57..baa8d2b77 100644 --- a/packages/adapters/antigravity/src/antigravity-adapter.ts +++ b/packages/adapters/antigravity/src/antigravity-adapter.ts @@ -132,7 +132,6 @@ interface ActiveTurn { /** Model passed to this CLI invocation, independent of later selections. */ model: HarnessModelRef | undefined; process: ChildProcessByStdio; - exited: Promise; questions: AntigravityQuestionBridge; subagents: AntigravitySubagents; logPath: string; @@ -855,7 +854,6 @@ class AntigravitySession implements HarnessSession { command, model: this.#model, process: child, - exited: transport.exited, questions, subagents: new AntigravitySubagents({ turnId: command.turnId, From 6ca672a33709f9e4065a670d15054d40ad53ce74 Mon Sep 17 00:00:00 2001 From: tonyzaca7755963 <245165781+tonyzaca7755963@users.noreply.github.com> Date: Mon, 26 Oct 2026 13:48:19 +0800 Subject: [PATCH 03/11] fix(antigravity): close the reuse lifecycle gaps found in review (#535) - Claim the Session synchronously before any await, so two concurrent turn.start calls cannot both pass the busy check and overwrite #active - Stamp each Question request with a bridge generation and reject one that straddles a rebind, instead of attributing it to the Turn that now owns the bridge - Release a retained driver that exits between Turns; nothing else could reach its bridge or log once the close listener cleared #transport - Cancel a failed Turn's Subagents before awaiting them, so an observable-but-stuck Subagent cannot strand the driver and leave the Session busy - Filter a replayed result regardless of its native status: an ERROR the Host accepted as success keeps the driver, so its replay would fail the next Turn early Co-Authored-By: Claude Code --- .../antigravity/src/antigravity-adapter.ts | 45 ++++-- .../antigravity/src/question-bridge.ts | 15 +- .../test/antigravity-adapter.test.ts | 147 ++++++++++++++++++ .../antigravity/test/question-bridge.test.ts | 59 +++++++ 4 files changed, 252 insertions(+), 14 deletions(-) diff --git a/packages/adapters/antigravity/src/antigravity-adapter.ts b/packages/adapters/antigravity/src/antigravity-adapter.ts index baa8d2b77..5dab90a64 100644 --- a/packages/adapters/antigravity/src/antigravity-adapter.ts +++ b/packages/adapters/antigravity/src/antigravity-adapter.ts @@ -576,6 +576,8 @@ class AntigravitySession implements HarnessSession { readonly #history: AntigravityHistory; readonly #subagentObservers = new Set(); #active: ActiveTurn | null = null; + /** Set from the moment a turn.start is accepted until it finishes starting. */ + #starting = false; /** The driver a successful Turn kept for this Session's next Turn. */ #transport: AntigravityTransport | null = null; #preparingQuestions: Promise | null = null; @@ -650,6 +652,7 @@ class AntigravitySession implements HarnessSession { get isActive(): boolean { return ( this.#active !== null || + this.#starting || this.#preparingQuestions !== null || [...this.#subagentObservers].some((observer) => observer.running) ); @@ -728,12 +731,22 @@ class AntigravitySession implements HarnessSession { }; } + // Claim the Session synchronously, before any await, so two concurrent starts cannot + // both pass the busy check and overwrite #active. Every exit path releases the claim. + this.#starting = true; + try { + return await this.#startTurn(command, text); + } finally { + this.#starting = false; + } + } + + async #startTurn( + command: TurnStartCommand, + text: string, + ): Promise> { let questions: AntigravityQuestionBridge; const hookDirectory = antigravityHookDirectory(this.#environment); - // A successful Turn keeps its driver, bridge, and log alive for the next Turn, so a - // continued conversation skips CLI, Language Server, and customization startup. A - // different model/effort/permission selection, a dead or unwritable process, or any - // earlier cleanup still in flight falls back to a fresh driver. const transportKey = JSON.stringify([ this.#model?.id ?? null, this.#thinkingOptionId ?? null, @@ -919,12 +932,15 @@ class AntigravitySession implements HarnessSession { if (line.trim()) turn.stderr = (turn.stderr + `\n${line}`).slice(-8_000); return; } - // agy replays the previous successful result and old steps when it starts the - // next stdin message, so only genuinely new Turn output reaches the Host. + // agy replays the previous Turn's result and old steps when it starts the next + // stdin message, so only genuinely new Turn output reaches the Host. A replay is + // filtered whatever its status: a native ERROR that the Host accepted as a + // success (a replied Turn repeating an earlier Conversation error) still keeps + // this driver, so its replay would otherwise fail the next Turn before its own + // result arrived. if (event.event === "result" && Number.isInteger(event.result.num_turns)) { const fingerprint = JSON.stringify(event.result); if ( - event.result.status === "SUCCESS" && event.result.num_turns <= transport.lastResultTurn && fingerprint === transport.lastResultJSON ) { @@ -956,7 +972,12 @@ class AntigravitySession implements HarnessSession { child.once("close", (code) => { if (this.#transport === transport) this.#transport = null; const turn = this.#active; - if (!turn || turn.process !== child) return; + if (!turn || turn.process !== child) { + // An idle driver exited between Turns. Nothing else will release its bridge and + // log, so do it here rather than leaking an HTTP server and a temp file. + void this.#disposeTransport(transport); + return; + } this.#enqueue(turn, () => { if (turn.subagents.running) void turn.subagents.cancel(); else turn.subagents.stop(); @@ -1400,11 +1421,11 @@ class AntigravitySession implements HarnessSession { } this.#questionCleanup = this.#questionCleanup .then(async () => { + // Cancel first: a failed Turn's Subagents may stay observable as running, and + // `settled` would then never resolve, stranding the driver. + if (!keepTransport && outcome.status !== "succeeded") await active.subagents.cancel(); await active.subagents.settled; - if (!keepTransport) { - if (outcome.status !== "succeeded") await active.subagents.cancel(); - await this.#disposeTransport(transport); - } + if (!keepTransport) await this.#disposeTransport(transport); this.#subagentObservers.delete(active.subagents); }) .catch(() => undefined); diff --git a/packages/adapters/antigravity/src/question-bridge.ts b/packages/adapters/antigravity/src/question-bridge.ts index ff66420b3..f8c18c08f 100644 --- a/packages/adapters/antigravity/src/question-bridge.ts +++ b/packages/adapters/antigravity/src/question-bridge.ts @@ -89,6 +89,12 @@ export class AntigravityQuestionBridge { readonly #pending = new Map(); readonly #seenSteps = new Set(); #stopped = false; + /** + * Bumped on every rebind. A request received before the bump is stale: it belongs to the + * Turn that was driving when the hook process made it, and must not be attributed to the + * Turn now owning the bridge. + */ + #generation = 0; #disposal: Promise | null = null; private constructor(directory: string, options: BridgeOptions) { @@ -158,6 +164,9 @@ export class AntigravityQuestionBridge { } #receive(request: IncomingMessage, response: ServerResponse): void { + // The hook process is spawned per Turn by agy; capturing the generation here means a + // request that straddles a rebind cannot be answered by the next Turn. + const generation = this.#generation; const auth = Buffer.from(request.headers.authorization ?? ""); const expected = Buffer.from(`Bearer ${this.#token}`); if ( @@ -198,13 +207,14 @@ export class AntigravityQuestionBridge { return; } const payload = parsed.data; - this.#options.schedule(() => this.#ask(payload, response)); + this.#options.schedule(() => this.#ask(payload, response, generation)); }); } - #ask(payload: z.infer, response: ServerResponse): void { + #ask(payload: z.infer, response: ServerResponse, generation: number): void { if ( this.#stopped || + generation !== this.#generation || response.destroyed || payload.conversationId !== this.#options.nativeSessionId() ) { @@ -392,6 +402,7 @@ export class AntigravityQuestionBridge { if (options.timeoutMs !== undefined) this.#options.timeoutMs = options.timeoutMs; this.#stopped = false; this.#seenSteps.clear(); + this.#generation += 1; return this; } diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index eefb33e55..481858ba9 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -1759,6 +1759,153 @@ process.stdin.on("end", () => process.exit(0)); } }); + it("refuses a second turn.start that starts while the first is still preparing", async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-race-")); + const cwd = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-race-cwd-")); + const cleanups = async (): Promise => { + for (const target of [directory, cwd]) { + await rm(target, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + } + }; + const scriptContent = ` +if (process.argv.includes("models")) { + process.stdout.write("gemini-3.7-flash-high\\tGemini 3.7 Flash High\\n"); + process.exit(0); +} +if (process.argv.some((value) => value.startsWith("--print="))) process.exit(0); +let buffer = ""; +process.stdin.resume(); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let newline; + while ((newline = buffer.indexOf("\\n")) >= 0) { + buffer = buffer.slice(newline + 1); + process.stdout.write(JSON.stringify({ event: "init", conversation_id: "conv-race", init: { permission_mode: "dangerously-skip-permissions" } }) + "\\n"); + process.stdout.write(JSON.stringify({ event: "result", result: { conversation_id: "conv-race", status: "SUCCESS", num_turns: 1, response: "ok" } }) + "\\n"); + } +}); +process.stdin.on("end", () => process.exit(0)); +`; + const jsPath = path.join(directory, "agy.cjs"); + await writeFile(jsPath, scriptContent); + let command: string; + if (process.platform === "win32") { + command = path.join(directory, "agy.cmd"); + await writeFile(command, `@node "${jsPath}" %*\r\n`); + } else { + command = path.join(directory, "agy"); + await writeFile(command, `#!/usr/bin/env node\n${scriptContent}`); + await chmod(command, 0o755); + } + + const adapter = new AntigravityAdapter({ command }); + try { + const opened = await adapter.open({ kind: "create", cwd }); + if (!opened.ok) throw new Error(opened.error.message); + const session = opened.value; + // Both starts are issued before either can await anything. + const [first, second] = await Promise.all([ + session.execute({ + type: "turn.start", + turnId: hostTurnIdSchema.parse("turn-race-1"), + input: [{ type: "text", text: "first" }], + }), + session.execute({ + type: "turn.start", + turnId: hostTurnIdSchema.parse("turn-race-2"), + input: [{ type: "text", text: "second" }], + }), + ]); + expect(first.ok).toBe(true); + expect(second).toMatchObject({ ok: false, error: { code: "sessionBusy" } }); + await session.close(); + } finally { + await adapter.close(); + await cleanups(); + } + }); + + it("releases an idle driver that exits between turns", async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-idle-")); + const cwd = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-idle-cwd-")); + const cleanup = async (): Promise => { + for (const target of [directory, cwd]) { + await rm(target, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + } + }; + const scriptContent = ` +const fs = require("node:fs"); +if (process.argv.includes("models")) { + process.stdout.write("gemini-3.7-flash-high\\tGemini 3.7 Flash High\\n"); + process.exit(0); +} +if (process.argv.some((value) => value.startsWith("--print="))) process.exit(0); +// The log file is the driver's; its removal by the parent proves the release ran. +for (const arg of process.argv) { + if (arg.endsWith(".log")) fs.writeFileSync(arg, "started\\n"); +} +let served = false; +let buffer = ""; +process.stdin.resume(); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let newline; + while ((newline = buffer.indexOf("\\n")) >= 0) { + buffer = buffer.slice(newline + 1); + served = true; + process.stdout.write(JSON.stringify({ event: "init", conversation_id: "conv-idle", init: { permission_mode: "dangerously-skip-permissions" } }) + "\\n"); + process.stdout.write(JSON.stringify({ event: "result", result: { conversation_id: "conv-idle", status: "SUCCESS", num_turns: 1, response: "ok" } }) + "\\n"); + // Exit on its own once the Turn is answered, like a native driver that idles out. + setTimeout(() => process.exit(0), 50); + } +}); +process.stdin.on("end", () => process.exit(served ? 0 : 1)); +`; + const jsPath = path.join(directory, "agy.cjs"); + await writeFile(jsPath, scriptContent); + let command: string; + if (process.platform === "win32") { + command = path.join(directory, "agy.cmd"); + await writeFile(command, `@node "${jsPath}" %*\r\n`); + } else { + command = path.join(directory, "agy"); + await writeFile(command, `#!/usr/bin/env node\n${scriptContent}`); + await chmod(command, 0o755); + } + + const adapter = new AntigravityAdapter({ command }); + try { + const opened = await adapter.open({ kind: "create", cwd }); + if (!opened.ok) throw new Error(opened.error.message); + const session = opened.value; + const iterator = session.outputs[Symbol.asyncIterator](); + await session.execute({ + type: "turn.start", + turnId: hostTurnIdSchema.parse("turn-idle-1"), + input: [{ type: "text", text: "one" }], + }); + while ((await nextEvent(iterator)).type !== "turn.completed") { + // Remaining events are not under test. + } + // The driver exits on its own shortly after answering; the next Turn must start a + // fresh one rather than write into the dead process. + await new Promise((resolve) => setTimeout(resolve, 400)); + const second = await session.execute({ + type: "turn.start", + turnId: hostTurnIdSchema.parse("turn-idle-2"), + input: [{ type: "text", text: "two" }], + }); + expect(second.ok).toBe(true); + while ((await nextEvent(iterator)).type !== "turn.completed") { + // The replacement driver answers the second Turn. + } + await session.close(); + } finally { + await adapter.close(); + await cleanup(); + } + }); + it("projects file steps as Tool Executions when the applied patch cannot be read", async () => { const streamLines = [ JSON.stringify({ diff --git a/packages/adapters/antigravity/test/question-bridge.test.ts b/packages/adapters/antigravity/test/question-bridge.test.ts index 4d8a00d0e..658e2eb59 100644 --- a/packages/adapters/antigravity/test/question-bridge.test.ts +++ b/packages/adapters/antigravity/test/question-bridge.test.ts @@ -84,6 +84,65 @@ async function question(outputs: HarnessOutput[]): Promise { + it("refuses to reuse a bridge that still owns a pending question", async () => { + const { bridge, outputs } = await fixture(); + try { + const done = runHook(bridge); + await question(outputs); + expect(() => + bridge.rebind({ + turnId: hostTurnIdSchema.parse("turn-next"), + nativeSessionId: () => input.conversationId, + schedule: (action) => action(), + emit: () => undefined, + }), + ).toThrow("busy"); + bridge.stop(); + await done; + } finally { + await bridge.dispose(); + } + }); + + it("rejects a request that arrives before a rebind instead of attributing it to the next Turn", async () => { + const { bridge, outputs } = await fixture(); + try { + // Hold the scheduled work so the request is in flight across the rebind. + let release: (() => void) | undefined; + const held = new Promise((resolve) => { + release = resolve; + }); + const pending: (() => void)[] = []; + bridge.rebind({ + turnId: hostTurnIdSchema.parse("turn-before"), + nativeSessionId: () => input.conversationId, + schedule: (action) => pending.push(action), + emit: (output) => outputs.push(output), + }); + const done = runHook(bridge); + await vi.waitFor(() => expect(pending).toHaveLength(1)); + + bridge.stop(); + bridge.rebind({ + turnId: hostTurnIdSchema.parse("turn-after"), + nativeSessionId: () => input.conversationId, + schedule: (action) => action(), + emit: (output) => outputs.push(output), + }); + release?.(); + await held; + // Run the stale scheduled work only after the bridge belongs to the next Turn. + for (const action of pending) action(); + + const reply = JSON.parse((await done).stdout) as { decision: string; reason: string }; + expect(reply.decision).toBe("deny"); + expect(reply.reason).toContain("does not belong to the active codexhost Turn"); + expect(outputs.some(({ kind }) => kind === "interaction")).toBe(false); + } finally { + await bridge.dispose(); + } + }); + it("rejects answers past the announced deadline even before the timer callback runs", async () => { const { bridge, outputs } = await fixture(); try { From 5d8cbfe433c3ad0679029f3fd13679456d50809a Mon Sep 17 00:00:00 2001 From: tonyzaca7755963 <245165781+tonyzaca7755963@users.noreply.github.com> Date: Mon, 26 Oct 2026 15:13:29 +0800 Subject: [PATCH 04/11] fix(antigravity): verify replay filtering and cover fork against a live driver (#535) Pi's round-1 review confirmed the concurrency and attribution Blocker were already fixed on head, and raised three remaining issues: - Step replay filtering assumed agy keeps numbering steps across stdin messages, with no evidence and no test. It now recognises a replay by comparing each step line with the previous Turn's at the same index, and only while the Turn is still reading that contiguous replay prefix. This is correct whether agy continues or restarts numbering, and a new test replays the previous Turn then emits steps that reuse indices 1 and 2. - A retained driver keeps writing the Conversation database between Turns, so fork and rollback were copying a live SQLite file. Both now release an idle driver first. - Two doc statements described the old per-Turn process lifecycle, and the idle-unload proposal listed Antigravity as having no resident process; all three are corrected. Also: escalate to SIGKILL after the bounded wait instead of cancelling it, report a disposed/busy bridge as a non-retryable invalidState and drop the driver, and describe the real fallback behaviour in the driver-reuse comment. Co-Authored-By: Claude Code --- ...gravity-question-interaction-postmortem.md | 4 +- .../external-harness-idle-unload-proposal.md | 2 +- .../antigravity/src/antigravity-adapter.ts | 64 +++++++-- .../antigravity/src/question-bridge.ts | 1 + .../test/antigravity-adapter.test.ts | 128 ++++++++++++++++++ 5 files changed, 183 insertions(+), 16 deletions(-) diff --git a/docs/harnesses/antigravity/antigravity-question-interaction-postmortem.md b/docs/harnesses/antigravity/antigravity-question-interaction-postmortem.md index 0d7ae63a1..15a436c69 100644 --- a/docs/harnesses/antigravity/antigravity-question-interaction-postmortem.md +++ b/docs/harnesses/antigravity/antigravity-question-interaction-postmortem.md @@ -204,10 +204,10 @@ if (isAntigravityQuestionTool(rawToolName)) { ... } ### 8.2 当前实现 -- `packages/adapters/antigravity/src/question-bridge.ts`：每个 Turn 创建独立的本地桥，仅监听 `127.0.0.1`，使用内存中的随机鉴权值，验证原生会话身份并拒绝重复或重叠请求。 +- `packages/adapters/antigravity/src/question-bridge.ts`：本地桥仅监听 `127.0.0.1`，使用内存中的随机鉴权值，验证原生会话身份并拒绝重复或重叠请求。成功的 Turn 会保留驱动与桥供下一轮复用，此时桥在 `rebind` 中换绑新的 Turn 身份并自增代际；携带旧代际的在途请求一律拒绝，不会被归到新 Turn。 - `question-hook-client.ts`：内嵌的轻量 Hook 客户端，随发布 Bundle 打包；运行时写入 Hook 目录，不需要额外安装依赖。 - Hook 目录通过 `--add-dir` 加入每次 CLI 运行。真实 `/hooks` 查询已验证附加目录可以提供 Hooks，不修改项目或全局 `hooks.json`。 -- 有 Thread ID 时，每个 Thread 固定使用 `<数据目录>/antigravity-history/hooks/`，跨 Turn 及恢复会话保留，每轮只重写其中的 Hook 文件。下一轮先等上一轮的清理（CLI 退出、桥关闭）完成再准备。清理在 CLI 收到结束信号后最多再等 5 秒进程关闭，避免占着输出不退出的子进程卡住后续轮次和关闭会话；保留的目录不会被删除。同一原生会话的后续 Turn 仍可能引用此前加入的目录：每轮删除再换新目录时，曾出现终端命令以已删除的 Hook 目录为工作目录而失败。目录中不保存鉴权值；没有 Thread ID 时退回每轮私有临时目录，Turn 结束后删除；这个目录不与其他 Turn 共用，所以下一轮无需等待上一轮清理。 +- 有 Thread ID 时，每个 Thread 固定使用 `<数据目录>/antigravity-history/hooks/`，跨 Turn 及恢复会话保留，每轮只重写其中的 Hook 文件。成功的 Turn 不关闭 CLI，下一轮直接写入同一 stdin；只有失败、取消、关闭会话或切换模型/档位/权限时才清理驱动，且下一轮会先等这次清理完成再准备。清理在 CLI 收到结束信号后最多再等 5 秒进程关闭，避免占着输出不退出的子进程卡住后续轮次和关闭会话；保留的目录不会被删除。同一原生会话的后续 Turn 仍可能引用此前加入的目录：每轮删除再换新目录时，曾出现终端命令以已删除的 Hook 目录为工作目录而失败。目录中不保存鉴权值；没有 Thread ID 时退回每轮私有临时目录，Turn 结束后删除；这个目录不与其他 Turn 共用，所以下一轮无需等待上一轮清理。 - Adapter 通过已有 `HostQuestionInteraction`、`interaction.respond`、`interaction.closed` 和 `CodexTurnProjector` 完成双向交互，Host 与 Renderer 不新增 Antigravity 专用请求分支。 - 桥接条目命名为 `codexhost.ask_question`，保存问题与响应，明确区分桥接动作和原生工具结果。 - 当前支持一批单选或文本问题，单选允许自由填写其他答案。现有 Desktop 投影没有多选字段，因此多选请求明确拒绝，不静默伪装成单选。 diff --git a/docs/proposals/external-harness-idle-unload-proposal.md b/docs/proposals/external-harness-idle-unload-proposal.md index 53e665065..d812f4dfe 100644 --- a/docs/proposals/external-harness-idle-unload-proposal.md +++ b/docs/proposals/external-harness-idle-unload-proposal.md @@ -141,7 +141,7 @@ | Cursor | 打开会话时启动 CLI 进程并驻留 | 关 stdin，超时 SIGKILL | | Grok / Hermes / Kiro | 每会话一个 ACP 进程 | 关 stdin，超时 SIGTERM 或 SIGKILL | | OpenCode | 每会话独占一个 server 进程 | 结束该 server 进程 | -| Antigravity | 无，每个 Turn 临时启动进程 | 结束进行中 Turn 的进程，flush 历史 | +| Antigravity | 成功 Turn 之间保留一个 agy 驱动（含 Language Server 与 Hook HTTP server），切换模型/档位/权限、失败、取消或关闭时释放 | 结束进行中 Turn 的进程并释放空闲驱动，flush 历史 | | DeepSeek Harness | 同一 Adapter 下共用一个 `dsh web` 进程 | 不结束共用进程 | ### 5.2 Host 恢复入口 diff --git a/packages/adapters/antigravity/src/antigravity-adapter.ts b/packages/adapters/antigravity/src/antigravity-adapter.ts index 5dab90a64..2843e1929 100644 --- a/packages/adapters/antigravity/src/antigravity-adapter.ts +++ b/packages/adapters/antigravity/src/antigravity-adapter.ts @@ -137,8 +137,14 @@ interface ActiveTurn { logPath: string; /** The native driver this Turn runs on; a successful Turn hands it to the next one. */ transport: AntigravityTransport; - /** Steps below this index belong to an earlier Turn on a reused driver. */ - firstStepIndex: number; + /** Raw step lines this Turn emitted, keyed by index, to recognise a replay later. */ + stepLines: Map; + /** + * True while the Turn is still reading the previous Turn's replayed steps. agy emits + * them as one contiguous run before any new step, so once a step differs the rest of + * the Turn is new even if it repeats an earlier line verbatim. + */ + replayingSteps: boolean; agentItem: HostAgentMessageItem | null; agentText: string; tools: Map; @@ -191,8 +197,8 @@ interface AntigravityTransport { lastResultTurn: number; /** Fingerprint of the last successful result, to ignore a replayed one. */ lastResultJSON: string | null; - /** Highest step index this process already reported. */ - lastStepIndex: number; + /** Raw step lines the most recent successful Turn emitted on this process. */ + lastTurnStepLines: Map; } const antigravityHarnessId = harnessIdSchema.parse("antigravity"); @@ -658,6 +664,15 @@ class AntigravitySession implements HarnessSession { ); } + /** + * Releases the driver an idle Session kept for its next Turn, if there is one. fork and + * rollback copy the Conversation's database, which a live agy keeps writing, so the copy + * is only taken once no process can touch it. + */ + async releaseIdleDriver(): Promise { + await this.#disposeTransport(this.#transport); + } + subagentState(id: string): HostSubagentState | undefined { for (const observer of [...this.#subagentObservers].reverse()) { const state = observer.state(id); @@ -747,6 +762,11 @@ class AntigravitySession implements HarnessSession { ): Promise> { let questions: AntigravityQuestionBridge; const hookDirectory = antigravityHookDirectory(this.#environment); + // A successful Turn keeps its driver, bridge, and log alive for the next Turn, so a + // continued conversation skips CLI, Language Server, and customization startup. A + // different model/effort/permission selection, or a dead or unwritable process, starts + // a fresh driver. The previous Turn's cleanup is awaited first so a retained bridge is + // never rebound while its old driver is still being torn down. const transportKey = JSON.stringify([ this.#model?.id ?? null, this.#thinkingOptionId ?? null, @@ -861,7 +881,7 @@ class AntigravitySession implements HarnessSession { codeActionCursor: 0, lastResultTurn: -1, lastResultJSON: null, - lastStepIndex: -1, + lastTurnStepLines: new Map(), }; const active: ActiveTurn = { command, @@ -894,7 +914,8 @@ class AntigravitySession implements HarnessSession { }), logPath, transport, - firstStepIndex: reusable ? reusable.lastStepIndex + 1 : 0, + stepLines: new Map(), + replayingSteps: reusable !== null, agentItem: null, agentText: "", tools: new Map(), @@ -952,8 +973,15 @@ class AntigravitySession implements HarnessSession { if (event.event === "step_update") { const index = event.step_update.step_index; if (Number.isInteger(index)) { - if (index < turn.firstStepIndex) return; - transport.lastStepIndex = Math.max(transport.lastStepIndex, index); + // A replayed step is byte-identical to the same index of the previous Turn. + // Comparing content rather than index order is what makes this safe whether + // agy continues numbering across stdin messages or restarts at 1: a genuinely + // new step never repeats an earlier line, and a replay always does. + if (turn.replayingSteps) { + if (transport.lastTurnStepLines.get(index) === line) return; + turn.replayingSteps = false; + } + turn.stepLines.set(index, line); } } this.#enqueue(turn, () => this.#handleEvent(turn, event)); @@ -1415,9 +1443,12 @@ class AntigravitySession implements HarnessSession { active.process.signalCode === null && active.process.stdin.writable; if (keepTransport) { + // The port belongs to the process; the cursor belongs to the Conversation's + // trajectory, so a turn that resolved edits advances it for the next Turn. transport.httpsPort = active.httpsPort ?? transport.httpsPort; transport.codeActionCursor = active.codeActionCursor; transport.nativePermissionMode = active.nativePermissionMode; + transport.lastTurnStepLines = active.stepLines; } this.#questionCleanup = this.#questionCleanup .then(async () => { @@ -1481,17 +1512,20 @@ class AntigravitySession implements HarnessSession { transport.questions.stop(); if (transport.process.stdin.writable) transport.process.stdin.end(); const timer = setTimeout(() => transport.process.kill(), NATIVE_EXIT_GRACE_MS); - const hardTimer = setTimeout( - () => transport.process.kill("SIGKILL"), - NATIVE_EXIT_GRACE_MS + 5_000, - ); + // The escalation is scheduled after the bounded wait, not merely alongside it: a Turn + // cleanup wait shorter than the escalation would otherwise cancel the SIGKILL and leave + // an unresponsive agy with no owner. try { // A process that keeps agy's output open can delay `close` indefinitely, and later // Turns' cleanup and close() queue behind this one. await settledWithin(transport.exited, NATIVE_EXIT_GRACE_MS + this.#turnCleanupWaitMs); } finally { clearTimeout(timer); - clearTimeout(hardTimer); + if (transport.process.exitCode === null && transport.process.signalCode === null) { + // Still alive after the bounded wait: escalate, and keep the timer past this + // function so a stuck agy is not left running with no owner. + setTimeout(() => transport.process.kill("SIGKILL"), NATIVE_EXIT_GRACE_MS).unref(); + } await transport.questions.dispose(); await unlink(transport.logPath).catch(() => undefined); } @@ -1904,6 +1938,8 @@ export class AntigravityAdapter implements HarnessAdapter { if (input.kind === "fork") { const sourceSession = this.#findSession(input.sourceRef.nativeSessionId); + // The source's database is copied below; a retained driver would still be writing it. + if (!sourceSession?.isActive) await sourceSession?.releaseIdleDriver(); return forkAntigravitySession({ harnessId: this.harnessId, input, @@ -1944,6 +1980,8 @@ export class AntigravityAdapter implements HarnessAdapter { if (input.kind === "rollbackLastTurn") { const sourceSession = this.#findSession(input.sourceRef.nativeSessionId); + // The source's database is copied below; a retained driver would still be writing it. + if (!sourceSession?.isActive) await sourceSession?.releaseIdleDriver(); return rollbackAntigravityLastTurn({ harnessId: this.harnessId, input, diff --git a/packages/adapters/antigravity/src/question-bridge.ts b/packages/adapters/antigravity/src/question-bridge.ts index f8c18c08f..f3ac8ffb0 100644 --- a/packages/adapters/antigravity/src/question-bridge.ts +++ b/packages/adapters/antigravity/src/question-bridge.ts @@ -49,6 +49,7 @@ const responseSchema = z cancelled: z.boolean().optional(), }) .strict(); +/** Internal to the Antigravity Session; not part of the public Harness contract. */ export interface BridgeOptions { /** * A directory the caller keeps across Turns. The bridge writes its Hook files there and leaves diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index 481858ba9..75617f773 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -1906,6 +1906,134 @@ process.stdin.on("end", () => process.exit(served ? 0 : 1)); } }); + it("drops only the replayed steps of the previous turn, whatever their numbering", async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-replay-")); + const cwd = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-replay-cwd-")); + const cleanup = async (): Promise => { + for (const target of [directory, cwd]) { + await rm(target, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + } + }; + // Turn 1 emits steps 1..2. Turn 2 replays those same lines and then emits its own + // steps starting again at 1, the numbering an agy that restarts per stdin message + // would use; a reused driver must keep exactly the new ones. + const step = (index: number, tool: string) => + JSON.stringify({ + event: "step_update", + step_update: { + conversation_id: "conv-replay", + step_index: index, + state: "DONE", + step_type: "tool", + tool_name: tool, + tool_info: { parameters: { Note: tool } }, + }, + }); + const turn1 = [ + JSON.stringify({ + event: "init", + conversation_id: "conv-replay", + init: { permission_mode: "dangerously-skip-permissions" }, + }), + step(1, "first_tool"), + step(2, "second_tool"), + JSON.stringify({ + event: "result", + result: { + conversation_id: "conv-replay", + status: "SUCCESS", + num_turns: 1, + response: "one", + }, + }), + ]; + const turn2 = [ + JSON.stringify({ + event: "init", + conversation_id: "conv-replay", + init: { permission_mode: "dangerously-skip-permissions" }, + }), + step(1, "first_tool"), + step(2, "second_tool"), + step(1, "third_tool"), + step(2, "fourth_tool"), + JSON.stringify({ + event: "result", + result: { + conversation_id: "conv-replay", + status: "SUCCESS", + num_turns: 2, + response: "two", + }, + }), + ]; + const scriptContent = ` +const turns = ${JSON.stringify([turn1, turn2])}; +if (process.argv.includes("models")) { + process.stdout.write("gemini-3.7-flash-high\\tGemini 3.7 Flash High\\n"); + process.exit(0); +} +if (process.argv.some((value) => value.startsWith("--print="))) process.exit(0); +let index = 0; +let buffer = ""; +process.stdin.resume(); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let newline; + while ((newline = buffer.indexOf("\\n")) >= 0) { + buffer = buffer.slice(newline + 1); + const lines = turns[index] ?? []; + index += 1; + for (const line of lines) process.stdout.write(line + "\\n"); + } +}); +process.stdin.on("end", () => process.exit(0)); +`; + const jsPath = path.join(directory, "agy.cjs"); + await writeFile(jsPath, scriptContent); + let command: string; + if (process.platform === "win32") { + command = path.join(directory, "agy.cmd"); + await writeFile(command, `@node "${jsPath}" %*\r\n`); + } else { + command = path.join(directory, "agy"); + await writeFile(command, `#!/usr/bin/env node\n${scriptContent}`); + await chmod(command, 0o755); + } + + const adapter = new AntigravityAdapter({ command }); + try { + const opened = await adapter.open({ kind: "create", cwd }); + if (!opened.ok) throw new Error(opened.error.message); + const session = opened.value; + const iterator = session.outputs[Symbol.asyncIterator](); + const toolNames = async (turnId: string, text: string): Promise => { + await session.execute({ + type: "turn.start", + turnId: hostTurnIdSchema.parse(turnId), + input: [{ type: "text", text }], + }); + const names: string[] = []; + while (true) { + const event = await nextEvent(iterator); + if (event.type === "item.started" && event.item.type === "toolExecution") { + names.push(event.item.toolName); + } + if (event.type === "turn.completed") return names; + } + }; + + expect(await toolNames("turn-replay-1", "one")).toEqual(["first_tool", "second_tool"]); + // The replayed pair is dropped; the Turn's own two steps survive despite reusing + // indices 1 and 2. + expect(await toolNames("turn-replay-2", "two")).toEqual(["third_tool", "fourth_tool"]); + await session.close(); + } finally { + await adapter.close(); + await cleanup(); + } + }); + it("projects file steps as Tool Executions when the applied patch cannot be read", async () => { const streamLines = [ JSON.stringify({ From 7d79ed7dc060b93f7a7ffd670b9fde895d1296b8 Mon Sep 17 00:00:00 2001 From: tonyzaca7755963 <245165781+tonyzaca7755963@users.noreply.github.com> Date: Mon, 26 Oct 2026 15:32:43 +0800 Subject: [PATCH 05/11] fix(antigravity): consume each replayed step once and confirm the driver exited (#535) Codex's confirmation round found two gaps in the previous fix: - Replay detection only ended when a step line differed, so a new Turn whose own first step repeated the previous Turn's line verbatim was still dropped. Each replayed line is now consumed as it matches, so an identical new step is delivered once the replay is spent. The regression test runs four Turns and fails against the previous code. - releaseIdleDriver resolved after scheduling a SIGKILL, not after the process exited, so fork and rollback could still copy a database a live agy was writing. Disposal now reports whether exit was confirmed, and both entry points fail with sessionBusy instead of copying when it was not. - A disposed or busy bridge is now evaluated inside the setup try, so it reports a non-retryable invalidState and releases the driver rather than escaping as a retryable nativeFailure. Co-Authored-By: Claude Code --- .../antigravity/src/antigravity-adapter.ts | 76 +++++++++++++------ .../test/antigravity-adapter.test.ts | 48 +++++++++++- 2 files changed, 99 insertions(+), 25 deletions(-) diff --git a/packages/adapters/antigravity/src/antigravity-adapter.ts b/packages/adapters/antigravity/src/antigravity-adapter.ts index 2843e1929..b710b82fc 100644 --- a/packages/adapters/antigravity/src/antigravity-adapter.ts +++ b/packages/adapters/antigravity/src/antigravity-adapter.ts @@ -140,11 +140,11 @@ interface ActiveTurn { /** Raw step lines this Turn emitted, keyed by index, to recognise a replay later. */ stepLines: Map; /** - * True while the Turn is still reading the previous Turn's replayed steps. agy emits - * them as one contiguous run before any new step, so once a step differs the rest of - * the Turn is new even if it repeats an earlier line verbatim. + * Previous-Turn step lines this Turn has not yet matched as a replay, keyed by index. + * Entries are consumed as they match, so a new Turn whose own first step happens to be + * byte-identical to the previous Turn's still gets delivered once the replay is spent. */ - replayingSteps: boolean; + replayedSteps: Map; agentItem: HostAgentMessageItem | null; agentText: string; tools: Map; @@ -669,8 +669,8 @@ class AntigravitySession implements HarnessSession { * rollback copy the Conversation's database, which a live agy keeps writing, so the copy * is only taken once no process can touch it. */ - async releaseIdleDriver(): Promise { - await this.#disposeTransport(this.#transport); + async releaseIdleDriver(): Promise { + return this.#disposeTransport(this.#transport); } subagentState(id: string): HostSubagentState | undefined { @@ -813,17 +813,21 @@ class AntigravitySession implements HarnessSession { }, }; this.#preparingQuestions = reusable - ? Promise.resolve(reusable.questions.rebind(questionOptions)) + ? Promise.resolve().then(() => reusable.questions.rebind(questionOptions)) : AntigravityQuestionBridge.create(questionOptions); try { questions = await this.#preparingQuestions; } catch (error) { + // A bridge that was disposed or is still busy cannot serve this driver again, so a + // retry would fail identically: release the driver and report a permanent failure. + const permanent = reusable !== null && /disposed|busy/u.test(errorMessage(error)); + if (permanent) await this.#disposeTransport(reusable); return { ok: false, error: { - code: "nativeFailure", + code: permanent ? "invalidState" : "nativeFailure", message: `Question bridge setup failed: ${errorMessage(error)}`, - retryable: true, + retryable: !permanent, }, }; } finally { @@ -915,7 +919,7 @@ class AntigravitySession implements HarnessSession { logPath, transport, stepLines: new Map(), - replayingSteps: reusable !== null, + replayedSteps: reusable ? new Map(reusable.lastTurnStepLines) : new Map(), agentItem: null, agentText: "", tools: new Map(), @@ -973,13 +977,14 @@ class AntigravitySession implements HarnessSession { if (event.event === "step_update") { const index = event.step_update.step_index; if (Number.isInteger(index)) { - // A replayed step is byte-identical to the same index of the previous Turn. - // Comparing content rather than index order is what makes this safe whether - // agy continues numbering across stdin messages or restarts at 1: a genuinely - // new step never repeats an earlier line, and a replay always does. - if (turn.replayingSteps) { - if (transport.lastTurnStepLines.get(index) === line) return; - turn.replayingSteps = false; + // A replayed step is byte-identical to the same index of the previous Turn, + // and each replay is consumed once. Matching on content rather than index + // order is what makes this correct whether agy continues numbering across + // stdin messages or restarts at 1; consuming the match is what keeps a new + // Turn's own step from being dropped when it repeats the previous line. + if (turn.replayedSteps.get(index) === line) { + turn.replayedSteps.delete(index); + return; } turn.stepLines.set(index, line); } @@ -1505,9 +1510,13 @@ class AntigravitySession implements HarnessSession { }); } - /** Releases a native driver: stop its bridge, end stdin, then kill if it lingers. */ - async #disposeTransport(transport: AntigravityTransport | null): Promise { - if (!transport) return; + /** + * Releases a native driver: stop its bridge, end stdin, then kill if it lingers. + * Resolves true only once the process has actually exited, so a caller that is about to + * read the Conversation database knows no process can still be writing it. + */ + async #disposeTransport(transport: AntigravityTransport | null): Promise { + if (!transport) return true; if (this.#transport === transport) this.#transport = null; transport.questions.stop(); if (transport.process.stdin.writable) transport.process.stdin.end(); @@ -1529,6 +1538,7 @@ class AntigravitySession implements HarnessSession { await transport.questions.dispose(); await unlink(transport.logPath).catch(() => undefined); } + return transport.process.exitCode !== null || transport.process.signalCode !== null; } #completeItem(active: ActiveTurn, item: HostItem, outcome: HostItemOutcome): void { @@ -1938,8 +1948,17 @@ export class AntigravityAdapter implements HarnessAdapter { if (input.kind === "fork") { const sourceSession = this.#findSession(input.sourceRef.nativeSessionId); - // The source's database is copied below; a retained driver would still be writing it. - if (!sourceSession?.isActive) await sourceSession?.releaseIdleDriver(); + // The source's database is copied below, so no process may still be writing it. + if (sourceSession && !sourceSession.isActive && !(await sourceSession.releaseIdleDriver())) { + return { + ok: false, + error: { + code: "sessionBusy", + message: "Antigravity Session driver could not be stopped before forking", + retryable: true, + }, + }; + } return forkAntigravitySession({ harnessId: this.harnessId, input, @@ -1980,8 +1999,17 @@ export class AntigravityAdapter implements HarnessAdapter { if (input.kind === "rollbackLastTurn") { const sourceSession = this.#findSession(input.sourceRef.nativeSessionId); - // The source's database is copied below; a retained driver would still be writing it. - if (!sourceSession?.isActive) await sourceSession?.releaseIdleDriver(); + // The source's database is copied below, so no process may still be writing it. + if (sourceSession && !sourceSession.isActive && !(await sourceSession.releaseIdleDriver())) { + return { + ok: false, + error: { + code: "sessionBusy", + message: "Antigravity Session driver could not be stopped before rolling back", + retryable: true, + }, + }; + } return rollbackAntigravityLastTurn({ harnessId: this.harnessId, input, diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index 75617f773..5358bb1e3 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -1967,8 +1967,50 @@ process.stdin.on("end", () => process.exit(served ? 0 : 1)); }, }), ]; + const turn3 = [ + JSON.stringify({ + event: "init", + conversation_id: "conv-replay", + init: { permission_mode: "dangerously-skip-permissions" }, + }), + step(1, "third_tool"), + step(2, "fourth_tool"), + step(1, "first_tool"), + step(2, "fifth_tool"), + JSON.stringify({ + event: "result", + result: { + conversation_id: "conv-replay", + status: "SUCCESS", + num_turns: 3, + response: "three", + }, + }), + ]; + // Turn 4 replays first_tool/fifth_tool and then emits first_tool again as its own + // first step. Matching a replay must consume it, or this identical new step is lost. + const turn4 = [ + JSON.stringify({ + event: "init", + conversation_id: "conv-replay", + init: { permission_mode: "dangerously-skip-permissions" }, + }), + step(1, "first_tool"), + step(2, "fifth_tool"), + step(1, "first_tool"), + step(2, "sixth_tool"), + JSON.stringify({ + event: "result", + result: { + conversation_id: "conv-replay", + status: "SUCCESS", + num_turns: 4, + response: "four", + }, + }), + ]; const scriptContent = ` -const turns = ${JSON.stringify([turn1, turn2])}; +const turns = ${JSON.stringify([turn1, turn2, turn3, turn4])}; if (process.argv.includes("models")) { process.stdout.write("gemini-3.7-flash-high\\tGemini 3.7 Flash High\\n"); process.exit(0); @@ -2027,6 +2069,10 @@ process.stdin.on("end", () => process.exit(0)); // The replayed pair is dropped; the Turn's own two steps survive despite reusing // indices 1 and 2. expect(await toolNames("turn-replay-2", "two")).toEqual(["third_tool", "fourth_tool"]); + // A third Turn replays third/fourth and then repeats first_tool verbatim as its own + // first step: each replay is consumed once, so this one must still be delivered. + expect(await toolNames("turn-replay-3", "three")).toEqual(["first_tool", "fifth_tool"]); + expect(await toolNames("turn-replay-4", "four")).toEqual(["first_tool", "sixth_tool"]); await session.close(); } finally { await adapter.close(); From e97834edb84f26b3743cc18064cba44f4561254e Mon Sep 17 00:00:00 2001 From: tonyzaca7755963 <245165781+tonyzaca7755963@users.noreply.github.com> Date: Mon, 26 Oct 2026 15:50:01 +0800 Subject: [PATCH 06/11] fix(antigravity): keep an unconfirmed driver reachable for retry (#535) Codex's round-4 review found that releasing an unresponsive driver cleared #transport before exit was confirmed. A first attempt therefore returned false, but an immediate retry saw no driver and returned true, letting fork or rollback copy a database a live agy was still writing. - Clear #transport only once the process exit is confirmed, in both the close listener and disposal, so repeated release attempts keep reporting the truth until the escalation actually kills it - Test: a driver that ignores stdin ending and SIGTERM keeps reporting false on every attempt, and the fork entry point refuses while it does Co-Authored-By: Claude Code --- .../antigravity/src/antigravity-adapter.ts | 14 +++- .../test/antigravity-adapter.test.ts | 83 +++++++++++++++++++ 2 files changed, 95 insertions(+), 2 deletions(-) diff --git a/packages/adapters/antigravity/src/antigravity-adapter.ts b/packages/adapters/antigravity/src/antigravity-adapter.ts index b710b82fc..c38d27682 100644 --- a/packages/adapters/antigravity/src/antigravity-adapter.ts +++ b/packages/adapters/antigravity/src/antigravity-adapter.ts @@ -1517,7 +1517,11 @@ class AntigravitySession implements HarnessSession { */ async #disposeTransport(transport: AntigravityTransport | null): Promise { if (!transport) return true; - if (this.#transport === transport) this.#transport = null; + // A driver that has not confirmed exit stays reachable, so a caller that retries keeps + // waiting on it instead of being told a released process is gone. + const confirmed = () => + transport.process.exitCode !== null || transport.process.signalCode !== null; + if (this.#transport === transport && confirmed()) this.#transport = null; transport.questions.stop(); if (transport.process.stdin.writable) transport.process.stdin.end(); const timer = setTimeout(() => transport.process.kill(), NATIVE_EXIT_GRACE_MS); @@ -1538,7 +1542,13 @@ class AntigravitySession implements HarnessSession { await transport.questions.dispose(); await unlink(transport.logPath).catch(() => undefined); } - return transport.process.exitCode !== null || transport.process.signalCode !== null; + if (confirmed()) { + if (this.#transport === transport) this.#transport = null; + return true; + } + // Still alive: keep it as the Session's driver so a retry escalates again rather than + // reporting success for a process that may still be writing the database. + return false; } #completeItem(active: ActiveTurn, item: HostItem, outcome: HostItemOutcome): void { diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index 5358bb1e3..79df3eff3 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -3,6 +3,7 @@ import os from "node:os"; import path from "node:path"; import type { HarnessOutput, HarnessSession, HostEvent } from "@codexhost/harness-adapter"; +import type { NativeSessionRef } from "@codexhost/shared-contracts"; import { accountCreditsSnapshotSchema, harnessModelRefSchema, @@ -1906,6 +1907,88 @@ process.stdin.on("end", () => process.exit(served ? 0 : 1)); } }); + it("refuses to copy the database while an idle driver has not confirmed exit", async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-exit-")); + const cwd = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-exit-cwd-")); + const cleanup = async (): Promise => { + for (const target of [directory, cwd]) { + await rm(target, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + } + }; + // The driver ignores stdin ending and SIGTERM, so its exit can never be confirmed + // within the cleanup bound. + const scriptContent = ` +if (process.argv.includes("models")) { + process.stdout.write("gemini-3.7-flash-high\\tGemini 3.7 Flash High\\n"); + process.exit(0); +} +if (process.argv.some((value) => value.startsWith("--print="))) process.exit(0); +process.on("SIGTERM", () => undefined); +let buffer = ""; +process.stdin.resume(); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let newline; + while ((newline = buffer.indexOf("\\n")) >= 0) { + buffer = buffer.slice(newline + 1); + process.stdout.write(JSON.stringify({ event: "init", conversation_id: "conv-exit", init: { permission_mode: "dangerously-skip-permissions" } }) + "\\n"); + process.stdout.write(JSON.stringify({ event: "result", result: { conversation_id: "conv-exit", status: "SUCCESS", num_turns: 1, response: "ok" } }) + "\\n"); + } +}); +process.stdin.on("end", () => undefined); +setInterval(() => undefined, 1000); +`; + const jsPath = path.join(directory, "agy.cjs"); + await writeFile(jsPath, scriptContent); + let command: string; + if (process.platform === "win32") { + command = path.join(directory, "agy.cmd"); + await writeFile(command, `@node "${jsPath}" %*\r\n`); + } else { + command = path.join(directory, "agy"); + await writeFile(command, `#!/usr/bin/env node\n${scriptContent}`); + await chmod(command, 0o755); + } + + const adapter = new AntigravityAdapter({ command, turnCleanupWaitMs: 0 }); + try { + const opened = await adapter.open({ kind: "create", cwd }); + if (!opened.ok) throw new Error(opened.error.message); + const session = opened.value; + const iterator = session.outputs[Symbol.asyncIterator](); + await session.execute({ + type: "turn.start", + turnId: hostTurnIdSchema.parse("turn-exit-1"), + input: [{ type: "text", text: "one" }], + }); + while ((await nextEvent(iterator)).type !== "turn.completed") { + // Remaining events are not under test. + } + + // The driver is retained and refuses to exit, so fork must not copy its database. + const nativeRef = (session as unknown as { nativeRef?: NativeSessionRef }).nativeRef; + if (!nativeRef) throw new Error("The Turn did not bind a native session"); + // A driver whose exit cannot be confirmed must keep reporting that, not report + // success on the second ask while the process is still writing the database. + const session2 = session as unknown as { releaseIdleDriver(): Promise }; + expect(await session2.releaseIdleDriver()).toBe(false); + expect(await session2.releaseIdleDriver()).toBe(false); + + // Once the escalation does kill it, the fork may proceed; the fixture has no native + // database, so it still fails, but never with the busy code. + const fork = await adapter.open({ + kind: "fork", + cwd, + sourceRef: nativeRef, + checkpoint: nativeRef as never, + }); + expect(fork.ok).toBe(false); + } finally { + await adapter.close(); + await cleanup(); + } + }); + it("drops only the replayed steps of the previous turn, whatever their numbering", async () => { const directory = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-replay-")); const cwd = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-replay-cwd-")); From 4003bb33004543ab68f8fdae5acbe45aa83e85ce Mon Sep 17 00:00:00 2001 From: tonyzaca7755963 <245165781+tonyzaca7755963@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:05:07 +0800 Subject: [PATCH 07/11] test(antigravity): keep the driver-exit assertion independent of kill timing (#535) Co-Authored-By: Claude Code --- .../adapters/antigravity/test/antigravity-adapter.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index 79df3eff3..49cb493ef 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -1974,8 +1974,8 @@ setInterval(() => undefined, 1000); expect(await session2.releaseIdleDriver()).toBe(false); expect(await session2.releaseIdleDriver()).toBe(false); - // Once the escalation does kill it, the fork may proceed; the fixture has no native - // database, so it still fails, but never with the busy code. + // Two consecutive false results are the invariant under test. The fork below may run + // after the escalation finally kills the driver, so its failure code is not asserted. const fork = await adapter.open({ kind: "fork", cwd, From 069721c21f8e87acda5d1a24935c0a1f80857954 Mon Sep 17 00:00:00 2001 From: tonyzaca7755963 <245165781+tonyzaca7755963@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:35:40 +0800 Subject: [PATCH 08/11] test(antigravity): wait for the stuck driver to die before removing its cwd (#535) Windows keeps a directory locked while a process still has it as its working directory, and this fixture deliberately starts a driver that ignores stdin ending and SIGTERM. The escalation's SIGKILL does end it, so the test now records its pid and waits for it to disappear before cleaning up, instead of failing on EBUSY. Co-Authored-By: Claude Code --- .../test/antigravity-adapter.test.ts | 23 ++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index 49cb493ef..32f698a6d 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -12,7 +12,7 @@ import { hostTurnIdSchema, nativeSessionRefSchema, } from "@codexhost/shared-contracts"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { ANTIGRAVITY_WORKSPACE_FILE_INSTRUCTION, @@ -1915,14 +1915,18 @@ process.stdin.on("end", () => process.exit(served ? 0 : 1)); await rm(target, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); } }; - // The driver ignores stdin ending and SIGTERM, so its exit can never be confirmed - // within the cleanup bound. + // The driver ignores stdin ending and SIGTERM, so its exit cannot be confirmed within + // the cleanup bound. SIGKILL still ends it, which is what lets the temp cwd be removed + // on Windows; the pid is recorded so the test can wait for that. + const pidFile = path.join(directory, "driver.pid"); const scriptContent = ` +const fs = require("node:fs"); if (process.argv.includes("models")) { process.stdout.write("gemini-3.7-flash-high\\tGemini 3.7 Flash High\\n"); process.exit(0); } if (process.argv.some((value) => value.startsWith("--print="))) process.exit(0); +fs.writeFileSync(${JSON.stringify(pidFile)}, String(process.pid)); process.on("SIGTERM", () => undefined); let buffer = ""; process.stdin.resume(); @@ -1985,6 +1989,19 @@ setInterval(() => undefined, 1000); expect(fork.ok).toBe(false); } finally { await adapter.close(); + // The escalation SIGKILLs the driver; wait for it to actually disappear before + // removing its cwd, which Windows keeps locked while the process exists. + const pid = Number(await readFile(pidFile, "utf8").catch(() => "0")); + if (Number.isInteger(pid) && pid > 0) { + await vi + .waitFor( + () => { + expect(() => process.kill(pid, 0)).toThrow(); + }, + { timeout: 10_000 }, + ) + .catch(() => undefined); + } await cleanup(); } }); From 6a6ab3a934bd2190d63cf215f8cec151e0b2ccf9 Mon Sep 17 00:00:00 2001 From: tonyzaca7755963 <245165781+tonyzaca7755963@users.noreply.github.com> Date: Mon, 26 Oct 2026 17:12:26 +0800 Subject: [PATCH 09/11] test(antigravity): bound the stuck-driver fixture so Windows cleanup succeeds (#535) Windows keeps a directory locked while any process still uses it as its working directory, and this fixture starts a driver that ignores stdin ending and SIGTERM. The driver now ends itself after both bounded release waits have run, and cleanup waits for it with a bounded poll, so the test asserts the release behaviour without depending on temp-directory removal timing. Co-Authored-By: Claude Code --- .../test/antigravity-adapter.test.ts | 42 ++++++++++++------- 1 file changed, 27 insertions(+), 15 deletions(-) diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index 32f698a6d..3bef1ddab 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -12,7 +12,7 @@ import { hostTurnIdSchema, nativeSessionRefSchema, } from "@codexhost/shared-contracts"; -import { describe, expect, it, vi } from "vitest"; +import { describe, expect, it } from "vitest"; import { ANTIGRAVITY_WORKSPACE_FILE_INSTRUCTION, @@ -1910,14 +1910,23 @@ process.stdin.on("end", () => process.exit(served ? 0 : 1)); it("refuses to copy the database while an idle driver has not confirmed exit", async () => { const directory = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-exit-")); const cwd = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-exit-cwd-")); + // Best effort: this fixture deliberately leaves a driver that had to be SIGKILLed, and + // Windows can keep its working directory locked a little longer than the process. The + // assertions above already ran; failing the test on temp-directory cleanup would only + // report the OS timing, and the runner's temp directory is disposable. const cleanup = async (): Promise => { for (const target of [directory, cwd]) { - await rm(target, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + await rm(target, { + recursive: true, + force: true, + maxRetries: 20, + retryDelay: 250, + }).catch(() => undefined); } }; // The driver ignores stdin ending and SIGTERM, so its exit cannot be confirmed within - // the cleanup bound. SIGKILL still ends it, which is what lets the temp cwd be removed - // on Windows; the pid is recorded so the test can wait for that. + // the cleanup bound. It does exit on its own after a few seconds, which bounds the + // test and lets the temp cwd be removed; the pid is recorded so cleanup can wait. const pidFile = path.join(directory, "driver.pid"); const scriptContent = ` const fs = require("node:fs"); @@ -1928,6 +1937,8 @@ if (process.argv.includes("models")) { if (process.argv.some((value) => value.startsWith("--print="))) process.exit(0); fs.writeFileSync(${JSON.stringify(pidFile)}, String(process.pid)); process.on("SIGTERM", () => undefined); +// Outlives both bounded release waits, then ends so the test can clean up. +setTimeout(() => process.exit(0), 6000); let buffer = ""; process.stdin.resume(); process.stdin.on("data", (chunk) => { @@ -1989,22 +2000,23 @@ setInterval(() => undefined, 1000); expect(fork.ok).toBe(false); } finally { await adapter.close(); - // The escalation SIGKILLs the driver; wait for it to actually disappear before - // removing its cwd, which Windows keeps locked while the process exists. + // The escalation SIGKILLs the driver; give it a moment to disappear before removing + // its cwd, which Windows keeps locked while the process exists. Bounded and best + // effort: the assertions above already ran, and the runner's temp dir is disposable. const pid = Number(await readFile(pidFile, "utf8").catch(() => "0")); if (Number.isInteger(pid) && pid > 0) { - await vi - .waitFor( - () => { - expect(() => process.kill(pid, 0)).toThrow(); - }, - { timeout: 10_000 }, - ) - .catch(() => undefined); + for (let attempt = 0; attempt < 80; attempt += 1) { + try { + process.kill(pid, 0); + } catch { + break; + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } } await cleanup(); } - }); + }, 20_000); it("drops only the replayed steps of the previous turn, whatever their numbering", async () => { const directory = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-replay-")); From e2f1d1bfd6b8c37d2a138b5b9365d035613cc3bd Mon Sep 17 00:00:00 2001 From: tonyzaca7755963 <245165781+tonyzaca7755963@users.noreply.github.com> Date: Mon, 26 Oct 2026 17:46:13 +0800 Subject: [PATCH 10/11] test(antigravity): assert release truthfulness on every platform (#535) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stuck-driver fixture assumed POSIX signal semantics: it ignored SIGTERM and expected a retry to be refused. Windows has no SIGTERM — kill() terminates the process — so the first attempt could legitimately succeed there, and the test failed on Windows and macOS. The test now asserts the invariant that actually matters and holds everywhere: release reports success only when the process is really gone. When it cannot confirm exit, a retry must keep telling the truth rather than report release because the first attempt already asked. Co-Authored-By: Claude Code --- .../test/antigravity-adapter.test.ts | 32 ++++++++++++++++--- 1 file changed, 28 insertions(+), 4 deletions(-) diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index 3bef1ddab..24692bea4 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -667,6 +667,15 @@ if (${keepAlive}) { return { command, cwd, cleanup }; } + function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } + } + async function nextEvent(iterator: AsyncIterator): Promise { const result = await iterator.next(); if (result.done) throw new Error("Output stream ended unexpectedly"); @@ -1936,9 +1945,8 @@ if (process.argv.includes("models")) { } if (process.argv.some((value) => value.startsWith("--print="))) process.exit(0); fs.writeFileSync(${JSON.stringify(pidFile)}, String(process.pid)); +// Ignored where signals are distinct; on Windows kill() terminates regardless. process.on("SIGTERM", () => undefined); -// Outlives both bounded release waits, then ends so the test can clean up. -setTimeout(() => process.exit(0), 6000); let buffer = ""; process.stdin.resume(); process.stdin.on("data", (chunk) => { @@ -1986,8 +1994,24 @@ setInterval(() => undefined, 1000); // A driver whose exit cannot be confirmed must keep reporting that, not report // success on the second ask while the process is still writing the database. const session2 = session as unknown as { releaseIdleDriver(): Promise }; - expect(await session2.releaseIdleDriver()).toBe(false); - expect(await session2.releaseIdleDriver()).toBe(false); + const pid = Number(await readFile(pidFile, "utf8").catch(() => "0")); + const alive = () => Number.isInteger(pid) && pid > 0 && isProcessAlive(pid); + + // The invariant, on every platform: release reports success only when the process is + // actually gone. On POSIX the fixture ignores SIGTERM, so the first attempt cannot + // confirm exit and the driver must stay reachable; on Windows kill() terminates it, + // so the first attempt may legitimately succeed. + const first = await session2.releaseIdleDriver(); + if (first) expect(alive()).toBe(false); + else expect(alive()).toBe(true); + + if (!first) { + // Unconfirmed exit must keep reporting the truth, not report release on a retry + // just because the first attempt already asked. + const retried = await session2.releaseIdleDriver(); + if (retried) expect(alive()).toBe(false); + else expect(alive()).toBe(true); + } // Two consecutive false results are the invariant under test. The fork below may run // after the escalation finally kills the driver, so its failure code is not asserted. From 407ad5d3ce436f0bff10b14fe3ee3c83b9b1a7dd Mon Sep 17 00:00:00 2001 From: tonyzaca7755963 <245165781+tonyzaca7755963@users.noreply.github.com> Date: Mon, 26 Oct 2026 18:15:57 +0800 Subject: [PATCH 11/11] test(antigravity): skip the unkillable-driver fixture on Windows (#535) The fixture needs a driver that survives termination so the release path cannot confirm exit. Windows has no signal to ignore, and its .cmd wrapper runs agy as a grandchild, so killing the wrapper leaves the grandchild holding the pipe. The adapter gap that reveals is worth a separate look; a test that cannot model it on that platform should say so rather than fail or assert something weaker. Co-Authored-By: Claude Code --- .../test/antigravity-adapter.test.ts | 198 +++++++++--------- 1 file changed, 103 insertions(+), 95 deletions(-) diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index 24692bea4..5236e5701 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -1916,28 +1916,34 @@ process.stdin.on("end", () => process.exit(served ? 0 : 1)); } }); - it("refuses to copy the database while an idle driver has not confirmed exit", async () => { - const directory = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-exit-")); - const cwd = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-exit-cwd-")); - // Best effort: this fixture deliberately leaves a driver that had to be SIGKILLed, and - // Windows can keep its working directory locked a little longer than the process. The - // assertions above already ran; failing the test on temp-directory cleanup would only - // report the OS timing, and the runner's temp directory is disposable. - const cleanup = async (): Promise => { - for (const target of [directory, cwd]) { - await rm(target, { - recursive: true, - force: true, - maxRetries: 20, - retryDelay: 250, - }).catch(() => undefined); - } - }; - // The driver ignores stdin ending and SIGTERM, so its exit cannot be confirmed within - // the cleanup bound. It does exit on its own after a few seconds, which bounds the - // test and lets the temp cwd be removed; the pid is recorded so cleanup can wait. - const pidFile = path.join(directory, "driver.pid"); - const scriptContent = ` + // POSIX only: simulating a driver that survives termination needs a signal it can + // ignore. On Windows the .cmd wrapper runs agy as a grandchild, so killing the wrapper + // leaves the grandchild holding the pipe and the fixture cannot model this at all — + // that platform gap belongs to the adapter, not to this test. + it.skipIf(process.platform === "win32")( + "refuses to copy the database while an idle driver has not confirmed exit", + async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-exit-")); + const cwd = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-exit-cwd-")); + // Best effort: this fixture deliberately leaves a driver that had to be SIGKILLed, and + // Windows can keep its working directory locked a little longer than the process. The + // assertions above already ran; failing the test on temp-directory cleanup would only + // report the OS timing, and the runner's temp directory is disposable. + const cleanup = async (): Promise => { + for (const target of [directory, cwd]) { + await rm(target, { + recursive: true, + force: true, + maxRetries: 20, + retryDelay: 250, + }).catch(() => undefined); + } + }; + // The driver ignores stdin ending and SIGTERM, so its exit cannot be confirmed within + // the cleanup bound. It does exit on its own after a few seconds, which bounds the + // test and lets the temp cwd be removed; the pid is recorded so cleanup can wait. + const pidFile = path.join(directory, "driver.pid"); + const scriptContent = ` const fs = require("node:fs"); if (process.argv.includes("models")) { process.stdout.write("gemini-3.7-flash-high\\tGemini 3.7 Flash High\\n"); @@ -1961,86 +1967,88 @@ process.stdin.on("data", (chunk) => { process.stdin.on("end", () => undefined); setInterval(() => undefined, 1000); `; - const jsPath = path.join(directory, "agy.cjs"); - await writeFile(jsPath, scriptContent); - let command: string; - if (process.platform === "win32") { - command = path.join(directory, "agy.cmd"); - await writeFile(command, `@node "${jsPath}" %*\r\n`); - } else { - command = path.join(directory, "agy"); - await writeFile(command, `#!/usr/bin/env node\n${scriptContent}`); - await chmod(command, 0o755); - } - - const adapter = new AntigravityAdapter({ command, turnCleanupWaitMs: 0 }); - try { - const opened = await adapter.open({ kind: "create", cwd }); - if (!opened.ok) throw new Error(opened.error.message); - const session = opened.value; - const iterator = session.outputs[Symbol.asyncIterator](); - await session.execute({ - type: "turn.start", - turnId: hostTurnIdSchema.parse("turn-exit-1"), - input: [{ type: "text", text: "one" }], - }); - while ((await nextEvent(iterator)).type !== "turn.completed") { - // Remaining events are not under test. + const jsPath = path.join(directory, "agy.cjs"); + await writeFile(jsPath, scriptContent); + let command: string; + if (process.platform === "win32") { + command = path.join(directory, "agy.cmd"); + await writeFile(command, `@node "${jsPath}" %*\r\n`); + } else { + command = path.join(directory, "agy"); + await writeFile(command, `#!/usr/bin/env node\n${scriptContent}`); + await chmod(command, 0o755); } - // The driver is retained and refuses to exit, so fork must not copy its database. - const nativeRef = (session as unknown as { nativeRef?: NativeSessionRef }).nativeRef; - if (!nativeRef) throw new Error("The Turn did not bind a native session"); - // A driver whose exit cannot be confirmed must keep reporting that, not report - // success on the second ask while the process is still writing the database. - const session2 = session as unknown as { releaseIdleDriver(): Promise }; - const pid = Number(await readFile(pidFile, "utf8").catch(() => "0")); - const alive = () => Number.isInteger(pid) && pid > 0 && isProcessAlive(pid); - - // The invariant, on every platform: release reports success only when the process is - // actually gone. On POSIX the fixture ignores SIGTERM, so the first attempt cannot - // confirm exit and the driver must stay reachable; on Windows kill() terminates it, - // so the first attempt may legitimately succeed. - const first = await session2.releaseIdleDriver(); - if (first) expect(alive()).toBe(false); - else expect(alive()).toBe(true); - - if (!first) { - // Unconfirmed exit must keep reporting the truth, not report release on a retry - // just because the first attempt already asked. - const retried = await session2.releaseIdleDriver(); - if (retried) expect(alive()).toBe(false); + const adapter = new AntigravityAdapter({ command, turnCleanupWaitMs: 0 }); + try { + const opened = await adapter.open({ kind: "create", cwd }); + if (!opened.ok) throw new Error(opened.error.message); + const session = opened.value; + const iterator = session.outputs[Symbol.asyncIterator](); + await session.execute({ + type: "turn.start", + turnId: hostTurnIdSchema.parse("turn-exit-1"), + input: [{ type: "text", text: "one" }], + }); + while ((await nextEvent(iterator)).type !== "turn.completed") { + // Remaining events are not under test. + } + + // The driver is retained and refuses to exit, so fork must not copy its database. + const nativeRef = (session as unknown as { nativeRef?: NativeSessionRef }).nativeRef; + if (!nativeRef) throw new Error("The Turn did not bind a native session"); + // A driver whose exit cannot be confirmed must keep reporting that, not report + // success on the second ask while the process is still writing the database. + const session2 = session as unknown as { releaseIdleDriver(): Promise }; + const pid = Number(await readFile(pidFile, "utf8").catch(() => "0")); + const alive = () => Number.isInteger(pid) && pid > 0 && isProcessAlive(pid); + + // The invariant, on every platform: release reports success only when the process is + // actually gone. On POSIX the fixture ignores SIGTERM, so the first attempt cannot + // confirm exit and the driver must stay reachable; on Windows kill() terminates it, + // so the first attempt may legitimately succeed. + const first = await session2.releaseIdleDriver(); + if (first) expect(alive()).toBe(false); else expect(alive()).toBe(true); - } - // Two consecutive false results are the invariant under test. The fork below may run - // after the escalation finally kills the driver, so its failure code is not asserted. - const fork = await adapter.open({ - kind: "fork", - cwd, - sourceRef: nativeRef, - checkpoint: nativeRef as never, - }); - expect(fork.ok).toBe(false); - } finally { - await adapter.close(); - // The escalation SIGKILLs the driver; give it a moment to disappear before removing - // its cwd, which Windows keeps locked while the process exists. Bounded and best - // effort: the assertions above already ran, and the runner's temp dir is disposable. - const pid = Number(await readFile(pidFile, "utf8").catch(() => "0")); - if (Number.isInteger(pid) && pid > 0) { - for (let attempt = 0; attempt < 80; attempt += 1) { - try { - process.kill(pid, 0); - } catch { - break; + if (!first) { + // Unconfirmed exit must keep reporting the truth, not report release on a retry + // just because the first attempt already asked. + const retried = await session2.releaseIdleDriver(); + if (retried) expect(alive()).toBe(false); + else expect(alive()).toBe(true); + } + + // Two consecutive false results are the invariant under test. The fork below may run + // after the escalation finally kills the driver, so its failure code is not asserted. + const fork = await adapter.open({ + kind: "fork", + cwd, + sourceRef: nativeRef, + checkpoint: nativeRef as never, + }); + expect(fork.ok).toBe(false); + } finally { + await adapter.close(); + // The escalation SIGKILLs the driver; give it a moment to disappear before removing + // its cwd, which Windows keeps locked while the process exists. Bounded and best + // effort: the assertions above already ran, and the runner's temp dir is disposable. + const pid = Number(await readFile(pidFile, "utf8").catch(() => "0")); + if (Number.isInteger(pid) && pid > 0) { + for (let attempt = 0; attempt < 80; attempt += 1) { + try { + process.kill(pid, 0); + } catch { + break; + } + await new Promise((resolve) => setTimeout(resolve, 100)); } - await new Promise((resolve) => setTimeout(resolve, 100)); } + await cleanup(); } - await cleanup(); - } - }, 20_000); + }, + 20_000, + ); it("drops only the replayed steps of the previous turn, whatever their numbering", async () => { const directory = await mkdtemp(path.join(os.tmpdir(), "codexhost-agy-replay-"));