diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 0fc1b9a2bf82d..df827e3e74fef 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1933,7 +1933,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { }), ); - it.effect("waits for native start before interrupting an acknowledged queued turn", () => + it.effect("waits for native start before Stop", () => Effect.gen(function* () { const nativeThreadId = "early-stop-thread"; const nativeTurnId = "early-stop-turn"; @@ -1985,7 +1985,10 @@ describe("CodexAdapterV2 post-settle continuation", () => { method: "turn/completed", params: { threadId: nativeThreadId, - turn: makeCodexReplayTurn({ id: nativeTurnId, status: "interrupted" }), + turn: makeCodexReplayTurn({ + id: nativeTurnId, + status: "interrupted", + }), }, }, }, diff --git a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts index 4a4726bcb9bd6..2faa33ec8fcba 100644 --- a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts @@ -11,6 +11,7 @@ import { ProviderTurnId, RunAttemptId, RunId, + RuntimeRequestId, ThreadId, TurnItemId, type OrchestrationV2DomainEvent, @@ -385,6 +386,79 @@ const stopEarlierBackgroundWork = ({ }, ], }); + // Shaped like CodexAdapterV2's buildApprovalRequestArtifacts output. + const approvalRequestId = RuntimeRequestId.make("request:stalled-approval"); + const approvalNodeId = NodeId.make("node:stalled-approval"); + const approvalItemId = TurnItemId.make("turn-item:stalled-approval"); + if (stalledRun === "returned-interrupt") { + yield* sink.write({ + events: [ + { + id: EventId.make("stalled-approval-node"), + type: "node.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + ...node, + id: approvalNodeId, + parentNodeId: node.id, + kind: "approval_request", + status: "waiting", + countsForRun: false, + runtimeRequestId: approvalRequestId, + }, + }, + { + id: EventId.make("stalled-approval-request"), + type: "runtime-request.updated", + threadId, + occurredAt: now, + payload: { + id: approvalRequestId, + nodeId: approvalNodeId, + providerTurnId: codexTurn.id, + nativeRequestRef: { driver, nativeId: "approval", strength: "strong" }, + kind: "command", + status: "pending", + responseCapability: { + type: "live", + providerSessionId: first.providerThread.providerSessionId!, + }, + createdAt: now, + resolvedAt: null, + }, + }, + { + id: EventId.make("stalled-approval-item"), + type: "turn-item.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + id: approvalItemId, + threadId, + runId: run.id, + nodeId: approvalNodeId, + providerThreadId: codexTurn.providerThreadId, + providerTurnId: codexTurn.id, + nativeItemRef: null, + parentItemId: null, + ordinal: 101, + status: "waiting", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "approval_request", + requestId: approvalRequestId, + requestKind: "command", + prompt: "rm -rf build", + }, + }, + ], + }); + } if (stalledRun === "superseded-attempt") { yield* sink.write({ events: [ @@ -446,6 +520,16 @@ const stopEarlierBackgroundWork = ({ interrupted ? 1 : 0, ); assert.isEmpty(after.runs.filter((candidate) => candidate.status === "waiting")); + if (stalledRun === "returned-interrupt") { + assert.deepEqual( + [ + after.runtimeRequests.find((candidate) => candidate.id === approvalRequestId)?.status, + after.nodes.find((candidate) => candidate.id === approvalNodeId)?.status, + after.turnItems.find((candidate) => candidate.id === approvalItemId)?.status, + ], + ["cancelled", "interrupted", "interrupted"], + ); + } return; } const settled = yield* watch( diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c732e0ca67ccf..00a0473a0cc02 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12213,6 +12213,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio .filter((subagent) => subagent.origin === "app_owned") .map((subagent) => subagent.id), ); + // Callers load only tool and message items; the request cards close with the requests below. + const requestCards = yield* projectionStore + .getThreadRecords(run.threadId, ["turnItems"], { + turnItemTypes: ["approval_request", "user_input_request"], + turnItemRunIds: [run.id], + turnItemStatuses: ["waiting"], + }) + .pipe(mapDispatchError(input.command)); const cascaded = yield* RunExecutionService.cascadeTerminalizeRunOwnedSubagents({ run, status: "interrupted", @@ -12235,13 +12243,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), turnItems: new Map(), childTurnItems: new Map( - input.projection.turnItems - .filter( + [ + ...input.projection.turnItems.filter( (item) => item.runId === run.id && (item.type === "assistant_message" || item.type === "reasoning"), - ) - .map((item) => [item.id, item]), + ), + ...requestCards.turnItems, + ].map((item) => [item.id, item]), ), linkedChildThreadIds: new Set([run.threadId]), },