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 8431b4396035..f79f5da6fdfd 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.test.ts +++ b/extensions/codex/src/app-server/native-subagent-monitor.test.ts @@ -5,7 +5,7 @@ import type { AgentHarnessTaskRecord, AgentHarnessTaskRuntimeScope, } from "openclaw/plugin-sdk/agent-harness-task-runtime"; -import { describe, expect, it, vi } from "vitest"; +import { describe, expect, it, onTestFinished, vi } from "vitest"; import { claimCodexAppServerLiveThread, consumeCodexAppServerLiveThread, @@ -15,6 +15,12 @@ import { retainCodexAppServerLiveThread, } from "./client-runtime.js"; import { createFakeCodexAppServerClient } from "./codex-app-server.test-fixtures.js"; +import { + buildEmptyToolTelemetry, + CodexAppServerEventProjector, + createParams, + registerCodexEventProjectorTestLifecycle, +} from "./event-projector.test-harness.js"; import { codexNativeSubagentMonitorRuntime } from "./native-subagent-monitor.js"; import type { CodexAppServerRequestResult, @@ -87,6 +93,10 @@ function createClient() { } return typeof response === "function" ? await response(readParams) : response; }); + onTestFinished(async () => { + fixture.close(); + await Promise.resolve(); + }); return { request: fixture.request, setThreadRead(childThreadId: string, response: CodexThreadReadResponse | Error) { @@ -205,12 +215,22 @@ async function notifyChildStarted( return notification; } +async function registerDetachedChild( + client: ReturnType, + monitor: CodexNativeSubagentMonitorInstance, +): Promise { + const owner = registerParent(monitor); + await notifyChildStarted(client); + owner.unregister(); +} + function nativeCompletionNotification( params: { agentPath?: string; statusLabel?: string; result?: string | null; parentThreadId?: string; + turnId?: string; } = {}, ): CodexServerNotification { const agentPath = params.agentPath ?? "child-thread"; @@ -224,6 +244,7 @@ function nativeCompletionNotification( method: "rawResponseItem/completed", params: { threadId: params.parentThreadId ?? "parent-thread", + ...(params.turnId ? { turnId: params.turnId } : {}), item: { type: "message", role: "assistant", @@ -391,6 +412,223 @@ function taskRecord(params: { } describe("CodexNativeSubagentMonitor", () => { + describe("native completion delivery ownership", () => { + registerCodexEventProjectorTestLifecycle(); + + function deliveredNativeCompletion(): CodexServerNotification { + return { + method: "rawResponseItem/completed", + params: { + threadId: "parent-thread", + turnId: "parent-turn", + item: { + type: "agent_message", + author: "/root/worker", + recipient: "/root", + content: [ + { + type: "input_text", + text: "Message Type: FINAL_ANSWER\nTask name: /root\nSender: /root/worker\nPayload:\nThe build passed.", + }, + ], + }, + }, + }; + } + + const completedChild = () => + childTurnCompletedNotification({ + status: "completed", + items: [ + { + type: "agentMessage", + id: "child-final", + phase: "final_answer", + text: "The build passed.", + }, + ], + }); + + it.each([ + { order: "native-first", final: "The build passed. The change is ready." }, + { order: "terminal-first", final: "The build passed. The change is ready." }, + { order: "native-first", final: "NO_REPLY" }, + ])( + "preserves $final when native delivery and child completion arrive $order", + async ({ order, final }) => { + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client as never, runtime); + const owner = registerParent(monitor); + owner.bindTurn("parent-turn"); + await notifyChildStarted(client, "parent-thread", "child-thread", "/root/worker"); + const projector = new CodexAppServerEventProjector( + await createParams(), + "parent-thread", + "parent-turn", + ); + let lastAnswer = ""; + const answer = async (text: string, id: string) => { + lastAnswer = text; + await projector.handleNotification({ + method: "item/completed", + params: { + threadId: "parent-thread", + turnId: "parent-turn", + item: { type: "agentMessage", id, phase: "final_answer", text }, + }, + }); + }; + runtime.deliverAgentHarnessTaskCompletion.mockImplementation(async () => { + await answer("NO_REPLY", "duplicate-answer"); + return { delivered: true, path: "steered" }; + }); + try { + if (order === "terminal-first") { + await client.notify(completedChild()); + } + await client.notify(deliveredNativeCompletion()); + await answer(final, "parent-answer"); + if (order === "native-first") { + await client.notify(completedChild()); + } + await projector.handleNotification({ + method: "turn/completed", + params: { + threadId: "parent-thread", + turn: { + id: "parent-turn", + status: "completed", + items: [{ type: "agentMessage", id: "last-answer", text: lastAnswer }], + error: null, + }, + }, + }); + owner.unregister(); + expect(projector.buildResult(buildEmptyToolTelemetry()).assistantTexts).toEqual([final]); + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + expect(runtime.setDetachedTaskDeliveryStatusByRunId).toHaveBeenLastCalledWith({ + runId: "codex-thread:child-thread", + deliveryStatus: "delivered", + }); + } finally { + owner.unregister(); + client.close(); + } + }, + ); + + it("defers delivery during unbound parent startup and drains it if startup is released", async () => { + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client as never, runtime); + const owner = registerParent(monitor); + await notifyChildStarted(client); + await client.notify(completedChild()); + try { + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + owner.unregister(); + expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledOnce(); + expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith( + expect.objectContaining({ result: "The build passed." }), + ); + } finally { + owner.unregister(); + client.close(); + } + }); + + it("defers completion when turn/started races ahead of the turn/start response", async () => { + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client as never, runtime); + const owner = registerParent(monitor); + try { + await client.notify({ + method: "turn/started", + params: { + threadId: "parent-thread", + turn: { id: "parent-turn", status: "inProgress", items: [] }, + }, + }); + await notifyChildStarted(client, "parent-thread", "child-thread", "/root/worker"); + await client.notify(completedChild()); + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + await client.notify(deliveredNativeCompletion()); + owner.bindTurn("parent-turn"); + owner.unregister(); + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + expect(runtime.setDetachedTaskDeliveryStatusByRunId).toHaveBeenLastCalledWith({ + runId: "codex-thread:child-thread", + deliveryStatus: "delivered", + }); + } finally { + owner.unregister(); + client.close(); + } + }); + + it.each([ + "other-turn", + "other-parent", + "other-child", + "ordinary-message", + "user-text", + ] as const)("does not acknowledge a completion from %s", async (source) => { + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client as never, runtime); + const owner = registerParent(monitor); + owner.bindTurn("parent-turn"); + await notifyChildStarted(client, "parent-thread", "child-thread", "/root/worker"); + const receipt = deliveredNativeCompletion(); + const params = receipt.params as JsonObject; + const item = params.item as JsonObject; + if (source === "other-turn") { + params.turnId = "older-turn"; + } else if (source === "other-parent") { + params.threadId = "another-parent"; + } else if (source === "other-child") { + item.author = "/root/another-child"; + item.content = [ + { + type: "input_text", + text: "Message Type: FINAL_ANSWER\nTask name: /root\nSender: /root/another-child\nPayload:\nThe build passed.", + }, + ]; + } else if (source === "ordinary-message") { + item.content = [{ type: "input_text", text: "Still working on the build." }]; + } else { + item.type = "message"; + item.role = "user"; + } + try { + await client.notify(completedChild()); + await client.notify(receipt); + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + owner.unregister(); + expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledOnce(); + } finally { + owner.unregister(); + client.close(); + } + }); + + it("delivers a deferred completion if the parent client closes", async () => { + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client as never, runtime); + const owner = registerParent(monitor); + owner.bindTurn("parent-turn"); + await notifyChildStarted(client); + await client.notify(completedChild()); + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + client.close(); + expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledOnce(); + owner.unregister(); + }); + }); + it("pins a parent subscription until its final independently running child settles", async () => { const client = createClient(); const runtime = createRuntime(); @@ -422,8 +660,7 @@ describe("CodexNativeSubagentMonitor", () => { const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { retainParentThread: () => releaseParentThread, }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); client.close(); @@ -441,10 +678,10 @@ describe("CodexNativeSubagentMonitor", () => { retainChildThread, retainParentThread, }); - registerParent(monitor); + registerParent(monitor).bindTurn("parent-turn"); await notifyChildStarted(client); - await client.notify(nativeCompletionNotification()); + await client.notify(nativeCompletionNotification({ turnId: "parent-turn" })); expect(claimChildThread).toHaveBeenCalledExactlyOnceWith("child-thread"); expect(retainChildThread).toHaveBeenCalledExactlyOnceWith("child-thread"); @@ -479,10 +716,10 @@ describe("CodexNativeSubagentMonitor", () => { retainChildThread, releaseChildThread, }); - registerParent(monitor); + registerParent(monitor).bindTurn("parent-turn"); await notifyChildStarted(client); - await client.notify(nativeCompletionNotification()); + await client.notify(nativeCompletionNotification({ turnId: "parent-turn" })); expect(releaseParentThread).toHaveBeenCalledOnce(); await client.notify(closeAgentNotification({ method: "item/started" })); @@ -492,7 +729,7 @@ describe("CodexNativeSubagentMonitor", () => { expect(releaseParentThread).toHaveBeenCalledOnce(); expect(retainChildThread).toHaveBeenCalledExactlyOnceWith("child-thread"); expect(releaseChildThread).toHaveBeenCalledExactlyOnceWith("child-thread"); - expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledOnce(); + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); monitor.dispose(); }); @@ -634,6 +871,8 @@ describe("CodexNativeSubagentMonitor", () => { terminalSummary: "child v2 result", }), ); + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + owner.unregister(); expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith( expect.objectContaining({ childSessionId: "child-v2", @@ -1088,7 +1327,7 @@ describe("CodexNativeSubagentMonitor", () => { const client = createClient(); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - monitor.registerParent({ + const parent = monitor.registerParent({ parentThreadId: "parent-thread", requesterSessionKey: "agent:main:discord:channel:C123", taskRuntimeScope: createTaskScope(), @@ -1096,6 +1335,7 @@ describe("CodexNativeSubagentMonitor", () => { }); await notifyChildStarted(client); + parent.unregister(); await client.notify({ method: "thread/status/changed", params: { @@ -1134,8 +1374,7 @@ describe("CodexNativeSubagentMonitor", () => { const client = createClient(); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify({ method: "item/started", params: { @@ -1284,8 +1523,7 @@ describe("CodexNativeSubagentMonitor", () => { const client = createClient(); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify( childTurnCompletedNotification({ @@ -1313,8 +1551,7 @@ describe("CodexNativeSubagentMonitor", () => { client.setThreadRead("child-thread", threadRead({ result: "history final result" })); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify(childTurnCompletedNotification({ status: "completed" })); @@ -1412,8 +1649,7 @@ describe("CodexNativeSubagentMonitor", () => { const client = createClient(); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); const completion = nativeCompletionNotification(); await client.notify(completion); @@ -1436,8 +1672,7 @@ describe("CodexNativeSubagentMonitor", () => { client.setThreadRead("child-thread", threadRead({ result: "history final result" })); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify(nativeCompletionNotification({ result: null })); @@ -1466,8 +1701,7 @@ describe("CodexNativeSubagentMonitor", () => { const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { recoveryPollDelaysMs: [10], }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify(nativeCompletionNotification({ result: null })); await vi.advanceTimersByTimeAsync(20); @@ -1493,8 +1727,7 @@ describe("CodexNativeSubagentMonitor", () => { const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { recoveryPollDelaysMs: [10], }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify(nativeCompletionNotification({ result: null })); await vi.advanceTimersByTimeAsync(20); @@ -1516,8 +1749,7 @@ describe("CodexNativeSubagentMonitor", () => { const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { recoveryPollDelaysMs: [10], }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify(nativeCompletionNotification({ result: null })); client.setThreadRead( @@ -1561,8 +1793,7 @@ describe("CodexNativeSubagentMonitor", () => { ); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await expect(monitor.reconcileChildThread("child-thread")).resolves.toBe(true); @@ -1578,8 +1809,7 @@ describe("CodexNativeSubagentMonitor", () => { const releaseClient = vi.fn(); const retainClient = vi.fn(() => releaseClient); const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { retainClient }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify(childTurnCompletedNotification({ status: "interrupted" })); @@ -1615,8 +1845,7 @@ describe("CodexNativeSubagentMonitor", () => { ); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await expect(monitor.reconcileChildThread("child-thread")).resolves.toBe(false); @@ -1636,8 +1865,7 @@ describe("CodexNativeSubagentMonitor", () => { ); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await expect(monitor.reconcileChildThread("child-thread")).resolves.toBe(false); @@ -1664,8 +1892,7 @@ describe("CodexNativeSubagentMonitor", () => { const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { recoveryPollDelaysMs: [10], }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await expect(monitor.reconcileChildThread("child-thread")).resolves.toBe(false); await vi.advanceTimersByTimeAsync(30); @@ -1698,8 +1925,7 @@ describe("CodexNativeSubagentMonitor", () => { }); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await expect(monitor.reconcileChildThread("child-thread")).resolves.toBe(false); @@ -1729,8 +1955,7 @@ describe("CodexNativeSubagentMonitor", () => { }); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await expect(monitor.reconcileChildThread("child-thread")).resolves.toBe(true); @@ -1758,8 +1983,7 @@ describe("CodexNativeSubagentMonitor", () => { recoveryPollDelaysMs: [10], retainClient: () => releaseClient, }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify({ method: "thread/status/changed", @@ -1808,8 +2032,7 @@ describe("CodexNativeSubagentMonitor", () => { recoveryPollDelaysMs: [10], retainClient: () => releaseClient, }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify({ method: "thread/status/changed", @@ -1852,8 +2075,7 @@ describe("CodexNativeSubagentMonitor", () => { recoveryPollDelaysMs: [10], retainClient: () => releaseClient, }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify({ method: "thread/status/changed", @@ -1898,8 +2120,7 @@ describe("CodexNativeSubagentMonitor", () => { ); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await expect(monitor.reconcileChildThread("child-thread")).resolves.toBe(true); @@ -1913,10 +2134,11 @@ describe("CodexNativeSubagentMonitor", () => { const client = createClient(); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); + const parent = registerParent(monitor); await notifyChildStarted(client, "parent-thread", "child-thread", "1.2", { directParentField: false, }); + parent.unregister(); await client.notify(nativeCompletionNotification({ agentPath: "1.2" })); @@ -1942,8 +2164,7 @@ describe("CodexNativeSubagentMonitor", () => { const client = createClient(); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); // Trust boundary: only assistant commentary carries inter-agent envelopes. // User-authored text quoting the markup must never finalize a real child. @@ -1975,7 +2196,7 @@ describe("CodexNativeSubagentMonitor", () => { const client = createClient(); const runtime = createRuntime(); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor, "parent-a", "agent:main:a"); + const parent = registerParent(monitor, "parent-a", "agent:main:a"); registerParent(monitor, "parent-b", "agent:main:b"); await notifyChildStarted(client, "parent-a", "child-thread"); await notifyChildStarted(client, "parent-b", "child-thread"); @@ -1988,6 +2209,7 @@ describe("CodexNativeSubagentMonitor", () => { ); expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + parent.unregister(); await client.notify( nativeCompletionNotification({ parentThreadId: "parent-a", @@ -2018,8 +2240,7 @@ describe("CodexNativeSubagentMonitor", () => { replacementClient as never, replacementRuntime, ); - registerParent(replacementMonitor); - await notifyChildStarted(replacementClient); + await registerDetachedChild(replacementClient, replacementMonitor); await replacementClient.notify(nativeCompletionNotification()); expect(replacementRuntime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1); @@ -2040,8 +2261,7 @@ describe("CodexNativeSubagentMonitor", () => { completionDeliveryRetryDelaysMs: [10], retainClient: () => releaseClient, }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify(nativeCompletionNotification()); expect(releaseClient).toHaveBeenCalledTimes(1); client.close(); @@ -2069,16 +2289,17 @@ describe("CodexNativeSubagentMonitor", () => { const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { completionDeliveryRetryDelaysMs: [10], }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify(nativeCompletionNotification()); expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1); - registerParent(monitor); + const parent = registerParent(monitor); await vi.advanceTimersByTimeAsync(0); expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1); await vi.advanceTimersByTimeAsync(10); + expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1); + parent.unregister(); expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(2); client.close(); } finally { @@ -2118,8 +2339,7 @@ describe("CodexNativeSubagentMonitor", () => { const firstMonitor = new CodexNativeSubagentMonitor(firstClient as never, runtime, { completionDeliveryRetryDelaysMs: [10], }); - registerParent(firstMonitor); - await notifyChildStarted(firstClient); + await registerDetachedChild(firstClient, firstMonitor); await firstClient.notify(nativeCompletionNotification()); expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1); @@ -2158,8 +2378,7 @@ describe("CodexNativeSubagentMonitor", () => { completionDeliveryMaxRetries: 2, retainClient: () => releaseClient, }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await client.notify(nativeCompletionNotification()); expect(releaseClient).toHaveBeenCalledTimes(1); @@ -2246,7 +2465,9 @@ describe("CodexNativeSubagentMonitor", () => { taskRecord({ childThreadId: "foreign-child", requesterSessionKey: "agent:main:other" }), ]); const monitor = new CodexNativeSubagentMonitor(client as never, runtime); - registerParent(monitor); + const parent = registerParent(monitor); + await vi.waitFor(() => expect(runtime.finalizeTaskRunByRunId).toHaveBeenCalledTimes(1)); + parent.unregister(); await vi.waitFor(() => expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1), ); @@ -2315,13 +2536,14 @@ describe("CodexNativeSubagentMonitor", () => { const second = registerParent(monitor); expect(client.request).toHaveBeenCalledTimes(1); releaseRead(); + await vi.waitFor(() => expect(runtime.finalizeTaskRunByRunId).toHaveBeenCalledTimes(1)); + first.unregister(); + second.unregister(); await vi.waitFor(() => expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1), ); expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1); - first.unregister(); - second.unregister(); client.close(); }); @@ -2339,7 +2561,7 @@ describe("CodexNativeSubagentMonitor", () => { const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { recoveryPollDelaysMs: [10], }); - registerParent(monitor); + const parent = registerParent(monitor); await Promise.resolve(); expect(client.request).toHaveBeenCalledTimes(1); @@ -2354,6 +2576,7 @@ describe("CodexNativeSubagentMonitor", () => { client.setThreadRead("child-thread", threadRead({ result: "fresh completed result" })); await vi.advanceTimersByTimeAsync(10); + parent.unregister(); expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1); expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith( @@ -2387,7 +2610,7 @@ describe("CodexNativeSubagentMonitor", () => { const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { recoveryPollDelaysMs: [10], }); - registerParent(monitor); + const parent = registerParent(monitor); await vi.advanceTimersByTimeAsync(0); expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); expect(client.request).toHaveBeenCalledWith( @@ -2397,6 +2620,7 @@ describe("CodexNativeSubagentMonitor", () => { ); await vi.advanceTimersByTimeAsync(10); + parent.unregister(); expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith( expect.objectContaining({ result: "eventual history result" }), @@ -2492,8 +2716,7 @@ describe("CodexNativeSubagentMonitor", () => { const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { recoveryPollDelaysMs: [10], }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); await vi.advanceTimersByTimeAsync(10); await vi.advanceTimersByTimeAsync(10); @@ -2530,11 +2753,12 @@ describe("CodexNativeSubagentMonitor", () => { runtime, }); first.unregister(); + second.bindTurn("parent-turn"); await notifyChildStarted(client); await vi.waitFor(() => expect(isCodexAppServerLiveThreadClaimed(client as never, "child-thread")).toBe(true), ); - await client.notify(nativeCompletionNotification()); + await client.notify(nativeCompletionNotification({ turnId: "parent-turn" })); await vi.waitFor(() => expect(isCodexAppServerLiveThreadClaimed(client as never, "child-thread")).toBe(false), ); @@ -2544,7 +2768,7 @@ describe("CodexNativeSubagentMonitor", () => { expect(childRelease).toHaveBeenCalledOnce(); expect(runtime.createRunningTaskRun).toHaveBeenCalledTimes(1); - expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1); + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); second.unregister(); await notifyChildStarted(client, "parent-thread", "late-child"); expect(runtime.createRunningTaskRun).toHaveBeenCalledTimes(1); @@ -2568,6 +2792,7 @@ describe("CodexNativeSubagentMonitor", () => { taskRuntimeScope: createTaskScope("agent:main:main"), runtime, }); + parent.bindTurn("parent-turn"); await notifyChildStarted(client); await vi.waitFor(() => @@ -2581,7 +2806,7 @@ describe("CodexNativeSubagentMonitor", () => { ).resolves.toBeUndefined(); expect(client.request).not.toHaveBeenCalled(); - await client.notify(nativeCompletionNotification()); + await client.notify(nativeCompletionNotification({ turnId: "parent-turn" })); await vi.waitFor(() => expect(isCodexAppServerLiveThreadClaimed(client as never, "child-thread")).toBe(false), ); @@ -2625,12 +2850,13 @@ describe("CodexNativeSubagentMonitor", () => { runtime, retainParentThread: () => releaseParentThread, }); + parent.bindTurn("parent-turn"); await notifyChildStarted(client); await vi.waitFor(() => expect(isCodexAppServerLiveThreadClaimed(client as never, "child-thread")).toBe(true), ); - await client.notify(nativeCompletionNotification()); + await client.notify(nativeCompletionNotification({ turnId: "parent-turn" })); await vi.waitFor(() => expect(client.request).toHaveBeenCalledExactlyOnceWith( @@ -2675,12 +2901,13 @@ describe("CodexNativeSubagentMonitor", () => { taskRuntimeScope: createTaskScope("agent:main:main"), runtime, }); + parent.bindTurn("parent-turn"); await notifyChildStarted(client); await vi.waitFor(() => expect(isCodexAppServerLiveThreadClaimed(client as never, "child-thread")).toBe(true), ); - await client.notify(nativeCompletionNotification()); + await client.notify(nativeCompletionNotification({ turnId: "parent-turn" })); await vi.waitFor(() => expect(isCodexAppServerLiveThreadClaimed(client as never, "child-thread")).toBe(false), ); @@ -2718,12 +2945,13 @@ describe("CodexNativeSubagentMonitor", () => { taskRuntimeScope: createTaskScope("agent:main:main"), runtime, }); + parent.bindTurn("parent-turn"); await notifyChildStarted(client); await vi.waitFor(() => expect(isCodexAppServerLiveThreadClaimed(client as never, "child-thread")).toBe(true), ); - await client.notify(nativeCompletionNotification()); + await client.notify(nativeCompletionNotification({ turnId: "parent-turn" })); await vi.waitFor(() => expect(isCodexAppServerLiveThreadClaimed(client as never, "child-thread")).toBe(false), ); @@ -2762,8 +2990,7 @@ describe("CodexNativeSubagentMonitor", () => { const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { recoveryPollDelaysMs: [10], }); - registerParent(monitor); - await notifyChildStarted(client); + await registerDetachedChild(client, monitor); client.close(); await vi.advanceTimersByTimeAsync(30); diff --git a/extensions/codex/src/app-server/native-subagent-monitor.ts b/extensions/codex/src/app-server/native-subagent-monitor.ts index 38fff64711be..7a710aa06a6b 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.ts +++ b/extensions/codex/src/app-server/native-subagent-monitor.ts @@ -68,6 +68,9 @@ type ParentState = { // Overlapping runs share this parent; the last owner releases it only after // detached children finish recovery and delivery. owners: Map; + // turn/started can precede bindTurn; retain receipt ownership until the + // foreground run has finalized its reply and releases this registration. + turnIds: Set; requesterSessionKey?: string; taskRuntimeScope?: AgentHarnessTaskRuntimeScope; agentId?: string; @@ -93,6 +96,7 @@ type ChildState = { terminal: boolean; fallbackCompletion?: RecoveredCompletion; pendingCompletion?: CodexNativeSubagentCompletion; + nativeCompletionDelivered: boolean; completionDeliveryAttempt: number; completionDeliveryTimer?: ReturnType; deliveringCompletion: boolean; @@ -346,6 +350,8 @@ class Monitor { this.parentThreadRetentions.clear(); for (const state of this.parentStates.values()) { state.owners.clear(); + state.turnIds.clear(); + this.deliverDetachedCompletions(state); } this.pendingDirectSpawnEvidence.clear(); for (const [parentThreadId] of this.parentStates) { @@ -384,7 +390,7 @@ class Monitor { throw new Error(`Codex thread ${parentThreadId} is already bound to another session`); } if (!state) { - state = { parentThreadId, owners: new Map() }; + state = { parentThreadId, owners: new Map(), turnIds: new Set() }; this.parentStates.set(parentThreadId, state); } state.requesterSessionKey ??= params.requesterSessionKey; @@ -428,6 +434,7 @@ class Monitor { return; } current.turnId = turnId; + registeredState.turnIds.add(turnId); this.drainPendingDirectSpawnEvidence(registeredState, current, turnId); this.clearUnconsumablePendingDirectSpawnEvidence(); }, @@ -438,8 +445,16 @@ class Monitor { registered = false; const current = this.parentStates.get(parentThreadId); if (current === registeredState) { + const turnId = current.owners.get(owner)?.turnId; current.owners.delete(owner); + if (turnId) { + current.turnIds.delete(turnId); + } + if (current.owners.size === 0) { + current.turnIds.clear(); + } this.clearUnconsumablePendingDirectSpawnEvidence(); + this.deliverDetachedCompletions(current); this.pruneParentIfUnused(current); } }, @@ -500,6 +515,13 @@ class Monitor { const threadStatus = isJsonObject(params?.status) ? normalizeIdentifier(readString(params.status, "type")) : undefined; + const parent = threadId ? this.parentStates.get(threadId) : undefined; + if (parent && parent.owners.size > 0 && notification.method === "turn/started") { + const turnId = isJsonObject(params?.turn) ? readString(params.turn, "id") : undefined; + if (turnId) { + parent.turnIds.add(turnId); + } + } const tracksRecoveryRevision = Boolean(threadId && this.threadStatusRevisions.has(threadId)); if ( RECOVERY_REVISION_NOTIFICATION_METHODS.has(notification.method) && @@ -532,8 +554,12 @@ class Monitor { } const childState = threadId ? this.childStates.get(threadId) : undefined; if (notification.method === "turn/started" && childState) { + childState.nativeCompletionDelivered = false; this.resumeChild(childState); } + if (parent && parent.turnIds.has(readString(params, "turnId") ?? "")) { + this.recordNativeCompletionDelivery(parent, notification); + } if (childState && !childState.terminal) { this.emitChildTaskActivity(notification, childState); } @@ -1196,6 +1222,10 @@ class Monitor { this.unregisterChild(childState); return; } + if (childState.nativeCompletionDelivered) { + this.finishCompletionDelivery(state, childState); + return; + } childState.pendingCompletion = completion; state.taskRuntime?.setDetachedTaskDeliveryStatusByRunId({ runId: codexNativeSubagentRunId(completion.childThreadId), @@ -1213,6 +1243,11 @@ class Monitor { if (!completion || !state.requesterSessionKey || !state.taskRuntimeScope) { return; } + // Codex owns completion input from registration through reply finalization, + // including turn startup before its id is bound. Only wake detached parents. + if (state.owners.size > 0) { + return; + } if (childState.deliveringCompletion || childState.completionDeliveryTimer) { return; } @@ -1238,13 +1273,7 @@ class Monitor { return; } if (isDurableAgentHarnessCompletionDelivery(delivery)) { - childState.pendingCompletion = undefined; - childState.completionDeliveryAttempt = 0; - state.taskRuntime?.setDetachedTaskDeliveryStatusByRunId({ - runId: codexNativeSubagentRunId(completion.childThreadId), - deliveryStatus: "delivered", - }); - this.unregisterChild(childState); + this.finishCompletionDelivery(state, childState); return; } const error = delivery.error ?? "completion delivery did not produce a parent response"; @@ -1278,6 +1307,43 @@ class Monitor { } } + private recordNativeCompletionDelivery( + state: ParentState, + notification: CodexServerNotification, + ): void { + for (const agentPath of nativeSubagentNotifications.deliveredAgentPaths(notification)) { + const childThreadId = this.childThreadIdsByAgentPath.get( + buildParentAgentPathKey(state.parentThreadId, agentPath), + ); + const child = childThreadId ? this.childStates.get(childThreadId) : undefined; + if (!child || child.parentThreadId !== state.parentThreadId) { + continue; + } + child.nativeCompletionDelivered = true; + if (child.pendingCompletion && !child.deliveringCompletion) { + this.finishCompletionDelivery(state, child); + } + } + } + + private finishCompletionDelivery(state: ParentState, child: ChildState): void { + child.pendingCompletion = undefined; + child.completionDeliveryAttempt = 0; + state.taskRuntime?.setDetachedTaskDeliveryStatusByRunId({ + runId: codexNativeSubagentRunId(child.childThreadId), + deliveryStatus: "delivered", + }); + this.unregisterChild(child); + } + + private deliverDetachedCompletions(state: ParentState): void { + for (const child of this.childStates.values()) { + if (child.parentThreadId === state.parentThreadId && child.pendingCompletion) { + void this.deliverPendingCompletion(state, child); + } + } + } + private scheduleCompletionDeliveryRetry(childState: ChildState, error: string): void { if ( !childState.pendingCompletion || @@ -1359,6 +1425,7 @@ class Monitor { assistantMessagesByTurn: new Map(), recoveryAttempt: 0, terminal: false, + nativeCompletionDelivered: false, settledWithoutCompletion: false, completionDeliveryAttempt: 0, deliveringCompletion: false, @@ -1918,6 +1985,7 @@ class Monitor { state = { parentThreadId, owners: new Map(), + turnIds: new Set(), requesterSessionKey: candidate.requesterSessionKey, taskRuntimeScope: candidate.taskRuntimeScope, agentId: candidate.agentId, diff --git a/extensions/codex/src/app-server/native-subagent-notification.test.ts b/extensions/codex/src/app-server/native-subagent-notification.test.ts index bc7a192733ce..3ba6aeb9e218 100644 --- a/extensions/codex/src/app-server/native-subagent-notification.test.ts +++ b/extensions/codex/src/app-server/native-subagent-notification.test.ts @@ -36,6 +36,40 @@ function trustedInterAgentNotification(params: { } describe("Codex native subagent notifications", () => { + it("recognizes a native completion receipt without treating its payload as a status", () => { + expect( + codexNativeSubagentNotifications.deliveredAgentPaths({ + method: "rawResponseItem/completed", + params: { + threadId: "parent-thread", + turnId: "parent-turn", + item: { + type: "agent_message", + author: "/root/worker", + recipient: "/root", + content: [ + { + type: "input_text", + text: "Message Type: FINAL_ANSWER\nTask name: /root\nSender: /root/worker\nPayload:\nBuild result", + }, + ], + }, + }, + }), + ).toEqual(["/root/worker"]); + }); + + it("recognizes the earlier trusted inter-agent completion envelope as a delivery receipt", () => { + expect( + codexNativeSubagentNotifications.deliveredAgentPaths( + trustedInterAgentNotification({ + agentPath: "child-thread", + text: '{"agent_path":"child-thread","status":{"completed":"done"}}', + }), + ), + ).toEqual(["child-thread"]); + }); + it("parses completed child results from Codex notification XML", () => { expect( extractCodexNativeSubagentCompletionsFromText( diff --git a/extensions/codex/src/app-server/native-subagent-notification.ts b/extensions/codex/src/app-server/native-subagent-notification.ts index ef63ea332302..265a668fe8e1 100644 --- a/extensions/codex/src/app-server/native-subagent-notification.ts +++ b/extensions/codex/src/app-server/native-subagent-notification.ts @@ -78,8 +78,41 @@ function extractCodexNativeSubagentCompletionsFromText( export const codexNativeSubagentNotifications = { fromNotification: extractCodexNativeSubagentCompletions, fromText: extractCodexNativeSubagentCompletionsFromText, + deliveredAgentPaths: readDeliveredNativeCompletionPaths, }; +/** Reads native delivery receipts, leaving status and result ownership with the child lifecycle. */ +function readDeliveredNativeCompletionPaths(notification: CodexServerNotification): string[] { + if (notification.method !== "rawResponseItem/completed") { + return []; + } + const params = isJsonObject(notification.params) ? notification.params : undefined; + const item = isJsonObject(params?.item) ? params.item : undefined; + if (!item || readString(item, "type") !== "agent_message") { + return extractCodexNativeSubagentCompletions(notification).map( + (completion) => completion.agentPath, + ); + } + const author = readString(item, "author"); + const recipient = readString(item, "recipient"); + const content = item.content; + if (!author || !recipient || !Array.isArray(content) || content.length !== 1) { + return []; + } + const part = content[0]; + if (!isJsonObject(part) || readString(part, "type") !== "input_text") { + return []; + } + const text = readString(part, "text"); + // Codex's native completion envelope identifies both endpoints outside the + // payload. Ordinary messages and quoted completion text are not receipts. + return text?.startsWith( + `Message Type: FINAL_ANSWER\nTask name: ${recipient}\nSender: ${author}\nPayload:\n`, + ) + ? [author] + : []; +} + function parseCodexNativeSubagentNotificationBody( body: string, ): CodexNativeSubagentNotificationCompletion | undefined {