Skip to content
Merged
Changes from 1 commit
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
101 changes: 60 additions & 41 deletions apps/server/src/provider/Layers/CodexSessionRuntime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -615,47 +615,60 @@ function readRouteFields(notification: CodexServerNotification): {
}
}

function rememberCollabReceiverTurns(
collabReceiverTurns: Map<string, TurnId>,
function rememberChildThreadTurns(
childThreadTurns: Map<string, TurnId>,
notification: CodexServerNotification,
parentTurnId: TurnId | undefined,
): void {
if (!parentTurnId) {
if (parentTurnId === undefined) {
return;
}

if (notification.method !== "item/started" && notification.method !== "item/completed") {
return;
}

if (notification.params.item.type !== "collabAgentToolCall") {
const item = notification.params.item;

if (item.type === "collabAgentToolCall") {
for (const receiverThreadId of item.receiverThreadIds) {
childThreadTurns.set(receiverThreadId, parentTurnId);
}
return;
}

for (const receiverThreadId of notification.params.item.receiverThreadIds) {
collabReceiverTurns.set(receiverThreadId, parentTurnId);
if (item.type === "subAgentActivity") {
childThreadTurns.set(item.agentThreadId, parentTurnId);
}
}

function shouldSuppressChildConversationNotification(
method: CodexRpc.ServerNotificationMethod,
): boolean {
return (
method === "thread/started" ||
method === "thread/status/changed" ||
method === "thread/archived" ||
method === "thread/unarchived" ||
method === "thread/closed" ||
method === "thread/compacted" ||
method === "thread/name/updated" ||
method === "thread/tokenUsage/updated" ||
method === "thread/goal/updated" ||
method === "thread/goal/cleared" ||
method === "turn/started" ||
method === "turn/completed" ||
method === "turn/plan/updated" ||
method === "item/plan/delta"
);
function shouldKeepChildNotification(notification: CodexServerNotification): boolean {

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Add focused tests for child notification filtering

This changes backend event routing without adding a focused CodexSessionRuntime test for the new subAgentActivity registration and allowlist. Please exercise a synthetic child stream and assert that goal/message/plan events are suppressed while command, file-change, tool, and diff events are retained and reparented; otherwise the repeated parent goal-card regression this change targets remains unverified.

AGENTS.md reference: AGENTS.md:L101-L101

Useful? React with 👍 / 👎.

switch (notification.method) {
case "item/started":
case "item/completed":
switch (notification.params.item.type) {
case "commandExecution":
case "fileChange":
case "mcpToolCall":
case "dynamicToolCall":
case "collabAgentToolCall":
case "webSearch":
case "imageView":
return true;
default:
return false;
}
case "item/commandExecution/outputDelta":
case "item/commandExecution/terminalInteraction":
case "item/fileChange/outputDelta":
case "item/fileChange/patchUpdated":
case "item/mcpToolCall/progress":
case "serverRequest/resolved":
case "turn/diff/updated":
return true;
default:
return false;
}
}

function toCodexUserInputAnswer(
Expand Down Expand Up @@ -739,7 +752,7 @@ export const makeCodexSessionRuntime = (
const pendingApprovalsRef = yield* Ref.make(new Map<ApprovalRequestId, PendingApproval>());
const approvalCorrelationsRef = yield* Ref.make(new Map<string, ApprovalCorrelation>());
const pendingUserInputsRef = yield* Ref.make(new Map<ApprovalRequestId, PendingUserInput>());
const collabReceiverTurnsRef = yield* Ref.make(new Map<string, TurnId>());
const childThreadTurnsRef = yield* Ref.make(new Map<string, TurnId>());
const closedRef = yield* Ref.make(false);

// `~` is not shell-expanded when env vars are set via
Expand Down Expand Up @@ -856,27 +869,35 @@ export const makeCodexSessionRuntime = (
),
);

const currentSessionProviderThreadId = Effect.map(Ref.get(sessionRef), currentProviderThreadId);

const handleRawNotification = (notification: CodexServerNotification) =>
Effect.gen(function* () {
const payload = notification.params;
const route = readRouteFields(notification);
const collabReceiverTurns = yield* Ref.get(collabReceiverTurnsRef);
const childParentTurnId = (() => {
const providerConversationId = readNotificationThreadId(notification);
return providerConversationId
? collabReceiverTurns.get(providerConversationId)
: undefined;
})();

rememberCollabReceiverTurns(collabReceiverTurns, notification, route.turnId);
if (childParentTurnId && shouldSuppressChildConversationNotification(notification.method)) {
yield* Ref.set(collabReceiverTurnsRef, collabReceiverTurns);
const providerThreadId = readNotificationThreadId(notification);
const rootProviderThreadId = yield* currentSessionProviderThreadId;
const childThreadTurns = yield* Ref.get(childThreadTurnsRef);
const childParentTurnId =
providerThreadId === undefined ? undefined : childThreadTurns.get(providerThreadId);
const parentTurnId = childParentTurnId ?? route.turnId;

rememberChildThreadTurns(childThreadTurns, notification, parentTurnId);

const belongsToChild =
childParentTurnId !== undefined ||
(providerThreadId !== undefined &&
rootProviderThreadId !== undefined &&
providerThreadId !== rootProviderThreadId);

if (belongsToChild && !shouldKeepChildNotification(notification)) {
yield* Ref.set(childThreadTurnsRef, childThreadTurns);
return;
}

let requestId: ApprovalRequestId | undefined;
let requestKind: ProviderRequestKind | undefined;
let turnId = childParentTurnId ?? route.turnId;
let turnId = parentTurnId;
let itemId = route.itemId;

if (notification.method === "serverRequest/resolved") {
Expand All @@ -900,7 +921,7 @@ export const makeCodexSessionRuntime = (
}
}

yield* Ref.set(collabReceiverTurnsRef, collabReceiverTurns);
yield* Ref.set(childThreadTurnsRef, childThreadTurns);
yield* emitEvent({
kind: "notification",
threadId: options.threadId,
Expand All @@ -916,8 +937,6 @@ export const makeCodexSessionRuntime = (
});
});

const currentSessionProviderThreadId = Effect.map(Ref.get(sessionRef), currentProviderThreadId);

yield* client.handleServerNotification("thread/started", (payload) =>
currentSessionProviderThreadId.pipe(
Effect.flatMap((providerThreadId) => {
Expand Down
Loading