diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 4947f4da8a44..3c946d3c7d02 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 1ed636154920..4f4440a4a407 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 2e9adb9c8a54..ae77bf140ebd 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/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 2129fcaf375e..379ce0d3ab99 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); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 95bc974a8c80..3f3cf8b0d394 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; 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; // 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) && @@ -1877,7 +1935,13 @@ export const layer: Layer.Layer< expectedStatus: "running", }, openRunOwnedSubagents: openSubagents, - ...(abandonedProviderTurn === null ? {} : { abandonedProviderTurn }), + ...(abandonedProviderTurn === null + ? {} + : { + abandonedProviderTurn, + // The interrupt restarted the runtime, so these items cannot end. + abandonedTurnItems: [...openRootTurnItems.values()], + }), completedAt, terminal: makeFailedTerminalEvent( makeProviderFailure({ diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 80d3ef8a5285..928731c62a81 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,