History

maybe fix codex subturns

Julius Marminge committed 7 months ago

Commit ef843d41 parent 654ca0d4 files · +334 −16Browse files

Unified · Split

4 files
  1. apps/server/src/codexAppServerManager.test.ts +187 −0
  2. apps/server/src/codexAppServerManager.ts +94 −10
  3. apps/server/src/provider/Layers/CodexAdapter.test.ts +39 −0
  4. apps/server/src/provider/Layers/CodexAdapter.ts +14 −6
apps/server/src/codexAppServerManager.test.ts+187−0
@@ -37,6 +37,7 @@ function createSendTurnHarness() { · 36 unmodified lines
3737 planType: null,
3838 sparkEnabled: true,
3939 },
40+ collabReceiverTurns: new Map(),
4041 };
4142
4243 const requireSession = vi
@@ -75,6 +76,7 @@ function createThreadControlHarness() { · 32 unmodified lines
7576 createdAt: "2026-02-10T00:00:00.000Z",
7677 updatedAt: "2026-02-10T00:00:00.000Z",
7778 },
79+ collabReceiverTurns: new Map(),
7880 };
7981
8082 const requireSession = vi
@@ -117,6 +119,7 @@ function createPendingUserInputHarness() { · 36 unmodified lines
117119 },
118120 ],
119121 ]),
122+ collabReceiverTurns: new Map(),
120123 };
121124
122125 const requireSession = vi
@@ -135,6 +138,43 @@ function createPendingUserInputHarness() { · 12 unmodified lines
135138 return { manager, context, requireSession, writeMessage, emitEvent };
136139 }
137140
141+function createCollabNotificationHarness() {
142+ const manager = new CodexAppServerManager();
143+ const context = {
144+ session: {
145+ provider: "codex",
146+ status: "running",
147+ threadId: asThreadId("thread_1"),
148+ runtimeMode: "full-access",
149+ model: "gpt-5.3-codex",
150+ activeTurnId: "turn_parent",
151+ resumeCursor: { threadId: "provider_parent" },
152+ createdAt: "2026-02-10T00:00:00.000Z",
153+ updatedAt: "2026-02-10T00:00:00.000Z",
154+ },
155+ account: {
156+ type: "unknown",
157+ planType: null,
158+ sparkEnabled: true,
159+ },
160+ pending: new Map(),
161+ pendingApprovals: new Map(),
162+ pendingUserInputs: new Map(),
163+ collabReceiverTurns: new Map<string, string>(),
164+ nextRequestId: 1,
165+ stopping: false,
166+ };
167+
168+ const emitEvent = vi
169+ .spyOn(manager as unknown as { emitEvent: (...args: unknown[]) => void }, "emitEvent")
170+ .mockImplementation(() => {});
171+ const updateSession = vi
172+ .spyOn(manager as unknown as { updateSession: (...args: unknown[]) => void }, "updateSession")
173+ .mockImplementation(() => {});
174+
175+ return { manager, context, emitEvent, updateSession };
176+}
177+
138178 describe("classifyCodexStderrLine", () => {
139179 it("ignores empty lines", () => {
140180 expect(classifyCodexStderrLine(" ")).toBeNull();
@@ -721,6 +761,7 @@ describe("respondToUserInput", () => { · 580 unmodified lines
721761 },
722762 pendingApprovals: new Map(),
723763 pendingUserInputs: new Map(),
764+ collabReceiverTurns: new Map(),
724765 };
725766 type ApprovalRequestContext = {
726767 session: typeof context.session;
@@ -748,6 +789,152 @@ describe("respondToUserInput", () => { · 21 unmodified lines
748789 });
749790 });
750791
792+describe("collab child conversation routing", () => {
793+ it("rewrites child notification turn ids onto the parent turn", () => {
794+ const { manager, context, emitEvent } = createCollabNotificationHarness();
795+
796+ (
797+ manager as unknown as {
798+ handleServerNotification: (context: unknown, notification: Record<string, unknown>) => void;
799+ }
800+ ).handleServerNotification(context, {
801+ method: "item/completed",
802+ params: {
803+ item: {
804+ type: "collabAgentToolCall",
805+ id: "call_collab_1",
806+ receiverThreadIds: ["child_provider_1"],
807+ },
808+ threadId: "provider_parent",
809+ turnId: "turn_parent",
810+ },
811+ });
812+
813+ (
814+ manager as unknown as {
815+ handleServerNotification: (context: unknown, notification: Record<string, unknown>) => void;
816+ }
817+ ).handleServerNotification(context, {
818+ method: "item/agentMessage/delta",
819+ params: {
820+ threadId: "child_provider_1",
821+ turnId: "turn_child_1",
822+ itemId: "msg_child_1",
823+ delta: "working",
824+ },
825+ });
826+
827+ expect(emitEvent).toHaveBeenLastCalledWith(
828+ expect.objectContaining({
829+ method: "item/agentMessage/delta",
830+ turnId: "turn_parent",
831+ itemId: "msg_child_1",
832+ }),
833+ );
834+ });
835+
836+ it("suppresses child lifecycle notifications so they cannot replace the parent turn", () => {
837+ const { manager, context, emitEvent, updateSession } = createCollabNotificationHarness();
838+
839+ (
840+ manager as unknown as {
841+ handleServerNotification: (context: unknown, notification: Record<string, unknown>) => void;
842+ }
843+ ).handleServerNotification(context, {
844+ method: "item/completed",
845+ params: {
846+ item: {
847+ type: "collabAgentToolCall",
848+ id: "call_collab_1",
849+ receiverThreadIds: ["child_provider_1"],
850+ },
851+ threadId: "provider_parent",
852+ turnId: "turn_parent",
853+ },
854+ });
855+ emitEvent.mockClear();
856+ updateSession.mockClear();
857+
858+ (
859+ manager as unknown as {
860+ handleServerNotification: (context: unknown, notification: Record<string, unknown>) => void;
861+ }
862+ ).handleServerNotification(context, {
863+ method: "turn/started",
864+ params: {
865+ threadId: "child_provider_1",
866+ turn: { id: "turn_child_1" },
867+ },
868+ });
869+
870+ (
871+ manager as unknown as {
872+ handleServerNotification: (context: unknown, notification: Record<string, unknown>) => void;
873+ }
874+ ).handleServerNotification(context, {
875+ method: "turn/completed",
876+ params: {
877+ threadId: "child_provider_1",
878+ turn: { id: "turn_child_1", status: "completed" },
879+ },
880+ });
881+
882+ expect(emitEvent).not.toHaveBeenCalled();
883+ expect(updateSession).not.toHaveBeenCalled();
884+ });
885+
886+ it("rewrites child approval requests onto the parent turn", () => {
887+ const { manager, context, emitEvent } = createCollabNotificationHarness();
888+
889+ (
890+ manager as unknown as {
891+ handleServerNotification: (context: unknown, notification: Record<string, unknown>) => void;
892+ }
893+ ).handleServerNotification(context, {
894+ method: "item/completed",
895+ params: {
896+ item: {
897+ type: "collabAgentToolCall",
898+ id: "call_collab_1",
899+ receiverThreadIds: ["child_provider_1"],
900+ },
901+ threadId: "provider_parent",
902+ turnId: "turn_parent",
903+ },
904+ });
905+ emitEvent.mockClear();
906+
907+ (
908+ manager as unknown as {
909+ handleServerRequest: (context: unknown, request: Record<string, unknown>) => void;
910+ }
911+ ).handleServerRequest(context, {
912+ id: 42,
913+ method: "item/commandExecution/requestApproval",
914+ params: {
915+ threadId: "child_provider_1",
916+ turnId: "turn_child_1",
917+ itemId: "call_child_1",
918+ command: "bun install",
919+ },
920+ });
921+
922+ expect(Array.from(context.pendingApprovals.values())[0]).toEqual(
923+ expect.objectContaining({
924+ turnId: "turn_parent",
925+ itemId: "call_child_1",
926+ }),
927+ );
928+ expect(emitEvent).toHaveBeenCalledWith(
929+ expect.objectContaining({
930+ method: "item/commandExecution/requestApproval",
931+ turnId: "turn_parent",
932+ itemId: "call_child_1",
933+ }),
934+ );
935+ });
936+});
937+
751938 describe.skipIf(!process.env.CODEX_BINARY_PATH)("startSession live Codex resume", () => {
752939 it("keeps prior thread history when resuming with a changed runtime mode", async () => {
753940 const workspaceDir = mkdtempSync(path.join(os.tmpdir(), "codex-live-resume-"));
apps/server/src/codexAppServerManager.ts+94−10
@@ -70,6 +70,7 @@ interface CodexSessionContext { · 69 unmodified lines
7070 pending: Map<PendingRequestKey, PendingRequest>;
7171 pendingApprovals: Map<ApprovalRequestId, PendingApprovalRequest>;
7272 pendingUserInputs: Map<ApprovalRequestId, PendingUserInputRequest>;
73+ collabReceiverTurns: Map<string, TurnId>;
7374 nextRequestId: number;
7475 stopping: boolean;
7576 }
@@ -571,6 +572,7 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 495 unmodified lines
571572 pending: new Map(),
572573 pendingApprovals: new Map(),
573574 pendingUserInputs: new Map(),
575+ collabReceiverTurns: new Map(),
574576 nextRequestId: 1,
575577 stopping: false,
576578 };
@@ -732,6 +734,7 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 155 unmodified lines
732734
733735 async sendTurn(input: CodexAppServerSendTurnInput): Promise<ProviderTurnStartResult> {
734736 const context = this.requireSession(input.threadId);
737+ context.collabReceiverTurns.clear();
735738
736739 const turnInput: Array<
737740 { type: "text"; text: string; text_elements: [] } | { type: "image"; url: string }
@@ -1121,7 +1124,16 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 383 unmodified lines
11211124 context: CodexSessionContext,
11221125 notification: JsonRpcNotification,
11231126 ): void {
1124− const route = this.readRouteFields(notification.params);
1127+ const rawRoute = this.readRouteFields(notification.params);
1128+ this.rememberCollabReceiverTurns(context, notification.params, rawRoute.turnId);
1129+ const childParentTurnId = this.readChildParentTurnId(context, notification.params);
1130+ const isChildConversation = childParentTurnId !== undefined;
1131+ if (
1132+ isChildConversation &&
1133+ this.shouldSuppressChildConversationNotification(notification.method)
1134+ ) {
1135+ return;
1136+ }
11251137 const textDelta =
11261138 notification.method === "item/agentMessage/delta"
11271139 ? this.readString(notification.params, "delta")
@@ -1134,8 +1146,10 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 6 unmodified lines
11341146 threadId: context.session.threadId,
11351147 createdAt: new Date().toISOString(),
11361148 method: notification.method,
1137− turnId: route.turnId,
1138− itemId: route.itemId,
1149+ ...((childParentTurnId ?? rawRoute.turnId)
1150+ ? { turnId: childParentTurnId ?? rawRoute.turnId }
1151+ : {}),
1152+ ...(rawRoute.itemId ? { itemId: rawRoute.itemId } : {}),
11391153 textDelta,
11401154 payload: notification.params,
11411155 });
@@ -1151,6 +1165,9 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 9 unmodified lines
11511165 }
11521166
11531167 if (notification.method === "turn/started") {
1168+ if (isChildConversation) {
1169+ return;
1170+ }
11541171 const turnId = toTurnId(this.readString(this.readObject(notification.params)?.turn, "id"));
11551172 this.updateSession(context, {
11561173 status: "running",
@@ -1160,6 +1177,10 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 3 unmodified lines
11601177 }
11611178
11621179 if (notification.method === "turn/completed") {
1180+ if (isChildConversation) {
1181+ return;
1182+ }
1183+ context.collabReceiverTurns.clear();
11631184 const turn = this.readObject(notification.params, "turn");
11641185 const status = this.readString(turn, "status");
11651186 const errorMessage = this.readString(this.readObject(turn, "error"), "message");
@@ -1172,6 +1193,9 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 6 unmodified lines
11721193 }
11731194
11741195 if (notification.method === "error") {
1196+ if (isChildConversation) {
1197+ return;
1198+ }
11751199 const message = this.readString(this.readObject(notification.params)?.error, "message");
11761200 const willRetry = this.readBoolean(notification.params, "willRetry");
11771201
@@ -1183,7 +1207,9 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 5 unmodified lines
11831207 }
11841208
11851209 private handleServerRequest(context: CodexSessionContext, request: JsonRpcRequest): void {
1186− const route = this.readRouteFields(request.params);
1210+ const rawRoute = this.readRouteFields(request.params);
1211+ const childParentTurnId = this.readChildParentTurnId(context, request.params);
1212+ const effectiveTurnId = childParentTurnId ?? rawRoute.turnId;
11871213 const requestKind = this.requestKindForMethod(request.method);
11881214 let requestId: ApprovalRequestId | undefined;
11891215 if (requestKind) {
@@ -1199,8 +1225,8 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 9 unmodified lines
11991225 : "item/fileChange/requestApproval",
12001226 requestKind,
12011227 threadId: context.session.threadId,
1202− ...(route.turnId ? { turnId: route.turnId } : {}),
1203− ...(route.itemId ? { itemId: route.itemId } : {}),
1228+ ...(effectiveTurnId ? { turnId: effectiveTurnId } : {}),
1229+ ...(rawRoute.itemId ? { itemId: rawRoute.itemId } : {}),
12041230 };
12051231 context.pendingApprovals.set(requestId, pendingRequest);
12061232 }
@@ -1211,8 +1237,8 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 4 unmodified lines
12111237 requestId,
12121238 jsonRpcId: request.id,
12131239 threadId: context.session.threadId,
1214− ...(route.turnId ? { turnId: route.turnId } : {}),
1215− ...(route.itemId ? { itemId: route.itemId } : {}),
1240+ ...(effectiveTurnId ? { turnId: effectiveTurnId } : {}),
1241+ ...(rawRoute.itemId ? { itemId: rawRoute.itemId } : {}),
12161242 });
12171243 }
12181244
@@ -1223,8 +1249,8 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 4 unmodified lines
12231249 threadId: context.session.threadId,
12241250 createdAt: new Date().toISOString(),
12251251 method: request.method,
1226− turnId: route.turnId,
1227− itemId: route.itemId,
1252+ ...(effectiveTurnId ? { turnId: effectiveTurnId } : {}),
1253+ ...(rawRoute.itemId ? { itemId: rawRoute.itemId } : {}),
12281254 requestId,
12291255 requestKind,
12301256 payload: request.params,
@@ -1451,6 +1477,64 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve · 220 unmodified lines
14511477 return route;
14521478 }
14531479
1480+ private readProviderConversationId(params: unknown): string | undefined {
1481+ return (
1482+ this.readString(params, "threadId") ??
1483+ this.readString(this.readObject(params, "thread"), "id") ??
1484+ this.readString(params, "conversationId")
1485+ );
1486+ }
1487+
1488+ private readChildParentTurnId(context: CodexSessionContext, params: unknown): TurnId | undefined {
1489+ const providerConversationId = this.readProviderConversationId(params);
1490+ if (!providerConversationId) {
1491+ return undefined;
1492+ }
1493+ return context.collabReceiverTurns.get(providerConversationId);
1494+ }
1495+
1496+ private rememberCollabReceiverTurns(
1497+ context: CodexSessionContext,
1498+ params: unknown,
1499+ parentTurnId: TurnId | undefined,
1500+ ): void {
1501+ if (!parentTurnId) {
1502+ return;
1503+ }
1504+ const payload = this.readObject(params);
1505+ const item = this.readObject(payload, "item") ?? payload;
1506+ const itemType = this.readString(item, "type") ?? this.readString(item, "kind");
1507+ if (itemType !== "collabAgentToolCall") {
1508+ return;
1509+ }
1510+
1511+ const receiverThreadIds =
1512+ this.readArray(item, "receiverThreadIds")
1513+ ?.map((value) => (typeof value === "string" ? value : null))
1514+ .filter((value): value is string => value !== null) ?? [];
1515+ for (const receiverThreadId of receiverThreadIds) {
1516+ context.collabReceiverTurns.set(receiverThreadId, parentTurnId);
1517+ }
1518+ }
1519+
1520+ private shouldSuppressChildConversationNotification(method: string): boolean {
1521+ return (
1522+ method === "thread/started" ||
1523+ method === "thread/status/changed" ||
1524+ method === "thread/archived" ||
1525+ method === "thread/unarchived" ||
1526+ method === "thread/closed" ||
1527+ method === "thread/compacted" ||
1528+ method === "thread/name/updated" ||
1529+ method === "thread/tokenUsage/updated" ||
1530+ method === "turn/started" ||
1531+ method === "turn/completed" ||
1532+ method === "turn/aborted" ||
1533+ method === "turn/plan/updated" ||
1534+ method === "item/plan/delta"
1535+ );
1536+ }
1537+
14541538 private readObject(value: unknown, key?: string): Record<string, unknown> | undefined {
14551539 const target =
14561540 key === undefined
apps/server/src/provider/Layers/CodexAdapter.test.ts+39−0
@@ -800,6 +800,45 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { · 799 unmodified lines
800800 }
801801 }),
802802 );
803+
804+ it.effect("prefers manager-assigned turn ids for Codex task events", () =>
805+ Effect.gen(function* () {
806+ const adapter = yield* CodexAdapter;
807+ const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild);
808+
809+ lifecycleManager.emit("event", {
810+ id: asEventId("evt-codex-task-started-parent-turn"),
811+ kind: "notification",
812+ provider: "codex",
813+ threadId: asThreadId("thread-1"),
814+ turnId: asTurnId("turn-parent"),
815+ createdAt: new Date().toISOString(),
816+ method: "codex/event/task_started",
817+ payload: {
818+ id: "turn-child",
819+ msg: {
820+ type: "task_started",
821+ turn_id: "turn-child",
822+ collaboration_mode_kind: "default",
823+ },
824+ conversationId: "child-provider-thread",
825+ },
826+ } satisfies ProviderEvent);
827+
828+ const firstEvent = yield* Fiber.join(firstEventFiber);
829+ assert.equal(firstEvent._tag, "Some");
830+ if (firstEvent._tag !== "Some") {
831+ return;
832+ }
833+ assert.equal(firstEvent.value.type, "task.started");
834+ if (firstEvent.value.type !== "task.started") {
835+ return;
836+ }
837+ assert.equal(firstEvent.value.turnId, "turn-parent");
838+ assert.equal(firstEvent.value.providerRefs?.providerTurnId, "turn-parent");
839+ assert.equal(firstEvent.value.payload.taskId, "turn-child");
840+ }),
841+ );
803842 });
804843
805844 afterAll(() => {
apps/server/src/provider/Layers/CodexAdapter.ts+14−6
@@ -109,6 +109,14 @@ function asNumber(value: unknown): number | undefined { · 108 unmodified lines
109109 return typeof value === "number" && Number.isFinite(value) ? value : undefined;
110110 }
111111
112+function toTurnId(value: string | undefined): TurnId | undefined {
113+ return value?.trim() ? TurnId.makeUnsafe(value) : undefined;
114+}
115+
116+function toProviderItemId(value: string | undefined): ProviderItemId | undefined {
117+ return value?.trim() ? ProviderItemId.makeUnsafe(value) : undefined;
118+}
119+
112120 function toTurnStatus(value: unknown): "completed" | "failed" | "cancelled" | "interrupted" {
113121 switch (value) {
114122 case "completed":
@@ -415,27 +423,27 @@ function codexEventBase( · 300 unmodified lines
415423 ): Omit<ProviderRuntimeEvent, "type" | "payload"> {
416424 const payload = asObject(event.payload);
417425 const msg = codexEventMessage(payload);
418− const turnId = asString(msg?.turn_id) ?? asString(msg?.turnId);
419− const itemId = asString(msg?.item_id) ?? asString(msg?.itemId);
426+ const turnId = event.turnId ?? toTurnId(asString(msg?.turn_id) ?? asString(msg?.turnId));
427+ const itemId = event.itemId ?? toProviderItemId(asString(msg?.item_id) ?? asString(msg?.itemId));
420428 const requestId = asString(msg?.request_id) ?? asString(msg?.requestId);
421429 const base = runtimeEventBase(event, canonicalThreadId);
422430 const providerRefs = base.providerRefs
423431 ? {
424432 ...base.providerRefs,
425433 ...(turnId ? { providerTurnId: turnId } : {}),
426− ...(itemId ? { providerItemId: ProviderItemId.makeUnsafe(itemId) } : {}),
434+ ...(itemId ? { providerItemId: itemId } : {}),
427435 ...(requestId ? { providerRequestId: requestId } : {}),
428436 }
429437 : {
430438 ...(turnId ? { providerTurnId: turnId } : {}),
431− ...(itemId ? { providerItemId: ProviderItemId.makeUnsafe(itemId) } : {}),
439+ ...(itemId ? { providerItemId: itemId } : {}),
432440 ...(requestId ? { providerRequestId: requestId } : {}),
433441 };
434442
435443 return {
436444 ...base,
437− ...(turnId ? { turnId: TurnId.makeUnsafe(turnId) } : {}),
438− ...(itemId ? { itemId: asRuntimeItemId(ProviderItemId.makeUnsafe(itemId)) } : {}),
445+ ...(turnId ? { turnId } : {}),
446+ ...(itemId ? { itemId: asRuntimeItemId(itemId) } : {}),
439447 ...(requestId ? { requestId: asRuntimeRequestId(requestId) } : {}),
440448 ...(Object.keys(providerRefs).length > 0 ? { providerRefs } : {}),
441449 };