fix(server): work a dead provider leaves running ends when its event stream fails

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.

apps/server/src/orchestration-v2/ProviderSessionManager.test.ts

@@ -458,6 +458,7 @@ const makeProviderAdapter = Effect.fnUntraced(function* (
     readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
     readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean>;
     readonly backgroundTasksEndWithProcess?: boolean;
+    readonly nativeSubagentsEndWithProcess?: true;
     readonly hangSessionScopeClose?: boolean;
     readonly startTurn?: Effect.Effect<void>;
     readonly interruptTurn?: Effect.Effect<void>;
@@ -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<boolean>;
   readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean>;
   readonly backgroundTasksEndWithProcess?: boolean;
+  readonly nativeSubagentsEndWithProcess?: true;
   readonly hangSessionScopeClose?: boolean;
   readonly startTurn?: Effect.Effect<void>;
   readonly interruptTurn?: Effect.Effect<void>;
@@ -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,23 +3515,32 @@ 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" })
           : manager.close(providerSessionId);
 
         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 }
+              : {}),
           }),
         ),
       );

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"], {

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);

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<OrchestrationV2TurnItem>;
+  readonly status: RunOwnedSubagentTerminalStatus;
+  readonly completedAt: DateTime.Utc;
+  readonly allocateEventId: () => Effect.Effect<EventId, IdAllocator.IdAllocatorV2AllocationError>;
+}) {
+  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<string | undefined, never>;
         readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection;
         readonly openRunOwnedSubagentsTerminal?: ProviderTerminalEvent;
+        // Root-run items the dead provider can no longer settle.
+        readonly abandonedTurnItems?: Iterable<OrchestrationV2TurnItem>;
         // 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<OrchestrationV2TurnItem["id"]>
         >(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<TurnItemId, OrchestrationV2TurnItem>();
         // 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({

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,