Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions packages/agent/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,8 @@ start()

The two tapping layers are distinct. The inner tap (from `createAcpConnection`) persists to logs. The outer tap (in `AgentServer`) broadcasts to SSE. This means log persistence works for both cloud and local, while SSE broadcast is cloud-only.

Adapters must remove provider notifications that do not belong to the active parent session before writing ACP updates. Persisted ACP updates do not retain enough provider-specific identity to separate a subagent transcript or completion from its parent safely during cloud replay.

### HTTP endpoints

| Method | Path | Auth | Description |
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,167 @@ describe("CodexAppServerAgent", () => {
});
});

it("isolates subagent output, usage, compaction, and completion", async () => {
const stub = makeStubRpc({
initialize: {},
"thread/start": { thread: { id: "thr_1" } },
"turn/start": { turn: { id: "turn_1", status: "inProgress" } },
});
const { client, sessionUpdates, extNotifications } = makeFakeClient();
const structuredOutputs: Array<Record<string, unknown>> = [];
const agent = new CodexAppServerAgent(client, {
processOptions: { binaryPath: "/bundle/codex" },
model: "gpt-5.5",
rpcFactory: stub.factory,
onStructuredOutput: async (output) => {
structuredOutputs.push(output);
},
});
const schema = {
type: "object",
properties: { source: { type: "string" } },
required: ["source"],
};

await agent.initialize(init);
await agent.newSession({
cwd: "/repo",
_meta: {
environment: "cloud",
jsonSchema: schema,
taskRunId: "run_1",
},
} as unknown as NewSessionRequest);
const promptDone = agent.prompt({
sessionId: "thr_1",
prompt: [{ type: "text", text: "delegate this" }],
} as unknown as PromptRequest);
stub.emit("item/started", {
threadId: "thr_1",
turnId: "turn_1",
item: {
type: "collabAgentToolCall",
id: "spawn_1",
tool: "spawnAgent",
status: "inProgress",
senderThreadId: "thr_1",
receiverThreadIds: ["subagent_1"],
prompt: "Review the implementation",
},
});
const sessionUpdateCount = sessionUpdates.length;
const extNotificationCount = extNotifications.length;

stub.emit("item/agentMessage/delta", {
threadId: "subagent_1",
turnId: "subagent_turn_1",
itemId: "subagent_message_1",
delta: "subagent prose",
});
stub.emit("item/reasoning/textDelta", {
threadId: "subagent_1",
turnId: "subagent_turn_1",
itemId: "subagent_reasoning_1",
delta: "subagent reasoning",
});
stub.emit("item/completed", {
threadId: "subagent_1",
turnId: "subagent_turn_1",
item: {
type: "agentMessage",
id: "subagent_message_1",
text: '{"source":"child"}',
},
});
stub.emit("item/commandExecution/outputDelta", {
threadId: "subagent_1",
turnId: "subagent_turn_1",
itemId: "shared_command_id",
delta: "child command output",
});
stub.emit("thread/tokenUsage/updated", {
threadId: "subagent_1",
tokenUsage: {
total: { totalTokens: 9000 },
modelContextWindow: 10000,
},
});
stub.emit("item/started", {
threadId: "subagent_1",
turnId: "subagent_turn_1",
item: { type: "contextCompaction", id: "subagent_compaction_1" },
});
stub.emit("item/completed", {
threadId: "subagent_1",
turnId: "subagent_turn_1",
item: { type: "contextCompaction", id: "subagent_compaction_1" },
});
stub.emit("turn/completed", {
threadId: "subagent_1",
turn: { id: "subagent_turn_1", status: "completed" },
});

let promptSettled = false;
void promptDone.then(() => {
promptSettled = true;
});
await Promise.resolve();
expect({
extNotifications: extNotifications.length,
promptSettled,
sessionUpdates: sessionUpdates.length,
}).toEqual({
extNotifications: extNotificationCount,
promptSettled: false,
sessionUpdates: sessionUpdateCount,
});

stub.emit("item/agentMessage/delta", {
threadId: "thr_1",
turnId: "turn_1",
itemId: "message_1",
delta: "parent response",
});
stub.emit("item/completed", {
threadId: "thr_1",
turnId: "turn_1",
item: {
type: "commandExecution",
id: "shared_command_id",
command: "echo parent",
status: "completed",
aggregatedOutput: null,
},
});
stub.emit("item/completed", {
threadId: "thr_1",
turnId: "turn_1",
item: {
type: "agentMessage",
id: "message_1",
text: '{"source":"parent"}',
},
});
stub.emit("turn/completed", {
threadId: "thr_1",
turn: { id: "turn_1", status: "completed" },
});

await expect(promptDone).resolves.toMatchObject({ stopReason: "end_turn" });
const serializedUpdates = JSON.stringify(sessionUpdates);
expect(serializedUpdates).toContain("spawn_agent");
expect(serializedUpdates).toContain("parent response");
expect(serializedUpdates).not.toContain("subagent prose");
expect(serializedUpdates).not.toContain("subagent reasoning");
expect(serializedUpdates).not.toContain("child command output");
expect(structuredOutputs).toEqual([{ source: "parent" }]);
expect(
extNotifications.filter(
(notification) => notification.method === "_posthog/turn_complete",
),
).toHaveLength(1);
});

it("includes buffered command output when completion omits aggregatedOutput", async () => {
const stub = makeStubRpc({
initialize: {},
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -937,33 +937,43 @@ export class CodexAppServerAgent extends BaseAcpAgent {
}

private handleNotification(method: string, params: unknown): void {
const mappedParams = this.withBufferedCommandOutput(method, params);
const notificationThreadId = readNotificationThreadId(params);
const isMainThread =
!notificationThreadId || notificationThreadId === this.threadId;
const mappedParams = isMainThread
? this.withBufferedCommandOutput(method, params)
: params;

if (this.sessionId && !this.session.cancelled) {
const notification = mapAppServerNotification(
this.sessionId,
method,
mappedParams,
);
if (notification) {
void this.client
.sessionUpdate(notification)
.catch((err) => this.logger.warn("sessionUpdate failed", err));
this.appendNotification(this.sessionId, notification);
if (isMainThread) {
const notification = mapAppServerNotification(
this.sessionId,
method,
mappedParams,
);
if (notification) {
void this.client
.sessionUpdate(notification)
.catch((err) => this.logger.warn("sessionUpdate failed", err));
this.appendNotification(this.sessionId, notification);
}
}
}

if (method === APP_SERVER_NOTIFICATIONS.TURN_STARTED) {
// Capture the active turn id (steer precondition / interrupt target).
this.turns.onStarted((params as { turn?: { id?: string } })?.turn?.id);
}

if (method === APP_SERVER_NOTIFICATIONS.ITEM_STARTED) {
this.mcp.capture(params);
}
if (method === APP_SERVER_NOTIFICATIONS.ITEM_COMPLETED) {
this.mcp.release(params);
}

if (!isMainThread) return;

if (method === APP_SERVER_NOTIFICATIONS.TURN_STARTED) {
// Capture the active turn id (steer precondition / interrupt target).
this.turns.onStarted((params as { turn?: { id?: string } })?.turn?.id);
}

// codex auto-compaction surfaces as a contextCompaction item: item/started → in progress,
// item/completed → boundary (codex emits no separate thread/compacted; that's a guarded
// fallback). compactionActive dedupes to one boundary per compaction.
Expand Down Expand Up @@ -1437,6 +1447,12 @@ function mapTurnStopReason(status: string | undefined): StopReason {
return "end_turn";
}

function readNotificationThreadId(params: unknown): string | undefined {
if (!params || typeof params !== "object") return undefined;
const threadId = (params as { threadId?: unknown }).threadId;
return typeof threadId === "string" ? threadId : undefined;
}

/** The codex thread config override map: folds in MCP servers + makes extra workspace roots writable. Undefined when empty. */
function buildThreadConfig(
mcpServers: ReturnType<typeof toCodexMcpServers>,
Expand Down
68 changes: 68 additions & 0 deletions packages/agent/src/adapters/codex-app-server/mapping.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -230,6 +230,74 @@ describe("mapAppServerNotification", () => {
});
});

it("maps a spawned Codex agent to an explicit subagent tool call", () => {
const result = mapAppServerNotification(
"s-1",
APP_SERVER_NOTIFICATIONS.ITEM_STARTED,
{
item: {
type: "collabAgentToolCall",
id: "spawn-1",
tool: "spawnAgent",
status: "inProgress",
senderThreadId: "main-thread",
receiverThreadIds: ["child-thread"],
prompt: "Review the authentication changes\nFocus on security.",
model: "gpt-5.5",
reasoningEffort: "high",
},
},
);

expect(result).toEqual({
sessionId: "s-1",
update: {
sessionUpdate: "tool_call",
toolCallId: "spawn-1",
title: "Review the authentication changes",
kind: "other",
status: "in_progress",
rawInput: {
prompt: "Review the authentication changes\nFocus on security.",
receiverThreadIds: ["child-thread"],
model: "gpt-5.5",
reasoningEffort: "high",
},
_meta: { posthog: { toolName: "spawn_agent" } },
},
});
});

it("keeps a completed spawn tool call terminal while its subagent is running", () => {
const result = mapAppServerNotification(
"s-1",
APP_SERVER_NOTIFICATIONS.ITEM_COMPLETED,
{
item: {
type: "collabAgentToolCall",
id: "spawn-1",
tool: "spawnAgent",
status: "completed",
senderThreadId: "main-thread",
receiverThreadIds: ["child-thread"],
prompt: "Review the authentication changes",
agentsStates: {
"child-thread": { status: "running", message: null },
},
},
},
);

expect(result).toEqual({
sessionId: "s-1",
update: {
sessionUpdate: "tool_call_update",
toolCallId: "spawn-1",
status: "completed",
},
});
});

it("drops agent message items (their deltas already streamed)", () => {
expect(
mapAppServerNotification("s-1", APP_SERVER_NOTIFICATIONS.ITEM_COMPLETED, {
Expand Down
Loading
Loading