diff --git a/extensions/codex/src/app-server/native-subagent-monitor.test.ts b/extensions/codex/src/app-server/native-subagent-monitor.test.ts index f79f5da6fdfd..ae7f5eaab3a5 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.test.ts +++ b/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(); diff --git a/extensions/codex/src/app-server/native-subagent-monitor.ts b/extensions/codex/src/app-server/native-subagent-monitor.ts index 7a710aa06a6b..5ee48af1e7ac 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.ts +++ b/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; + nativeCompletionReceipts: Set; 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; 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); }