fix: accept empty Pi completions and stabilize CI

README.md

@@ -172,6 +172,16 @@ estimates with reported charges. Subscription cost estimates are API list-price
 comparisons, not subscription charges. Auth changes affect child environments
 only; Headless never replaces shared login files.
 
+### Empty Pi completions
+
+Pi can finish artifact-producing work with an empty final assistant message.
+Headless accepts that as success when Pi's native terminal event confirms a normal
+completion and the process exits successfully. Plain output contains no invented
+answer; `--usage` still reports usage, and SDK results contain `finalMessage: ""`.
+An incomplete lifecycle or native error remains a failure, including when earlier
+assistant progress text exists. Legacy message-only output remains supported.
+Artifact validation remains the caller's responsibility.
+
 ## Native TUI Completion
 
 Use `--tmux --wait --delete` when you want Headless to launch the agent in its native TUI, wait for the final native transcript message, print that message, and then terminate the tmux session after the prompt completes.

src/cli.ts

@@ -19,6 +19,7 @@ import { randomUUID } from "node:crypto";
 import { dirname, join } from "node:path";
 import { tmpdir } from "node:os";
 import { fileURLToPath } from "node:url";
+import { PiCompletionObserver } from "./pi-completion.js";
 
 import {
   buildAgentCommand,
@@ -4560,6 +4561,7 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise<number
     waitingSpinner?.start();
     let result: ExecuteResult | undefined;
     let billingResult: BillingRunResult | undefined;
+    const piCompletion = parsed.agent === "pi" ? new PiCompletionObserver() : undefined;
     let antigravityUsageTrace = "";
     let antigravityUsageCapture: AntigravityUsageCapture | undefined;
     if (parsed.agent === "antigravity" && parsed.usage && !parsed.docker && !parsed.modal) {
@@ -4622,7 +4624,11 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise<number
             invoke: (execute) => runBilling((attempt) => execute(
               buildAttemptCommand(attempt.env, attempt.options), attempt.env,
               attempt.timeoutSeconds ?? modalTimeoutSeconds,
-              (text) => { attempt.observe(text); if (stdoutHandling === "capture") commandStdoutLog?.(text); },
+              (text) => {
+                piCompletion?.write(text);
+                attempt.observe(text);
+                if (stdoutHandling === "capture") commandStdoutLog?.(text);
+              },
             )),
           })
           : await runBilling(async (attempt) => {
@@ -4648,7 +4654,11 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise<number
               ? buildAttemptCommand(attempt.env, attempt.options) : command,
             cwd, attempt.env, displayStderr, {
               stdout: commandStdout, stdoutHandling,
-              stdoutLog: (text) => { attempt.observe(text); commandStdoutLog?.(text); }, stderr: commandStderr,
+              stdoutLog: (text) => {
+                piCompletion?.write(text);
+                attempt.observe(text);
+                commandStdoutLog?.(text);
+              }, stderr: commandStderr,
               timeoutSeconds,
               captureFinalMessageTrace: Boolean(parsed.sdkFormat) || (parsed.agent === "antigravity" && parsed.json && Boolean(parsed.runId)),
               captureRelevantTrace: Boolean(parsed.sdkFormat) || parsed.usage || (parsed.json && (Boolean(parsed.runId) || Boolean(parsed.sessionAlias))),
@@ -4659,6 +4669,7 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise<number
             );
           });
       } finally {
+        piCompletion?.end();
         waitingSpinner?.stop();
         antigravityUsageTrace = antigravityUsageCapture?.read() ?? "";
         antigravityUsageCapture?.cleanup();
@@ -4671,8 +4682,15 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise<number
       if (!result) {
         throw new CliError("agent execution did not produce a result");
       }
-      billingHomeCompleted = result.code === 0;
       const commandTrace = result.stdout || result.finalMessageTrace || result.usageTrace || "";
+      const piOutcome = piCompletion?.outcome;
+      const piError = piOutcome?.status === "error" ? `pi error: ${piOutcome.error}`
+        : piOutcome?.status === "unknown" && piCompletion?.observedLifecycle
+          ? "pi error: native invocation ended without a successful final completion"
+          : undefined;
+      const piFinalMessage = piOutcome?.status === "success" ? piOutcome.finalMessage : undefined;
+      if (piError && result.code === 0) result = { ...result, code: 1 };
+      billingHomeCompleted = result.code === 0;
       const usageCommandTrace = result.stdout || result.usageTrace || result.finalMessageTrace || "";
       const usageTrace = antigravityUsageTrace
         ? `${usageCommandTrace}\n${antigravityUsageTrace}`
@@ -4700,26 +4718,27 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise<number
       }
       if (parsed.runId && parsed.role && nodeId) {
         const finalMessage =
-          extractFinalMessage(parsed.agent, commandTrace) ||
+          piFinalMessage ?? (extractFinalMessage(parsed.agent, commandTrace) ||
           (result.usageTrace && result.usageTrace !== commandTrace
             ? extractFinalMessage(parsed.agent, result.usageTrace)
-            : "");
+            : ""));
         const metrics = extractRunNodeMetrics(
           parsed.agent,
           usageTrace,
           usageContext(parsed.agent, configuredDefaults, env, effectiveProfile),
         );
-        updateNodeStatus(env, parsed.runId, nodeId, result.code === 0 ? "idle" : "failed", finalMessage || undefined, metrics);
-        if (result.code === 0 && parsed.role === "orchestrator" && finalMessage) {
-          completeIdleRunNodes(env, parsed.runId, nodeId, finalMessage);
+        const runMessage = piFinalMessage ?? (finalMessage || undefined);
+        updateNodeStatus(env, parsed.runId, nodeId, result.code === 0 ? "idle" : "failed", runMessage, metrics);
+        if (result.code === 0 && parsed.role === "orchestrator" && (finalMessage || piFinalMessage === "")) {
+          completeIdleRunNodes(env, parsed.runId, nodeId, runMessage);
         }
       }
       if (parsed.sdkFormat) {
         const finalMessage =
-          extractFinalMessage(parsed.agent, commandTrace) ||
-          sdkTraceWriter?.finalMessage;
-        const agentError = extractAgentError(parsed.agent, commandTrace);
-        if (!finalMessage) {
+          piFinalMessage ?? (extractFinalMessage(parsed.agent, commandTrace) ||
+          sdkTraceWriter?.finalMessage);
+        const agentError = piError ?? extractAgentError(parsed.agent, commandTrace);
+        if (piError || (!finalMessage && !(piFinalMessage === "" && result.code === 0))) {
           const exitCode = result.code || 1;
           stdout(
             renderSdkError(
@@ -4751,6 +4770,7 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise<number
         return result.code;
       }
       if (parsed.json) {
+        if (piError) stderr(`headless: ${piError}\n`);
         if (parsed.usage) {
           const stdoutEndsWithNewline = result.stdoutEndsWithNewline ?? result.stdout.endsWith("\n");
           const stdoutReceived = result.stdoutReceived ?? Boolean(result.stdout);
@@ -4764,14 +4784,14 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise<number
         return result.code;
       }
 
-      const finalMessage = extractFinalMessage(parsed.agent, result.stdout);
-      if (finalMessage) {
-        if (parsed.debug) {
+      const finalMessage = piFinalMessage ?? extractFinalMessage(parsed.agent, result.stdout);
+      if (!piError && (finalMessage || piFinalMessage === "")) {
+        if (parsed.debug && finalMessage) {
           if (!result.stdout.endsWith("\n")) {
             stdout("\n");
           }
           stdout(`--- final message ---\n${finalMessage}\n`);
-        } else {
+        } else if (finalMessage) {
           stdout(`${finalMessage}\n`);
         }
         if (parsed.usage) {
@@ -4781,7 +4801,8 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise<number
         }
         return result.code;
       }
-      const agentError = extractAgentError(parsed.agent, result.stdout);
+      if (piCompletion && parsed.usage) stdout(await finalUsageOutput());
+      const agentError = piError ?? extractAgentError(parsed.agent, result.stdout);
       if (agentError) {
         stderr(`headless: ${agentError}\n`);
         return result.code === 0 ? 1 : result.code;

src/pi-completion.ts

@@ -0,0 +1,145 @@
+import { PiTerminalJson } from "./pi-terminal-json.js";
+
+export type PiCompletionOutcome =
+  | { status: "unknown" }
+  | { status: "success"; finalMessage: string }
+  | { status: "error"; error: string };
+
+const MAX_LINE_BYTES = 4 * 1024 * 1024;
+const MAX_ERROR_CHARS = 4096;
+type JsonRecord = Record<string, unknown>;
+
+function record(value: unknown): JsonRecord | undefined {
+  return value !== null && typeof value === "object" && !Array.isArray(value)
+    ? value as JsonRecord : undefined;
+}
+
+function assistantFailure(message: JsonRecord): string | undefined {
+  const detail = typeof message.errorMessage === "string" ? message.errorMessage.trim() : "";
+  if (detail) return detail.slice(0, MAX_ERROR_CHARS);
+  if (message.stopReason === "error") return "Pi assistant request failed.";
+  if (message.stopReason === "aborted") return "Pi assistant request aborted.";
+  return undefined;
+}
+
+function finalText(message: JsonRecord): string | undefined {
+  if (message.stopReason !== "stop" || !Array.isArray(message.content)) return undefined;
+  if (message.errorMessage !== undefined && typeof message.errorMessage !== "string") return undefined;
+  const text: string[] = [];
+  for (const value of message.content) {
+    const block = record(value);
+    if (block?.type === "text" && typeof block.text === "string") text.push(block.text);
+    else if (block?.type !== "thinking" || typeof block.thinking !== "string") return undefined;
+  }
+  return text.join("").trim();
+}
+
+/** Native JSONL completion only: never recursively inspect tool results or assistant prose. */
+export class PiCompletionObserver {
+  outcome: PiCompletionOutcome = { status: "unknown" };
+  private pending = "";
+  private pendingBytes = 0;
+  private oversizedRecord?: PiTerminalJson;
+  private lifecycleSeen = false;
+  private compactionCompletion?: Extract<PiCompletionOutcome, { status: "success" }>;
+
+  get observedLifecycle(): boolean {
+    return this.lifecycleSeen;
+  }
+
+  write(chunk: string): void {
+    let start = 0;
+    while (start < chunk.length) {
+      const newline = chunk.indexOf("\n", start);
+      const end = newline < 0 ? chunk.length : newline;
+      const segment = chunk.slice(start, end);
+      if (!this.oversizedRecord) {
+        this.pendingBytes += Buffer.byteLength(segment);
+        if (this.pendingBytes > MAX_LINE_BYTES) {
+          this.oversizedRecord = new PiTerminalJson();
+          this.oversizedRecord.write(this.pending);
+          this.pending = "";
+          this.invalidateSuccess();
+        } else {
+          this.pending += segment;
+        }
+      }
+      this.oversizedRecord?.write(segment);
+      if (newline < 0) break;
+      this.finishLine();
+      this.resetLine();
+      start = newline + 1;
+    }
+  }
+
+  end(): void {
+    this.finishLine();
+    this.resetLine();
+  }
+
+  private resetLine(): void {
+    this.pending = "";
+    this.pendingBytes = 0;
+    this.oversizedRecord = undefined;
+  }
+
+  private finishLine(): void {
+    if (!this.oversizedRecord && !this.pending.trim()) return;
+    let event: JsonRecord | undefined;
+    if (this.oversizedRecord) {
+      event = this.oversizedRecord.end();
+      if (event?.type !== "agent_end") event = undefined;
+    } else {
+      try { event = record(JSON.parse(this.pending)); } catch { /* Incomplete native evidence cannot preserve success. */ }
+    }
+    this.consume(event);
+  }
+
+  private invalidateSuccess(): void {
+    this.compactionCompletion = undefined;
+    if (this.outcome.status === "success") this.outcome = { status: "unknown" };
+  }
+
+  private observeCompaction(event: JsonRecord): boolean {
+    if (event.type === "compaction_start" && event.reason === "threshold" && this.outcome.status === "success") {
+      const completion = this.outcome;
+      this.invalidateSuccess();
+      this.compactionCompletion = completion;
+      return true;
+    }
+    if (event.type !== "compaction_end" || !this.compactionCompletion) return false;
+    const completion = this.compactionCompletion;
+    this.invalidateSuccess();
+    if (event.reason === "threshold" && event.aborted === false && event.willRetry === false
+      && record(event.result) && event.errorMessage === undefined) {
+      this.outcome = completion;
+    }
+    return true;
+  }
+
+  private consume(event: JsonRecord | undefined): void {
+    if (!event || typeof event.type !== "string") {
+      this.invalidateSuccess();
+      return;
+    }
+    if (["agent_start", "agent_end", "agent_settled"].includes(event.type)) this.lifecycleSeen = true;
+    if (event.type === "agent_settled") return;
+    if (this.observeCompaction(event)) return;
+    this.invalidateSuccess();
+    let message: JsonRecord | undefined;
+    if (event.type === "agent_end" && Array.isArray(event.messages)) {
+      message = record(event.messages.at(-1));
+    } else if (event.type === "message_end" || event.type === "turn_end") {
+      message = record(event.message);
+    }
+    if (message?.role !== "assistant") return;
+    const error = assistantFailure(message);
+    if (error) {
+      this.outcome = { status: "error", error };
+      return;
+    }
+    if (event.type !== "agent_end" || (event.willRetry !== undefined && event.willRetry !== false)) return;
+    const finalMessage = finalText(message);
+    if (finalMessage !== undefined) this.outcome = { status: "success", finalMessage };
+  }
+}