fix(codex): complete native subagent turns

2026-08-08 19:12:22 +00:00 · 2026-06-19 12:38:14 +08:00
parent 8e66d7aad3
commit e172f64f3f
2 changed files with 416 additions and 116 deletions
--- a/extensions/codex/src/app-server/native-subagent-monitor.test.ts
+++ b/extensions/codex/src/app-server/native-subagent-monitor.test.ts
@@ -11,7 +11,7 @@ import {
  CodexNativeSubagentMonitor,
  registerCodexNativeSubagentMonitor,
 } from "./native-subagent-monitor.js";
-import type { CodexServerNotification } from "./protocol.js";
+import type { CodexServerNotification, JsonValue } from "./protocol.js";

 function createClient() {
  const handlers = new Set<(notification: CodexServerNotification) => Promise<void> | void>();
@@ -158,6 +158,27 @@ function nativeCompletionNotification(params: {
  };
 }

+function childTurnCompletedNotification(params: {
+  status: "completed" | "failed" | "interrupted";
+  error?: string;
+  turnId?: string;
+  items?: JsonValue[];
+}): CodexServerNotification {
+  const turnId = params.turnId ?? "child-turn";
+  return {
+    method: "turn/completed",
+    params: {
+      threadId: "child-thread",
+      turn: {
+        id: turnId,
+        status: params.status,
+        ...(params.items ? { items: params.items } : {}),
+        ...(params.error ? { error: { message: params.error } } : {}),
+      },
+    },
+  };
+}
+
 describe("CodexNativeSubagentMonitor", () => {
  it("keeps native subagent task mirroring alive on the shared client", async () => {
    const client = createClient();
@@ -314,12 +335,10 @@ describe("CodexNativeSubagentMonitor", () => {
    );
  });

-  it("delivers child agent-message completion when a native subagent becomes idle", async () => {
+  it("delivers a completed child turn with its final agent message", async () => {
    const client = createClient();
    const runtime = createRuntime();
-    const monitor = new CodexNativeSubagentMonitor(client, runtime, {
-      codexHome: "/tmp/codex-home",
-    });
+    const monitor = new CodexNativeSubagentMonitor(client, runtime);
    monitor.registerParent({
      parentThreadId: "parent-thread",
      requesterSessionKey: "agent:main:discord:channel:C123",
@@ -329,15 +348,21 @@ describe("CodexNativeSubagentMonitor", () => {

    await notifyChildStarted(client);
    await client.notify({
-      method: "item/completed",
+      method: "item/agentMessage/delta",
      params: {
        threadId: "child-thread",
-        item: {
-          type: "agentMessage",
-          id: "msg-child-final",
-          phase: "final_answer",
-          text: "child final result",
-        },
+        turnId: "child-turn",
+        itemId: "msg-child-final",
+        delta: "child ",
+      },
+    });
+    await client.notify({
+      method: "item/agentMessage/delta",
+      params: {
+        threadId: "child-thread",
+        turnId: "child-turn",
+        itemId: "msg-child-final",
+        delta: "final result",
      },
    });

@@ -351,6 +376,9 @@ describe("CodexNativeSubagentMonitor", () => {
      },
    });

+    expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled();
+    await client.notify(childTurnCompletedNotification({ status: "completed" }));
+
    expect(runtime.finalizeTaskRunByRunId).toHaveBeenCalledWith(
      expect.objectContaining({
        runId: "codex-thread:child-thread",
@@ -362,7 +390,7 @@ describe("CodexNativeSubagentMonitor", () => {
      expect.objectContaining({
        childSessionId: "child-thread",
        status: "succeeded",
-        statusLabel: "agent_message",
+        statusLabel: "turn_completed",
        result: "child final result",
      }),
    );
@@ -370,12 +398,10 @@ describe("CodexNativeSubagentMonitor", () => {
    client.close();
  });

-  it("does not deliver commentary-only child messages as native subagent completion", async () => {
+  it("does not complete commentary-only child messages before a terminal turn", async () => {
    const client = createClient();
    const runtime = createRuntime();
-    const monitor = new CodexNativeSubagentMonitor(client, runtime, {
-      codexHome: "/tmp/codex-home",
-    });
+    const monitor = new CodexNativeSubagentMonitor(client, runtime);
    monitor.registerParent({
      parentThreadId: "parent-thread",
      requesterSessionKey: "agent:main:discord:channel:C123",
@@ -384,10 +410,20 @@ describe("CodexNativeSubagentMonitor", () => {
    });

    await notifyChildStarted(client);
+    await client.notify({
+      method: "item/agentMessage/delta",
+      params: {
+        threadId: "child-thread",
+        turnId: "child-turn",
+        itemId: "msg-child-commentary",
+        delta: "checking now",
+      },
+    });
    await client.notify({
      method: "item/completed",
      params: {
        threadId: "child-thread",
+        turnId: "child-turn",
        item: {
          type: "agentMessage",
          id: "msg-child-commentary",
@@ -407,6 +443,162 @@ describe("CodexNativeSubagentMonitor", () => {
    expect(runtime.finalizeTaskRunByRunId).not.toHaveBeenCalled();
    expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled();

+    await client.notify(childTurnCompletedNotification({ status: "completed" }));
+
+    expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith(
+      expect.objectContaining({
+        childSessionId: "child-thread",
+        result: "Codex native subagent completed without a final assistant message.",
+      }),
+    );
+
+    client.close();
+  });
+
+  it("delivers a completed child turn with its snapshot-only final message", async () => {
+    const client = createClient();
+    const runtime = createRuntime();
+    const monitor = new CodexNativeSubagentMonitor(client, runtime);
+    monitor.registerParent({
+      parentThreadId: "parent-thread",
+      requesterSessionKey: "agent:main:discord:channel:C123",
+      taskRuntimeScope: createTaskScope(),
+      agentId: "main",
+    });
+
+    await notifyChildStarted(client);
+    await client.notify(
+      childTurnCompletedNotification({
+        status: "completed",
+        items: [
+          {
+            id: "msg-child-snapshot",
+            type: "agentMessage",
+            phase: "final_answer",
+            text: "snapshot final result",
+          },
+        ],
+      }),
+    );
+
+    expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith(
+      expect.objectContaining({
+        childSessionId: "child-thread",
+        result: "snapshot final result",
+      }),
+    );
+
+    client.close();
+  });
+
+  it("reconciles transcript text for a completed child turn without a final message", async () => {
+    const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-codex-subagent-"));
+    const codexHome = path.join(tempDir, "codex-home");
+    const transcriptDir = path.join(codexHome, "sessions", "2026", "06", "09");
+    await fs.mkdir(transcriptDir, { recursive: true });
+    await fs.writeFile(
+      path.join(transcriptDir, "rollout-2026-06-09T10-11-12-child-thread.jsonl"),
+      [
+        JSON.stringify({
+          type: "session_meta",
+          payload: {
+            source: {
+              subagent: {
+                thread_spawn: {
+                  parent_thread_id: "parent-thread",
+                  depth: 1,
+                },
+              },
+            },
+          },
+        }),
+        JSON.stringify({
+          timestamp: "2026-06-09T10:12:00.000Z",
+          type: "event_msg",
+          payload: {
+            type: "task_complete",
+            last_agent_message: "child turn transcript result",
+            completed_at: 1781009520,
+          },
+        }),
+        "",
+      ].join("\n"),
+    );
+    const client = createClient();
+    const runtime = createRuntime();
+    const monitor = new CodexNativeSubagentMonitor(client, runtime, {
+      codexHome,
+      transcriptPollDelaysMs: [60_000],
+    });
+    monitor.registerParent({
+      parentThreadId: "parent-thread",
+      requesterSessionKey: "agent:main:discord:channel:C123",
+      taskRuntimeScope: createTaskScope(),
+      agentId: "main",
+    });
+
+    await notifyChildStarted(client);
+    await client.notify(childTurnCompletedNotification({ status: "completed" }));
+
+    expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith(
+      expect.objectContaining({
+        childSessionId: "child-thread",
+        statusLabel: "task_complete",
+        result: "child turn transcript result",
+      }),
+    );
+
+    client.close();
+  });
+
+  it("does not reuse an interrupted child turn's message after resuming", async () => {
+    const client = createClient();
+    const runtime = createRuntime();
+    const monitor = new CodexNativeSubagentMonitor(client, runtime);
+    monitor.registerParent({
+      parentThreadId: "parent-thread",
+      requesterSessionKey: "agent:main:discord:channel:C123",
+      taskRuntimeScope: createTaskScope(),
+      agentId: "main",
+    });
+
+    await notifyChildStarted(client);
+    await client.notify({
+      method: "item/completed",
+      params: {
+        threadId: "child-thread",
+        turnId: "child-turn",
+        item: {
+          type: "agentMessage",
+          id: "msg-child-partial",
+          text: "partial child result",
+        },
+      },
+    });
+    await client.notify({
+      method: "thread/status/changed",
+      params: {
+        threadId: "child-thread",
+        status: { type: "idle" },
+      },
+    });
+    await client.notify(childTurnCompletedNotification({ status: "interrupted" }));
+
+    expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled();
+
+    await client.notify(
+      childTurnCompletedNotification({ status: "completed", turnId: "resumed-child-turn" }),
+    );
+
+    expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1);
+    expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith(
+      expect.objectContaining({
+        childSessionId: "child-thread",
+        status: "succeeded",
+        result: "Codex native subagent completed without a final assistant message.",
+      }),
+    );
+
    client.close();
  });

--- a/extensions/codex/src/app-server/native-subagent-monitor.ts
+++ b/extensions/codex/src/app-server/native-subagent-monitor.ts
@@ -51,14 +51,11 @@ type ParentState = {
 type ChildState = {
  childThreadId: string;
  parentThreadId: string;
+  assistantMessagesByTurn: Map<string, ChildAssistantMessages>;
  transcriptPath?: string;
  transcriptPollAttempt: number;
  transcriptPollTimer?: ReturnType<typeof setTimeout>;
  transcriptTerminal: boolean;
-  idle: boolean;
-  lastAgentMessage?: string;
-  lastAgentMessageAt?: number;
-  agentMessageCompletionDelivered: boolean;
  pendingCompletion?: CodexNativeSubagentCompletion;
  pendingCompletionEventAt?: number;
  completionDeliveryAttempt: number;
@@ -67,6 +64,12 @@ type ChildState = {
  noFinalCompletionFallbackTimer?: ReturnType<typeof setTimeout>;
 };

+type ChildAssistantMessages = {
+  texts: Map<string, string>;
+  order: string[];
+  commentaryIds: Set<string>;
+};
+
 type TranscriptCompletion = CodexNativeSubagentCompletion & {
  parentThreadId?: string;
  completedAt?: number;
@@ -215,10 +218,9 @@ export class CodexNativeSubagentMonitor {
        });
      }
    }
-    const childThreadId = this.recordChildAgentMessage(notification);
-    const idleChildThreadId = this.recordChildIdle(notification);
+    this.captureChildAssistantMessage(notification);
+    await this.handleChildTurnCompletion(notification);
    await this.handleCompletionNotification(notification);
-    await this.processChildAgentMessageCompletion(childThreadId ?? idleChildThreadId);
  }

  private ensureParentTaskRuntime(state: ParentState): void {
@@ -299,7 +301,11 @@ export class CodexNativeSubagentMonitor {
        nativeCompletion.agentPath,
      );
      const childState = childThreadId ? this.childStates.get(childThreadId) : undefined;
-      if (!childState || childState.parentThreadId !== state.parentThreadId) {
+      if (
+        !childState ||
+        childState.parentThreadId !== state.parentThreadId ||
+        childState.transcriptTerminal
+      ) {
        embeddedAgentLog.warn(
          "Ignoring Codex native subagent completion for unknown child thread",
          {
@@ -310,21 +316,148 @@ export class CodexNativeSubagentMonitor {
        continue;
      }
      const completion = toThreadCompletion(nativeCompletion, childState.childThreadId);
-      if (shouldWaitForTranscriptCompletion(completion, this.codexHome)) {
-        // Codex can notify `completed: null` before the child transcript exposes
-        // its final assistant message; poll briefly before delivering the no-final fallback.
-        const eventAt = Date.now();
-        const reconciled = await this.reconcileChildTranscript(childState.childThreadId);
-        if (!reconciled) {
-          this.scheduleTranscriptPoll(childState);
-          this.scheduleNoFinalCompletionFallback(state, childState, completion, eventAt);
-        }
-        continue;
-      }
-      await this.processCompletion(state, completion);
+      await this.processChildCompletion(state, childState, completion);
    }
  }

+  private captureChildAssistantMessage(notification: CodexServerNotification): void {
+    const params = isJsonObject(notification.params) ? notification.params : undefined;
+    const childThreadId = readString(params, "threadId")?.trim();
+    const childState = childThreadId ? this.childStates.get(childThreadId) : undefined;
+    if (!childState || childState.transcriptTerminal) {
+      return;
+    }
+    if (notification.method === "item/agentMessage/delta") {
+      const turnId = readString(params, "turnId");
+      const itemId = readString(params, "itemId");
+      const delta = readString(params, "delta");
+      if (turnId && itemId && delta) {
+        this.recordChildAssistantMessage(childState, turnId, itemId, delta);
+      }
+      return;
+    }
+    if (notification.method !== "item/completed") {
+      return;
+    }
+    const turnId = readString(params, "turnId");
+    const item = isJsonObject(params?.item) ? params.item : undefined;
+    this.captureChildAssistantMessageItem(childState, turnId, item);
+  }
+
+  private captureChildAssistantMessageItem(
+    childState: ChildState,
+    turnId: string | undefined,
+    item: JsonObject | undefined,
+  ): void {
+    if (readString(item, "type") !== "agentMessage") {
+      return;
+    }
+    const itemId = readString(item, "id");
+    if (!turnId || !itemId) {
+      return;
+    }
+    const assistantMessages = this.getChildAssistantMessages(childState, turnId);
+    const phase = readString(item, "phase");
+    if (phase === "commentary") {
+      assistantMessages.commentaryIds.add(itemId);
+    }
+    const text = readString(item, "text");
+    if (text) {
+      this.recordChildAssistantMessage(childState, turnId, itemId, text, { replace: true });
+    }
+  }
+
+  private captureChildTurnAssistantMessages(childState: ChildState, turn: JsonObject): void {
+    const turnId = readString(turn, "id");
+    if (!turnId || !Array.isArray(turn.items)) {
+      return;
+    }
+    for (const item of turn.items) {
+      this.captureChildAssistantMessageItem(
+        childState,
+        turnId,
+        isJsonObject(item) ? item : undefined,
+      );
+    }
+  }
+
+  private recordChildAssistantMessage(
+    childState: ChildState,
+    turnId: string,
+    itemId: string,
+    text: string,
+    options: { replace?: boolean } = {},
+  ): void {
+    const assistantMessages = this.getChildAssistantMessages(childState, turnId);
+    if (!assistantMessages.texts.has(itemId)) {
+      assistantMessages.order.push(itemId);
+    }
+    const existing = assistantMessages.texts.get(itemId) ?? "";
+    assistantMessages.texts.set(itemId, options.replace ? text : `${existing}${text}`);
+  }
+
+  private getChildAssistantMessages(
+    childState: ChildState,
+    turnId: string,
+  ): ChildAssistantMessages {
+    const existing = childState.assistantMessagesByTurn.get(turnId);
+    if (existing) {
+      return existing;
+    }
+    const assistantMessages: ChildAssistantMessages = {
+      texts: new Map<string, string>(),
+      order: [],
+      commentaryIds: new Set<string>(),
+    };
+    childState.assistantMessagesByTurn.set(turnId, assistantMessages);
+    return assistantMessages;
+  }
+
+  private async handleChildTurnCompletion(notification: CodexServerNotification): Promise<void> {
+    if (notification.method !== "turn/completed") {
+      return;
+    }
+    const params = isJsonObject(notification.params) ? notification.params : undefined;
+    const childThreadId = readString(params, "threadId")?.trim();
+    const childState = childThreadId ? this.childStates.get(childThreadId) : undefined;
+    const state = childState ? this.parentStates.get(childState.parentThreadId) : undefined;
+    const turn = isJsonObject(params?.turn) ? params.turn : undefined;
+    if (childState && turn && readString(turn, "status") === "interrupted") {
+      const turnId = readString(turn, "id");
+      if (turnId) {
+        childState.assistantMessagesByTurn.delete(turnId);
+      }
+      return;
+    }
+    if (childState && turn) {
+      this.captureChildTurnAssistantMessages(childState, turn);
+    }
+    const completion = childState && turn ? toChildTurnCompletion(childState, turn) : undefined;
+    if (!state || !childState || childState.transcriptTerminal || !completion) {
+      return;
+    }
+    await this.processChildCompletion(state, childState, completion);
+  }
+
+  private async processChildCompletion(
+    state: ParentState,
+    childState: ChildState,
+    completion: CodexNativeSubagentCompletion,
+  ): Promise<void> {
+    if (shouldWaitForTranscriptCompletion(completion, this.codexHome)) {
+      // Codex can notify `completed: null` before the child transcript exposes
+      // its final assistant message; poll briefly before delivering the no-final fallback.
+      const eventAt = Date.now();
+      const reconciled = await this.reconcileChildTranscript(childState.childThreadId);
+      if (!reconciled) {
+        this.scheduleTranscriptPoll(childState);
+        this.scheduleNoFinalCompletionFallback(state, childState, completion, eventAt);
+      }
+      return;
+    }
+    await this.processCompletion(state, completion);
+  }
+
  async reconcileChildTranscript(
    childThreadId: string,
    options: { allowTreeScan?: boolean } = {},
@@ -557,10 +690,9 @@ export class CodexNativeSubagentMonitor {
      childState = {
        childThreadId: normalizedChildThreadId,
        parentThreadId: normalizedParentThreadId,
+        assistantMessagesByTurn: new Map<string, ChildAssistantMessages>(),
        transcriptPollAttempt: 0,
        transcriptTerminal: false,
-        idle: false,
-        agentMessageCompletionDelivered: false,
        completionDeliveryAttempt: 0,
      };
      this.childStates.set(normalizedChildThreadId, childState);
@@ -570,83 +702,6 @@ export class CodexNativeSubagentMonitor {
    }
  }

-  private recordChildAgentMessage(notification: CodexServerNotification): string | undefined {
-    if (notification.method !== "item/completed") {
-      return undefined;
-    }
-    const params = isJsonObject(notification.params) ? notification.params : undefined;
-    const item = isJsonObject(params?.item) ? params.item : undefined;
-    if (!params || !item || readString(item, "type") !== "agentMessage") {
-      return undefined;
-    }
-    const childThreadId = readString(params, "threadId")?.trim();
-    const childState = childThreadId ? this.childStates.get(childThreadId) : undefined;
-    if (!childState || childState.transcriptTerminal) {
-      return undefined;
-    }
-    // Codex app-server can report the child final answer as the child thread's
-    // own agentMessage without also emitting a parent subagent notification.
-    // Pair it with idle below so commentary does not become a false terminal.
-    const phase = readString(item, "phase");
-    if (phase === "commentary") {
-      return undefined;
-    }
-    const text = readString(item, "text")?.trim();
-    if (!text) {
-      return undefined;
-    }
-    childState.lastAgentMessage = text;
-    childState.lastAgentMessageAt = Date.now();
-    return childState.childThreadId;
-  }
-
-  private recordChildIdle(notification: CodexServerNotification): string | undefined {
-    if (notification.method !== "thread/status/changed") {
-      return undefined;
-    }
-    const params = isJsonObject(notification.params) ? notification.params : undefined;
-    if (!params || !isJsonObject(params.status) || readString(params.status, "type") !== "idle") {
-      return undefined;
-    }
-    const childThreadId = readString(params, "threadId")?.trim();
-    const childState = childThreadId ? this.childStates.get(childThreadId) : undefined;
-    if (!childState || childState.transcriptTerminal) {
-      return undefined;
-    }
-    childState.idle = true;
-    return childState.childThreadId;
-  }
-
-  private async processChildAgentMessageCompletion(
-    childThreadId: string | undefined,
-  ): Promise<void> {
-    const childState = childThreadId ? this.childStates.get(childThreadId) : undefined;
-    if (
-      !childState ||
-      !childState.idle ||
-      childState.transcriptTerminal ||
-      childState.agentMessageCompletionDelivered ||
-      !childState.lastAgentMessage
-    ) {
-      return;
-    }
-    const state = this.parentStates.get(childState.parentThreadId);
-    if (!state) {
-      return;
-    }
-    childState.agentMessageCompletionDelivered = true;
-    await this.processCompletion(
-      state,
-      {
-        childThreadId: childState.childThreadId,
-        status: "succeeded",
-        statusLabel: "agent_message",
-        result: childState.lastAgentMessage,
-      },
-      childState.lastAgentMessageAt,
-    );
-  }
-
  private ensureChildState(parentThreadId: string, childThreadId: string): ChildState {
    this.registerChildThread(parentThreadId, childThreadId);
    return this.childStates.get(childThreadId.trim())!;
@@ -970,6 +1025,59 @@ function buildCompletionDedupeKey(
  return `${parentThreadId}:${completion.childThreadId}:${completion.status}:${hash}`;
 }

+function toChildTurnCompletion(
+  childState: ChildState,
+  turn: JsonObject,
+): CodexNativeSubagentCompletion | undefined {
+  const status = readString(turn, "status");
+  if (status === "completed") {
+    const turnId = readString(turn, "id");
+    const result = turnId ? lastChildAssistantMessage(childState, turnId) : undefined;
+    return {
+      childThreadId: childState.childThreadId,
+      status: "succeeded",
+      statusLabel: result ? "turn_completed" : "completed_without_final_message",
+      result: result ?? "Codex native subagent completed without a final assistant message.",
+    };
+  }
+  if (status === "failed") {
+    return {
+      childThreadId: childState.childThreadId,
+      status: "failed",
+      statusLabel: "turn_failed",
+      result: readTurnErrorMessage(turn) ?? "Codex native subagent failed.",
+    };
+  }
+  return undefined;
+}
+
+function lastChildAssistantMessage(childState: ChildState, turnId: string): string | undefined {
+  const assistantMessages = childState.assistantMessagesByTurn.get(turnId);
+  if (!assistantMessages) {
+    return undefined;
+  }
+  for (let index = assistantMessages.order.length - 1; index >= 0; index -= 1) {
+    const itemId = assistantMessages.order[index];
+    if (!assistantMessages.commentaryIds.has(itemId)) {
+      const text = normalizeOptionalString(assistantMessages.texts.get(itemId));
+      if (text) {
+        return text;
+      }
+    }
+  }
+  return undefined;
+}
+
+function readTurnErrorMessage(turn: JsonObject): string | undefined {
+  const error = isJsonObject(turn.error) ? turn.error : undefined;
+  return (
+    normalizeOptionalString(readString(error, "message")) ??
+    normalizeOptionalString(
+      isJsonObject(error?.codexErrorInfo) ? readString(error.codexErrorInfo, "message") : undefined,
+    )
+  );
+}
+
 function buildParentAgentPathKey(parentThreadId: string, agentPath: string): string {
  return `${parentThreadId}\0${agentPath}`;
 }