mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-30 05:02:58 +00:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c299598aea | ||
|
|
f110243b4c | ||
|
|
1bb0798857 | ||
|
|
198d8a5c68 |
@@ -58,6 +58,7 @@ import {
|
||||
mapCodexToolCallEnvelope,
|
||||
mapCodexToolCallFromThreadItem,
|
||||
normalizeCommandExecutionCommand,
|
||||
normalizeCollabAgentChildStatus,
|
||||
splitCodexMcpToolResultImages,
|
||||
} from "./codex/tool-call-mapper.js";
|
||||
import {
|
||||
@@ -1860,6 +1861,65 @@ function readCodexSubAgentActivity(item: unknown): CodexSubAgentActivity | null
|
||||
};
|
||||
}
|
||||
|
||||
interface CodexCollabAgentUpdate {
|
||||
threadId: string;
|
||||
status: ToolCallTimelineItem["status"];
|
||||
preservesTerminal: boolean;
|
||||
reopen: boolean;
|
||||
}
|
||||
|
||||
interface CodexCollabAgentControl {
|
||||
childThreadIds: string[];
|
||||
updates: CodexCollabAgentUpdate[];
|
||||
}
|
||||
|
||||
function readCodexCollabAgentControl(item: unknown): CodexCollabAgentControl | null {
|
||||
const record = toObjectRecord(item);
|
||||
if (
|
||||
!record ||
|
||||
normalizeCodexThreadItemType(typeof record.type === "string" ? record.type : undefined) !==
|
||||
"collabAgentToolCall" ||
|
||||
record.tool === "spawnAgent"
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
const updates = Object.entries(toObjectRecord(record.agentsStates) ?? {}).flatMap(
|
||||
([threadId, value]): CodexCollabAgentUpdate[] => {
|
||||
const state = toObjectRecord(value);
|
||||
if (typeof state?.status !== "string") return [];
|
||||
const status = normalizeCollabAgentChildStatus(state.status);
|
||||
return [
|
||||
{
|
||||
threadId,
|
||||
status: record.tool === "closeAgent" && status === "running" ? "canceled" : status,
|
||||
preservesTerminal: state.status === "shutdown" || record.tool === "closeAgent",
|
||||
reopen: record.tool === "sendInput" && record.status === "completed",
|
||||
},
|
||||
];
|
||||
},
|
||||
);
|
||||
const receivers = Array.isArray(record.receiverThreadIds)
|
||||
? record.receiverThreadIds.filter((id): id is string => typeof id === "string")
|
||||
: [];
|
||||
return {
|
||||
childThreadIds: [...new Set([...receivers, ...updates.map((update) => update.threadId)])],
|
||||
updates,
|
||||
};
|
||||
}
|
||||
|
||||
function resolveCodexCollabAgentUpdate(
|
||||
previous: ToolCallTimelineItem["status"],
|
||||
update: CodexCollabAgentUpdate,
|
||||
): ToolCallTimelineItem["status"] {
|
||||
if (
|
||||
isTerminalSubAgentStatus(previous) &&
|
||||
(update.preservesTerminal || (update.status === "running" && !update.reopen))
|
||||
) {
|
||||
return previous;
|
||||
}
|
||||
return update.status;
|
||||
}
|
||||
|
||||
function shouldIgnoreMirroredLifecycleItem(source: "item" | "codex_event", item: unknown): boolean {
|
||||
return source === "codex_event" && !readCodexSubAgentActivity(item);
|
||||
}
|
||||
@@ -2093,6 +2153,30 @@ async function requestCodexThreadHistory(
|
||||
return CodexThreadReadResponseSchema.parse(response);
|
||||
}
|
||||
|
||||
function updateHistoricalCollabAgentCalls(
|
||||
timeline: PersistedTimelineEntry[],
|
||||
timelineIndexByThreadId: Map<string, number>,
|
||||
item: unknown,
|
||||
): boolean {
|
||||
const control = readCodexCollabAgentControl(item);
|
||||
if (!control) return false;
|
||||
for (const update of control.updates) {
|
||||
const existingIndex = timelineIndexByThreadId.get(update.threadId);
|
||||
const existing = existingIndex === undefined ? undefined : timeline[existingIndex];
|
||||
if (existingIndex === undefined || existing?.item.type !== "tool_call") continue;
|
||||
const status = resolveCodexCollabAgentUpdate(existing.item.status, update);
|
||||
const toolCall: ToolCallTimelineItem =
|
||||
status === "failed"
|
||||
? { ...existing.item, status, error: { message: "Sub-agent failed" } }
|
||||
: { ...existing.item, status, error: null };
|
||||
timeline[existingIndex] = { ...existing, item: toolCall };
|
||||
}
|
||||
return (
|
||||
control.childThreadIds.length > 0 &&
|
||||
control.childThreadIds.every((id) => timelineIndexByThreadId.has(id))
|
||||
);
|
||||
}
|
||||
|
||||
async function loadCodexThreadHistoryTimeline(params: {
|
||||
threadId: string;
|
||||
cwd: string | null;
|
||||
@@ -2103,6 +2187,9 @@ async function loadCodexThreadHistoryTimeline(params: {
|
||||
const subAgentTimelineIndexByThreadId = new Map<string, number>();
|
||||
for (const turn of response.thread.turns) {
|
||||
for (const item of turn.items) {
|
||||
if (updateHistoricalCollabAgentCalls(timeline, subAgentTimelineIndexByThreadId, item)) {
|
||||
continue;
|
||||
}
|
||||
const historicalSubAgentActivity = readCodexSubAgentActivity(item);
|
||||
if (historicalSubAgentActivity) {
|
||||
const existingIndex = subAgentTimelineIndexByThreadId.get(
|
||||
@@ -5761,6 +5848,23 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
}
|
||||
|
||||
private handleRegisteredSubAgentActivity(rawItem: { [key: string]: unknown }): boolean {
|
||||
const control = readCodexCollabAgentControl(rawItem);
|
||||
if (control) {
|
||||
for (const update of control.updates) {
|
||||
const callId = this.subAgentCallIdByChildThreadId.get(update.threadId);
|
||||
if (!callId) continue;
|
||||
const state = this.subAgentCallsByCallId.get(callId);
|
||||
if (!state) continue;
|
||||
const status = resolveCodexCollabAgentUpdate(state.toolCall.status, update);
|
||||
if (status !== state.toolCall.status) {
|
||||
this.emitSubAgentActivityUpdate(callId, status, { reopen: update.reopen });
|
||||
}
|
||||
}
|
||||
return (
|
||||
control.childThreadIds.length > 0 &&
|
||||
control.childThreadIds.every((id) => this.subAgentCallIdByChildThreadId.has(id))
|
||||
);
|
||||
}
|
||||
const activity = readCodexSubAgentActivity(rawItem);
|
||||
if (!activity) {
|
||||
return false;
|
||||
|
||||
@@ -0,0 +1,200 @@
|
||||
import { describe, expect, test, vi } from "vitest";
|
||||
import { CodexAppServerAgentSession } from "../codex-app-server-agent.js";
|
||||
import { createTestLogger } from "../../../../test-utils/test-logger.js";
|
||||
import { createFakeCodexAppServer } from "./test-utils/fake-app-server.js";
|
||||
|
||||
function collab(tool: string, states: Record<string, string>, id = tool) {
|
||||
return {
|
||||
type: "collabAgentToolCall",
|
||||
id,
|
||||
tool,
|
||||
status: "completed",
|
||||
prompt: tool === "spawnAgent" ? "Reply with a short result." : null,
|
||||
receiverThreadIds: Object.keys(states),
|
||||
agentsStates: Object.fromEntries(
|
||||
Object.entries(states).map(([threadId, status]) => [threadId, { status }]),
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
async function replay(items: unknown[]) {
|
||||
const appServer = createFakeCodexAppServer({
|
||||
"thread/read": (params) => ({
|
||||
thread: {
|
||||
turns: [
|
||||
{
|
||||
items: (params as { threadId: string }).threadId === "parent" ? items : [],
|
||||
},
|
||||
],
|
||||
},
|
||||
}),
|
||||
});
|
||||
const session = new CodexAppServerAgentSession(
|
||||
{ provider: "codex", cwd: process.cwd(), modeId: "full-access" },
|
||||
{ sessionId: "parent" },
|
||||
createTestLogger(),
|
||||
async () => appServer.child,
|
||||
{},
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
undefined,
|
||||
"history",
|
||||
);
|
||||
try {
|
||||
await session.connect();
|
||||
return await Array.fromAsync(session.streamHistory());
|
||||
} finally {
|
||||
await session.close();
|
||||
appServer.assertNoErrors();
|
||||
}
|
||||
}
|
||||
|
||||
describe("Codex collaboration history", () => {
|
||||
test("restores completed children from nullable wait and close results", async () => {
|
||||
const history = await replay([
|
||||
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
|
||||
collab("spawnAgent", { second: "pendingInit" }, "spawn-second"),
|
||||
collab("wait", { first: "completed" }, "wait-first"),
|
||||
collab("wait", { second: "completed" }, "wait-second"),
|
||||
collab("closeAgent", { first: "completed" }, "close-first"),
|
||||
collab("closeAgent", { second: "completed" }, "close-second"),
|
||||
]);
|
||||
expect(
|
||||
history.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert"
|
||||
? [{ id: event.event.id, status: event.event.status }]
|
||||
: [],
|
||||
),
|
||||
).toEqual([
|
||||
{ id: "first", status: "completed" },
|
||||
{ id: "second", status: "completed" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("stops counting a child closed while running", async () => {
|
||||
const history = await replay([
|
||||
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
|
||||
collab("closeAgent", { first: "running" }),
|
||||
]);
|
||||
expect(
|
||||
history.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert"
|
||||
? [{ id: event.event.id, status: event.event.status }]
|
||||
: [],
|
||||
),
|
||||
).toEqual([{ id: "first", status: "canceled" }]);
|
||||
});
|
||||
|
||||
test("stops counting an errored child whose close record reports failed", async () => {
|
||||
const history = await replay([
|
||||
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
|
||||
{ ...collab("closeAgent", { first: "errored" }), status: "failed" },
|
||||
]);
|
||||
expect(
|
||||
history.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert"
|
||||
? [{ id: event.event.id, status: event.event.status }]
|
||||
: [],
|
||||
),
|
||||
).toEqual([{ id: "first", status: "canceled" }]);
|
||||
});
|
||||
|
||||
test("preserves completion when a close carries an older running snapshot", async () => {
|
||||
const history = await replay([
|
||||
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
|
||||
collab("wait", { first: "completed" }),
|
||||
collab("closeAgent", { first: "running" }),
|
||||
]);
|
||||
expect(
|
||||
history.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert"
|
||||
? [{ id: event.event.id, status: event.event.status }]
|
||||
: [],
|
||||
),
|
||||
).toEqual([{ id: "first", status: "completed" }]);
|
||||
});
|
||||
|
||||
test.each([
|
||||
["shutdown", "canceled"],
|
||||
["notFound", "failed"],
|
||||
])("restores a %s child as %s and preserves its running sibling", async (status, expected) => {
|
||||
const history = await replay([
|
||||
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
|
||||
collab("spawnAgent", { second: "pendingInit" }, "spawn-second"),
|
||||
collab("closeAgent", { first: status }),
|
||||
]);
|
||||
expect(
|
||||
history.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert"
|
||||
? [{ id: event.event.id, status: event.event.status }]
|
||||
: [],
|
||||
),
|
||||
).toEqual([
|
||||
{ id: "first", status: expected },
|
||||
{ id: "second", status: "running" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("keeps completed live children settled until an explicit follow-up", async () => {
|
||||
const appServer = createFakeCodexAppServer();
|
||||
const session = new CodexAppServerAgentSession(
|
||||
{ provider: "codex", cwd: process.cwd(), modeId: "full-access" },
|
||||
null,
|
||||
createTestLogger(),
|
||||
async () => appServer.child,
|
||||
);
|
||||
const statuses = new Map<string, string>();
|
||||
const timelineCallIds = new Set<string>();
|
||||
session.subscribe((event) => {
|
||||
if (event.type === "provider_subagent" && event.event.type === "upsert") {
|
||||
statuses.set(event.event.id, event.event.status);
|
||||
}
|
||||
if (event.type === "timeline" && event.item.type === "tool_call") {
|
||||
timelineCallIds.add(event.item.callId);
|
||||
}
|
||||
});
|
||||
function complete(tool: string, states: Record<string, string>, id = tool) {
|
||||
if (tool !== "spawnAgent") {
|
||||
appServer.child.stdout.write(
|
||||
`${JSON.stringify({
|
||||
method: "item/started",
|
||||
params: {
|
||||
threadId: "thread-1",
|
||||
item: { ...collab(tool, states, id), status: "inProgress", agentsStates: {} },
|
||||
},
|
||||
})}\n`,
|
||||
);
|
||||
}
|
||||
appServer.child.stdout.write(
|
||||
`${JSON.stringify({
|
||||
method: "item/completed",
|
||||
params: { threadId: "thread-1", item: collab(tool, states, id) },
|
||||
})}\n`,
|
||||
);
|
||||
}
|
||||
try {
|
||||
const result = session.run("Delegate the work.");
|
||||
await appServer.waitForTurnStart();
|
||||
complete("spawnAgent", { first: "pendingInit" }, "spawn-first");
|
||||
complete("spawnAgent", { second: "pendingInit" }, "spawn-second");
|
||||
complete("wait", { first: "completed", second: "running" });
|
||||
await vi.waitFor(() => {
|
||||
expect(statuses.get("first")).toBe("completed");
|
||||
expect(statuses.get("second")).toBe("running");
|
||||
});
|
||||
complete("closeAgent", { first: "shutdown" });
|
||||
complete("wait", { first: "running" }, "stale-wait");
|
||||
appServer.says({ threadId: "thread-1", text: "Still waiting." });
|
||||
appServer.completeTurn();
|
||||
await result;
|
||||
expect(statuses.get("first")).toBe("completed");
|
||||
complete("sendInput", { first: "running" });
|
||||
await vi.waitFor(() => expect(statuses.get("first")).toBe("running"));
|
||||
expect([...timelineCallIds]).toEqual(["spawn-first", "spawn-second"]);
|
||||
} finally {
|
||||
await session.close();
|
||||
appServer.assertNoErrors();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -213,6 +213,20 @@ describe("codex tool-call mapper", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("accepts a nullable prompt in completed collaboration control items", () => {
|
||||
expect(
|
||||
mapCodexToolCallFromThreadItem({
|
||||
type: "collabAgentToolCall",
|
||||
id: "wait-child",
|
||||
tool: "wait",
|
||||
prompt: null,
|
||||
status: "completed",
|
||||
receiverThreadIds: ["child"],
|
||||
agentsStates: { child: { status: "completed" } },
|
||||
}),
|
||||
).toMatchObject({ callId: "wait-child", status: "completed" });
|
||||
});
|
||||
|
||||
it("maps collabAgentToolCall into canonical sub-agent detail", () => {
|
||||
const item = mapCodexToolCallFromThreadItem({
|
||||
type: "collabAgentToolCall",
|
||||
|
||||
@@ -186,7 +186,7 @@ const CodexCollabAgentToolCallItemSchema = z
|
||||
id: z.string().min(1),
|
||||
status: z.string().optional(),
|
||||
error: z.unknown().optional(),
|
||||
prompt: z.string().optional(),
|
||||
prompt: z.string().nullish(),
|
||||
tool: z.string().optional(),
|
||||
receiverThreadIds: z.array(z.string()).optional(),
|
||||
agentsStates: z.record(z.string(), z.unknown()).optional(),
|
||||
@@ -545,7 +545,7 @@ function readStatus(value: unknown): string | undefined {
|
||||
return typeof value.status === "string" ? value.status : undefined;
|
||||
}
|
||||
|
||||
function normalizeCollabAgentChildStatus(status: string): ToolCallTimelineItem["status"] {
|
||||
export function normalizeCollabAgentChildStatus(status: string): ToolCallTimelineItem["status"] {
|
||||
const normalized = status.trim().toLowerCase();
|
||||
switch (normalized) {
|
||||
case "error":
|
||||
|
||||
Reference in New Issue
Block a user