From bf266b7f7b8702cbad9b3e9bf04fa9acf1efab92 Mon Sep 17 00:00:00 2001 From: Alex Southwell <4596216+saphid@users.noreply.github.com> Date: Thu, 5 Nov 2026 19:17:19 +1100 Subject: [PATCH 1/2] fix(server): Stop closes native subagent question cards Cancel pending runless child requests alongside their stopped native subagent. Preserve requests associated with user runs on the child thread. Co-Authored-By: GPT-6.1 Sol --- .../BackgroundWorkStop.integration.test.ts | 95 ++++++++++++++++++- .../src/orchestration-v2/Orchestrator.ts | 65 ++++++++++++- 2 files changed, 154 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts index ffe8203c8cbbf..16182d56fb9c4 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, @@ -520,6 +521,81 @@ const stopEarlierBackgroundWork = ({ }); const { threadId: reviewerThreadId, rootId: reviewerRootId } = yield* subagentThread("reviewer"); + const childRoot = (yield* orchestrator.getThreadProjection(reviewerThreadId)).nodes.find( + (node) => node.id === reviewerRootId, + )!; + for (const userRun of [false, true]) { + const suffix = userRun ? ":user" : ""; + 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: reviewerThreadId, + occurredAt: now, + payload: { + ...childRoot, + id: questionNodeId, + runId: userRun ? RunId.make("run:user-question") : null, + parentNodeId: reviewerRootId, + kind: "user_input_request", + status: "waiting", + runtimeRequestId: questionRequestId, + }, + }, + { + id: EventId.make(`child-question-request${suffix}`), + type: "runtime-request.updated", + threadId: reviewerThreadId, + occurredAt: now, + payload: { + id: questionRequestId, + nodeId: questionNodeId, + providerTurnId: null, + nativeRequestRef: { driver, nativeId: "child-question", strength: "strong" }, + kind: "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: reviewerThreadId, + occurredAt: now, + payload: { + id: questionItemId, + threadId: reviewerThreadId, + runId: userRun ? RunId.make("run:user-question") : null, + nodeId: questionNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "waiting", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "user_input_request", + requestId: questionRequestId, + 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"); @@ -987,8 +1063,25 @@ 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"]], + ] as const) { + const actual = [ + reviewerThread.runtimeRequests.find( + (request) => request.id === `request:child-question${suffix}`, + )?.status, + reviewerThread.nodes.find((node) => node.id === `node:child-question${suffix}`)?.status, + reviewerThread.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( diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index a6b876db1de96..5199b299dd28c 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -10514,11 +10514,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), @@ -10534,6 +10539,56 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio allocateEventId, }).pipe(mapDispatchError(input.command)); yield* Ref.update(input.events, (events) => [...events, ...childEnded]); + for (const request of child.runtimeRequests) { + if (request.status !== "pending") continue; + const node = child.nodes.find((candidate) => candidate.id === request.nodeId); + // Runs on a native child's thread belong to the user, not the stopped subagent. + if (node === undefined || node.runId !== null) continue; + const scope = { + threadId: childThreadId, + 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 child.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 }); } From 02f2a79c3bc6f16bb8df421ffccbbaa6b799bab4 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 15 Nov 2026 03:46:23 +0000 Subject: [PATCH 2/2] fix(server): stop clears native child approvals after parent completion --- .../Adapters/CodexAdapterV2.test.ts | 177 +++++++++++------- .../Adapters/CodexAdapterV2.ts | 3 + .../BackgroundWorkStop.integration.test.ts | 107 ++++++++--- .../src/orchestration-v2/Orchestrator.ts | 126 ++++++++----- 4 files changed, 285 insertions(+), 128 deletions(-) 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 75044d1d49abc..c94ee3a9f1b9f 100644 --- a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts @@ -44,6 +44,7 @@ const stopEarlierBackgroundWork = ({ backgroundWorkAfterStop, stalledRun, workflowMember = false, + nativeAlreadyInterrupted = false, }: { readonly wakeAfterStop?: boolean; readonly failedStart?: boolean; @@ -51,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* () { @@ -59,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, @@ -146,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"), @@ -531,8 +543,14 @@ const stopEarlierBackgroundWork = ({ const childRoot = (yield* orchestrator.getThreadProjection(reviewerThreadId)).nodes.find( (node) => node.id === reviewerRootId, )!; - for (const userRun of [false, true]) { - const suffix = userRun ? ":user" : ""; + 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}`); @@ -541,14 +559,17 @@ const stopEarlierBackgroundWork = ({ { id: EventId.make(`child-question-node${suffix}`), type: "node.updated", - threadId: reviewerThreadId, + threadId: requestThreadId, occurredAt: now, payload: { ...childRoot, id: questionNodeId, - runId: userRun ? RunId.make("run:user-question") : null, - parentNodeId: reviewerRootId, - kind: "user_input_request", + 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, }, @@ -556,14 +577,14 @@ const stopEarlierBackgroundWork = ({ { id: EventId.make(`child-question-request${suffix}`), type: "runtime-request.updated", - threadId: reviewerThreadId, + threadId: requestThreadId, occurredAt: now, payload: { id: questionRequestId, nodeId: questionNodeId, providerTurnId: null, nativeRequestRef: { driver, nativeId: "child-question", strength: "strong" }, - kind: "user_input", + kind: onParent ? "command" : "user_input", status: "pending", responseCapability: { type: "live", @@ -576,12 +597,12 @@ const stopEarlierBackgroundWork = ({ { id: EventId.make(`child-question-item${suffix}`), type: "turn-item.updated", - threadId: reviewerThreadId, + threadId: requestThreadId, occurredAt: now, payload: { id: questionItemId, - threadId: reviewerThreadId, - runId: userRun ? RunId.make("run:user-question") : null, + threadId: requestThreadId, + runId: requestRunId, nodeId: questionNodeId, providerThreadId: null, providerTurnId: null, @@ -593,11 +614,19 @@ const stopEarlierBackgroundWork = ({ startedAt: now, completedAt: null, updatedAt: now, - type: "user_input_request", requestId: questionRequestId, - questions: [ - { id: "choice", header: "Choice", question: "Which option?", options: [] }, - ], + ...(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: [] }, + ], + }), }, }, ], @@ -1018,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" ? { @@ -1071,13 +1131,17 @@ const stopEarlierBackgroundWork = ({ 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 = [ - reviewerThread.runtimeRequests.find( + requestThread.runtimeRequests.find( (request) => request.id === `request:child-question${suffix}`, )?.status, - reviewerThread.nodes.find((node) => node.id === `node:child-question${suffix}`)?.status, - reviewerThread.turnItems.find((item) => item.id === `item:child-question${suffix}`) + 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]); @@ -1119,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 deec39bfbb80e..2a5ea4315aba4 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11971,54 +11971,74 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio allocateEventId, }).pipe(mapDispatchError(input.command)); yield* Ref.update(input.events, (events) => [...events, ...childEnded]); - for (const request of child.runtimeRequests) { - if (request.status !== "pending") continue; - const node = child.nodes.find((candidate) => candidate.id === request.nodeId); - // Runs on a native child's thread belong to the user, not the stopped subagent. - if (node === undefined || node.runId !== null) continue; - const scope = { - threadId: childThreadId, - 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 child.turnItems) { - if ( - (card.type !== "approval_request" && card.type !== "user_input_request") || - card.requestId !== request.id || - card.status !== "waiting" - ) + 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: "turn-item.updated", + type: "runtime-request.updated", payload: { - ...card, - status: "interrupted", - completedAt: input.now, - updatedAt: input.now, + ...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) { @@ -12122,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) {