Keep native completion receipts through child recovery (#130347)

extensions/codex/src/app-server/native-subagent-monitor.test.ts

@@ -315,6 +315,7 @@ function threadRead(
   params: {
     childThreadId?: string;
     parentThreadId?: string;
+    agentPath?: string;
     status?: "completed" | "failed" | "interrupted" | "inProgress";
     result?: string;
     error?: string;
@@ -357,7 +358,11 @@ function threadRead(
       ...(params.directParentField === false ? {} : { parentThreadId }),
       source: {
         subAgent: {
-          thread_spawn: { parent_thread_id: parentThreadId, depth: 1 },
+          thread_spawn: {
+            parent_thread_id: parentThreadId,
+            depth: 1,
+            ...(params.agentPath ? { agent_path: params.agentPath } : {}),
+          },
         },
       },
       status: { type: params.threadStatus ?? "idle" },
@@ -614,6 +619,95 @@ describe("CodexNativeSubagentMonitor", () => {
       }
     });
 
+    it.each(["before", "after"])(
+      "retains a native receipt when task recovery finishes %s parent release",
+      async (order) => {
+        const client = createClient();
+        let releaseRead!: (response: CodexThreadReadResponse) => void;
+        client.setThreadReadFactory(
+          "child-thread",
+          () =>
+            new Promise((resolve) => {
+              releaseRead = resolve;
+            }),
+        );
+        const runtime = createRuntime();
+        runtime.listTaskRecords.mockReturnValue([taskRecord({ childThreadId: "child-thread" })]);
+        const monitor = new CodexNativeSubagentMonitor(client as never, runtime);
+        const owner = registerParent(monitor);
+        owner.bindTurn("parent-turn");
+        expect(client.request).toHaveBeenCalledOnce();
+        await client.notify(deliveredNativeCompletion());
+        if (order === "after") {
+          owner.unregister();
+        }
+        releaseRead(threadRead({ agentPath: "/root/worker", result: "The build passed." }));
+        await vi.waitFor(() => expect(runtime.finalizeTaskRunByRunId).toHaveBeenCalledOnce());
+        owner.unregister();
+        expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled();
+        expect(runtime.setDetachedTaskDeliveryStatusByRunId).toHaveBeenLastCalledWith({
+          runId: "codex-thread:child-thread",
+          deliveryStatus: "delivered",
+        });
+        client.close();
+      },
+    );
+
+    it.each(["other-turn", "other-lineage"])(
+      "does not acknowledge recovered delivery from %s",
+      async (source) => {
+        const client = createClient();
+        let releaseRead!: (response: CodexThreadReadResponse) => void;
+        client.setThreadReadFactory(
+          "child-thread",
+          () =>
+            new Promise((resolve) => {
+              releaseRead = resolve;
+            }),
+        );
+        const runtime = createRuntime();
+        runtime.listTaskRecords.mockReturnValue([taskRecord({ childThreadId: "child-thread" })]);
+        const monitor = new CodexNativeSubagentMonitor(client as never, runtime);
+        const owner = registerParent(monitor);
+        owner.bindTurn("parent-turn");
+        const receipt = deliveredNativeCompletion();
+        if (source === "other-turn") {
+          (receipt.params as JsonObject).turnId = "old-turn";
+        }
+        await client.notify(receipt);
+        owner.unregister();
+        releaseRead(
+          threadRead({
+            agentPath: "/root/worker",
+            parentThreadId: source === "other-lineage" ? "old-parent" : "parent-thread",
+            result: "The build passed.",
+          }),
+        );
+        await vi.waitFor(() =>
+          expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledOnce(),
+        );
+        client.close();
+      },
+    );
+
+    it("does not carry an unmatched receipt into a later parent run", async () => {
+      const client = createClient();
+      const runtime = createRuntime();
+      const monitor = new CodexNativeSubagentMonitor(client as never, runtime);
+      const first = registerParent(monitor);
+      first.bindTurn("parent-turn");
+      await notifyChildStarted(client, "parent-thread", "waiting-child");
+      await client.notify(deliveredNativeCompletion());
+      first.unregister();
+      const second = registerParent(monitor);
+      second.bindTurn("next-turn");
+      await notifyChildStarted(client, "parent-thread", "child-thread", "/root/worker");
+      await client.notify(completedChild());
+      second.unregister();
+      expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledOnce();
+      client.close();
+    });
+
     it("delivers a deferred completion if the parent client closes", async () => {
       const client = createClient();
       const runtime = createRuntime();

extensions/codex/src/app-server/native-subagent-monitor.ts

@@ -71,6 +71,7 @@ type ParentState = {
   // turn/started can precede bindTurn; retain receipt ownership until the
   // foreground run has finalized its reply and releases this registration.
   turnIds: Set<string>;
+  nativeCompletionReceipts: Set<string>;
   requesterSessionKey?: string;
   taskRuntimeScope?: AgentHarnessTaskRuntimeScope;
   agentId?: string;
@@ -118,6 +119,7 @@ type RecoveredCompletion = CodexNativeSubagentCompletion & {
 
 type ThreadRecovery = {
   parentThreadId?: string;
+  agentPath?: string;
   completion?: RecoveredCompletion;
   fallbackCompletion?: RecoveredCompletion;
   resumable: boolean;
@@ -133,6 +135,7 @@ type ThreadStatusRevision = {
 
 type TaskRecoveryCandidate = {
   parentState: ParentState;
+  nativeCompletionReceipts: Set<string>;
   childThreadId: string;
   recoveryAttempt: number;
   requesterSessionKey: string;
@@ -390,7 +393,12 @@ class Monitor {
       throw new Error(`Codex thread ${parentThreadId} is already bound to another session`);
     }
     if (!state) {
-      state = { parentThreadId, owners: new Map(), turnIds: new Set() };
+      state = {
+        parentThreadId,
+        owners: new Map(),
+        turnIds: new Set(),
+        nativeCompletionReceipts: new Set(),
+      };
       this.parentStates.set(parentThreadId, state);
     }
     state.requesterSessionKey ??= params.requesterSessionKey;
@@ -452,6 +460,9 @@ class Monitor {
           }
           if (current.owners.size === 0) {
             current.turnIds.clear();
+            // In-flight recovery retains this run's receipts; a later run must
+            // not inherit them merely because it reuses an agent path.
+            current.nativeCompletionReceipts = new Set();
           }
           this.clearUnconsumablePendingDirectSpawnEvidence();
           this.deliverDetachedCompletions(current);
@@ -555,6 +566,9 @@ class Monitor {
     const childState = threadId ? this.childStates.get(threadId) : undefined;
     if (notification.method === "turn/started" && childState) {
       childState.nativeCompletionDelivered = false;
+      for (const key of childState.agentPathKeys) {
+        state?.nativeCompletionReceipts.delete(key);
+      }
       this.resumeChild(childState);
     }
     if (parent && parent.turnIds.has(readString(params, "turnId") ?? "")) {
@@ -1070,6 +1084,9 @@ class Monitor {
         this.unregisterChild(childState);
         return false;
       }
+      if (recovery.agentPath) {
+        this.registerAgentPath(childState, recovery.agentPath);
+      }
       if (recovery.threadState === "active") {
         this.observeActiveChild(childState);
         return false;
@@ -1181,6 +1198,7 @@ class Monitor {
     }
     return {
       parentThreadId: readThreadParentThreadId(thread),
+      agentPath: normalizeOptionalString(readString(readThreadSpawnSource(thread), "agent_path")),
       completion,
       fallbackCompletion,
       resumable,
@@ -1312,9 +1330,11 @@ class Monitor {
     notification: CodexServerNotification,
   ): void {
     for (const agentPath of nativeSubagentNotifications.deliveredAgentPaths(notification)) {
-      const childThreadId = this.childThreadIdsByAgentPath.get(
-        buildParentAgentPathKey(state.parentThreadId, agentPath),
-      );
+      const key = buildParentAgentPathKey(state.parentThreadId, agentPath);
+      // Native input can arrive before asynchronous history restores the child
+      // mapping. Preserve the observed delivery, not a guess from task status.
+      state.nativeCompletionReceipts.add(key);
+      const childThreadId = this.childThreadIdsByAgentPath.get(key);
       const child = childThreadId ? this.childStates.get(childThreadId) : undefined;
       if (!child || child.parentThreadId !== state.parentThreadId) {
         continue;
@@ -1588,6 +1608,9 @@ class Monitor {
     }
     this.childThreadIdsByAgentPath.set(key, childState.childThreadId);
     childState.agentPathKeys.add(key);
+    if (this.parentStates.get(childState.parentThreadId)?.nativeCompletionReceipts.has(key)) {
+      childState.nativeCompletionDelivered = true;
+    }
   }
 
   private unregisterChild(
@@ -1876,6 +1899,7 @@ class Monitor {
       const childThreadId = task.runId!.slice(CODEX_NATIVE_SUBAGENT_RUN_ID_PREFIX.length).trim();
       candidates.set(childThreadId, {
         parentState: state,
+        nativeCompletionReceipts: state.nativeCompletionReceipts,
         requesterSessionKey: state.requesterSessionKey,
         childThreadId,
         recoveryAttempt: 0,
@@ -1986,6 +2010,7 @@ class Monitor {
           parentThreadId,
           owners: new Map(),
           turnIds: new Set(),
+          nativeCompletionReceipts: new Set(),
           requesterSessionKey: candidate.requesterSessionKey,
           taskRuntimeScope: candidate.taskRuntimeScope,
           agentId: candidate.agentId,
@@ -1994,11 +2019,20 @@ class Monitor {
         this.prepareParentTaskRuntime(state);
         this.parentStates.set(parentThreadId, state);
       }
-      const childState = this.registerChildThread(state, candidate.childThreadId);
+      const childState = this.registerChildThread(
+        state,
+        candidate.childThreadId,
+        recovery.agentPath ? { agentPath: recovery.agentPath } : {},
+      );
       if (!childState) {
         this.pruneParentIfUnused(state);
         return;
       }
+      if (
+        [...childState.agentPathKeys].some((key) => candidate.nativeCompletionReceipts.has(key))
+      ) {
+        childState.nativeCompletionDelivered = true;
+      }
       if (recovery.threadState === "active") {
         this.observeActiveChild(childState);
       }