From 0f81fa2fdd5a99398b2686076d1607e455f494d0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 15 Nov 2026 16:57:53 +0000 Subject: [PATCH 1/5] test(server): a stream failure mid-turn leaves the root command running Fails on main: when the provider event stream fails while a root command_execution item is running, the failed-run finalization writes no terminal update for that item (issue #37290, second case). Co-Authored-By: Claude Opus 5.5 --- .../RunExecutionService.test.ts | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 2129fcaf375e1..379ce0d3ab99d 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -3521,6 +3521,51 @@ it.effect("does not interrupt an ended provider when its terminal write fails", }), ); +it.effect("ends a running root command when the provider stream fails mid-turn", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const { written, observed } = yield* captureRootRunTermination({ + key: "stream-error-running-command", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.make( + { + 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: "running" as const, + startedAt: now, + completedAt: null, + }, + }, + backgroundTurnItemEvent(ids, "command_execution", "running", 1), + ).pipe( + Stream.concat( + Stream.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make("session:stream-error-running-command"), + cause: "provider process exited", + }), + ), + ), + ), + }); + assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); + const ids = backgroundScenarioIds("stream-error-running-command"); + const endedCommand = written.find( + (item) => item.id === ids.itemId && item.status !== "running", + ); + assert.isDefined(endedCommand); + }), +); + it.effect("does not interrupt a provider when ingestion fails before the root turn is known", () => Effect.gen(function* () { const interrupts = yield* Ref.make(0); From a72763345e4d997370b15c250b15759964f09dd2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 15 Nov 2026 17:07:46 +0000 Subject: [PATCH 2/5] fix(server): work a dead provider leaves running ends when its event stream fails When the provider event stream failed, the run's open root commands and tools were never ended, and a subagent that outlived its finished root was skipped entirely, so both stayed running until Stop or a restart. Ingestion now tracks the run's open root-thread commands and tools and ends them as failed with the run. When the root already settled, it ends the run-owned subagents and items that outlived it. Fixes pingdotgg/t3code#37290 Co-Authored-By: Claude Opus 5.5 --- .../RunExecutionService.test.ts | 56 ++++++++++ .../orchestration-v2/RunExecutionService.ts | 100 +++++++++++++++++- 2 files changed, 155 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 379ce0d3ab99d..96a95479422f0 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -3566,6 +3566,62 @@ it.effect("ends a running root command when the provider stream fails mid-turn", }), ); +it.effect("ends a subagent that outlived its root when the provider stream fails", () => + Effect.gen(function* () { + const providerInstanceId = ProviderInstanceId.make("codex"); + const { finalEvents } = yield* captureRootRunTermination({ + key: "stream-error-outlived-subagent", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.make( + { + type: "subagent.updated" as const, + driver, + subagent: makeRunOwnedSubagentFixture({ + ids, + providerInstanceId, + childThreadId: ids.childThreadId, + driver, + status: "running", + }), + }, + { + type: "turn_item.updated" as const, + driver, + turnItem: makeRunOwnedSubagentTurnItemFixture({ + ids, + providerInstanceId, + childThreadId: ids.childThreadId, + driver, + status: "running", + }), + }, + rootTerminalEvent(ids, "completed"), + ).pipe( + Stream.concat( + Stream.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make("session:stream-error-outlived-subagent"), + cause: "provider process exited", + }), + ), + ), + ), + }); + assert.deepEqual( + finalEvents.flatMap((event) => + event.type === "run.updated" || + event.type === "subagent.updated" || + event.type === "turn-item.updated" + ? [`${event.type}:${event.payload.status}`] + : [], + ), + ["run.updated:waiting", "subagent.updated:failed", "turn-item.updated:failed"], + ); + }), +); + it.effect("does not interrupt a provider when ingestion fails before the root turn is known", () => 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 95bc974a8c80b..5b6599d47ae8f 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -329,6 +329,34 @@ export const cascadeTerminalizeRunOwnedSubagents = Effect.fn( return events; }); +function endAbandonedTurnItems(input: { + readonly run: OrchestrationV2Run; + readonly items: Iterable; + readonly status: RunOwnedSubagentTerminalStatus; + readonly completedAt: DateTime.Utc; + readonly allocateEventId: () => Effect.Effect; +}) { + return Effect.forEach(input.items, (item) => + input.allocateEventId().pipe( + Effect.map((id): OrchestrationV2DomainEvent => ({ + id, + type: "turn-item.updated", + threadId: item.threadId, + runId: input.run.id, + ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), + providerInstanceId: input.run.providerInstanceId, + occurredAt: input.completedAt, + payload: { + ...item, + status: input.status, + completedAt: input.completedAt, + updatedAt: input.completedAt, + }, + })), + ), + ); +} + function finalProviderThreadStatus( disposition: ProviderTerminalEvent["threadDisposition"], ): OrchestrationV2ProviderThread["status"] { @@ -678,6 +706,8 @@ export const layer: Layer.Layer< readonly loadRunInterruptReason?: () => Effect.Effect; readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection; readonly openRunOwnedSubagentsTerminal?: ProviderTerminalEvent; + // Root-run items the dead provider can no longer settle. + readonly abandonedTurnItems?: Iterable; // Root turn snapshot written with the run when T3 ends the run itself. readonly abandonedProviderTurn?: OrchestrationV2ProviderTurn; readonly terminal: ProviderTerminalEvent; @@ -762,6 +792,17 @@ export const layer: Layer.Layer< allocateEventId, }) : []; + const abandonedTurnItemEvents = + input.abandonedTurnItems !== undefined && + isRunOwnedSubagentTerminalStatus(input.terminal.status) + ? yield* endAbandonedTurnItems({ + run: input.run, + items: input.abandonedTurnItems, + status: input.terminal.status, + completedAt, + allocateEventId, + }) + : []; const persistedStatus = input.terminal.status === "completed" ? "waiting" : input.terminal.status; // Omit delegatedCompletion; the startup snapshot must not overwrite newer acknowledgements, successors, or Stop barriers. @@ -810,6 +851,7 @@ export const layer: Layer.Layer< : [], events: [ ...cascadedSubagentEvents, + ...abandonedTurnItemEvents, ...(input.abandonedProviderTurn === undefined ? [] : [ @@ -1150,6 +1192,8 @@ export const layer: Layer.Layer< ReadonlySet >(new Set(inheritedBackgroundTurnItemsById.keys())); const openRunOwnedSubagents = yield* Ref.make(emptyOpenRunOwnedSubagentProjection()); + // Open root-thread commands and tools of this run; only an ingestion failure ends them. + const openRootTurnItems = new Map(); // The last stored root snapshot's pendingBackgroundWork; adapters that omit it leave it false. let rootSnapshotPendingWork = false; // Owned child threads whose last snapshot reported pending work, such as a queued native @@ -1222,6 +1266,7 @@ export const layer: Layer.Layer< } if (isRunOwnedSubagentTerminalStatus(terminal.status) && !childTurnsContinue(terminal)) { yield* Ref.set(openRunOwnedSubagents, emptyOpenRunOwnedSubagentProjection()); + openRootTurnItems.clear(); idleFlushClosed = true; } yield* Ref.set(rootRunFinalized, true); @@ -1334,6 +1379,19 @@ export const layer: Layer.Layer< return next; }); } + if ( + belongsToRootRun && + event.turnItem.threadId === input.run.threadId && + event.turnItem.type !== "subagent" && + backgroundCapableTurnItemTypes.has(event.turnItem.type) && + deliverable + ) { + if (isSettledTurnItemStatus(event.turnItem.status)) { + openRootTurnItems.delete(event.turnItem.id); + } else { + openRootTurnItems.set(event.turnItem.id, event.turnItem); + } + } if ( belongsToOwnedChildThread && (event.turnItem.runId === null || belongsToRootRun) && @@ -1826,7 +1884,46 @@ export const layer: Layer.Layer< runId: input.run.id, cause, }); - if (yield* Ref.get(rootRunFinalized)) return; + if (yield* Ref.get(rootRunFinalized)) { + // The root settled earlier, but work that outlived it has no stream to end it. + const allocateEventId = () => + idAllocator.allocate.event({ threadId: input.run.threadId }); + const open = yield* Ref.getAndSet( + openRunOwnedSubagents, + emptyOpenRunOwnedSubagentProjection(), + ); + const abandoned = [...openRootTurnItems.values()]; + openRootTurnItems.clear(); + const events = [ + ...(yield* cascadeTerminalizeRunOwnedSubagents({ + run: input.run, + open, + status: "failed", + completedAt, + allocateEventId, + })), + ...(yield* endAbandonedTurnItems({ + run: input.run, + items: abandoned, + status: "failed", + completedAt, + allocateEventId, + })), + ]; + if (events.length === 0) return; + yield* eventSink.writeWithEffects({ effects: [], events }).pipe( + Effect.andThen(refreshAfterTurn), + Effect.catchCauseIf( + (writeCause) => !Cause.hasInterruptsOnly(writeCause), + (writeCause) => + Effect.logWarning( + "orchestration V2 failed to end work that outlived its run", + { runId: input.run.id, cause: writeCause }, + ), + ), + ); + return; + } const providerThread = yield* Ref.get(latestProviderThread); // Stop background work too; the failed consumer cannot persist terminal events. const abandonedProviderTurn = @@ -1877,6 +1974,7 @@ export const layer: Layer.Layer< expectedStatus: "running", }, openRunOwnedSubagents: openSubagents, + abandonedTurnItems: [...openRootTurnItems.values()], ...(abandonedProviderTurn === null ? {} : { abandonedProviderTurn }), completedAt, terminal: makeFailedTerminalEvent( From 69cb6b55ac5d330bbf7708a526879527f5b48f6b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 15 Nov 2026 17:20:43 +0000 Subject: [PATCH 3/5] fix(server): Codex session release ends the work its dead process left running Ending a subagent that outlived its root from the run's event consumer trusted any ingestion failure as a dead provider and wrote cached running snapshots over newer records, such as a committed Stop. Drop that branch. The session manager already settles background work when a session whose tasks end with its process is released: it reads current projections and writes each end conditioned on the record being unchanged. Codex's app-server is the session's child process, so opt Codex in. The mid-turn root command cleanup now applies only when the failure handler's interrupt restarted the runtime. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.ts | 2 + .../RunExecutionService.test.ts | 56 ------------------- .../orchestration-v2/RunExecutionService.ts | 52 +++-------------- 3 files changed, 11 insertions(+), 99 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 4947f4da8a440..896fc552cf1b7 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7850,6 +7850,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( providerSessionId: input.providerSessionId, providerSession: session, events: Stream.fromEffectRepeat(Queue.take(events)), + // The app-server is this session's child process; its commands and subagents die with it. + backgroundTasksEndWithProcess: true, canReuseContextUsage: canReuseCodexContextUsage, normalizeThreadLoadModelSelection: normalizeCodexThreadLoadModelSelection, // Idle release can beat delayed subagent resumes: Codex reports them completed between turns with no resume-expected signal. diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 96a95479422f0..379ce0d3ab99d 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -3566,62 +3566,6 @@ it.effect("ends a running root command when the provider stream fails mid-turn", }), ); -it.effect("ends a subagent that outlived its root when the provider stream fails", () => - Effect.gen(function* () { - const providerInstanceId = ProviderInstanceId.make("codex"); - const { finalEvents } = yield* captureRootRunTermination({ - key: "stream-error-outlived-subagent", - shouldFinalizeRun: () => Effect.succeed(true), - events: (ids) => - Stream.make( - { - type: "subagent.updated" as const, - driver, - subagent: makeRunOwnedSubagentFixture({ - ids, - providerInstanceId, - childThreadId: ids.childThreadId, - driver, - status: "running", - }), - }, - { - type: "turn_item.updated" as const, - driver, - turnItem: makeRunOwnedSubagentTurnItemFixture({ - ids, - providerInstanceId, - childThreadId: ids.childThreadId, - driver, - status: "running", - }), - }, - rootTerminalEvent(ids, "completed"), - ).pipe( - Stream.concat( - Stream.fail( - new ProviderAdapter.ProviderAdapterEventStreamError({ - driver, - providerSessionId: ProviderSessionId.make("session:stream-error-outlived-subagent"), - cause: "provider process exited", - }), - ), - ), - ), - }); - assert.deepEqual( - finalEvents.flatMap((event) => - event.type === "run.updated" || - event.type === "subagent.updated" || - event.type === "turn-item.updated" - ? [`${event.type}:${event.payload.status}`] - : [], - ), - ["run.updated:waiting", "subagent.updated:failed", "turn-item.updated:failed"], - ); - }), -); - it.effect("does not interrupt a provider when ingestion fails before the root turn is known", () => 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 5b6599d47ae8f..3f3cf8b0d394b 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1192,7 +1192,7 @@ export const layer: Layer.Layer< ReadonlySet >(new Set(inheritedBackgroundTurnItemsById.keys())); const openRunOwnedSubagents = yield* Ref.make(emptyOpenRunOwnedSubagentProjection()); - // Open root-thread commands and tools of this run; only an ingestion failure ends them. + // Open root-thread commands and tools of this run; an ingestion failure that stops the provider ends them. const openRootTurnItems = new Map(); // The last stored root snapshot's pendingBackgroundWork; adapters that omit it leave it false. let rootSnapshotPendingWork = false; @@ -1884,46 +1884,7 @@ export const layer: Layer.Layer< runId: input.run.id, cause, }); - if (yield* Ref.get(rootRunFinalized)) { - // The root settled earlier, but work that outlived it has no stream to end it. - const allocateEventId = () => - idAllocator.allocate.event({ threadId: input.run.threadId }); - const open = yield* Ref.getAndSet( - openRunOwnedSubagents, - emptyOpenRunOwnedSubagentProjection(), - ); - const abandoned = [...openRootTurnItems.values()]; - openRootTurnItems.clear(); - const events = [ - ...(yield* cascadeTerminalizeRunOwnedSubagents({ - run: input.run, - open, - status: "failed", - completedAt, - allocateEventId, - })), - ...(yield* endAbandonedTurnItems({ - run: input.run, - items: abandoned, - status: "failed", - completedAt, - allocateEventId, - })), - ]; - if (events.length === 0) return; - yield* eventSink.writeWithEffects({ effects: [], events }).pipe( - Effect.andThen(refreshAfterTurn), - Effect.catchCauseIf( - (writeCause) => !Cause.hasInterruptsOnly(writeCause), - (writeCause) => - Effect.logWarning( - "orchestration V2 failed to end work that outlived its run", - { runId: input.run.id, cause: writeCause }, - ), - ), - ); - return; - } + if (yield* Ref.get(rootRunFinalized)) return; const providerThread = yield* Ref.get(latestProviderThread); // Stop background work too; the failed consumer cannot persist terminal events. const abandonedProviderTurn = @@ -1974,8 +1935,13 @@ export const layer: Layer.Layer< expectedStatus: "running", }, openRunOwnedSubagents: openSubagents, - abandonedTurnItems: [...openRootTurnItems.values()], - ...(abandonedProviderTurn === null ? {} : { abandonedProviderTurn }), + ...(abandonedProviderTurn === null + ? {} + : { + abandonedProviderTurn, + // The interrupt restarted the runtime, so these items cannot end. + abandonedTurnItems: [...openRootTurnItems.values()], + }), completedAt, terminal: makeFailedTerminalEvent( makeProviderFailure({ From 90a350a940a6c8d072de8bb69cf6f760707c4597 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 15 Nov 2026 17:45:15 +0000 Subject: [PATCH 4/5] fix(server): Codex keeps its open work until its processes are reaped Codex does not reap the commands its app-server started when the process exits, so they can outlive it. Claiming they end with the process let a session release show live commands as ended. Drop the flag. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 896fc552cf1b7..4947f4da8a440 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7850,8 +7850,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( providerSessionId: input.providerSessionId, providerSession: session, events: Stream.fromEffectRepeat(Queue.take(events)), - // The app-server is this session's child process; its commands and subagents die with it. - backgroundTasksEndWithProcess: true, canReuseContextUsage: canReuseCodexContextUsage, normalizeThreadLoadModelSelection: normalizeCodexThreadLoadModelSelection, // Idle release can beat delayed subagent resumes: Codex reports them completed between turns with no resume-expected signal. From 06f09564955b5c025524dbef8ad9297d5a7dc0fa Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 15 Nov 2026 18:04:53 +0000 Subject: [PATCH 5/5] fix(server): a Codex subagent that outlived its run ends when the app-server dies Codex runs native subagents as threads of its app-server, so they end with the process, but Codex cannot claim the same for commands it does not reap. Session release settled native subagents only for runtimes whose background tasks and tool calls all end with the process, so a Codex subagent that outlived its completed root stayed Running after the app-server died. Add nativeSubagentsEndWithProcess: release then ends the session's native subagents through the existing conditional writes, while the task roster and tool calls stay as they are. Codex sets it. Fixes pingdotgg/t3code#37290 Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.ts | 2 ++ .../ProviderSessionManager.test.ts | 28 ++++++++++++++++--- .../ProviderSessionManager.ts | 9 ++++-- .../src/server/ProviderAdapter.ts | 2 ++ 4 files changed, 35 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 4947f4da8a440..3c946d3c7d020 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7850,6 +7850,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( providerSessionId: input.providerSessionId, providerSession: session, events: Stream.fromEffectRepeat(Queue.take(events)), + // Native subagents are app-server threads; commands it spawned are not reaped and may outlive it. + nativeSubagentsEndWithProcess: true, canReuseContextUsage: canReuseCodexContextUsage, normalizeThreadLoadModelSelection: normalizeCodexThreadLoadModelSelection, // Idle release can beat delayed subagent resumes: Codex reports them completed between turns with no resume-expected signal. diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 1ed6361549202..4f4440a4a4073 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -458,6 +458,7 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( readonly hasPendingBackgroundWork?: Effect.Effect; readonly hasPendingBackgroundWorkForThread?: Effect.Effect; readonly backgroundTasksEndWithProcess?: boolean; + readonly nativeSubagentsEndWithProcess?: true; readonly hangSessionScopeClose?: boolean; readonly startTurn?: Effect.Effect; readonly interruptTurn?: Effect.Effect; @@ -562,6 +563,9 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( ...(options.backgroundTasksEndWithProcess === false ? {} : { backgroundTasksEndWithProcess: true as const }), + ...(options.nativeSubagentsEndWithProcess === undefined + ? {} + : { nativeSubagentsEndWithProcess: options.nativeSubagentsEndWithProcess }), ...(options.hasPendingBackgroundWork === undefined ? {} : { hasPendingBackgroundWork: options.hasPendingBackgroundWork }), @@ -640,6 +644,7 @@ function layerTest(input: { readonly hasPendingBackgroundWork?: Effect.Effect; readonly hasPendingBackgroundWorkForThread?: Effect.Effect; readonly backgroundTasksEndWithProcess?: boolean; + readonly nativeSubagentsEndWithProcess?: true; readonly hangSessionScopeClose?: boolean; readonly startTurn?: Effect.Effect; readonly interruptTurn?: Effect.Effect; @@ -686,6 +691,9 @@ function layerTest(input: { ...(input.backgroundTasksEndWithProcess === undefined ? {} : { backgroundTasksEndWithProcess: input.backgroundTasksEndWithProcess }), + ...(input.nativeSubagentsEndWithProcess === undefined + ? {} + : { nativeSubagentsEndWithProcess: input.nativeSubagentsEndWithProcess }), ...(input.hangSessionScopeClose === undefined ? {} : { hangSessionScopeClose: input.hangSessionScopeClose }), @@ -3507,16 +3515,19 @@ it.effect.each(["detach", "stream failure", "provider stop", "early close"] as c }), ); -it.effect.each(["a runtime whose tasks can outlive it", "server shutdown"] as const)( +it.effect.each([ + "a runtime whose tasks can outlive it", + "a runtime whose only native subagents end with it", + "server shutdown", +] as const)( "ProviderSessionManagerV2 keeps the background work after the release of %s", (release) => Effect.gen(function* () { const state = yield* Ref.make(emptyState); const effect = Effect.gen(function* () { const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; - const { providerSessionId, rosters, toolCalls } = yield* openSessionWithBackgroundCommand( - "provider-session-manager-outliving-roster", - ); + const { providerSessionId, rosters, toolCalls, nativeSubagentStatus } = + yield* openSessionWithBackgroundCommand("provider-session-manager-outliving-roster"); yield* release === "server shutdown" ? manager.release({ providerSessionId, reason: "server_shutdown" }) @@ -3524,6 +3535,12 @@ it.effect.each(["a runtime whose tasks can outlive it", "server shutdown"] as co assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]); assert.equal((yield* toolCalls).cutOff, "running"); + assert.equal( + yield* nativeSubagentStatus, + release === "a runtime whose only native subagents end with it" + ? "interrupted" + : "running", + ); }); yield* effect.pipe( @@ -3532,6 +3549,9 @@ it.effect.each(["a runtime whose tasks can outlive it", "server shutdown"] as co state, idleTimeoutMs: 60_000, ...(release === "server shutdown" ? {} : { backgroundTasksEndWithProcess: false }), + ...(release === "a runtime whose only native subagents end with it" + ? { nativeSubagentsEndWithProcess: true as const } + : {}), }), ), ); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 2e9adb9c8a54c..ae77bf140ebdc 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -914,9 +914,10 @@ export const layerWithOptions = ( readonly reason: ProviderSessionReleaseReason; readonly detachedThreadId?: ThreadId; }) { + const tasksEnd = input.entry.runtime.backgroundTasksEndWithProcess === true; if ( input.reason === "server_shutdown" || - input.entry.runtime.backgroundTasksEndWithProcess !== true + (!tasksEnd && input.entry.runtime.nativeSubagentsEndWithProcess !== true) ) { return []; } @@ -933,8 +934,11 @@ export const layerWithOptions = ( providerSessionId, providerThreadId: providerThread.id, driver: providerThread.driver, + toolCallsEnd: tasksEnd, taskIds: new Set( - (providerThread.pendingBackgroundTasks ?? []).map((task) => task.taskId), + tasksEnd + ? (providerThread.pendingBackgroundTasks ?? []).map((task) => task.taskId) + : [], ), }); } @@ -1006,6 +1010,7 @@ export const layerWithOptions = ( const writeReleasedToolCallEvents = Effect.fn( "ProviderSessionManager.writeReleasedToolCallEvents", )(function* (released: ReleasedProviderThread, releasedAt: DateTime.Utc) { + if (!released.toolCallsEnd) return; const { threadId, providerSessionId, providerThreadId, driver } = released; const readCutOff = projectionStore .getThreadRecords(threadId, ["turnItems"], { diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 80d3ef8a5285a..928731c62a811 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -711,6 +711,8 @@ export interface ProviderAdapterV2SessionRuntime { * must keep Waiting. */ readonly backgroundTasksEndWithProcess?: true; + /** Set if native subagents run inside the process, so releasing the session ends them. */ + readonly nativeSubagentsEndWithProcess?: true; readonly getModelContextWindow?: ( modelSelection: ModelSelection, cwd?: string | null,