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 024ac3952f95..ce7da2a4b368 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.test.ts +++ b/extensions/codex/src/app-server/native-subagent-monitor.test.ts @@ -11,7 +11,7 @@ import { CodexNativeSubagentMonitor, registerCodexNativeSubagentMonitor, } from "./native-subagent-monitor.js"; -import type { CodexServerNotification } from "./protocol.js"; +import type { CodexServerNotification, JsonValue } from "./protocol.js"; function createClient() { const handlers = new Set<(notification: CodexServerNotification) => Promise | void>(); @@ -158,6 +158,27 @@ function nativeCompletionNotification(params: { }; } +function childTurnCompletedNotification(params: { + status: "completed" | "failed" | "interrupted"; + error?: string; + turnId?: string; + items?: JsonValue[]; +}): CodexServerNotification { + const turnId = params.turnId ?? "child-turn"; + return { + method: "turn/completed", + params: { + threadId: "child-thread", + turn: { + id: turnId, + status: params.status, + ...(params.items ? { items: params.items } : {}), + ...(params.error ? { error: { message: params.error } } : {}), + }, + }, + }; +} + describe("CodexNativeSubagentMonitor", () => { it("keeps native subagent task mirroring alive on the shared client", async () => { const client = createClient(); @@ -314,12 +335,10 @@ describe("CodexNativeSubagentMonitor", () => { ); }); - it("delivers child agent-message completion when a native subagent becomes idle", async () => { + it("delivers a completed child turn with its final agent message", async () => { const client = createClient(); const runtime = createRuntime(); - const monitor = new CodexNativeSubagentMonitor(client, runtime, { - codexHome: "/tmp/codex-home", - }); + const monitor = new CodexNativeSubagentMonitor(client, runtime); monitor.registerParent({ parentThreadId: "parent-thread", requesterSessionKey: "agent:main:discord:channel:C123", @@ -329,15 +348,21 @@ describe("CodexNativeSubagentMonitor", () => { await notifyChildStarted(client); await client.notify({ - method: "item/completed", + method: "item/agentMessage/delta", params: { threadId: "child-thread", - item: { - type: "agentMessage", - id: "msg-child-final", - phase: "final_answer", - text: "child final result", - }, + turnId: "child-turn", + itemId: "msg-child-final", + delta: "child ", + }, + }); + await client.notify({ + method: "item/agentMessage/delta", + params: { + threadId: "child-thread", + turnId: "child-turn", + itemId: "msg-child-final", + delta: "final result", }, }); @@ -351,6 +376,9 @@ describe("CodexNativeSubagentMonitor", () => { }, }); + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + await client.notify(childTurnCompletedNotification({ status: "completed" })); + expect(runtime.finalizeTaskRunByRunId).toHaveBeenCalledWith( expect.objectContaining({ runId: "codex-thread:child-thread", @@ -362,7 +390,7 @@ describe("CodexNativeSubagentMonitor", () => { expect.objectContaining({ childSessionId: "child-thread", status: "succeeded", - statusLabel: "agent_message", + statusLabel: "turn_completed", result: "child final result", }), ); @@ -370,12 +398,10 @@ describe("CodexNativeSubagentMonitor", () => { client.close(); }); - it("does not deliver commentary-only child messages as native subagent completion", async () => { + it("does not complete commentary-only child messages before a terminal turn", async () => { const client = createClient(); const runtime = createRuntime(); - const monitor = new CodexNativeSubagentMonitor(client, runtime, { - codexHome: "/tmp/codex-home", - }); + const monitor = new CodexNativeSubagentMonitor(client, runtime); monitor.registerParent({ parentThreadId: "parent-thread", requesterSessionKey: "agent:main:discord:channel:C123", @@ -384,10 +410,20 @@ describe("CodexNativeSubagentMonitor", () => { }); await notifyChildStarted(client); + await client.notify({ + method: "item/agentMessage/delta", + params: { + threadId: "child-thread", + turnId: "child-turn", + itemId: "msg-child-commentary", + delta: "checking now", + }, + }); await client.notify({ method: "item/completed", params: { threadId: "child-thread", + turnId: "child-turn", item: { type: "agentMessage", id: "msg-child-commentary", @@ -407,6 +443,162 @@ describe("CodexNativeSubagentMonitor", () => { expect(runtime.finalizeTaskRunByRunId).not.toHaveBeenCalled(); expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + await client.notify(childTurnCompletedNotification({ status: "completed" })); + + expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith( + expect.objectContaining({ + childSessionId: "child-thread", + result: "Codex native subagent completed without a final assistant message.", + }), + ); + + client.close(); + }); + + it("delivers a completed child turn with its snapshot-only final message", async () => { + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client, runtime); + monitor.registerParent({ + parentThreadId: "parent-thread", + requesterSessionKey: "agent:main:discord:channel:C123", + taskRuntimeScope: createTaskScope(), + agentId: "main", + }); + + await notifyChildStarted(client); + await client.notify( + childTurnCompletedNotification({ + status: "completed", + items: [ + { + id: "msg-child-snapshot", + type: "agentMessage", + phase: "final_answer", + text: "snapshot final result", + }, + ], + }), + ); + + expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith( + expect.objectContaining({ + childSessionId: "child-thread", + result: "snapshot final result", + }), + ); + + client.close(); + }); + + it("reconciles transcript text for a completed child turn without a final message", async () => { + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-codex-subagent-")); + const codexHome = path.join(tempDir, "codex-home"); + const transcriptDir = path.join(codexHome, "sessions", "2026", "06", "09"); + await fs.mkdir(transcriptDir, { recursive: true }); + await fs.writeFile( + path.join(transcriptDir, "rollout-2026-06-09T10-11-12-child-thread.jsonl"), + [ + JSON.stringify({ + type: "session_meta", + payload: { + source: { + subagent: { + thread_spawn: { + parent_thread_id: "parent-thread", + depth: 1, + }, + }, + }, + }, + }), + JSON.stringify({ + timestamp: "2026-06-09T10:12:00.000Z", + type: "event_msg", + payload: { + type: "task_complete", + last_agent_message: "child turn transcript result", + completed_at: 1781009520, + }, + }), + "", + ].join("\n"), + ); + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client, runtime, { + codexHome, + transcriptPollDelaysMs: [60_000], + }); + monitor.registerParent({ + parentThreadId: "parent-thread", + requesterSessionKey: "agent:main:discord:channel:C123", + taskRuntimeScope: createTaskScope(), + agentId: "main", + }); + + await notifyChildStarted(client); + await client.notify(childTurnCompletedNotification({ status: "completed" })); + + expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith( + expect.objectContaining({ + childSessionId: "child-thread", + statusLabel: "task_complete", + result: "child turn transcript result", + }), + ); + + client.close(); + }); + + it("does not reuse an interrupted child turn's message after resuming", async () => { + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client, runtime); + monitor.registerParent({ + parentThreadId: "parent-thread", + requesterSessionKey: "agent:main:discord:channel:C123", + taskRuntimeScope: createTaskScope(), + agentId: "main", + }); + + await notifyChildStarted(client); + await client.notify({ + method: "item/completed", + params: { + threadId: "child-thread", + turnId: "child-turn", + item: { + type: "agentMessage", + id: "msg-child-partial", + text: "partial child result", + }, + }, + }); + await client.notify({ + method: "thread/status/changed", + params: { + threadId: "child-thread", + status: { type: "idle" }, + }, + }); + await client.notify(childTurnCompletedNotification({ status: "interrupted" })); + + expect(runtime.deliverAgentHarnessTaskCompletion).not.toHaveBeenCalled(); + + await client.notify( + childTurnCompletedNotification({ status: "completed", turnId: "resumed-child-turn" }), + ); + + expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledTimes(1); + expect(runtime.deliverAgentHarnessTaskCompletion).toHaveBeenCalledWith( + expect.objectContaining({ + childSessionId: "child-thread", + status: "succeeded", + result: "Codex native subagent completed without a final assistant message.", + }), + ); + client.close(); }); diff --git a/extensions/codex/src/app-server/native-subagent-monitor.ts b/extensions/codex/src/app-server/native-subagent-monitor.ts index a5bfdd6732b0..2551f1f53f57 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.ts +++ b/extensions/codex/src/app-server/native-subagent-monitor.ts @@ -51,14 +51,11 @@ type ParentState = { type ChildState = { childThreadId: string; parentThreadId: string; + assistantMessagesByTurn: Map; transcriptPath?: string; transcriptPollAttempt: number; transcriptPollTimer?: ReturnType; transcriptTerminal: boolean; - idle: boolean; - lastAgentMessage?: string; - lastAgentMessageAt?: number; - agentMessageCompletionDelivered: boolean; pendingCompletion?: CodexNativeSubagentCompletion; pendingCompletionEventAt?: number; completionDeliveryAttempt: number; @@ -67,6 +64,12 @@ type ChildState = { noFinalCompletionFallbackTimer?: ReturnType; }; +type ChildAssistantMessages = { + texts: Map; + order: string[]; + commentaryIds: Set; +}; + type TranscriptCompletion = CodexNativeSubagentCompletion & { parentThreadId?: string; completedAt?: number; @@ -215,10 +218,9 @@ export class CodexNativeSubagentMonitor { }); } } - const childThreadId = this.recordChildAgentMessage(notification); - const idleChildThreadId = this.recordChildIdle(notification); + this.captureChildAssistantMessage(notification); + await this.handleChildTurnCompletion(notification); await this.handleCompletionNotification(notification); - await this.processChildAgentMessageCompletion(childThreadId ?? idleChildThreadId); } private ensureParentTaskRuntime(state: ParentState): void { @@ -299,7 +301,11 @@ export class CodexNativeSubagentMonitor { nativeCompletion.agentPath, ); const childState = childThreadId ? this.childStates.get(childThreadId) : undefined; - if (!childState || childState.parentThreadId !== state.parentThreadId) { + if ( + !childState || + childState.parentThreadId !== state.parentThreadId || + childState.transcriptTerminal + ) { embeddedAgentLog.warn( "Ignoring Codex native subagent completion for unknown child thread", { @@ -310,21 +316,148 @@ export class CodexNativeSubagentMonitor { continue; } const completion = toThreadCompletion(nativeCompletion, childState.childThreadId); - if (shouldWaitForTranscriptCompletion(completion, this.codexHome)) { - // Codex can notify `completed: null` before the child transcript exposes - // its final assistant message; poll briefly before delivering the no-final fallback. - const eventAt = Date.now(); - const reconciled = await this.reconcileChildTranscript(childState.childThreadId); - if (!reconciled) { - this.scheduleTranscriptPoll(childState); - this.scheduleNoFinalCompletionFallback(state, childState, completion, eventAt); - } - continue; - } - await this.processCompletion(state, completion); + await this.processChildCompletion(state, childState, completion); } } + private captureChildAssistantMessage(notification: CodexServerNotification): void { + 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.transcriptTerminal) { + return; + } + if (notification.method === "item/agentMessage/delta") { + const turnId = readString(params, "turnId"); + const itemId = readString(params, "itemId"); + const delta = readString(params, "delta"); + if (turnId && itemId && delta) { + this.recordChildAssistantMessage(childState, turnId, itemId, delta); + } + return; + } + if (notification.method !== "item/completed") { + return; + } + const turnId = readString(params, "turnId"); + const item = isJsonObject(params?.item) ? params.item : undefined; + this.captureChildAssistantMessageItem(childState, turnId, item); + } + + private captureChildAssistantMessageItem( + childState: ChildState, + turnId: string | undefined, + item: JsonObject | undefined, + ): void { + if (readString(item, "type") !== "agentMessage") { + return; + } + const itemId = readString(item, "id"); + if (!turnId || !itemId) { + return; + } + const assistantMessages = this.getChildAssistantMessages(childState, turnId); + const phase = readString(item, "phase"); + if (phase === "commentary") { + assistantMessages.commentaryIds.add(itemId); + } + const text = readString(item, "text"); + if (text) { + this.recordChildAssistantMessage(childState, turnId, itemId, text, { replace: true }); + } + } + + private captureChildTurnAssistantMessages(childState: ChildState, turn: JsonObject): void { + const turnId = readString(turn, "id"); + if (!turnId || !Array.isArray(turn.items)) { + return; + } + for (const item of turn.items) { + this.captureChildAssistantMessageItem( + childState, + turnId, + isJsonObject(item) ? item : undefined, + ); + } + } + + private recordChildAssistantMessage( + childState: ChildState, + turnId: string, + itemId: string, + text: string, + options: { replace?: boolean } = {}, + ): void { + const assistantMessages = this.getChildAssistantMessages(childState, turnId); + if (!assistantMessages.texts.has(itemId)) { + assistantMessages.order.push(itemId); + } + const existing = assistantMessages.texts.get(itemId) ?? ""; + assistantMessages.texts.set(itemId, options.replace ? text : `${existing}${text}`); + } + + private getChildAssistantMessages( + childState: ChildState, + turnId: string, + ): ChildAssistantMessages { + const existing = childState.assistantMessagesByTurn.get(turnId); + if (existing) { + return existing; + } + const assistantMessages: ChildAssistantMessages = { + texts: new Map(), + order: [], + commentaryIds: new Set(), + }; + childState.assistantMessagesByTurn.set(turnId, assistantMessages); + return assistantMessages; + } + + private async handleChildTurnCompletion(notification: CodexServerNotification): Promise { + if (notification.method !== "turn/completed") { + return; + } + const params = isJsonObject(notification.params) ? notification.params : undefined; + const childThreadId = readString(params, "threadId")?.trim(); + const childState = childThreadId ? this.childStates.get(childThreadId) : undefined; + const state = childState ? this.parentStates.get(childState.parentThreadId) : undefined; + const turn = isJsonObject(params?.turn) ? params.turn : undefined; + if (childState && turn && readString(turn, "status") === "interrupted") { + const turnId = readString(turn, "id"); + if (turnId) { + childState.assistantMessagesByTurn.delete(turnId); + } + return; + } + if (childState && turn) { + this.captureChildTurnAssistantMessages(childState, turn); + } + const completion = childState && turn ? toChildTurnCompletion(childState, turn) : undefined; + if (!state || !childState || childState.transcriptTerminal || !completion) { + return; + } + await this.processChildCompletion(state, childState, completion); + } + + private async processChildCompletion( + state: ParentState, + childState: ChildState, + completion: CodexNativeSubagentCompletion, + ): Promise { + if (shouldWaitForTranscriptCompletion(completion, this.codexHome)) { + // Codex can notify `completed: null` before the child transcript exposes + // its final assistant message; poll briefly before delivering the no-final fallback. + const eventAt = Date.now(); + const reconciled = await this.reconcileChildTranscript(childState.childThreadId); + if (!reconciled) { + this.scheduleTranscriptPoll(childState); + this.scheduleNoFinalCompletionFallback(state, childState, completion, eventAt); + } + return; + } + await this.processCompletion(state, completion); + } + async reconcileChildTranscript( childThreadId: string, options: { allowTreeScan?: boolean } = {}, @@ -557,10 +690,9 @@ export class CodexNativeSubagentMonitor { childState = { childThreadId: normalizedChildThreadId, parentThreadId: normalizedParentThreadId, + assistantMessagesByTurn: new Map(), transcriptPollAttempt: 0, transcriptTerminal: false, - idle: false, - agentMessageCompletionDelivered: false, completionDeliveryAttempt: 0, }; this.childStates.set(normalizedChildThreadId, childState); @@ -570,83 +702,6 @@ export class CodexNativeSubagentMonitor { } } - private recordChildAgentMessage(notification: CodexServerNotification): string | undefined { - if (notification.method !== "item/completed") { - return undefined; - } - const params = isJsonObject(notification.params) ? notification.params : undefined; - const item = isJsonObject(params?.item) ? params.item : undefined; - if (!params || !item || readString(item, "type") !== "agentMessage") { - return undefined; - } - const childThreadId = readString(params, "threadId")?.trim(); - const childState = childThreadId ? this.childStates.get(childThreadId) : undefined; - if (!childState || childState.transcriptTerminal) { - return undefined; - } - // Codex app-server can report the child final answer as the child thread's - // own agentMessage without also emitting a parent subagent notification. - // Pair it with idle below so commentary does not become a false terminal. - const phase = readString(item, "phase"); - if (phase === "commentary") { - return undefined; - } - const text = readString(item, "text")?.trim(); - if (!text) { - return undefined; - } - childState.lastAgentMessage = text; - childState.lastAgentMessageAt = Date.now(); - return childState.childThreadId; - } - - private recordChildIdle(notification: CodexServerNotification): string | undefined { - if (notification.method !== "thread/status/changed") { - return undefined; - } - const params = isJsonObject(notification.params) ? notification.params : undefined; - if (!params || !isJsonObject(params.status) || readString(params.status, "type") !== "idle") { - return undefined; - } - const childThreadId = readString(params, "threadId")?.trim(); - const childState = childThreadId ? this.childStates.get(childThreadId) : undefined; - if (!childState || childState.transcriptTerminal) { - return undefined; - } - childState.idle = true; - return childState.childThreadId; - } - - private async processChildAgentMessageCompletion( - childThreadId: string | undefined, - ): Promise { - const childState = childThreadId ? this.childStates.get(childThreadId) : undefined; - if ( - !childState || - !childState.idle || - childState.transcriptTerminal || - childState.agentMessageCompletionDelivered || - !childState.lastAgentMessage - ) { - return; - } - const state = this.parentStates.get(childState.parentThreadId); - if (!state) { - return; - } - childState.agentMessageCompletionDelivered = true; - await this.processCompletion( - state, - { - childThreadId: childState.childThreadId, - status: "succeeded", - statusLabel: "agent_message", - result: childState.lastAgentMessage, - }, - childState.lastAgentMessageAt, - ); - } - private ensureChildState(parentThreadId: string, childThreadId: string): ChildState { this.registerChildThread(parentThreadId, childThreadId); return this.childStates.get(childThreadId.trim())!; @@ -970,6 +1025,59 @@ function buildCompletionDedupeKey( return `${parentThreadId}:${completion.childThreadId}:${completion.status}:${hash}`; } +function toChildTurnCompletion( + childState: ChildState, + turn: JsonObject, +): CodexNativeSubagentCompletion | undefined { + const status = readString(turn, "status"); + if (status === "completed") { + const turnId = readString(turn, "id"); + const result = turnId ? lastChildAssistantMessage(childState, turnId) : undefined; + return { + childThreadId: childState.childThreadId, + status: "succeeded", + statusLabel: result ? "turn_completed" : "completed_without_final_message", + result: result ?? "Codex native subagent completed without a final assistant message.", + }; + } + if (status === "failed") { + return { + childThreadId: childState.childThreadId, + status: "failed", + statusLabel: "turn_failed", + result: readTurnErrorMessage(turn) ?? "Codex native subagent failed.", + }; + } + return undefined; +} + +function lastChildAssistantMessage(childState: ChildState, turnId: string): string | undefined { + const assistantMessages = childState.assistantMessagesByTurn.get(turnId); + if (!assistantMessages) { + return undefined; + } + for (let index = assistantMessages.order.length - 1; index >= 0; index -= 1) { + const itemId = assistantMessages.order[index]; + if (!assistantMessages.commentaryIds.has(itemId)) { + const text = normalizeOptionalString(assistantMessages.texts.get(itemId)); + if (text) { + return text; + } + } + } + return undefined; +} + +function readTurnErrorMessage(turn: JsonObject): string | undefined { + const error = isJsonObject(turn.error) ? turn.error : undefined; + return ( + normalizeOptionalString(readString(error, "message")) ?? + normalizeOptionalString( + isJsonObject(error?.codexErrorInfo) ? readString(error.codexErrorInfo, "message") : undefined, + ) + ); +} + function buildParentAgentPathKey(parentThreadId: string, agentPath: string): string { return `${parentThreadId}\0${agentPath}`; }