diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 0fc1b9a2bf82d..e0c7798005beb 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -6324,77 +6324,124 @@ describe("CodexAdapterV2 post-settle continuation", () => { ); }; - it.effect("contains descendant commands and keeps Stop authoritative", () => - Effect.gen(function* () { - const harness = yield* makeCodexReplayHarness(interruptSubagentCommandTranscript); - const now = yield* DateTime.now; + it.effect.each([false, true])( + "contains descendant commands after parent completion=%s", + (parentCompleted) => + Effect.gen(function* () { + const entries = [...interruptSubagentCommandTranscript.entries]; + if (parentCompleted) { + const stopIndex = entries.findIndex( + (entry) => entry.type === "expect_outbound" && entry.label === "turn/interrupt/root", + ); + entries.splice(stopIndex, 0, { + type: "emit_inbound", + label: "turn/completed/parent-before-stop", + frame: { + method: "turn/completed", + params: { + threadId: INTERRUPT_NATIVE_THREAD, + turn: makeCodexReplayTurn({ id: INTERRUPT_NATIVE_TURN, status: "completed" }), + }, + }, + }); + } + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "codex-interrupt-subagent-command", + entries: parentCompleted + ? entries + .filter( + (entry) => + !("label" in entry) || + (entry.label !== "turn/interrupt/root" && + entry.label !== "turn/completed/root"), + ) + .map((entry) => + "frame" in entry && + typeof entry.frame === "object" && + entry.frame !== null && + "id" in entry.frame && + typeof entry.frame.id === "number" && + entry.frame.id >= 5 + ? { ...entry, frame: { ...entry.frame, id: entry.frame.id - 1 } } + : entry, + ) + : entries, + }), + ); + const now = yield* DateTime.now; - yield* harness.runtime.startTurn( - makeCodexTestTurnInput({ - threadId: harness.threadId, + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-codex-interrupt-subagent-command"), + text: INTERRUPT_PROMPT, + }), + ); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && + event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM && + event.turnItem.status === "running", + ), + ); + const providerTurnId = harness.events.find( + ( + event, + ): event is Extract< + ProviderAdapter.ProviderAdapterV2Event, + { type: "provider_turn.updated" } + > => event.type === "provider_turn.updated" && event.threadId === harness.threadId, + )?.providerTurn.id; + assert.isDefined(providerTurnId); + if (parentCompleted) yield* harness.awaitTerminal(1); + + yield* harness.runtime.interruptTurn({ providerThread: harness.providerThread, - now, - attemptId: RunAttemptId.make("attempt-codex-interrupt-subagent-command"), - text: INTERRUPT_PROMPT, - }), - ); - yield* harness.until(() => - harness.events.some( - (event) => + providerTurnId, + }); + + yield* harness.awaitTerminal(1); + assert.equal( + harness.terminalEvents()[0]?.status, + parentCompleted ? "completed" : "interrupted", + ); + const childCommandUpdates = harness.events.filter( + ( + event, + ): event is Extract< + ProviderAdapter.ProviderAdapterV2Event, + { type: "turn_item.updated" } + > => event.type === "turn_item.updated" && event.turnItem.type === "command_execution" && - event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM && - event.turnItem.status === "running", - ), - ); - const providerTurnId = harness.events.find( - ( - event, - ): event is Extract< - ProviderAdapter.ProviderAdapterV2Event, - { type: "provider_turn.updated" } - > => event.type === "provider_turn.updated" && event.threadId === harness.threadId, - )?.providerTurn.id; - assert.isDefined(providerTurnId); - - yield* harness.runtime.interruptTurn({ - providerThread: harness.providerThread, - providerTurnId, - }); - - yield* harness.awaitTerminal(1); - assert.equal(harness.terminalEvents()[0]?.status, "interrupted"); - const childCommandUpdates = harness.events.filter( - ( - event, - ): event is Extract< - ProviderAdapter.ProviderAdapterV2Event, - { type: "turn_item.updated" } - > => - event.type === "turn_item.updated" && - event.turnItem.type === "command_execution" && - event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM, - ); - assert.equal(childCommandUpdates.at(-1)?.turnItem.status, "interrupted"); - assert.equal(harness.subagentUpdates().at(-1)?.subagent.status, "interrupted"); - assert.isDefined( - harness - .subagentUpdates() - .find((event) => event.subagent.status === "running" && event.subagent.progress), - "a running native subagent reports its child's current command as progress", - ); - assertChildProviderTerminalBeforeRoot(harness.events, harness.threadId); - assert.isFalse(yield* harness.hasPendingBackgroundWork); - }).pipe( - Effect.provide( - Layer.mergeAll( - IdAllocator.layer, - McpProviderSessions.layer, - NodeServices.layer, - NodeCrypto.layer, + event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM, + ); + assert.equal(childCommandUpdates.at(-1)?.turnItem.status, "interrupted"); + assert.equal(harness.subagentUpdates().at(-1)?.subagent.status, "interrupted"); + assert.isDefined( + harness + .subagentUpdates() + .find((event) => event.subagent.status === "running" && event.subagent.progress), + "a running native subagent reports its child's current command as progress", + ); + if (!parentCompleted) + assertChildProviderTerminalBeforeRoot(harness.events, harness.threadId); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), ), ), - ), ); const rootCompletionIndex = interruptSubagentCommandTranscript.entries.findIndex( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index e26b3dc5860c2..fdfa7cff125d8 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7971,6 +7971,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( activeTurnContexts.find( (candidate) => candidate.providerTurnId === turnInput.providerTurnId, ) ?? + activeTurnContexts + .map((context) => approvalOwnerCodexTurn(context).owner) + .find((candidate) => candidate.providerTurnId === turnInput.providerTurnId) ?? (turnInput.requestRuntimeRestart === true ? settledTurnContexts.find( (candidate) => candidate.providerThread.id === turnInput.providerThread.id, diff --git a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts index 4a4726bcb9bd6..c94ee3a9f1b9f 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, @@ -43,6 +44,7 @@ const stopEarlierBackgroundWork = ({ backgroundWorkAfterStop, stalledRun, workflowMember = false, + nativeAlreadyInterrupted = false, }: { readonly wakeAfterStop?: boolean; readonly failedStart?: boolean; @@ -50,6 +52,7 @@ const stopEarlierBackgroundWork = ({ readonly stopWithQueue?: "thread.stop" | "run.interrupt"; readonly olderStart?: boolean; readonly workflowMember?: boolean; + readonly nativeAlreadyInterrupted?: boolean | "item-only"; readonly stalledRun?: "missing-session-terminal" | "returned-interrupt" | "superseded-attempt"; }) => Effect.gen(function* () { @@ -58,6 +61,7 @@ const stopEarlierBackgroundWork = ({ const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = []; const interrupts: ProviderAdapter.ProviderAdapterV2InterruptInput[] = []; const interruptCalls = yield* Queue.unbounded(); + let beforeInterrupt = Effect.void; const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = { instanceId, driver, @@ -145,7 +149,16 @@ const stopEarlierBackgroundWork = ({ interruptTurn: (interrupt) => Effect.sync(() => { interrupts.push(interrupt); - }).pipe(Effect.andThen(Queue.offer(interruptCalls, undefined))), + }).pipe( + Effect.andThen( + Effect.suspend(() => { + const pending = beforeInterrupt; + beforeInterrupt = Effect.void; + return pending; + }), + ), + Effect.andThen(Queue.offer(interruptCalls, undefined)), + ), respondToRuntimeRequest: () => Effect.die("unused"), readThreadSnapshot: () => Effect.die("unused"), rollbackThread: () => Effect.die("unused"), @@ -527,6 +540,98 @@ const stopEarlierBackgroundWork = ({ }); const { threadId: reviewerThreadId, rootId: reviewerRootId } = yield* subagentThread("reviewer"); + const childRoot = (yield* orchestrator.getThreadProjection(reviewerThreadId)).nodes.find( + (node) => node.id === reviewerRootId, + )!; + for (const suffix of ["", ":user", ":parent", ":unrelated"] as const) { + const onParent = suffix === ":parent" || suffix === ":unrelated"; + const requestThreadId = onParent ? threadId : reviewerThreadId; + const requestRunId = onParent + ? RunId.make("run:3") + : suffix === ":user" + ? RunId.make("run:user-question") + : null; + const questionNodeId = NodeId.make(`node:child-question${suffix}`); + const questionRequestId = RuntimeRequestId.make(`request:child-question${suffix}`); + const questionItemId = TurnItemId.make(`item:child-question${suffix}`); + yield* sink.write({ + events: [ + { + id: EventId.make(`child-question-node${suffix}`), + type: "node.updated", + threadId: requestThreadId, + occurredAt: now, + payload: { + ...childRoot, + id: questionNodeId, + threadId: requestThreadId, + runId: requestRunId, + parentNodeId: onParent + ? NodeId.make(suffix === ":parent" ? "subagent:3" : "node:3") + : reviewerRootId, + kind: onParent ? "approval_request" : "user_input_request", + status: "waiting", + runtimeRequestId: questionRequestId, + }, + }, + { + id: EventId.make(`child-question-request${suffix}`), + type: "runtime-request.updated", + threadId: requestThreadId, + occurredAt: now, + payload: { + id: questionRequestId, + nodeId: questionNodeId, + providerTurnId: null, + nativeRequestRef: { driver, nativeId: "child-question", strength: "strong" }, + kind: onParent ? "command" : "user_input", + status: "pending", + responseCapability: { + type: "live", + providerSessionId: first.providerThread.providerSessionId!, + }, + createdAt: now, + resolvedAt: null, + }, + }, + { + id: EventId.make(`child-question-item${suffix}`), + type: "turn-item.updated", + threadId: requestThreadId, + occurredAt: now, + payload: { + id: questionItemId, + threadId: requestThreadId, + runId: requestRunId, + nodeId: questionNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "waiting", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + requestId: questionRequestId, + ...(onParent + ? { + type: "approval_request" as const, + requestKind: "command" as const, + prompt: "Allow network access?", + } + : { + type: "user_input_request" as const, + questions: [ + { id: "choice", header: "Choice", question: "Which option?", options: [] }, + ], + }), + }, + }, + ], + }); + } const tester = yield* subagentThread("tester"); const nestedSubagentId = NodeId.make("subagent:nested"); const nestedItemId = TurnItemId.make("turn-item:nested-subagent"); @@ -942,6 +1047,37 @@ const stopEarlierBackgroundWork = ({ }); } + if (nativeAlreadyInterrupted) { + const before = yield* orchestrator.getThreadProjection(threadId); + const task = before.subagents.find((candidate) => candidate.id === "subagent:3")!; + const item = before.turnItems.find((candidate) => candidate.id === reviewerId)!; + beforeInterrupt = sink + .write({ + events: [ + { + id: EventId.make("native-interrupted-before-settle"), + type: "subagent.updated", + threadId, + occurredAt: now, + payload: { + ...task, + ...(nativeAlreadyInterrupted === "item-only" + ? {} + : { status: "interrupted" as const, completedAt: now }), + }, + }, + { + id: EventId.make("native-item-interrupted-before-settle"), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { ...item, status: "interrupted", completedAt: now }, + }, + ], + }) + .pipe(Effect.asVoid, Effect.orDie); + } + yield* orchestrator.dispatch( stopWithQueue === "thread.stop" ? { @@ -987,8 +1123,29 @@ const stopEarlierBackgroundWork = ({ ); assert.equal(after.subagents[0]?.status, "interrupted"); const reviewerThread = yield* orchestrator.getThreadProjection(reviewerThreadId); - assert.equal(reviewerThread.nodes[0]?.status, "interrupted"); + assert.equal( + reviewerThread.nodes.find((node) => node.id === reviewerRootId)?.status, + "interrupted", + ); assert.equal(reviewerThread.subagents[0]?.status, "interrupted"); + for (const [suffix, statuses] of [ + ["", ["cancelled", "interrupted", "interrupted"]], + [":user", ["pending", "waiting", "waiting"]], + [":parent", ["cancelled", "interrupted", "interrupted"]], + [":unrelated", ["pending", "waiting", "waiting"]], + ] as const) { + const requestThread = + suffix === ":parent" || suffix === ":unrelated" ? after : reviewerThread; + const actual = [ + requestThread.runtimeRequests.find( + (request) => request.id === `request:child-question${suffix}`, + )?.status, + requestThread.nodes.find((node) => node.id === `node:child-question${suffix}`)?.status, + requestThread.turnItems.find((item) => item.id === `item:child-question${suffix}`) + ?.status, + ]; + assert.deepEqual(actual, [...statuses]); + } if (workflowMember) { assert.isUndefined(reviewerThread.turnItems.find((item) => item.id === nestedItemId)); assert.equal( @@ -1026,8 +1183,9 @@ const stopEarlierBackgroundWork = ({ ); }); -it.effect("Stop reaches background work an earlier provider thread still runs", () => - stopEarlierBackgroundWork({}), +it.effect.each([false, true, "item-only"] as const)( + "Stop reaches background work with native already interrupted=%s", + (nativeAlreadyInterrupted) => stopEarlierBackgroundWork({ nativeAlreadyInterrupted }), ); it.effect("Stop settles workflow members without subagent turn items", () => diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c732e0ca67ccf..2a5ea4315aba4 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11923,6 +11923,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio threadId: input.command.threadId, commandId: input.command.commandId, }); + const emitEvent = emit(input.events, input.command); const dying = [...input.dying]; const { childThreadIds } = input; for (const { projection, subagent } of dying) { @@ -11945,11 +11946,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } childThreadIds.add(childThreadId); const child = yield* projectionStore - .getThreadRecords(childThreadId, ["nodes", "providerTurns", "subagents", "turnItems"], { - // Runless items are the subagent's; any run there is the user's own. - turnItemRunIds: [null], - turnItemStatuses: ["pending", "running", "waiting"], - }) + .getThreadRecords( + childThreadId, + ["nodes", "providerTurns", "subagents", "turnItems", "runtimeRequests"], + { + runtimeRequestStatuses: ["pending"], + // Runless items are the subagent's; any run there is the user's own. + turnItemRunIds: [null], + turnItemStatuses: ["pending", "running", "waiting"], + }, + ) .pipe( Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }), mapDispatchError(input.command), @@ -11965,6 +11971,76 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio allocateEventId, }).pipe(mapDispatchError(input.command)); yield* Ref.update(input.events, (events) => [...events, ...childEnded]); + const parentThreadId = + projection.subagents.find((candidate) => candidate.id === subagent.subagentId) + ?.threadId ?? + projection.turnItems.find( + (item) => item.type === "subagent" && item.subagentId === subagent.subagentId, + )?.threadId; + const parent = + parentThreadId === undefined + ? null + : yield* projectionStore + .getThreadRecords(parentThreadId, ["nodes", "runtimeRequests", "turnItems"], { + runtimeRequestStatuses: ["pending"], + turnItemTypes: ["approval_request"], + turnItemStatuses: ["waiting"], + }) + .pipe(mapDispatchError(input.command)); + for (const owner of parent === null ? [child] : [child, parent]) { + for (const request of owner.runtimeRequests) { + if (request.status !== "pending") continue; + const node = owner.nodes.find((candidate) => candidate.id === request.nodeId); + if (node === undefined) continue; + // Codex puts child approvals on the parent; runs on the child belong to the user. + if (owner === child ? node.runId !== null : node.parentNodeId !== subagent.subagentId) + continue; + const scope = { + threadId: node.threadId, + nodeId: node.id, + providerInstanceId: subagent.providerInstanceId, + occurredAt: input.now, + }; + yield* emitEvent({ + ...scope, + type: "runtime-request.updated", + payload: { + ...request, + status: "cancelled", + resolvedAt: input.now, + responseCapability: { + type: "not_resumable", + reason: "The subagent was interrupted.", + }, + }, + }); + if (isOrchestrationV2WorkActive(node.status)) { + yield* emitEvent({ + ...scope, + type: "node.updated", + payload: { ...node, status: "interrupted", completedAt: input.now }, + }); + } + for (const card of owner.turnItems) { + if ( + (card.type !== "approval_request" && card.type !== "user_input_request") || + card.requestId !== request.id || + card.status !== "waiting" + ) + continue; + yield* emitEvent({ + ...scope, + type: "turn-item.updated", + payload: { + ...card, + status: "interrupted", + completedAt: input.now, + updatedAt: input.now, + }, + }); + } + } + } for (const nested of child.turnItems) { if (nested.type === "subagent") dying.push({ projection: child, subagent: nested }); } @@ -12066,6 +12142,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio now: input.now, }); } + // Native task records and their timeline items can settle in either order. + for (const subagent of input.projection.subagents) { + if ( + subagent.origin !== "provider_native" || + (!isOrchestrationV2WorkActive(subagent.status) && subagent.status !== "interrupted") + ) + continue; + const run = input.projection.runs.find((candidate) => candidate.id === subagent.runId); + if ( + run === undefined || + run.ordinal > input.throughRunOrdinal || + run.providerThreadId !== input.stoppedProviderThreadId + ) + continue; + yield* endNativeSubagentTrees({ + command: input.command, + events: input.events, + dying: [{ projection: input.projection, subagent: nativeSubagentRef(subagent) }], + childThreadIds, + now: input.now, + }); + } for (const providerThread of input.projection.providerThreads) { if ((providerThread.pendingBackgroundTasks?.length ?? 0) === 0) continue; if (providerThread.id === input.stoppedProviderThreadId) {