fix(codex): preserve yielded native subagent delivery

vincentkoc · vincentkoc · commit 2f89de81655a · 2026-07-06T11:58:00.000-07:00
diff --git 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
@@ -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
@@ -45,6 +45,7 @@ type ParentState = {
   agentId?: string;
   taskRuntime?: AgentHarnessTaskRuntime;
   mirror?: CodexNativeSubagentTaskMirror;
+  deferredSettlement?: () => Promise<void> | void;
   deliveredCompletionKeys: Set<string>;
 };
 
@@ -62,6 +63,7 @@ type ChildState = {
   completionDeliveryTimer?: ReturnType<typeof setTimeout>;
   deliveringCompletionKey?: string;
   noFinalCompletionFallbackTimer?: ReturnType<typeof setTimeout>;
+  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> | 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<void> {
+    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<void> {
+    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> | void,
+  ): Promise<void> {
+    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
@@ -1432,6 +1432,9 @@ export async function runCodexAppServerAttempt(
     trajectoryEndRecorded = true;
   };
   let nativeHookRelay: NativeHookRelayRegistrationHandle | undefined;
+  const nativeSubagentMonitorRef: {
+    current?: ReturnType<typeof registerCodexNativeSubagentMonitor>;
+  } = {};
   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
