From 39bc46897f65184118fb823541e828d1e353a207 Mon Sep 17 00:00:00 2001 From: yuwk <1729065730@qq.com> Date: Wed, 4 Nov 2026 16:50:02 +0800 Subject: [PATCH 1/2] fix(antigravity): project planner thinking steps as reasoning Items (#738) --- .../antigravity/src/antigravity-adapter.ts | 78 +++++++++- .../antigravity/src/code-action-diff.ts | 86 +++++++++++ packages/adapters/antigravity/src/index.ts | 2 + .../adapters/antigravity/src/stream-events.ts | 2 + .../antigravity/src/subagent-transcript.ts | 60 ++++++++ .../test/antigravity-adapter.test.ts | 140 +++++++++++++++++- .../antigravity/test/code-action-diff.test.ts | 35 +++++ .../antigravity/test/subagents.test.ts | 74 +++++++++ 8 files changed, 475 insertions(+), 2 deletions(-) diff --git a/packages/adapters/antigravity/src/antigravity-adapter.ts b/packages/adapters/antigravity/src/antigravity-adapter.ts index 3fca920ac..c6c6c19f6 100644 --- a/packages/adapters/antigravity/src/antigravity-adapter.ts +++ b/packages/adapters/antigravity/src/antigravity-adapter.ts @@ -88,13 +88,18 @@ import { rollbackAntigravityLastTurn } from "./rollback.js"; import { ANTIGRAVITY_COMMAND_CATALOG, parseAndFormatAntigravityCommand } from "./slash-commands.js"; import { codeActionFileChange, + requestAntigravityStepThinking, requestAntigravityTrajectorySteps, type AntigravityCodeAction, } from "./code-action-diff.js"; import { fetchAntigravityQuota, type AntigravityQuotaSnapshot } from "./quota.js"; import { AntigravityQuestionBridge } from "./question-bridge.js"; import { AntigravitySubagents } from "./subagents.js"; -import { nativeSubagentIdSchema, readSubagentTranscript } from "./subagent-transcript.js"; +import { + nativeSubagentIdSchema, + readSubagentTranscript, + readTranscriptStepThinking, +} from "./subagent-transcript.js"; import { createAntigravityUsageStatistics } from "./usage-statistics.js"; import { antigravityToolErrorMessage, @@ -170,6 +175,8 @@ interface ActiveTurn { /** First sighting of a step whose Item was deferred: agy names the tool and * its parameters when the step opens, not when it ends. */ pendingSteps: Map; + thinkingSteps: Set; + checkedActiveThinkingSteps: Set; httpsPort: number | null; } @@ -866,6 +873,8 @@ class AntigravitySession implements HarnessSession { codeActionCursor: 0, fileChanges: new Map(), pendingSteps: new Map(), + thinkingSteps: new Set(), + checkedActiveThinkingSteps: new Set(), httpsPort: null, }; this.#active = active; @@ -1188,6 +1197,7 @@ class AntigravitySession implements HarnessSession { step = { ...step, step_type: "tool" }; } if (step.step_type === "agent_response") { + await this.#maybeProjectStepThinking(active, step); if (typeof step.text_delta === "string" && step.text_delta.length > 0) { this.#appendOrSyncAgentText(active, step.text_delta, true); return; @@ -1304,6 +1314,72 @@ class AntigravitySession implements HarnessSession { return port; } + /** + * `agy --output-format stream-json` strips `` blocks from step text + * and omits `thinking` from `step_update`, reporting only `usage.thinking_tokens` + * when a planner step finishes. Resolve the step's thinking from inline fields, + * the Language Server trajectory, or the native brain transcript before + * projecting the step's visible response or subsequent tool calls. + */ + async #maybeProjectStepThinking( + active: ActiveTurn, + step: AntigravityStepUpdateEvent["step_update"], + ): Promise { + if (active.thinkingSteps.has(step.step_index)) return; + const isTerminal = step.state === "DONE" || step.state === "ERROR"; + if (!isTerminal && active.checkedActiveThinkingSteps.has(step.step_index)) return; + if (!isTerminal) active.checkedActiveThinkingSteps.add(step.step_index); + + const inlineThinking = + typeof step.thinking_delta === "string" && step.thinking_delta.trim().length > 0 + ? step.thinking_delta + : typeof step.thinking === "string" && step.thinking.trim().length > 0 + ? step.thinking + : null; + let thinking = inlineThinking; + if (thinking === null) { + const thinkingTokens = safeToken(step.usage?.thinking_tokens); + if (!isTerminal || thinkingTokens === undefined || thinkingTokens > 0) { + const port = await this.#languageServerPort(active); + if (port !== null) { + thinking = await requestAntigravityStepThinking( + port, + step.conversation_id, + step.step_index, + TRAJECTORY_TIMEOUT_MS, + ); + } + if (thinking === null) { + const home = + (process.platform === "win32" + ? this.#environment.USERPROFILE + : this.#environment.HOME) || undefined; + thinking = await readTranscriptStepThinking( + step.conversation_id, + step.step_index, + this.#toolOutputLimit, + home, + ); + } + } + } + if (isTerminal) active.thinkingSteps.add(step.step_index); + if (thinking === null || thinking.trim().length === 0) return; + active.thinkingSteps.add(step.step_index); + if (active.agentItem) { + this.#completeItem(active, active.agentItem, { status: "succeeded" }); + active.agentItem = null; + active.agentText = ""; + } + const item: HostItem = { + type: "reasoning", + itemId: this.#newItemId(), + text: thinking.slice(0, this.#toolOutputLimit), + }; + this.#event({ type: "item.started", turnId: active.command.turnId, item }); + this.#completeItem(active, item, { status: "succeeded" }); + } + #appendOrSyncAgentText(active: ActiveTurn, text: string, isExplicitDelta: boolean): void { if (!text) return; if (isExplicitDelta || !active.agentItem) { diff --git a/packages/adapters/antigravity/src/code-action-diff.ts b/packages/adapters/antigravity/src/code-action-diff.ts index 866f5b35d..800bb8f56 100644 --- a/packages/adapters/antigravity/src/code-action-diff.ts +++ b/packages/adapters/antigravity/src/code-action-diff.ts @@ -20,6 +20,7 @@ import type { HostFileChange } from "@codexhost/harness-adapter"; import { displayPath } from "./tool-projection.js"; const DIFF_CONTEXT_LINES = 3; +const TRAJECTORY_PAGE_SIZE = 256; type UnifiedDiffLineType = "insert" | "delete" | "unchanged"; @@ -239,3 +240,88 @@ export function requestAntigravityTrajectorySteps( request.end(JSON.stringify({ cascadeId: conversationId })); }); } + +function plannerStepThinking(step: unknown): string | null { + if (!isRecord(step) || step.type !== "CORTEX_STEP_TYPE_PLANNER_RESPONSE") return null; + const plannerResponse = isRecord(step.plannerResponse) ? step.plannerResponse : null; + if (!plannerResponse) return null; + for (const candidate of [ + plannerResponse.thinking, + plannerResponse.rawThinking, + plannerResponse.raw_thinking, + ]) { + if (typeof candidate === "string" && candidate.trim().length > 0) return candidate; + } + return null; +} + +/** + * Reads the thinking text of one `CORTEX_STEP_TYPE_PLANNER_RESPONSE` step out + * of a `GetCascadeTrajectorySteps` response. `agy` strips `` blocks + * from `stream-json` step updates while recording the full thinking text on + * the trajectory step at `stepIndex`. + */ +export function parseAntigravityStepThinking( + value: unknown, + stepIndex: number, + stepOffset = 0, +): string | null { + if (!isRecord(value) || !Array.isArray(value.steps)) return null; + const relativeIndex = stepIndex - stepOffset; + if (relativeIndex >= 0 && relativeIndex < value.steps.length) { + const direct = plannerStepThinking(value.steps[relativeIndex]); + if (direct !== null) return direct; + } + for (const step of value.steps) { + if (isRecord(step) && step.stepIndex === stepIndex) { + const matched = plannerStepThinking(step); + if (matched !== null) return matched; + } + } + return null; +} + +export function requestAntigravityStepThinking( + port: number, + conversationId: string, + stepIndex: number, + timeoutMs: number, +): Promise { + const stepOffset = stepIndex >= TRAJECTORY_PAGE_SIZE ? stepIndex : 0; + return new Promise((resolve) => { + const request = https.request( + { + hostname: "127.0.0.1", + port, + path: "/exa.language_server_pb.LanguageServerService/GetCascadeTrajectorySteps", + method: "POST", + rejectUnauthorized: false, + timeout: timeoutMs, + headers: { "content-type": "application/json" }, + }, + (response) => { + let body = ""; + response.setEncoding("utf8"); + response.on("data", (chunk: string) => { + body += chunk; + }); + response.on("end", () => { + if (response.statusCode !== 200) return resolve(null); + try { + resolve(parseAntigravityStepThinking(JSON.parse(body), stepIndex, stepOffset)); + } catch { + resolve(null); + } + }); + }, + ); + request.on("timeout", () => request.destroy()); + request.on("error", () => resolve(null)); + request.end( + JSON.stringify({ + cascadeId: conversationId, + ...(stepOffset > 0 ? { stepOffset } : {}), + }), + ); + }); +} diff --git a/packages/adapters/antigravity/src/index.ts b/packages/adapters/antigravity/src/index.ts index e8ba9c688..440122a52 100644 --- a/packages/adapters/antigravity/src/index.ts +++ b/packages/adapters/antigravity/src/index.ts @@ -31,6 +31,8 @@ export type { AntigravityStreamEvent } from "./stream-events.js"; export { codeActionFileChange, parseAntigravityCodeActions, + parseAntigravityStepThinking, + requestAntigravityStepThinking, requestAntigravityTrajectorySteps, } from "./code-action-diff.js"; export type { AntigravityCodeAction } from "./code-action-diff.js"; diff --git a/packages/adapters/antigravity/src/stream-events.ts b/packages/adapters/antigravity/src/stream-events.ts index 2094b350e..6a297bcd2 100644 --- a/packages/adapters/antigravity/src/stream-events.ts +++ b/packages/adapters/antigravity/src/stream-events.ts @@ -30,6 +30,8 @@ export interface AntigravityStepUpdateEvent { step_index: number; state: "ACTIVE" | "DONE" | "ERROR" | string; step_type: string; + thinking_delta?: string; + thinking?: string; text_delta?: string; text?: string; content?: string; diff --git a/packages/adapters/antigravity/src/subagent-transcript.ts b/packages/adapters/antigravity/src/subagent-transcript.ts index 8923cd3d5..8934dfdc0 100644 --- a/packages/adapters/antigravity/src/subagent-transcript.ts +++ b/packages/adapters/antigravity/src/subagent-transcript.ts @@ -24,6 +24,7 @@ const transcriptStepSchema = z.object({ type: z.string(), status: z.string(), content: z.string().optional(), + thinking: z.string().optional(), tool_calls: z.array(z.object({ name: z.string(), args: z.unknown().optional() })).optional(), }); @@ -156,6 +157,16 @@ export function parseSubagentTranscript( } : { status: "succeeded" }; if (step.type === "PLANNER_RESPONSE") { + if (step.thinking?.trim()) { + current.items.push({ + item: { + type: "reasoning", + itemId: hostItemIdSchema.parse(`${itemId}-thinking`), + text: step.thinking.slice(0, outputLimit), + }, + outcome, + }); + } if (step.content) { current.items.push({ item: { type: "agentMessage", itemId, text: step.content.slice(0, outputLimit) }, @@ -260,3 +271,52 @@ export async function readSubagentTranscript(input: { }; } } + +/** + * Reads the persisted thinking text for one `PLANNER_RESPONSE` step from the + * native brain transcript when the Language Server trajectory is unavailable. + */ +export async function readTranscriptStepThinking( + conversationId: string, + stepIndex: number, + outputLimit: number, + home?: string, +): Promise { + if (!conversationId || /[/\\]|^\.\.?$/u.test(conversationId)) return null; + const logsDir = path.join(nativeBrainDirPath(conversationId, home), ".system_generated", "logs"); + for (const filename of ["transcript_full.jsonl", "transcript.jsonl"]) { + try { + const handle = await open(path.join(logsDir, filename), "r"); + let text: string; + try { + const size = (await handle.stat()).size; + if (size === 0 || size > MAX_TRANSCRIPT_BYTES) continue; + const buffer = Buffer.alloc(size); + const { bytesRead } = await handle.read(buffer, 0, size, 0); + text = buffer.subarray(0, bytesRead).toString("utf8"); + } finally { + await handle.close(); + } + for (const line of text.split("\n")) { + if (!line.trim()) continue; + try { + const parsed = JSON.parse(line) as unknown; + if ( + isRecord(parsed) && + parsed.step_index === stepIndex && + parsed.type === "PLANNER_RESPONSE" && + typeof parsed.thinking === "string" && + parsed.thinking.trim().length > 0 + ) { + return parsed.thinking.slice(0, outputLimit); + } + } catch { + // Ignore an incomplete trailing line while the Turn is still writing. + } + } + } catch { + // Try next transcript candidate. + } + } + return null; +} diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index 92016b2bd..998ec550b 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -1,4 +1,4 @@ -import { access, chmod, mkdtemp, readFile, realpath, rm, writeFile } from "node:fs/promises"; +import { access, chmod, mkdir, mkdtemp, readFile, realpath, rm, writeFile } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -2258,5 +2258,143 @@ if (count === 0) { expect(formatAntigravityTurnPrompt(slashCommand, project)).toBe(slashCommand); expect(formatAntigravityTurnPrompt(located, project)).toBe(located); }); + + it("projects planner thinking steps as reasoning Items before tools and responses", async () => { + const conversationId = "conv-thinking-projection"; + const streamLines = [ + JSON.stringify({ + event: "init", + init: { permission_mode: "dangerously-skip-permissions" }, + conversation_id: conversationId, + }), + JSON.stringify({ + event: "step_update", + step_update: { + conversation_id: conversationId, + step_index: 1, + state: "DONE", + step_type: "agent_response", + usage: { + input_tokens: 100, + output_tokens: 40, + thinking_tokens: 180, + total_tokens: 320, + }, + }, + }), + JSON.stringify({ + event: "step_update", + step_update: { + conversation_id: conversationId, + step_index: 2, + state: "DONE", + step_type: "tool", + tool_name: "read_file", + tool_info: { name: "read_file", parameters: { Path: "/src/app.ts" }, output: "ok" }, + }, + }), + JSON.stringify({ + event: "step_update", + step_update: { + conversation_id: conversationId, + step_index: 3, + state: "DONE", + step_type: "agent_response", + text_delta: "All checks passed.", + usage: { input_tokens: 140, output_tokens: 30, thinking_tokens: 95, total_tokens: 265 }, + }, + }), + JSON.stringify({ + event: "result", + result: { + conversation_id: conversationId, + status: "SUCCESS", + num_turns: 1, + response: "All checks passed.", + }, + }), + ]; + const { command, cwd, cleanup } = await fakeStreamingAgy(streamLines); + const logsDir = path.join( + cwd, + ".gemini", + "antigravity-cli", + "brain", + conversationId, + ".system_generated", + "logs", + ); + await mkdir(logsDir, { recursive: true }); + await writeFile( + path.join(logsDir, "transcript_full.jsonl"), + [ + JSON.stringify({ + step_index: 1, + source: "MODEL", + type: "PLANNER_RESPONSE", + status: "DONE", + thinking: "Reading /src/app.ts to inspect its exports.", + tool_calls: [{ name: "read_file", args: { Path: "/src/app.ts" } }], + }), + JSON.stringify({ + step_index: 3, + source: "MODEL", + type: "PLANNER_RESPONSE", + status: "DONE", + thinking: "Verified /src/app.ts exports; formulating final answer.", + content: "All checks passed.", + }), + ].join("\n") + "\n", + ); + const adapter = new AntigravityAdapter({ + command, + environment: { ...process.env, HOME: cwd, USERPROFILE: cwd, CODEXHOST_DATA_DIR: cwd }, + }); + 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-thinking"), + input: [{ type: "text", text: "check app.ts" }], + }); + const completedItems: Array<{ type: string; text?: string; toolName?: string }> = []; + while (true) { + const event = await nextEvent(iterator); + if (event.type === "item.completed") { + const { item } = event.snapshot; + completedItems.push({ + type: item.type, + ...("text" in item ? { text: item.text } : {}), + ...("toolName" in item ? { toolName: item.toolName } : {}), + }); + } + if (event.type === "turn.completed") break; + } + expect(completedItems).toEqual([ + { type: "reasoning", text: "Reading /src/app.ts to inspect its exports." }, + { type: "toolExecution", toolName: "read_file" }, + { + type: "reasoning", + text: "Verified /src/app.ts exports; formulating final answer.", + }, + { type: "agentMessage", text: "All checks passed." }, + ]); + const snapshot = await session.readSnapshot(); + if (!snapshot.ok) throw new Error(snapshot.error.message); + expect(snapshot.value.turns[0]?.items.map(({ item }) => item.type)).toEqual([ + "reasoning", + "toolExecution", + "reasoning", + "agentMessage", + ]); + await session.close(); + } finally { + await adapter.close(); + await cleanup(); + } + }); }); }); diff --git a/packages/adapters/antigravity/test/code-action-diff.test.ts b/packages/adapters/antigravity/test/code-action-diff.test.ts index 7c02ff60f..76507b7ac 100644 --- a/packages/adapters/antigravity/test/code-action-diff.test.ts +++ b/packages/adapters/antigravity/test/code-action-diff.test.ts @@ -5,6 +5,7 @@ import { describe, expect, it } from "vitest"; import { codeActionFileChange, parseAntigravityCodeActions, + parseAntigravityStepThinking, type AntigravityCodeAction, } from "../src/index.js"; @@ -191,4 +192,38 @@ describe("Antigravity Code Action diffs", () => { expect(change?.unifiedDiff).toContain(`+++ ${outside.replaceAll("\\", "/")}`); }); }); + + describe("parseAntigravityStepThinking", () => { + it("reads plannerResponse thinking or rawThinking at the step index or offset", () => { + const payload = { + steps: [ + { type: "CORTEX_STEP_TYPE_USER_INPUT", status: "CORTEX_STEP_STATUS_DONE" }, + { + type: "CORTEX_STEP_TYPE_PLANNER_RESPONSE", + status: "CORTEX_STEP_STATUS_DONE", + plannerResponse: { + thinking: "Inspecting the binary symbols before running tools.", + response: "Done.", + }, + }, + { + type: "CORTEX_STEP_TYPE_PLANNER_RESPONSE", + status: "CORTEX_STEP_STATUS_DONE", + plannerResponse: { + rawThinking: "Fallback raw thinking block.", + }, + }, + ], + }; + expect(parseAntigravityStepThinking(payload, 0)).toBeNull(); + expect(parseAntigravityStepThinking(payload, 1)).toBe( + "Inspecting the binary symbols before running tools.", + ); + expect(parseAntigravityStepThinking(payload, 2)).toBe("Fallback raw thinking block."); + expect(parseAntigravityStepThinking({ steps: [payload.steps[1]] }, 256, 256)).toBe( + "Inspecting the binary symbols before running tools.", + ); + expect(parseAntigravityStepThinking(null, 1)).toBeNull(); + }); + }); }); diff --git a/packages/adapters/antigravity/test/subagents.test.ts b/packages/adapters/antigravity/test/subagents.test.ts index 6ee08aa33..ba6e1e882 100644 --- a/packages/adapters/antigravity/test/subagents.test.ts +++ b/packages/adapters/antigravity/test/subagents.test.ts @@ -131,6 +131,80 @@ describe("Antigravity Subagents", () => { expect(otherParent.turns[0]?.items[0]?.item.itemId).not.toBe(ids[0]); }); + it("projects PLANNER_RESPONSE thinking as a reasoning Item before tool calls and final text", () => { + const withThinking = + row({ + step_index: 0, + source: "USER_EXPLICIT", + type: "USER_INPUT", + status: "DONE", + content: "\nInspect fixture\n", + }) + + row({ + step_index: 1, + source: "MODEL", + type: "PLANNER_RESPONSE", + status: "DONE", + thinking: "Checking directory contents first.", + tool_calls: [{ name: "list_dir", args: { DirectoryPath: "fixture" } }], + }) + + row({ + step_index: 2, + source: "MODEL", + type: "GENERIC", + status: "DONE", + content: "marker.txt", + }) + + row({ + step_index: 3, + source: "MODEL", + type: "PLANNER_RESPONSE", + status: "DONE", + thinking: "Found marker.txt; summarizing result.", + content: "AGY_CHILD_PROBE_OK", + }); + const snapshot = parseSubagentTranscript( + withThinking, + parent, + child, + "completed", + process.cwd(), + 64_000, + ); + expect(snapshot.turns[0]?.items).toMatchObject([ + { + item: { + type: "reasoning", + itemId: `agy-${parent}-${child}-1-thinking`, + text: "Checking directory contents first.", + }, + outcome: { status: "succeeded" }, + }, + { + item: { + type: "toolExecution", + toolName: "list_dir", + output: { content: [{ text: "marker.txt" }] }, + }, + }, + { + item: { + type: "reasoning", + itemId: `agy-${parent}-${child}-3-thinking`, + text: "Found marker.txt; summarizing result.", + }, + outcome: { status: "succeeded" }, + }, + { + item: { + type: "agentMessage", + itemId: `agy-${parent}-${child}-3`, + text: "AGY_CHILD_PROBE_OK", + }, + }, + ]); + }); + it("ignores an incomplete final line, rejects interior corruption and does not invent terminal success", () => { const snapshot = parseSubagentTranscript( transcript + '{"step_index":4', From 1a1a8e235c0b8e49d46425115223310cb19604dc Mon Sep 17 00:00:00 2001 From: Vincent <1729065730@qq.com> Date: Thu, 5 Nov 2026 15:45:51 +0800 Subject: [PATCH 2/2] fix(antigravity): keep streamed replies below late planner thinking When thinking is enabled, agy often streams agent_response text before the step reaches DONE, while the thinking text is only available from the transcript or Language Server at DONE. Projecting that late thinking closed the open agentMessage and emitted a reasoning Item after the answer. Buffer agent_response text for a step until its thinking is resolved (or known absent at DONE), then flush the reply so reasoning stays above it. Sessions without a thinking option still stream immediately. Signed-off-by: Vincent <1729065730@qq.com> --- .../antigravity/src/antigravity-adapter.ts | 75 +++++++++-- .../test/antigravity-adapter.test.ts | 126 ++++++++++++++++++ 2 files changed, 188 insertions(+), 13 deletions(-) diff --git a/packages/adapters/antigravity/src/antigravity-adapter.ts b/packages/adapters/antigravity/src/antigravity-adapter.ts index c6c6c19f6..418d274e0 100644 --- a/packages/adapters/antigravity/src/antigravity-adapter.ts +++ b/packages/adapters/antigravity/src/antigravity-adapter.ts @@ -177,6 +177,12 @@ interface ActiveTurn { pendingSteps: Map; thinkingSteps: Set; checkedActiveThinkingSteps: Set; + /** + * `agent_response` text held until that step's thinking is projected (or + * known absent at DONE/ERROR). Streaming reply deltas otherwise outrank a + * reasoning Item that only becomes available when the step finishes. + */ + pendingAgentText: Map; httpsPort: number | null; } @@ -875,6 +881,7 @@ class AntigravitySession implements HarnessSession { pendingSteps: new Map(), thinkingSteps: new Set(), checkedActiveThinkingSteps: new Set(), + pendingAgentText: new Map(), httpsPort: null, }; this.#active = active; @@ -1198,21 +1205,21 @@ class AntigravitySession implements HarnessSession { } if (step.step_type === "agent_response") { await this.#maybeProjectStepThinking(active, step); - if (typeof step.text_delta === "string" && step.text_delta.length > 0) { - this.#appendOrSyncAgentText(active, step.text_delta, true); - return; - } - const fullOrDelta = - step.text ?? - (typeof step.content === "string" ? step.content : undefined) ?? - (typeof step.message === "string" ? step.message : undefined); - if (typeof fullOrDelta === "string" && fullOrDelta.length > 0) { - this.#appendOrSyncAgentText(active, fullOrDelta, false); - return; + const isDelta = typeof step.text_delta === "string" && step.text_delta.length > 0; + const text = isDelta + ? step.text_delta + : (step.text ?? + (typeof step.content === "string" ? step.content : undefined) ?? + (typeof step.message === "string" ? step.message : undefined)); + if (typeof text === "string" && text.length > 0) { + this.#emitOrBufferAgentResponseText(active, step.step_index, text, isDelta); + } else if (active.thinkingSteps.has(step.step_index)) { + this.#flushPendingAgentText(active, step.step_index); } return; } if (step.step_type !== "tool") return; + this.#flushAllPendingAgentText(active); if (active.agentItem) { this.#completeItem(active, active.agentItem, { status: "succeeded" }); active.agentItem = null; @@ -1319,7 +1326,10 @@ class AntigravitySession implements HarnessSession { * and omits `thinking` from `step_update`, reporting only `usage.thinking_tokens` * when a planner step finishes. Resolve the step's thinking from inline fields, * the Language Server trajectory, or the native brain transcript before - * projecting the step's visible response or subsequent tool calls. + * projecting the step's visible response or subsequent tool calls. Reply text + * for the same step is buffered until this returns with the step recorded in + * `thinkingSteps`, so a late-fetched reasoning Item is not outranked by the + * streamed answer. */ async #maybeProjectStepThinking( active: ActiveTurn, @@ -1364,7 +1374,12 @@ class AntigravitySession implements HarnessSession { } } if (isTerminal) active.thinkingSteps.add(step.step_index); - if (thinking === null || thinking.trim().length === 0) return; + if (thinking === null || thinking.trim().length === 0) { + // Without a thinking option, do not stall streamed replies waiting for a + // DONE-only transcript fetch that is unlikely to appear. + if (!isTerminal && !this.#thinkingOptionId) active.thinkingSteps.add(step.step_index); + return; + } active.thinkingSteps.add(step.step_index); if (active.agentItem) { this.#completeItem(active, active.agentItem, { status: "succeeded" }); @@ -1380,6 +1395,39 @@ class AntigravitySession implements HarnessSession { this.#completeItem(active, item, { status: "succeeded" }); } + /** + * Emit agent_response text only after that step's thinking is resolved when + * a late reasoning Item is still possible. Otherwise buffer so the streamed + * answer cannot outrank thinking that only becomes available at DONE. + */ + #emitOrBufferAgentResponseText( + active: ActiveTurn, + stepIndex: number, + text: string, + isDelta: boolean, + ): void { + if (active.thinkingSteps.has(stepIndex) || !this.#thinkingOptionId) { + this.#flushPendingAgentText(active, stepIndex); + this.#appendOrSyncAgentText(active, text, isDelta); + return; + } + const previous = active.pendingAgentText.get(stepIndex) ?? ""; + active.pendingAgentText.set(stepIndex, isDelta ? previous + text : text); + } + + #flushPendingAgentText(active: ActiveTurn, stepIndex: number): void { + const pending = active.pendingAgentText.get(stepIndex); + if (pending === undefined) return; + active.pendingAgentText.delete(stepIndex); + if (pending.length > 0) this.#appendOrSyncAgentText(active, pending, true); + } + + #flushAllPendingAgentText(active: ActiveTurn): void { + for (const stepIndex of [...active.pendingAgentText.keys()].sort((a, b) => a - b)) { + this.#flushPendingAgentText(active, stepIndex); + } + } + #appendOrSyncAgentText(active: ActiveTurn, text: string, isExplicitDelta: boolean): void { if (!text) return; if (isExplicitDelta || !active.agentItem) { @@ -1446,6 +1494,7 @@ class AntigravitySession implements HarnessSession { ? { status: "cancelled", ...(outcome.reason ? { reason: outcome.reason } : {}) } : { status: "succeeded" }; active.subagents.finish(itemOutcome); + this.#flushAllPendingAgentText(active); if (active.agentItem) this.#completeItem(active, active.agentItem, itemOutcome); // A step whose patch never resolved still ran, so an interrupted Turn // reports it as a Tool Execution rather than dropping it silently. diff --git a/packages/adapters/antigravity/test/antigravity-adapter.test.ts b/packages/adapters/antigravity/test/antigravity-adapter.test.ts index 998ec550b..f430dd2b0 100644 --- a/packages/adapters/antigravity/test/antigravity-adapter.test.ts +++ b/packages/adapters/antigravity/test/antigravity-adapter.test.ts @@ -2396,5 +2396,131 @@ if (count === 0) { await cleanup(); } }); + + it("keeps streamed agent_response text after a reasoning Item resolved only at DONE", async () => { + const conversationId = "conv-thinking-order"; + const streamLines = [ + JSON.stringify({ + event: "init", + init: { permission_mode: "dangerously-skip-permissions" }, + conversation_id: conversationId, + }), + JSON.stringify({ + event: "step_update", + step_update: { + conversation_id: conversationId, + step_index: 1, + state: "RUNNING", + step_type: "agent_response", + text_delta: "First half. ", + }, + }), + JSON.stringify({ + event: "step_update", + step_update: { + conversation_id: conversationId, + step_index: 1, + state: "RUNNING", + step_type: "agent_response", + text_delta: "Second half.", + }, + }), + JSON.stringify({ + event: "step_update", + step_update: { + conversation_id: conversationId, + step_index: 1, + state: "DONE", + step_type: "agent_response", + usage: { + input_tokens: 100, + output_tokens: 40, + thinking_tokens: 220, + total_tokens: 360, + }, + }, + }), + JSON.stringify({ + event: "result", + result: { + conversation_id: conversationId, + status: "SUCCESS", + num_turns: 1, + response: "First half. Second half.", + }, + }), + ]; + const { command, cwd, cleanup } = await fakeStreamingAgy(streamLines); + const logsDir = path.join( + cwd, + ".gemini", + "antigravity-cli", + "brain", + conversationId, + ".system_generated", + "logs", + ); + await mkdir(logsDir, { recursive: true }); + await writeFile( + path.join(logsDir, "transcript_full.jsonl"), + JSON.stringify({ + step_index: 1, + source: "MODEL", + type: "PLANNER_RESPONSE", + status: "DONE", + thinking: "Drafting the final answer after reviewing the codebase.", + content: "First half. Second half.", + }) + "\n", + ); + const adapter = new AntigravityAdapter({ + command, + environment: { ...process.env, HOME: cwd, USERPROFILE: cwd, CODEXHOST_DATA_DIR: cwd }, + }); + try { + const opened = await adapter.open({ + kind: "create", + cwd, + model: harnessModelRefSchema.parse({ id: "gemini-3.7-flash" }), + thinkingOptionId: harnessThinkingOptionIdSchema.parse("high"), + }); + 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-thinking-order"), + input: [{ type: "text", text: "summarize" }], + }); + const completedItems: Array<{ type: string; text?: string }> = []; + while (true) { + const event = await nextEvent(iterator); + if (event.type === "item.completed") { + const { item } = event.snapshot; + completedItems.push({ + type: item.type, + ...("text" in item ? { text: item.text } : {}), + }); + } + if (event.type === "turn.completed") break; + } + expect(completedItems).toEqual([ + { + type: "reasoning", + text: "Drafting the final answer after reviewing the codebase.", + }, + { type: "agentMessage", text: "First half. Second half." }, + ]); + const snapshot = await session.readSnapshot(); + if (!snapshot.ok) throw new Error(snapshot.error.message); + expect(snapshot.value.turns[0]?.items.map(({ item }) => item.type)).toEqual([ + "reasoning", + "agentMessage", + ]); + await session.close(); + } finally { + await adapter.close(); + await cleanup(); + } + }); }); });