Skip to content

Commit 955ff1b

Browse files
duclvzduclvz
committed
feat(subagent-lifecycle): report coherent agent and command updates
Present each sub-agent run as one activity and expose internal command output through portable ACP content.
1 parent 3196231 commit 955ff1b

16 files changed

Lines changed: 1304 additions & 185 deletions

‎src/CodexAcpServer.ts‎

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,6 @@ import {
5151
LEGACY_SET_SESSION_MODEL_METHOD,
5252
} from "./AcpExtensions";
5353
import {
54-
createCollabAgentToolCallUpdate,
5554
createCompletedContextCompactionUpdate,
5655
createCommandExecutionCompleteUpdate,
5756
createCommandExecutionUpdate,
@@ -60,7 +59,6 @@ import {
6059
createImageGenerationUpdate,
6160
createImageViewUpdate,
6261
createMcpToolCallUpdate,
63-
createSubAgentActivityUpdate,
6462
formatWebSearchTitle,
6563
} from "./CodexToolCallMapper";
6664
import {
@@ -86,6 +84,7 @@ import {
8684
type ThreadGoalSnapshot,
8785
toThreadGoalSnapshot,
8886
} from "./ThreadGoalSnapshot";
87+
import { getSubAgentActivityTracker } from "./SubAgentActivityTracker";
8988

9089
export interface SessionState {
9190
sessionId: string,
@@ -188,7 +187,7 @@ export class CodexAcpServer {
188187
this.getRecentStderr = getRecentStderr ?? (() => "");
189188
this.clientInfo = null;
190189
this.clientCapabilities = null;
191-
this.terminalOutputMode = "terminal_output_delta";
190+
this.terminalOutputMode = "content";
192191
this.booleanConfigOptionsSupported = false;
193192
this.availableCommands = new CodexCommands(
194193
connection,
@@ -1183,7 +1182,7 @@ export class CodexAcpServer {
11831182
case "sleep":
11841183
return [];
11851184
case "subAgentActivity":
1186-
return [createSubAgentActivityUpdate(item, "completed", "tool_call")];
1185+
return getSubAgentActivityTracker(sessionState).mapSubAgentActivity(item, "completed");
11871186
case "agentMessage": {
11881187
const meta = createCodexMessagePhaseMeta(item.phase);
11891188
return [{
@@ -1198,7 +1197,7 @@ export class CodexAcpServer {
11981197
case "fileChange":
11991198
return [await createFileChangeUpdate(item)];
12001199
case "commandExecution": {
1201-
const updates = [await createCommandExecutionUpdate(item)];
1200+
const updates = [await createCommandExecutionUpdate(item, sessionState.terminalOutputMode)];
12021201
const completeUpdate = createCommandExecutionCompleteUpdate(item, sessionState.terminalOutputMode);
12031202
if (completeUpdate) {
12041203
updates.push(completeUpdate);
@@ -1210,7 +1209,7 @@ export class CodexAcpServer {
12101209
case "dynamicToolCall":
12111210
return [await createDynamicToolCallUpdate(item)];
12121211
case "collabAgentToolCall":
1213-
return [createCollabAgentToolCallUpdate(item)];
1212+
return getSubAgentActivityTracker(sessionState).mapCollabAgentToolCall(item, "completed");
12141213
case "webSearch":
12151214
return [this.createWebSearchUpdate(item)];
12161215
case "imageView":
@@ -1890,9 +1889,9 @@ function mergeHistoryUpdates(
18901889
const seen = new Set<string>();
18911890
let fallbackIndex = 0;
18921891

1893-
const pushUpdate = (update: UpdateSessionEvent) => {
1892+
const pushUpdate = (update: UpdateSessionEvent, dedupe: boolean = true) => {
18941893
const key = historyUpdateKey(update);
1895-
if (key && seen.has(key)) {
1894+
if (dedupe && key && seen.has(key)) {
18961895
return;
18971896
}
18981897
if (key) {
@@ -1928,7 +1927,10 @@ function mergeHistoryUpdates(
19281927

19291928
for (const update of threadUpdates) {
19301929
flushFallbackBeforeMatchingDuplicate(update);
1931-
pushUpdate(update);
1930+
// Thread history is authoritative and can legitimately contain several
1931+
// progress updates for the same tool call. Only fallback records are
1932+
// deduplicated against those updates.
1933+
pushUpdate(update, false);
19321934
}
19331935

19341936
while (fallbackIndex < responseItemFallbackUpdates.length) {

‎src/CodexAppServerClient.ts‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -144,6 +144,7 @@ export class CodexAppServerClient {
144144
private readonly threadGoalClearedCaptures = new Map<string, Set<() => void>>();
145145
private readonly threadSettings = new Map<string, ThreadSettings>();
146146
private readonly staleTurnIds = new Map<string, Set<string>>();
147+
private readonly childThreadParents = new Map<string, string>();
147148

148149
constructor(connection: MessageConnection) {
149150
this.connection = connection;
@@ -180,6 +181,7 @@ export class CodexAppServerClient {
180181
if (this.handleStaleTurnNotification(serverNotification, routing)) {
181182
return;
182183
}
184+
this.recordChildThreadParent(serverNotification);
183185
this.recordTurnRouting(routing);
184186
if (this.handleStaleTurnNotification(serverNotification, routing)) {
185187
return;
@@ -258,6 +260,12 @@ export class CodexAppServerClient {
258260
this.notificationHandlers.delete(threadId);
259261
this.approvalHandlers.delete(threadId);
260262
this.elicitationHandlers.delete(threadId);
263+
this.childThreadParents.delete(threadId);
264+
for (const [childThreadId, parentThreadId] of this.childThreadParents) {
265+
if (parentThreadId === threadId) {
266+
this.childThreadParents.delete(childThreadId);
267+
}
268+
}
261269
}
262270

263271
async initialize(params: InitializeParams): Promise<InitializeResponse> {
@@ -675,13 +683,41 @@ export class CodexAppServerClient {
675683
if (handler) {
676684
handler(notification);
677685
}
686+
const parentThreadId = this.childThreadParents.get(threadId);
687+
const parentHandler = parentThreadId ? this.notificationHandlers.get(parentThreadId) : undefined;
688+
if (
689+
parentHandler
690+
&& parentHandler !== handler
691+
&& isChildActivityNotification(notification)
692+
) {
693+
parentHandler(notification);
694+
}
678695
return;
679696
}
680697
for (const notificationHandler of this.notificationHandlers.values()) {
681698
notificationHandler(notification);
682699
}
683700
}
684701

702+
private recordChildThreadParent(notification: ServerNotification): void {
703+
if (notification.method !== "item/started" && notification.method !== "item/completed") {
704+
return;
705+
}
706+
const item = notification.params.item;
707+
if (item.type === "subAgentActivity") {
708+
this.childThreadParents.set(item.agentThreadId, notification.params.threadId);
709+
return;
710+
}
711+
if (
712+
item.type === "collabAgentToolCall"
713+
&& (item.tool === "spawnAgent" || item.tool === "resumeAgent")
714+
) {
715+
for (const childThreadId of item.receiverThreadIds) {
716+
this.childThreadParents.set(childThreadId, notification.params.threadId);
717+
}
718+
}
719+
}
720+
685721
private recordTurnCompleted(event: TurnCompletedNotification): void {
686722
const threadResolvers = this.pendingTurnCompletionResolvers.get(event.threadId);
687723
const resolve = threadResolvers?.get(event.turn.id);
@@ -1006,6 +1042,14 @@ function isTurnCompletedNotification(notification: ServerNotification): notifica
10061042
return notification.method === "turn/completed";
10071043
}
10081044

1045+
function isChildActivityNotification(notification: ServerNotification): boolean {
1046+
if (notification.method === "turn/completed") {
1047+
return true;
1048+
}
1049+
return notification.method === "item/completed"
1050+
&& notification.params.item.type === "agentMessage";
1051+
}
1052+
10091053
function isThreadStatusChangedNotification(notification: ServerNotification): notification is {
10101054
method: "thread/status/changed";
10111055
params: ThreadStatusChangedNotification;

‎src/CodexEventHandler.ts‎

Lines changed: 52 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -33,9 +33,8 @@ import type { McpStartupCompleteEvent } from "./app-server";
3333
import {toTokenCount} from "./TokenCount";
3434
import {
3535
commandExecutionUsesTerminalOutput,
36-
createCollabAgentToolCallCompleteUpdate,
37-
createCollabAgentToolCallUpdate,
3836
createCommandExecutionUpdate,
37+
createCommandExecutionCompleteUpdate,
3938
createContextCompactionCompleteUpdate,
4039
createContextCompactionStartUpdate,
4140
createDynamicToolCallUpdate,
@@ -51,7 +50,6 @@ import {
5150
createFuzzyFileSearchComplete,
5251
createFuzzyFileSearchStartOrUpdate,
5352
createMcpToolCallUpdate,
54-
createSubAgentActivityUpdate,
5553
createWebSearchCompleteUpdate,
5654
createWebSearchStartUpdate,
5755
fuzzyFileSearchToolCallId,
@@ -64,9 +62,12 @@ import {
6462
createAgentTextThoughtChunk,
6563
} from "./ContentChunks";
6664
import {sameThreadGoalSnapshot, toThreadGoalSnapshot} from "./ThreadGoalSnapshot";
65+
import { getSubAgentActivityTracker } from "./SubAgentActivityTracker";
6766

6867
export { stripShellPrefix };
6968

69+
type UpdateResult = UpdateSessionEvent | UpdateSessionEvent[] | null;
70+
7071
export class CodexEventHandler {
7172

7273
private readonly connection: AcpClientConnection;
@@ -80,7 +81,6 @@ export class CodexEventHandler {
8081
private readonly terminalCommandIds = new Set<string>();
8182
private readonly terminalCommandOutputIds = new Set<string>();
8283
private readonly agentMessagePhases = new Map<string, string | null>();
83-
private readonly activeSubAgentActivities = new Set<string>();
8484

8585
constructor(connection: AcpClientConnection, sessionState: SessionState) {
8686
this.connection = connection;
@@ -93,13 +93,14 @@ export class CodexEventHandler {
9393

9494
async handleNotification(notification: ServerNotification) {
9595
const session = new ACPSessionConnection(this.connection, this.sessionState.sessionId);
96-
const updateEvent = await this.createUpdateEvent(notification);
97-
if (updateEvent) {
96+
const result = await this.createUpdateEvent(notification);
97+
const updateEvents = Array.isArray(result) ? result : result ? [result] : [];
98+
for (const updateEvent of updateEvents) {
9899
await session.update(updateEvent);
99100
}
100101
}
101102

102-
private async createUpdateEvent(notification: ServerNotification): Promise<UpdateSessionEvent | null> {
103+
private async createUpdateEvent(notification: ServerNotification): Promise<UpdateResult> {
103104
/*
104105
TODO split UpdateSessionEvent to improve completion
105106
createUpdateEvent({
@@ -122,6 +123,12 @@ export class CodexEventHandler {
122123
this.sessionState.currentTurnId = notification.params.turn.id;
123124
return null;
124125
case "turn/completed":
126+
if (notification.params.threadId !== this.sessionState.sessionId) {
127+
return getSubAgentActivityTracker(this.sessionState).completeChildTurn(
128+
notification.params.threadId,
129+
notification.params.turn,
130+
);
131+
}
125132
this.sessionState.currentTurnId = null;
126133
return null;
127134
case "thread/tokenUsage/updated":
@@ -299,18 +306,21 @@ export class CodexEventHandler {
299306
return createAgentTextThoughtChunk(text, messageId);
300307
}
301308

302-
private async createItemEvent(event: ItemStartedNotification): Promise<UpdateSessionEvent | null> {
309+
private async createItemEvent(event: ItemStartedNotification): Promise<UpdateResult> {
303310
switch (event.item.type) {
304311
case "fileChange":
305312
return await createFileChangeUpdate(event.item);
306313
case "commandExecution": {
307-
if (commandExecutionUsesTerminalOutput(event.item)) {
314+
if (
315+
this.sessionState.terminalOutputMode !== "content"
316+
&& commandExecutionUsesTerminalOutput(event.item)
317+
) {
308318
this.terminalCommandIds.add(event.item.id);
309319
} else {
310320
this.terminalCommandIds.delete(event.item.id);
311321
this.terminalCommandOutputIds.delete(event.item.id);
312322
}
313-
return await createCommandExecutionUpdate(event.item);
323+
return await createCommandExecutionUpdate(event.item, this.sessionState.terminalOutputMode);
314324
}
315325
case "mcpToolCall":
316326
return await createMcpToolCallUpdate(event.item);
@@ -325,15 +335,14 @@ export class CodexEventHandler {
325335
this.activeImageGenerationItems.add(event.item.id);
326336
return createImageGenerationStartUpdate(event.item);
327337
case "collabAgentToolCall":
328-
return createCollabAgentToolCallUpdate(event.item);
338+
return getSubAgentActivityTracker(this.sessionState).mapCollabAgentToolCall(event.item, "started");
329339
case "agentMessage":
330340
this.rememberAgentMessagePhase(event.item);
331341
return null;
332342
case "contextCompaction":
333343
return createContextCompactionStartUpdate(event.item);
334344
case "subAgentActivity":
335-
this.activeSubAgentActivities.add(event.item.id);
336-
return createSubAgentActivityUpdate(event.item, "in_progress", "tool_call");
345+
return getSubAgentActivityTracker(this.sessionState).mapSubAgentActivity(event.item, "started");
337346
case "sleep":
338347
case "userMessage":
339348
case "hookPrompt":
@@ -345,7 +354,7 @@ export class CodexEventHandler {
345354
}
346355
}
347356

348-
private async completeItemEvent(event: ItemCompletedNotification): Promise<UpdateSessionEvent | null> {
357+
private async completeItemEvent(event: ItemCompletedNotification): Promise<UpdateResult> {
349358
switch (event.item.type) {
350359
case "fileChange":
351360
case "dynamicToolCall":
@@ -382,21 +391,24 @@ export class CodexEventHandler {
382391
case "webSearch":
383392
return createWebSearchCompleteUpdate(event.item);
384393
case "collabAgentToolCall":
385-
return createCollabAgentToolCallCompleteUpdate(event.item);
394+
return getSubAgentActivityTracker(this.sessionState).mapCollabAgentToolCall(event.item, "completed");
386395
case "agentMessage":
396+
if (event.threadId !== this.sessionState.sessionId) {
397+
getSubAgentActivityTracker(this.sessionState).recordChildMessage(
398+
event.threadId,
399+
event.item.text,
400+
);
401+
return null;
402+
}
387403
this.rememberAgentMessagePhase(event.item);
388404
return null;
389405
case "exitedReviewMode":
390406
return this.createExitedReviewModeEvent(event.item);
391407
case "contextCompaction":
392408
return createContextCompactionCompleteUpdate(event.item);
393409
//ignored types
394-
case "subAgentActivity": {
395-
const sessionUpdate = this.activeSubAgentActivities.delete(event.item.id)
396-
? "tool_call_update"
397-
: "tool_call";
398-
return createSubAgentActivityUpdate(event.item, "completed", sessionUpdate);
399-
}
410+
case "subAgentActivity":
411+
return getSubAgentActivityTracker(this.sessionState).mapSubAgentActivity(event.item, "completed");
400412
case "sleep":
401413
case "userMessage":
402414
case "hookPrompt":
@@ -433,7 +445,7 @@ export class CodexEventHandler {
433445
}
434446

435447
private createCommandOutputDeltaEvent(event: CommandExecutionOutputDeltaNotification): UpdateSessionEvent {
436-
if (this.terminalCommandIds.has(event.itemId) && event.delta.length > 0) {
448+
if (event.delta.length > 0) {
437449
this.terminalCommandOutputIds.add(event.itemId);
438450
}
439451
return this.createCommandOutputEvent(event.itemId, event.delta, this.commandOutputMode(event.itemId));
@@ -444,6 +456,16 @@ export class CodexEventHandler {
444456
data: string,
445457
terminalOutputMode: TerminalOutputMode
446458
): UpdateSessionEvent {
459+
if (terminalOutputMode === "content") {
460+
return {
461+
sessionUpdate: "tool_call_update",
462+
toolCallId: itemId,
463+
content: [{
464+
type: "content",
465+
content: { type: "text", text: data },
466+
}],
467+
};
468+
}
447469
return {
448470
sessionUpdate: "tool_call_update",
449471
toolCallId: itemId,
@@ -515,37 +537,16 @@ export class CodexEventHandler {
515537
}
516538

517539
private completeCommandExecutionEvent(item: ThreadItem & { "type": "commandExecution" }): UpdateSessionEvent {
518-
const update: UpdateSessionEvent = {
519-
sessionUpdate: "tool_call_update",
520-
toolCallId: item.id,
521-
status: item.status === "completed" ? "completed" : "failed",
522-
rawOutput: {
523-
formatted_output: item.aggregatedOutput ?? "",
524-
exit_code: item.exitCode
525-
},
526-
};
527-
528540
const commandHadTerminal = this.terminalCommandIds.delete(item.id);
529541
const commandHadOutput = this.terminalCommandOutputIds.delete(item.id);
530-
if (!commandHadTerminal) {
531-
return update;
532-
}
533-
const terminalMeta: Record<string, unknown> = {};
534-
if (!commandHadOutput && item.aggregatedOutput) {
535-
Object.assign(
536-
terminalMeta,
537-
createTerminalOutputMeta(this.sessionState.terminalOutputMode, item.id, item.aggregatedOutput)
538-
);
539-
}
540-
terminalMeta["terminal_exit"] = {
541-
exit_code: item.exitCode,
542-
signal: null,
543-
terminal_id: item.id
544-
};
545-
return {
546-
...update,
547-
_meta: terminalMeta,
548-
};
542+
return createCommandExecutionCompleteUpdate(
543+
item,
544+
this.sessionState.terminalOutputMode,
545+
{
546+
includeOutputContent: !commandHadOutput,
547+
includeTerminalMeta: commandHadTerminal,
548+
},
549+
)!;
549550
}
550551

551552
private async updatePlan(event: TurnPlanUpdatedNotification): Promise<UpdateSessionEvent> {

0 commit comments

Comments
 (0)