From 39755f613ee7ea62295b3f23e7c59ceac10745f6 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 25 Oct 2026 14:20:09 -0700 Subject: [PATCH 1/4] fix(server): ended roots no longer leave provider turns running Co-Authored-By: Claude Opus 5.5 --- .../ProjectionRecovery.test.ts | 100 ++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 24 +++-- .../RunExecutionService.test.ts | 100 ++++++++++++++++-- .../orchestration-v2/RunExecutionService.ts | 34 +++--- .../src/server/subagentProjection.ts | 23 +++- 5 files changed, 254 insertions(+), 27 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 6e0dd990ce388..6ed24558f01dc 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -12,6 +12,7 @@ import { ProviderInstanceId, ProviderSessionId, ProviderThreadId, + ProviderTurnId, RunId, RuntimeRequestId, ThreadId, @@ -966,3 +967,102 @@ it.effect("marks fork descendants unreadable when their source is missing or cor ); }).pipe(Effect.provide(layerTest)), ); + +it.effect("startup ends the provider turn of a native child root it cancels", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; + const now = yield* DateTime.now; + const parentThreadId = yield* createThread("native-child-root-turn-parent"); + const threadId = yield* createThread("native-child-root-turn"); + const rootNodeId = NodeId.make(`node:${threadId}:thread-root`); + const providerThreadId = ProviderThreadId.make(`provider-thread:${threadId}`); + const providerTurnId = ProviderTurnId.make(`provider-turn:${threadId}`); + yield* projections.apply({ + id: EventId.make(`event:${threadId}:root`), + type: "node.updated", + threadId, + nodeId: rootNodeId, + driver, + occurredAt: now, + payload: { + id: rootNodeId, + threadId, + runId: null, + parentNodeId: null, + rootNodeId, + kind: "root_turn", + status: "running", + countsForRun: false, + providerThreadId, + providerTurnId, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }, + }); + // A native subagent's turn belongs to no run attempt. + yield* projections.apply({ + id: EventId.make(`event:${threadId}:provider-turn`), + type: "provider-turn.updated", + threadId, + nodeId: rootNodeId, + driver, + occurredAt: now, + payload: { + id: providerTurnId, + providerThreadId, + nodeId: rootNodeId, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + + const subagentId = NodeId.make(`node:${parentThreadId}:subagent`); + yield* projections.apply({ + id: EventId.make(`event:${parentThreadId}:subagent`), + type: "subagent.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { + id: subagentId, + threadId: parentThreadId, + runId: null, + parentNodeId: NodeId.make(`node:${parentThreadId}:root`), + origin: "provider_native", + createdBy: "agent", + driver, + providerInstanceId, + providerThreadId, + childThreadId: threadId, + nativeTaskRef: null, + title: null, + model: null, + prompt: "sleep 300", + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, + }); + + yield* recovery.reconcile("startup"); + + const projection = yield* projections.getThreadProjection(threadId); + assert.deepEqual( + { + root: projection.nodes.find((node) => node.id === rootNodeId)?.status, + providerTurn: projection.providerTurns.find((turn) => turn.id === providerTurnId)?.status, + }, + { root: "cancelled", providerTurn: "cancelled" }, + "a root startup recovery cancelled left its provider turn running", + ); + }).pipe(Effect.provide(layerRecoveryTest)), +); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 700395facd79d..3c26ae3747179 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4605,14 +4605,26 @@ export const layer: Layer.Layer = FROM orchestration_v2_projection_provider_turns AS provider_turn WHERE provider_turn.thread_id = ${threadId} AND provider_turn.status IN ('pending', 'starting', 'running', 'waiting') - AND provider_turn.run_attempt_id IN ( - SELECT attempt_id FROM orchestration_v2_projection_run_attempts - WHERE thread_id = ${threadId} - AND run_id IN ( - SELECT run_id FROM orchestration_v2_projection_runs + AND ( + provider_turn.run_attempt_id IN ( + SELECT attempt_id FROM orchestration_v2_projection_run_attempts + WHERE thread_id = ${threadId} + AND run_id IN ( + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} + AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + ) + ) + OR ( + provider_turn.run_attempt_id IS NULL + AND provider_turn.node_id IN ( + SELECT node_id FROM orchestration_v2_projection_nodes WHERE thread_id = ${threadId} - AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + AND run_id IS NULL + AND kind = 'root_turn' + AND status IN ('pending', 'running', 'waiting') ) + ) ) ORDER BY provider_turn.provider_thread_id ASC, provider_turn.ordinal ASC `, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 03bbbf0506b50..9ad3be4c81d5f 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -11,6 +11,7 @@ import { type OrchestrationV2DomainEvent, type OrchestrationV2ExecutionNode, type OrchestrationV2ProviderThread, + type OrchestrationV2ProviderTurn, type OrchestrationV2Run, type OrchestrationV2RunAttempt, type OrchestrationV2Subagent, @@ -2613,17 +2614,102 @@ it.effect.each(["failure", "defect", "interrupt-failure", "ownership-lost"] as c ); assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); const repaired = finalEvents.find((event) => event.type === "provider-turn.updated"); - if (failureMode === "interrupt-failure") { - assert.isUndefined(repaired); - } else { - assert.equal(repaired?.payload.id, ids.rootProviderTurnId); - assert.equal(repaired?.payload.status, "interrupted"); - assert.isNotNull(repaired?.payload.completedAt); - } + assert.equal(repaired?.payload.id, ids.rootProviderTurnId); + assert.equal( + repaired?.payload.status, + failureMode === "interrupt-failure" ? "failed" : "interrupted", + ); + assert.isNotNull(repaired?.payload.completedAt); } }), ); +it.effect.each([ + { status: "completed", failure: "later-stream" }, + { status: "failed", failure: "later-stream" }, + { status: "completed", failure: "terminal-ingest" }, +] as const)( + "keeps a $status root provider turn durable after a $failure failure", + ({ status, failure }) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const terminalAt = DateTime.subtract(now, { seconds: 1 }); + const key = `terminal-snapshot:${status}:${failure}`; + const durableTurns = yield* Ref.make< + ReadonlyMap + >(new Map()); + const persist = (turn: OrchestrationV2ProviderTurn) => + Ref.update(durableTurns, (current) => new Map(current).set(turn.id, turn)); + const interrupts = yield* Ref.make< + ReadonlyArray + >([]); + const ids = backgroundScenarioIds(key); + const rootTurn = ( + turnStatus: OrchestrationV2ProviderTurn["status"], + completedAt: DateTime.Utc | null, + ) => ({ + type: "provider_turn.updated" as const, + driver, + providerTurn: { + id: ids.rootProviderTurnId, + providerThreadId: ids.providerThreadId, + nodeId: ids.rootNodeId, + runAttemptId: ids.attemptId, + nativeTurnRef: null, + ordinal: 1, + status: turnStatus, + startedAt: terminalAt, + completedAt, + }, + }); + const { observed, finalEvents } = yield* captureRootRunTermination({ + key, + shouldFinalizeRun: () => Effect.succeed(true), + events: () => + Stream.make(rootTurn("running", null), rootTurn(status, terminalAt)).pipe( + Stream.concat( + failure === "later-stream" + ? Stream.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make(`session:${key}`), + cause: "provider process exited", + }), + ) + : Stream.empty, + ), + ), + ingestNormalized: ({ event }) => + event.type !== "provider_turn.updated" + ? Effect.succeed([]) + : failure === "terminal-ingest" && event.providerTurn.status !== "running" + ? Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: ProviderSessionId.make(`session:${key}`), + eventCount: 1, + cause: new Error("SQLITE_FULL: database or disk is full"), + }), + ) + : persist(event.providerTurn).pipe(Effect.as([])), + interruptTurn: (input) => Ref.update(interrupts, (current) => [...current, input]), + beforeTerminalWrite: (request) => + Effect.forEach(request.events, (event) => + event.type === "provider-turn.updated" ? persist(event.payload) : Effect.void, + ).pipe(Effect.as(true)), + }); + const calls = yield* Ref.get(interrupts); + assert.equal(calls.length, 1); + assert.isTrue(calls[0]?.requestRuntimeRestart); + assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); + assert.isTrue(finalEvents.some((event) => event.type === "run.updated")); + const durable = (yield* Ref.get(durableTurns)).get(ids.rootProviderTurnId); + assert.deepEqual( + { status: durable?.status, completedAt: durable?.completedAt }, + { status, completedAt: terminalAt }, + ); + }), +); + it.effect("does not interrupt an ended provider when its terminal write fails", () => Effect.gen(function* () { const interrupts = yield* Ref.make(0); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 9ec61cc85b994..6209a8fa387c2 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -620,7 +620,8 @@ export const layer: Layer.Layer< readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly loadRunInterruptReason?: () => Effect.Effect; readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection; - readonly interruptedProviderTurn?: OrchestrationV2ProviderTurn; + // Only when T3 ends the run itself: the provider never ends this turn. + readonly abandonedProviderTurn?: OrchestrationV2ProviderTurn; readonly terminal: ProviderTerminalEvent; readonly completedAt?: DateTime.Utc; readonly failureItemPersisted: boolean; @@ -747,7 +748,7 @@ export const layer: Layer.Layer< : [], events: [ ...cascadedSubagentEvents, - ...(input.interruptedProviderTurn === undefined + ...(input.abandonedProviderTurn === undefined ? [] : [ { @@ -758,11 +759,7 @@ export const layer: Layer.Layer< nodeId: input.rootNode.id, providerInstanceId: input.run.providerInstanceId, occurredAt: completedAt, - payload: { - ...input.interruptedProviderTurn, - status: "interrupted" as const, - completedAt, - }, + payload: input.abandonedProviderTurn, }, ]), ...(finalizedAttempt === null @@ -1482,8 +1479,11 @@ export const layer: Layer.Layer< if (yield* Ref.get(rootRunFinalized)) return; const providerThread = yield* Ref.get(latestProviderThread); const routing = yield* Ref.get(eventRouting); - // Stop background work too; the failed consumer cannot persist terminal events. - const interruptedProviderTurn = + const rootProviderTurn = yield* Ref.get(latestRootProviderTurn); + // Stop background work too; the failed consumer cannot persist terminal + // events. The turn still ends with the run when the interrupt fails or a + // lost process never answers, so it cannot stay running. + const abandonedProviderTurn = routing.rootProviderTurnId !== null && !routing.rootTurnEnded ? yield* input.session .interruptTurn({ @@ -1492,14 +1492,24 @@ export const layer: Layer.Layer< requestRuntimeRestart: true, }) .pipe( - Effect.as(yield* Ref.get(latestRootProviderTurn)), + Effect.timeout("5 seconds"), + Effect.as("interrupted" as const), Effect.catchCauseIf( (interruptCause) => !Cause.hasInterruptsOnly(interruptCause), (interruptCause) => Effect.logWarning( "orchestration V2 failed to interrupt turn after ingestion failure", { runId: input.run.id, cause: interruptCause }, - ).pipe(Effect.as(null)), + ).pipe(Effect.as("failed" as const)), + ), + // The snapshot is cached before it is published, so it may not be + // stored. Replay a terminal snapshot as-is; only active turns end here. + Effect.map((status) => + rootProviderTurn === null + ? null + : isTerminalProviderTurnStatus(rootProviderTurn.status) + ? rootProviderTurn + : { ...rootProviderTurn, status, completedAt }, ), ) : null; @@ -1517,7 +1527,7 @@ export const layer: Layer.Layer< expectedStatus: "running", }, openRunOwnedSubagents: openSubagents, - ...(interruptedProviderTurn === null ? {} : { interruptedProviderTurn }), + ...(abandonedProviderTurn === null ? {} : { abandonedProviderTurn }), completedAt, terminal: makeFailedTerminalEvent( makeProviderFailure({ cause: Cause.squash(cause), class: "unknown" }), diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index 9393d3a551878..75612cfd538d9 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -271,7 +271,7 @@ export function delegatedTaskProgress(projection: { function writtenIds( emitted: ReadonlyArray, - type: "subagent.updated" | "node.updated" | "turn-item.updated", + type: "subagent.updated" | "node.updated" | "turn-item.updated" | "provider-turn.updated", ): ReadonlySet { return new Set(emitted.flatMap((event) => (event.type === type ? [event.payload.id] : []))); } @@ -359,7 +359,8 @@ export const endRunlessRootTurns = Effect.fn("subagentProjection.endRunlessRootT >(input: { readonly threadId: ThreadId; readonly providerInstanceId: ProviderInstanceId; - readonly projection: Pick; + readonly projection: Pick & + Partial>; readonly status: "completed" | "failed" | "interrupted" | "cancelled"; readonly completedAt?: DateTime.Utc; readonly now: DateTime.Utc; @@ -369,6 +370,7 @@ export const endRunlessRootTurns = Effect.fn("subagentProjection.endRunlessRootT const { now, status, completedAt = now } = input; const endedNodeIds = writtenIds(input.emitted, "node.updated"); const endedItemIds = writtenIds(input.emitted, "turn-item.updated"); + const endedProviderTurnIds = writtenIds(input.emitted, "provider-turn.updated"); const scope = { threadId: input.threadId, providerInstanceId: input.providerInstanceId, @@ -421,6 +423,23 @@ export const endRunlessRootTurns = Effect.fn("subagentProjection.endRunlessRootT }, }); } + for (const providerTurn of input.projection.providerTurns ?? []) { + if ( + providerTurn.nodeId !== node.id || + providerTurn.runAttemptId !== null || + !isOrchestrationV2WorkActive(providerTurn.status) || + endedProviderTurnIds.has(providerTurn.id) + ) { + continue; + } + events.push({ + ...scope, + id: yield* input.allocateEventId(), + type: "provider-turn.updated", + nodeId: node.id, + payload: { ...providerTurn, status, completedAt }, + }); + } } return events; }); From 9ed56dc40eb156bf8573a7d814dda318c5f4af95 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 25 Oct 2026 14:28:05 -0700 Subject: [PATCH 2/4] fix(server): keep failed interrupts retryable and end native child provider turns Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 4 ++-- .../RunExecutionService.test.ts | 13 +++++----- .../orchestration-v2/RunExecutionService.ts | 24 +++++++++++-------- .../SubagentProjection.test.ts | 1 + .../src/orchestration-v2/ThreadStop.test.ts | 23 ++++++++++++++++++ .../src/server/subagentProjection.ts | 8 ++++--- 6 files changed, 52 insertions(+), 21 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index aae5d1b68db83..3a9cb4facbd38 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -9112,7 +9112,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } childThreadIds.add(childThreadId); const child = yield* projectionStore - .getThreadRecords(childThreadId, ["nodes", "subagents", "turnItems"], { + .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"], @@ -9671,7 +9671,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* Ref.update(events, (current) => [...current, ...ended]); if (subagent.childThreadId !== null) { const child = yield* projectionStore - .getThreadRecords(subagent.childThreadId, ["nodes", "turnItems"], { + .getThreadRecords(subagent.childThreadId, ["nodes", "providerTurns", "turnItems"], { turnItemRunIds: [null], turnItemStatuses: ["pending", "running", "waiting"], }) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 9ad3be4c81d5f..a05e64b4b05e4 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -2614,12 +2614,13 @@ it.effect.each(["failure", "defect", "interrupt-failure", "ownership-lost"] as c ); assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); const repaired = finalEvents.find((event) => event.type === "provider-turn.updated"); - assert.equal(repaired?.payload.id, ids.rootProviderTurnId); - assert.equal( - repaired?.payload.status, - failureMode === "interrupt-failure" ? "failed" : "interrupted", - ); - assert.isNotNull(repaired?.payload.completedAt); + if (failureMode === "interrupt-failure") { + assert.isUndefined(repaired); + } else { + assert.equal(repaired?.payload.id, ids.rootProviderTurnId); + assert.equal(repaired?.payload.status, "interrupted"); + assert.isNotNull(repaired?.payload.completedAt); + } } }), ); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 6209a8fa387c2..f93651cf86e96 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -620,7 +620,7 @@ export const layer: Layer.Layer< readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly loadRunInterruptReason?: () => Effect.Effect; readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection; - // Only when T3 ends the run itself: the provider never ends this turn. + // Root turn snapshot written with the run when T3 ends the run itself. readonly abandonedProviderTurn?: OrchestrationV2ProviderTurn; readonly terminal: ProviderTerminalEvent; readonly completedAt?: DateTime.Utc; @@ -1480,9 +1480,7 @@ export const layer: Layer.Layer< const providerThread = yield* Ref.get(latestProviderThread); const routing = yield* Ref.get(eventRouting); const rootProviderTurn = yield* Ref.get(latestRootProviderTurn); - // Stop background work too; the failed consumer cannot persist terminal - // events. The turn still ends with the run when the interrupt fails or a - // lost process never answers, so it cannot stay running. + // Stop background work too; the failed consumer cannot persist terminal events. const abandonedProviderTurn = routing.rootProviderTurnId !== null && !routing.rootTurnEnded ? yield* input.session @@ -1492,24 +1490,30 @@ export const layer: Layer.Layer< requestRuntimeRestart: true, }) .pipe( - Effect.timeout("5 seconds"), - Effect.as("interrupted" as const), + Effect.as(true), Effect.catchCauseIf( (interruptCause) => !Cause.hasInterruptsOnly(interruptCause), (interruptCause) => Effect.logWarning( "orchestration V2 failed to interrupt turn after ingestion failure", { runId: input.run.id, cause: interruptCause }, - ).pipe(Effect.as("failed" as const)), + ).pipe(Effect.as(false)), ), // The snapshot is cached before it is published, so it may not be - // stored. Replay a terminal snapshot as-is; only active turns end here. - Effect.map((status) => + // stored: replay a terminal one as-is. An active turn whose interrupt + // failed stays running so the next start retries the interrupt. + Effect.map((interrupted) => rootProviderTurn === null ? null : isTerminalProviderTurnStatus(rootProviderTurn.status) ? rootProviderTurn - : { ...rootProviderTurn, status, completedAt }, + : interrupted + ? { + ...rootProviderTurn, + status: "interrupted" as const, + completedAt, + } + : null, ), ) : null; diff --git a/apps/server/src/orchestration-v2/SubagentProjection.test.ts b/apps/server/src/orchestration-v2/SubagentProjection.test.ts index e87fd7d9c6eea..d3506649f6a62 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.test.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.test.ts @@ -375,6 +375,7 @@ it.effect("ends a runless root turn's items when its node already ended", () => providerInstanceId: parentProviderInstanceId, projection: { nodes: [root], + providerTurns: [], turnItems: [item("turn-item:command"), item("turn-item:reasoning")], } as never, status: "cancelled", diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 0c38faa5929f1..23a52f39d33d3 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -1335,6 +1335,25 @@ it.effect.each(["missing session", "completed", "failed", "cancelled"] as const) occurredAt: now, payload: childRoot, }); + const childTurnId = ProviderTurnId.make("turn:native-stop-child"); + yield* projections.apply({ + id: EventId.make("event:native-child-stop:child-turn"), + type: "provider-turn.updated", + threadId: childRoot.threadId, + nodeId: childRoot.id, + occurredAt: now, + payload: { + id: childTurnId, + providerThreadId: ProviderThreadId.make("provider-thread:native-stop-child"), + nodeId: childRoot.id, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }); const providerThread = before.providerThreads.find( (thread) => thread.id === run.providerThreadId, )!; @@ -1499,6 +1518,10 @@ it.effect.each(["missing session", "completed", "failed", "cancelled"] as const) ); const child = yield* orchestrator.getThreadProjection(childRoot.threadId); assert.equal(child.nodes.find((node) => node.id === childRoot.id)?.status, confirmedStatus); + assert.equal( + child.providerTurns.find((turn) => turn.id === childTurnId)?.status, + confirmedStatus, + ); assert.isNotNull(child.nodes.find((node) => node.id === childRoot.id)?.completedAt); assert.equal(settled.subagents.find((task) => task.id === sibling.id)?.status, "running"); assert.deepEqual(yield* threadState(threadId), stateBefore); diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index 75612cfd538d9..f83a0b376233b 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -359,8 +359,10 @@ export const endRunlessRootTurns = Effect.fn("subagentProjection.endRunlessRootT >(input: { readonly threadId: ThreadId; readonly providerInstanceId: ProviderInstanceId; - readonly projection: Pick & - Partial>; + readonly projection: Pick< + OrchestrationV2ThreadProjection, + "nodes" | "providerTurns" | "turnItems" + >; readonly status: "completed" | "failed" | "interrupted" | "cancelled"; readonly completedAt?: DateTime.Utc; readonly now: DateTime.Utc; @@ -423,7 +425,7 @@ export const endRunlessRootTurns = Effect.fn("subagentProjection.endRunlessRootT }, }); } - for (const providerTurn of input.projection.providerTurns ?? []) { + for (const providerTurn of input.projection.providerTurns) { if ( providerTurn.nodeId !== node.id || providerTurn.runAttemptId !== null || From e1283a4ebf4c9631cfbb8c8b82745ee2cacb4934 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 25 Oct 2026 14:36:35 -0700 Subject: [PATCH 3/4] fix(server): a failed interrupt never replaces the stored running turn Co-Authored-By: Claude Opus 5.5 --- .../RunExecutionService.test.ts | 36 +++++++++++++------ .../orchestration-v2/RunExecutionService.ts | 20 +++++------ 2 files changed, 34 insertions(+), 22 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index a05e64b4b05e4..a1e62295be4ec 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -2626,16 +2626,17 @@ it.effect.each(["failure", "defect", "interrupt-failure", "ownership-lost"] as c ); it.effect.each([ - { status: "completed", failure: "later-stream" }, - { status: "failed", failure: "later-stream" }, - { status: "completed", failure: "terminal-ingest" }, + { status: "completed", failure: "later-stream", interrupt: "succeeds" }, + { status: "failed", failure: "later-stream", interrupt: "succeeds" }, + { status: "completed", failure: "terminal-ingest", interrupt: "succeeds" }, + { status: "completed", failure: "terminal-ingest", interrupt: "fails" }, ] as const)( - "keeps a $status root provider turn durable after a $failure failure", - ({ status, failure }) => + "keeps a $status root provider turn durable after a $failure failure when the interrupt $interrupt", + ({ status, failure, interrupt }) => Effect.gen(function* () { const now = yield* DateTime.now; const terminalAt = DateTime.subtract(now, { seconds: 1 }); - const key = `terminal-snapshot:${status}:${failure}`; + const key = `terminal-snapshot:${status}:${failure}:${interrupt}`; const durableTurns = yield* Ref.make< ReadonlyMap >(new Map()); @@ -2692,7 +2693,21 @@ it.effect.each([ }), ) : persist(event.providerTurn).pipe(Effect.as([])), - interruptTurn: (input) => Ref.update(interrupts, (current) => [...current, input]), + interruptTurn: (input) => + Ref.update(interrupts, (current) => [...current, input]).pipe( + Effect.andThen( + interrupt === "fails" + ? Effect.fail( + new ProviderAdapter.ProviderAdapterInterruptError({ + driver, + providerThreadId: input.providerThread.id, + providerTurnId: input.providerTurnId, + cause: "provider already exited", + }), + ) + : Effect.void, + ), + ), beforeTerminalWrite: (request) => Effect.forEach(request.events, (event) => event.type === "provider-turn.updated" ? persist(event.payload) : Effect.void, @@ -2704,10 +2719,9 @@ it.effect.each([ assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); assert.isTrue(finalEvents.some((event) => event.type === "run.updated")); const durable = (yield* Ref.get(durableTurns)).get(ids.rootProviderTurnId); - assert.deepEqual( - { status: durable?.status, completedAt: durable?.completedAt }, - { status, completedAt: terminalAt }, - ); + // A failed interrupt keeps the stored turn running for the next start to retry. + assert.equal(durable?.status, interrupt === "fails" ? "running" : status); + assert.deepEqual(durable?.completedAt, interrupt === "fails" ? null : terminalAt); }), ); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index f93651cf86e96..d7789df380b68 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1499,21 +1499,19 @@ export const layer: Layer.Layer< { runId: input.run.id, cause: interruptCause }, ).pipe(Effect.as(false)), ), - // The snapshot is cached before it is published, so it may not be - // stored: replay a terminal one as-is. An active turn whose interrupt - // failed stays running so the next start retries the interrupt. + // A failed interrupt leaves the stored turn running so the next start + // retries it. The snapshot is cached before it is published, so it may + // not be stored: replay a terminal one as-is. Effect.map((interrupted) => - rootProviderTurn === null + !interrupted || rootProviderTurn === null ? null : isTerminalProviderTurnStatus(rootProviderTurn.status) ? rootProviderTurn - : interrupted - ? { - ...rootProviderTurn, - status: "interrupted" as const, - completedAt, - } - : null, + : { + ...rootProviderTurn, + status: "interrupted" as const, + completedAt, + }, ), ) : null; From 9eca0643aebf5a9ede943a4de47cd9d3b8683cc0 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 25 Oct 2026 14:45:56 -0700 Subject: [PATCH 4/4] test(server): name the interrupt-failure snapshot case for what it asserts Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/RunExecutionService.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index a1e62295be4ec..f25d03bc7f527 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -2631,7 +2631,7 @@ it.effect.each([ { status: "completed", failure: "terminal-ingest", interrupt: "succeeds" }, { status: "completed", failure: "terminal-ingest", interrupt: "fails" }, ] as const)( - "keeps a $status root provider turn durable after a $failure failure when the interrupt $interrupt", + "stores a $status root provider turn snapshot after a $failure failure only if the interrupt succeeds (it $interrupt)", ({ status, failure, interrupt }) => Effect.gen(function* () { const now = yield* DateTime.now;