fix(antigravity): project planner thinking steps as reasoning Items

packages/adapters/antigravity/src/antigravity-adapter.ts

@@ -93,13 +93,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 { antigravityHome, createAntigravityUsageStatistics } from "./usage-statistics.js";
 import {
   antigravityToolErrorMessage,
@@ -180,6 +185,14 @@ 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<number, AntigravityStepUpdateEvent["step_update"]>;
+  thinkingSteps: Set<number>;
+  checkedActiveThinkingSteps: Set<number>;
+  /**
+   * `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<number, string>;
   httpsPort: number | null;
 }
 
@@ -956,6 +969,9 @@ class AntigravitySession implements HarnessSession {
       codeActionCursor: 0,
       fileChanges: new Map(),
       pendingSteps: new Map(),
+      thinkingSteps: new Set(),
+      checkedActiveThinkingSteps: new Set(),
+      pendingAgentText: new Map(),
       httpsPort: null,
     };
     this.#active = active;
@@ -1319,17 +1335,17 @@ class AntigravitySession implements HarnessSession {
       step = { ...step, step_type: "tool" };
     }
     if (step.step_type === "agent_response") {
-      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;
+      await this.#maybeProjectStepThinking(active, step);
+      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;
     }
@@ -1338,6 +1354,7 @@ class AntigravitySession implements HarnessSession {
       return;
     }
     if (step.step_type !== "tool") return;
+    this.#flushAllPendingAgentText(active);
     if (active.agentItem) {
       this.#completeItem(active, active.agentItem, { status: "succeeded" });
       active.agentItem = null;
@@ -1478,6 +1495,113 @@ class AntigravitySession implements HarnessSession {
     return port;
   }
 
+  /**
+   * `agy --output-format stream-json` strips `<thought>` 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. 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,
+    step: AntigravityStepUpdateEvent["step_update"],
+  ): Promise<void> {
+    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) {
+      // 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" });
+      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" });
+  }
+
+  /**
+   * 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) {
@@ -1544,6 +1668,7 @@ class AntigravitySession implements HarnessSession {
           ? { status: "cancelled", ...(outcome.reason ? { reason: outcome.reason } : {}) }
           : { status: "succeeded" };
     active.subagents.finish(itemOutcome);
+    this.#flushAllPendingAgentText(active);
     if (active.compactionItem) this.#completeItem(active, active.compactionItem, itemOutcome);
     active.compactionItem = null;
     if (active.agentItem) this.#completeItem(active, active.agentItem, itemOutcome);

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 `<thought>` 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<string | null> {
+  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 } : {}),
+      }),
+    );
+  });
+}

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

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;

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<string | null> {
+  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;
+}