feat(antigravity): reuse one stream-json driver across successful turns (#535)

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/<Thread ID>`，跨 Turn 及恢复会话保留，每轮只重写其中的 Hook 文件。下一轮先等上一轮的清理（CLI 退出、桥关闭）完成再准备。清理在 CLI 收到结束信号后最多再等 5 秒进程关闭，避免占着输出不退出的子进程卡住后续轮次和关闭会话；保留的目录不会被删除。同一原生会话的后续 Turn 仍可能引用此前加入的目录：每轮删除再换新目录时，曾出现终端命令以已删除的 Hook 目录为工作目录而失败。目录中不保存鉴权值；没有 Thread ID 时退回每轮私有临时目录，Turn 结束后删除；这个目录不与其他 Turn 共用，所以下一轮无需等待上一轮清理。
+- 有 Thread ID 时，每个 Thread 固定使用 `<数据目录>/antigravity-history/hooks/<Thread ID>`，跨 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 投影没有多选字段，因此多选请求明确拒绝，不静默伪装成单选。

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 恢复入口

packages/adapters/antigravity/src/question-bridge.ts

@@ -49,7 +49,8 @@ const responseSchema = z
     cancelled: z.boolean().optional(),
   })
   .strict();
-interface BridgeOptions {
+/** 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
    * the directory on dispose; without one it uses a private temporary directory it removes.
@@ -89,6 +90,12 @@ export class AntigravityQuestionBridge {
   readonly #pending = new Map<string, PendingQuestion>();
   readonly #seenSteps = new Set<string>();
   #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<void> | null = null;
 
   private constructor(directory: string, options: BridgeOptions) {
@@ -158,6 +165,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 +208,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<typeof requestSchema>, response: ServerResponse): void {
+  #ask(payload: z.infer<typeof requestSchema>, response: ServerResponse, generation: number): void {
     if (
       this.#stopped ||
+      generation !== this.#generation ||
       response.destroyed ||
       payload.conversationId !== this.#options.nativeSessionId()
     ) {
@@ -374,6 +385,28 @@ 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();
+    this.#generation += 1;
+    return this;
+  }
+
   stop(): void {
     if (this.#stopped) return;
     this.#stopped = true;

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");

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") {