mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-26 20:35:39 -06:00
Fix missing replies after native Codex subagents finish (#130311)
* Fix missing replies after native Codex subagents finish * Fence subagent completion during parent startup
This commit is contained in:
pash-openai
@@ -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<typeof createClient>,
|
||||
monitor: CodexNativeSubagentMonitorInstance,
|
||||
): Promise<void> {
|
||||
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);
|
||||
|
||||
@@ -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<symbol, ParentOwner>;
|
||||
// turn/started can precede bindTurn; retain receipt ownership until the
|
||||
// foreground run has finalized its reply and releases this registration.
|
||||
turnIds: Set<string>;
|
||||
requesterSessionKey?: string;
|
||||
taskRuntimeScope?: AgentHarnessTaskRuntimeScope;
|
||||
agentId?: string;
|
||||
@@ -93,6 +96,7 @@ type ChildState = {
|
||||
terminal: boolean;
|
||||
fallbackCompletion?: RecoveredCompletion;
|
||||
pendingCompletion?: CodexNativeSubagentCompletion;
|
||||
nativeCompletionDelivered: boolean;
|
||||
completionDeliveryAttempt: number;
|
||||
completionDeliveryTimer?: ReturnType<typeof setTimeout>;
|
||||
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<string, ChildAssistantMessages>(),
|
||||
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,
|
||||
|
||||
@@ -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: '<subagent_notification>{"agent_path":"child-thread","status":{"completed":"done"}}</subagent_notification>',
|
||||
}),
|
||||
),
|
||||
).toEqual(["child-thread"]);
|
||||
});
|
||||
|
||||
it("parses completed child results from Codex notification XML", () => {
|
||||
expect(
|
||||
extractCodexNativeSubagentCompletionsFromText(
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user