fix(server): prevent provider cancellation from wedging all threads

apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts

@@ -167,6 +167,7 @@ describe("ProviderCommandReactor", () => {
     readonly startSessionEffect?: (
       session: ProviderSession,
     ) => Effect.Effect<ProviderSession, ProviderAdapterRequestError>;
+    readonly interruptTurnEffect?: ProviderServiceShape["interruptTurn"];
   }) {
     const now = "2026-01-01T00:00:00.000Z";
     const baseDir =
@@ -252,7 +253,7 @@ describe("ProviderCommandReactor", () => {
         turnId: asTurnId("turn-1"),
       }),
     );
-    const interruptTurn = vi.fn((_: unknown) => Effect.void);
+    const interruptTurn = vi.fn(input?.interruptTurnEffect ?? ((_: unknown) => Effect.void));
     const respondToRequest = vi.fn<ProviderServiceShape["respondToRequest"]>(() => Effect.void);
     const respondToUserInput = vi.fn<ProviderServiceShape["respondToUserInput"]>(() => Effect.void);
     const stopSession = vi.fn((input: unknown) =>
@@ -2490,6 +2491,78 @@ describe("ProviderCommandReactor", () => {
     });
   });
 
+  it("bounds a hung provider interrupt so later thread starts still run", async () => {
+    const harness = await createHarness({
+      interruptTurnEffect: () => Effect.never,
+    });
+    const now = "2026-01-01T00:00:00.000Z";
+    const secondThreadId = ThreadId.make("thread-2");
+    const modelSelection: ModelSelection = {
+      instanceId: ProviderInstanceId.make("codex"),
+      model: "gpt-5-codex",
+    };
+
+    await harness.dispatch({
+      type: "thread.create",
+      commandId: CommandId.make("cmd-create-thread-after-hung-interrupt"),
+      threadId: secondThreadId,
+      projectId: asProjectId("project-1"),
+      title: "Independent thread",
+      modelSelection,
+      interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
+      runtimeMode: "approval-required",
+      branch: null,
+      worktreePath: null,
+      createdAt: now,
+    });
+    await harness.dispatch({
+      type: "thread.session.set",
+      commandId: CommandId.make("cmd-running-session-before-hung-interrupt"),
+      threadId: ThreadId.make("thread-1"),
+      session: {
+        threadId: ThreadId.make("thread-1"),
+        status: "running",
+        providerName: "codex",
+        runtimeMode: "approval-required",
+        activeTurnId: asTurnId("turn-hung-interrupt"),
+        lastError: null,
+        updatedAt: now,
+      },
+      createdAt: now,
+    });
+    await harness.dispatch({
+      type: "thread.turn.interrupt",
+      commandId: CommandId.make("cmd-hung-interrupt"),
+      threadId: ThreadId.make("thread-1"),
+      turnId: asTurnId("turn-hung-interrupt"),
+      createdAt: now,
+    });
+    await harness.dispatch({
+      type: "thread.turn.start",
+      commandId: CommandId.make("cmd-start-after-hung-interrupt"),
+      threadId: secondThreadId,
+      message: {
+        messageId: asMessageId("message-after-hung-interrupt"),
+        role: "user",
+        text: "this thread must still start",
+        attachments: [],
+      },
+      interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
+      runtimeMode: "approval-required",
+      createdAt: now,
+    });
+
+    await waitFor(async () => {
+      const readModel = await harness.readModel();
+      const interrupted = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
+      return interrupted?.session?.status === "ready";
+    });
+    await waitFor(() => harness.sendTurn.mock.calls.length === 1, 8_000);
+    expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({
+      threadId: secondThreadId,
+    });
+  });
+
   it("starts a fresh session when only projected session state exists", async () => {
     const harness = await createHarness();
     const now = "2026-01-01T00:00:00.000Z";

apps/server/src/orchestration/Layers/ProviderCommandReactor.ts

@@ -65,6 +65,8 @@ import {
 } from "../../serverSettings.ts";
 import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts";
 import { GitWorkflowService } from "../../git/GitWorkflowService.ts";
+
+const PROVIDER_CONTROL_TIMEOUT = Duration.seconds(5);
 const isProviderAdapterRequestError = Schema.is(ProviderAdapterRequestError);
 const isProviderDriverKind = Schema.is(ProviderDriverKind);
 
@@ -1180,30 +1182,38 @@ const make = Effect.gen(function* () {
     }
 
     // Orchestration turn ids are not provider turn ids, so interrupt by session.
-    // Clearing the projection here is authoritative: a persisted session may
-    // say "running" even though its provider process disappeared yesterday.
-    // Provider cancellation remains best-effort so Abort always restores a
-    // usable composer without mistaking a quiet, live process for a dead one.
-    yield* providerService
+    // Clear the projection before touching the provider. This state transition
+    // is authoritative and must not depend on a cooperative protocol peer.
+    yield* setThreadSession({
+      threadId: event.payload.threadId,
+      session: {
+        ...thread.session,
+        status: "ready",
+        activeTurnId: null,
+        updatedAt: event.payload.createdAt,
+      },
+      createdAt: event.payload.createdAt,
+    });
+
+    // Provider cancellation is best-effort and bounded. Some protocol peers
+    // never answer cancellation; an interruptible timeout releases this
+    // thread's command lane even in that case.
+    const interruptResult = yield* providerService
       .interruptTurn({ threadId: event.payload.threadId })
       .pipe(
+        Effect.interruptible,
+        Effect.timeoutOption(PROVIDER_CONTROL_TIMEOUT),
         Effect.catchCause((cause) =>
-          Effect.logWarning("provider turn interrupt failed", { cause }),
+          Effect.logWarning("provider turn interrupt failed", {
+            threadId: event.payload.threadId,
+            cause: Cause.pretty(cause),
+          }).pipe(Effect.as(Option.some(undefined))),
         ),
       );
-
-    const latestThread = yield* resolveThread(event.payload.threadId);
-    const session = latestThread?.session;
-    if (session && session.status !== "stopped") {
-      yield* setThreadSession({
+    if (Option.isNone(interruptResult)) {
+      yield* Effect.logWarning("provider turn interrupt timed out", {
         threadId: event.payload.threadId,
-        session: {
-          ...session,
-          status: "ready",
-          activeTurnId: null,
-          updatedAt: event.payload.createdAt,
-        },
-        createdAt: event.payload.createdAt,
+        timeout: Duration.format(PROVIDER_CONTROL_TIMEOUT),
       });
     }
   });
@@ -1344,10 +1354,7 @@ const make = Effect.gen(function* () {
     }
 
     const now = event.payload.createdAt;
-    if (context.session && context.session.status !== "stopped") {
-      yield* providerService.stopSession({ threadId: context.threadId });
-    }
-
+    const shouldStopProvider = context.session && context.session.status !== "stopped";
     yield* setThreadSession({
       threadId: context.threadId,
       session: {
@@ -1364,6 +1371,25 @@ const make = Effect.gen(function* () {
       },
       createdAt: now,
     });
+
+    if (shouldStopProvider) {
+      const stopResult = yield* providerService.stopSession({ threadId: context.threadId }).pipe(
+        Effect.interruptible,
+        Effect.timeoutOption(PROVIDER_CONTROL_TIMEOUT),
+        Effect.catchCause((cause) =>
+          Effect.logWarning("provider session stop failed", {
+            threadId: context.threadId,
+            cause: Cause.pretty(cause),
+          }).pipe(Effect.as(Option.some(undefined))),
+        ),
+      );
+      if (Option.isNone(stopResult)) {
+        yield* Effect.logWarning("provider session stop timed out", {
+          threadId: context.threadId,
+          timeout: Duration.format(PROVIDER_CONTROL_TIMEOUT),
+        });
+      }
+    }
   });
 
   const setRecoveryFailureState = Effect.fn("setRecoveryFailureState")(function* (input: {