From 2f89de81655accf71f5fac5cdabcdf0c3488eab7 Mon Sep 17 00:00:00 2001 From: Vincent Koc <25068+vincentkoc@users.noreply.github.com> Date: Mon, 6 Jul 2026 11:57:40 -0700 Subject: [PATCH] fix(codex): preserve yielded native subagent delivery --- CHANGELOG.md | 1 + .../native-subagent-monitor.test.ts | 128 ++++++++++++++++++ .../src/app-server/native-subagent-monitor.ts | 96 ++++++++++++- .../codex/src/app-server/run-attempt.ts | 24 +++- 4 files changed, 245 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d2c2923fac0..d9257669f3e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ Docs: https://docs.openclaw.ai ### Fixes +- **Codex yielded native subagents:** keep the parent app-server subscription and shared client alive until yielded native subagent completion delivery settles, preventing lost wakeups and leaked one-shot cleanup. - **Discord streamed finals:** send completion replies as fresh messages so inactive channels become unread, while preserving targeted mentions without escalating `@everyone` or `@here`. (#99711, #99662) Thanks @davelutztx. - **Cron edit delivery:** preserve each job's implicit delivery mode when applying partial delivery updates, so disabling best-effort delivery no longer turns detached job announcements off. (#100846) Thanks @machine3at. - **Control UI session creation:** keep newly created sessions at the front of the stable sidebar order after selecting another session. Thanks @shakkernerd. 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 3e4dfa27769..82bfe741d27 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.test.ts +++ b/extensions/codex/src/app-server/native-subagent-monitor.test.ts @@ -793,6 +793,134 @@ describe("CodexNativeSubagentMonitor", () => { ); }); + it("runs deferred parent cleanup after native subagent delivery settles", async () => { + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client, runtime); + const cleanup = vi.fn(); + monitor.registerParent({ + parentThreadId: "parent-thread", + requesterSessionKey: "agent:main:main", + taskRuntimeScope: createTaskScope("agent:main:main"), + agentId: "main", + }); + + await notifyChildStarted(client); + monitor.deferUntilParentSettles("parent-thread", cleanup); + expect(cleanup).not.toHaveBeenCalled(); + + await client.notify( + nativeCompletionNotification({ + agentPath: "child-thread", + statusLabel: "completed", + result: "child final result", + }), + ); + + expect(cleanup).toHaveBeenCalledOnce(); + }); + + it("leaves immediate parent cleanup with the caller", () => { + const client = createClient(); + const monitor = new CodexNativeSubagentMonitor(client, createRuntime()); + const cleanup = vi.fn(); + monitor.registerParent({ + parentThreadId: "parent-thread", + requesterSessionKey: "agent:main:main", + taskRuntimeScope: createTaskScope("agent:main:main"), + agentId: "main", + }); + + expect(monitor.deferUntilParentSettles("parent-thread", cleanup)).toBe(false); + expect(cleanup).not.toHaveBeenCalled(); + }); + + it("runs deferred parent cleanup when an interrupted child has no completion to deliver", async () => { + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client, runtime); + const cleanup = vi.fn(); + monitor.registerParent({ + parentThreadId: "parent-thread", + requesterSessionKey: "agent:main:main", + taskRuntimeScope: createTaskScope("agent:main:main"), + agentId: "main", + }); + + await notifyChildStarted(client); + monitor.deferUntilParentSettles("parent-thread", cleanup); + await client.notify(childTurnCompletedNotification({ status: "interrupted" })); + + expect(cleanup).toHaveBeenCalledOnce(); + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + }); + + it("runs deferred parent cleanup when a child ends in a system error", async () => { + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client, runtime); + const cleanup = vi.fn(); + monitor.registerParent({ + parentThreadId: "parent-thread", + requesterSessionKey: "agent:main:main", + taskRuntimeScope: createTaskScope("agent:main:main"), + agentId: "main", + }); + + await notifyChildStarted(client); + monitor.deferUntilParentSettles("parent-thread", cleanup); + await client.notify({ + method: "thread/status/changed", + params: { + threadId: "child-thread", + status: { type: "systemError" }, + }, + }); + + expect(cleanup).toHaveBeenCalledOnce(); + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + }); + + it("waits again when an interrupted child starts another turn before cleanup is deferred", async () => { + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client, runtime); + const cleanup = vi.fn(); + monitor.registerParent({ + parentThreadId: "parent-thread", + requesterSessionKey: "agent:main:main", + taskRuntimeScope: createTaskScope("agent:main:main"), + agentId: "main", + }); + + await notifyChildStarted(client); + await client.notify(childTurnCompletedNotification({ status: "interrupted" })); + await client.notify({ + method: "turn/started", + params: { + threadId: "child-thread", + turn: { + id: "resumed-child-turn", + status: "inProgress", + items: [], + error: null, + }, + }, + }); + monitor.deferUntilParentSettles("parent-thread", cleanup); + expect(cleanup).not.toHaveBeenCalled(); + + await client.notify( + nativeCompletionNotification({ + agentPath: "child-thread", + statusLabel: "completed", + result: "resumed child final result", + }), + ); + + expect(cleanup).toHaveBeenCalledOnce(); + }); + it("reconciles transcript final text before delivering empty Codex completion notifications", async () => { const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-codex-subagent-")); const codexHome = path.join(tempDir, "codex-home"); diff --git a/extensions/codex/src/app-server/native-subagent-monitor.ts b/extensions/codex/src/app-server/native-subagent-monitor.ts index 0407a27ee0d..5d5efa5a03c 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.ts +++ b/extensions/codex/src/app-server/native-subagent-monitor.ts @@ -45,6 +45,7 @@ type ParentState = { agentId?: string; taskRuntime?: AgentHarnessTaskRuntime; mirror?: CodexNativeSubagentTaskMirror; + deferredSettlement?: () => Promise | void; deliveredCompletionKeys: Set; }; @@ -62,6 +63,7 @@ type ChildState = { completionDeliveryTimer?: ReturnType; deliveringCompletionKey?: string; noFinalCompletionFallbackTimer?: ReturnType; + settledWithoutCompletion: boolean; }; type ChildAssistantMessages = { @@ -111,7 +113,7 @@ export function registerCodexNativeSubagentMonitor(params: { agentId?: string; codexHome?: string; runtime?: NativeSubagentMonitorRuntime; -}): void { +}): CodexNativeSubagentMonitor { let monitor = monitors.get(params.client); if (!monitor) { monitor = new CodexNativeSubagentMonitor(params.client, params.runtime ?? defaultRuntime, { @@ -127,6 +129,7 @@ export function registerCodexNativeSubagentMonitor(params: { taskRuntimeScope: params.taskRuntimeScope, agentId: params.agentId, }); + return monitor; } /** Tracks native subagent thread notifications, transcript completions, and task delivery. */ @@ -168,6 +171,18 @@ export class CodexNativeSubagentMonitor { this.transcriptPathsByChildThreadId.clear(); } + deferUntilParentSettles(parentThreadId: string, callback: () => Promise | void): boolean { + const normalizedParentThreadId = parentThreadId.trim(); + const state = this.parentStates.get(normalizedParentThreadId); + if (!state || !this.hasUnsettledChildren(normalizedParentThreadId)) { + return false; + } + // A yielded one-shot turn must keep this monitor alive until its child + // result reaches the parent; cleanup ownership transfers back afterward. + state.deferredSettlement = callback; + return true; + } + configure(options: MonitorOptions): void { const codexHome = normalizeOptionalString(options.codexHome); if (codexHome) { @@ -222,11 +237,42 @@ export class CodexNativeSubagentMonitor { }); } } + this.markChildTurnStarted(notification); + await this.handleChildSystemError(notification); this.captureChildAssistantMessage(notification); await this.handleChildTurnCompletion(notification); await this.handleCompletionNotification(notification); } + private markChildTurnStarted(notification: CodexServerNotification): void { + if (notification.method !== "turn/started") { + return; + } + 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.settledWithoutCompletion = false; + } + } + + private async handleChildSystemError(notification: CodexServerNotification): Promise { + if (notification.method !== "thread/status/changed") { + return; + } + const params = isJsonObject(notification.params) ? notification.params : undefined; + const status = isJsonObject(params?.status) ? params.status : undefined; + if (readString(status, "type") !== "systemError") { + return; + } + const childThreadId = readString(params, "threadId")?.trim(); + const childState = childThreadId ? this.childStates.get(childThreadId) : undefined; + if (childState) { + childState.settledWithoutCompletion = true; + await this.flushDeferredParentSettlements(childState.parentThreadId); + } + } + private ensureParentTaskRuntime(state: ParentState): void { if (state.taskRuntime || !state.requesterSessionKey || !state.taskRuntimeScope) { return; @@ -434,6 +480,10 @@ export class CodexNativeSubagentMonitor { if (turnId) { childState.assistantMessagesByTurn.delete(turnId); } + // Codex keeps interrupted agents resumable but intentionally sends no + // parent completion, so one-shot cleanup may settle until another turn starts. + childState.settledWithoutCompletion = true; + await this.flushDeferredParentSettlements(childState.parentThreadId); return; } if (childState && turn) { @@ -516,6 +566,7 @@ export class CodexNativeSubagentMonitor { } } if (!state.requesterSessionKey) { + await this.flushDeferredParentSettlements(state.parentThreadId); return; } const completionKey = buildCompletionDedupeKey(state.parentThreadId, completion); @@ -585,6 +636,7 @@ export class CodexNativeSubagentMonitor { }); } finally { childState.deliveringCompletionKey = undefined; + await this.flushDeferredParentSettlements(state.parentThreadId); } } @@ -635,6 +687,47 @@ export class CodexNativeSubagentMonitor { unrefTimer(childState.completionDeliveryTimer); } + private hasUnsettledChildren(parentThreadId: string): boolean { + for (const childState of this.childStates.values()) { + if ( + childState.parentThreadId === parentThreadId && + (childState.pendingCompletion !== undefined || + childState.deliveringCompletionKey !== undefined || + (!childState.transcriptTerminal && !childState.settledWithoutCompletion)) + ) { + return true; + } + } + return false; + } + + private async flushDeferredParentSettlements(parentThreadId: string): Promise { + if (this.hasUnsettledChildren(parentThreadId)) { + return; + } + const state = this.parentStates.get(parentThreadId); + const callback = state?.deferredSettlement; + if (!state || !callback) { + return; + } + state.deferredSettlement = undefined; + await this.runDeferredParentSettlement(parentThreadId, callback); + } + + private async runDeferredParentSettlement( + parentThreadId: string, + callback: () => Promise | void, + ): Promise { + try { + await callback(); + } catch (error) { + embeddedAgentLog.warn("Failed to finish deferred Codex app-server cleanup", { + parentThreadId, + error: formatErrorMessage(error), + }); + } + } + private finalizeCompletionTask(completion: CodexNativeSubagentCompletion, eventAt: number): void { const taskRuntime = this.getTaskRuntimeForChild(completion.childThreadId); if (!taskRuntime) { @@ -701,6 +794,7 @@ export class CodexNativeSubagentMonitor { transcriptPollAttempt: 0, transcriptTerminal: false, completionDeliveryAttempt: 0, + settledWithoutCompletion: false, }; this.childStates.set(normalizedChildThreadId, childState); } diff --git a/extensions/codex/src/app-server/run-attempt.ts b/extensions/codex/src/app-server/run-attempt.ts index 14b2f074fab..68ea3d5c42b 100644 --- a/extensions/codex/src/app-server/run-attempt.ts +++ b/extensions/codex/src/app-server/run-attempt.ts @@ -1432,6 +1432,9 @@ export async function runCodexAppServerAttempt( trajectoryEndRecorded = true; }; let nativeHookRelay: NativeHookRelayRegistrationHandle | undefined; + const nativeSubagentMonitorRef: { + current?: ReturnType; + } = {}; const pendingNativePreToolUseFailures: CodexNativePreToolUseFailure[] = []; const projectorRef: { current?: CodexAppServerEventProjector } = {}; let nativePreToolUseFailureFallbackActive = false; @@ -2360,7 +2363,7 @@ export async function runCodexAppServerAttempt( appServer.start.transport === "stdio" ? (appServer.start.env?.CODEX_HOME ?? resolveCodexAppServerHomeDir(agentDir)) : undefined; - registerCodexNativeSubagentMonitor({ + nativeSubagentMonitorRef.current = registerCodexNativeSubagentMonitor({ client, parentThreadId: thread.threadId, requesterSessionKey: params.sessionKey, @@ -3596,7 +3599,20 @@ export async function runCodexAppServerAttempt( if (!timedOut && !runAbortController.signal.aborted) { await steeringQueueRef.current?.flushPending(); } - if (!timedOut) { + const yieldedOneShotCleanupDeferred = + !timedOut && + params.cleanupBundleMcpOnRunEnd === true && + yieldDetected && + nativeSubagentMonitorRef.current?.deferUntilParentSettles(thread.threadId, async () => { + // Keep the parent subscription alive until native child delivery; + // unsubscribing first drops the completion signal that settles cleanup. + await unsubscribeCodexThreadBestEffort(client, { + threadId: thread.threadId, + timeoutMs: CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, + }); + await releaseSharedClientLeaseAndRetireOneShotClient(); + }); + if (!timedOut && !yieldedOneShotCleanupDeferred) { await unsubscribeCodexThreadBestEffort(client, { threadId: thread.threadId, timeoutMs: CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, @@ -3607,7 +3623,9 @@ export async function runCodexAppServerAttempt( notificationCleanup(); requestCleanup(); closeCleanup?.(); - await releaseSharedClientLeaseAndRetireOneShotClient(); + if (!yieldedOneShotCleanupDeferred) { + await releaseSharedClientLeaseAndRetireOneShotClient(); + } if (nativeHookRelay) { if (shouldDelayNativeHookRelayUnregister) { // Codex hook subprocesses can outlive a completed app-server turn by a