fix(server): one silent provider no longer stalls every thread

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts

@@ -2285,6 +2285,209 @@ describe("CodexAdapterV2 post-settle continuation", () => {
     }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
   );
 
+  it.effect("stops a turn Codex starts after its turn/start was abandoned", () =>
+    Effect.gen(function* () {
+      const nativeThreadId = "abandoned-start-thread";
+      const nativeTurnId = "abandoned-start-turn";
+      const prompt = "Run a command.";
+      const preamble = codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt });
+      const transcript = makeCodexReplayTranscript({
+        scenario: "abandoned-turn-start",
+        entries: [
+          // Codex never answers turn/start, then starts the turn anyway.
+          ...preamble.slice(0, -2),
+          {
+            type: "emit_inbound",
+            label: "turn/started",
+            afterMs: 1000,
+            frame: {
+              method: "turn/started",
+              params: {
+                threadId: nativeThreadId,
+                turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }),
+              },
+            },
+          },
+          {
+            type: "expect_outbound",
+            label: "turn/interrupt",
+            frame: {
+              id: 4,
+              method: "turn/interrupt",
+              params: { threadId: nativeThreadId, turnId: nativeTurnId },
+            },
+          },
+          { type: "emit_inbound", label: "turn/interrupt", frame: { id: 4, result: {} } },
+        ],
+      });
+      const interruptSent = yield* Deferred.make<void>();
+      const harness = yield* makeCodexReplayHarness(
+        transcript,
+        () => Effect.void,
+        (method) =>
+          method === "turn/interrupt"
+            ? Deferred.succeed(interruptSent, undefined).pipe(Effect.asVoid)
+            : Effect.void,
+      );
+      const starting = yield* harness.runtime
+        .startTurn(
+          makeCodexTestTurnInput({
+            threadId: harness.threadId,
+            providerThread: harness.providerThread,
+            now: yield* DateTime.now,
+            attemptId: RunAttemptId.make("abandoned-start-attempt"),
+            text: prompt,
+          }),
+        )
+        .pipe(Effect.forkScoped);
+      yield* TestClock.adjust("500 millis");
+      // The provider request deadline gives up on the start.
+      yield* Fiber.interrupt(starting);
+      assert.isFalse(yield* Deferred.isDone(interruptSent));
+      yield* TestClock.adjust("500 millis");
+      yield* Deferred.await(interruptSent);
+    }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
+  );
+
+  it.effect("stops a turn Codex reported started before its turn/start was abandoned", () =>
+    Effect.gen(function* () {
+      const nativeThreadId = "abandoned-registered-thread";
+      const nativeTurnId = "abandoned-registered-turn";
+      const prompt = "Run a command.";
+      const preamble = codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt });
+      const transcript = makeCodexReplayTranscript({
+        scenario: "abandoned-registered-turn-start",
+        entries: [
+          // Codex reports the turn started but never answers turn/start.
+          ...preamble.slice(0, -2),
+          preamble.at(-1)!,
+          {
+            type: "expect_outbound",
+            label: "turn/interrupt",
+            frame: {
+              id: 4,
+              method: "turn/interrupt",
+              params: { threadId: nativeThreadId, turnId: nativeTurnId },
+            },
+          },
+          { type: "emit_inbound", label: "turn/interrupt", frame: { id: 4, result: {} } },
+        ],
+      });
+      const interruptSent = yield* Deferred.make<void>();
+      const turnRegistered = yield* Deferred.make<void>();
+      const harness = yield* makeCodexReplayHarness(
+        transcript,
+        (event) =>
+          event.type === "provider_turn.updated" &&
+          event.providerTurn.nativeTurnRef?.nativeId === nativeTurnId
+            ? Deferred.succeed(turnRegistered, undefined)
+            : Effect.void,
+        (method) =>
+          method === "turn/interrupt"
+            ? Deferred.succeed(interruptSent, undefined).pipe(Effect.asVoid)
+            : Effect.void,
+      );
+      const starting = yield* harness.runtime
+        .startTurn(
+          makeCodexTestTurnInput({
+            threadId: harness.threadId,
+            providerThread: harness.providerThread,
+            now: yield* DateTime.now,
+            attemptId: RunAttemptId.make("abandoned-registered-attempt"),
+            text: prompt,
+          }),
+        )
+        .pipe(Effect.forkScoped);
+      yield* Deferred.await(turnRegistered);
+      assert.isFalse(yield* Deferred.isDone(interruptSent));
+      yield* Fiber.interrupt(starting);
+      yield* Deferred.await(interruptSent);
+    }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
+  );
+
+  it.effect("keeps a newer start from adopting the turn of an abandoned start", () =>
+    Effect.gen(function* () {
+      const nativeThreadId = "abandoned-then-retried-thread";
+      const abandonedTurnId = "abandoned-turn";
+      const retriedTurnId = "retried-turn";
+      const prompt = "Run a command.";
+      const preamble = codexReplayPreamble({
+        nativeThreadId,
+        nativeTurnId: abandonedTurnId,
+        prompt,
+      });
+      const firstStart = preamble.at(-3)!;
+      if (firstStart.type !== "expect_outbound") return yield* Effect.die("expected turn/start");
+      const retryFrame = { ...(firstStart.frame as object), id: 4 };
+      const transcript = makeCodexReplayTranscript({
+        scenario: "abandoned-turn-start-then-retry",
+        entries: [
+          ...preamble.slice(0, -2),
+          // The retry is sent before Codex gets to the abandoned start.
+          { type: "expect_outbound", label: "turn/start/retry", frame: retryFrame },
+          {
+            type: "emit_inbound",
+            label: "turn/started/abandoned",
+            frame: {
+              method: "turn/started",
+              params: {
+                threadId: nativeThreadId,
+                turn: makeCodexReplayTurn({ id: abandonedTurnId, status: "inProgress" }),
+              },
+            },
+          },
+          {
+            type: "expect_outbound",
+            label: "turn/interrupt",
+            frame: {
+              id: 5,
+              method: "turn/interrupt",
+              params: { threadId: nativeThreadId, turnId: abandonedTurnId },
+            },
+          },
+          { type: "emit_inbound", label: "turn/interrupt", frame: { id: 5, result: {} } },
+          {
+            type: "emit_inbound",
+            label: "turn/start/retry",
+            frame: {
+              id: 4,
+              result: { turn: makeCodexReplayTurn({ id: retriedTurnId, status: "inProgress" }) },
+            },
+          },
+        ],
+      });
+      const registeredTurns: Array<string> = [];
+      const retriedTurnRegistered = yield* Deferred.make<void>();
+      const harness = yield* makeCodexReplayHarness(transcript, (event) => {
+        if (event.type !== "provider_turn.updated") return Effect.void;
+        const nativeId = event.providerTurn.nativeTurnRef?.nativeId;
+        if (nativeId == null || registeredTurns.includes(nativeId)) return Effect.void;
+        registeredTurns.push(nativeId);
+        return nativeId === retriedTurnId
+          ? Deferred.succeed(retriedTurnRegistered, undefined)
+          : Effect.void;
+      });
+      const turnInput = (attempt: string) =>
+        Effect.map(DateTime.now, (now) =>
+          makeCodexTestTurnInput({
+            threadId: harness.threadId,
+            providerThread: harness.providerThread,
+            now,
+            attemptId: RunAttemptId.make(attempt),
+            text: prompt,
+          }),
+        );
+      const starting = yield* harness.runtime
+        .startTurn(yield* turnInput("abandoned-attempt"))
+        .pipe(Effect.forkScoped);
+      yield* TestClock.adjust("500 millis");
+      yield* Fiber.interrupt(starting);
+      yield* harness.runtime.startTurn(yield* turnInput("retried-attempt"));
+      yield* Deferred.await(retriedTurnRegistered);
+      assert.deepEqual(registeredTurns, [retriedTurnId]);
+    }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
+  );
+
   it.effect("settles Stop when a queued native turn fails before starting", () =>
     Effect.gen(function* () {
       const nativeThreadId = "early-stop-thread";

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts

@@ -1778,6 +1778,25 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
           ).hasSubagents = true;
         };
         const pendingRootTurns = yield* Ref.make(new Map<string, ProviderAdapterV2TurnInput>());
+        // Native threads whose turn/start was given up before Codex answered,
+        // by a deadline or a Stop. Codex takes a thread's starts in order, so
+        // the next turn it starts there belongs to the abandoned start, which
+        // has no run left to report to.
+        const abandonedTurnStarts = new Set<string>();
+        const stopAbandonedTurn = (nativeThreadId: string, nativeTurnId: string) =>
+          Effect.logWarning("orchestration-v2.codex-abandoned-turn-started", {
+            nativeThreadId,
+            nativeTurnId,
+          }).pipe(
+            Effect.andThen(
+              client.request("turn/interrupt", { threadId: nativeThreadId, turnId: nativeTurnId }),
+            ),
+            Effect.catch((cause) =>
+              Effect.logWarning("orchestration-v2.codex-abandoned-turn-stop-failed", { cause }),
+            ),
+            Effect.forkIn(scope),
+            Effect.asVoid,
+          );
         const turnWaiters = yield* Ref.make(new Map<string, Deferred.Deferred<void, never>>());
         const subagentThreads = yield* Ref.make(new Map<string, CodexSubagentThreadContext>());
         const subagentModels = new Map<string, string>();
@@ -4076,6 +4095,12 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
               }
               return;
             }
+            // Checked before a pending start: a newer start waiting on the same
+            // thread must not adopt the turn Codex started for the abandoned one.
+            if (abandonedTurnStarts.delete(payload.threadId)) {
+              yield* stopAbandonedTurn(payload.threadId, payload.turn.id);
+              return;
+            }
             const pendingRootTurn = (yield* Ref.get(pendingRootTurns)).get(payload.threadId);
             if (pendingRootTurn !== undefined) {
               yield* registerRootTurn({
@@ -5887,7 +5912,22 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
               else next.delete(threadId);
               return next;
             });
-            const started = yield* client.request("turn/start", turnStartParams);
+            const started = yield* client.request("turn/start", turnStartParams).pipe(
+              Effect.onInterrupt(() =>
+                Effect.gen(function* () {
+                  // turn/started can arrive before the response and register
+                  // the turn already; stop that one now instead of waiting.
+                  const registered = Array.from((yield* Ref.get(activeTurns)).values()).find(
+                    (context) => context.input === turnInput,
+                  );
+                  if (registered !== undefined) {
+                    return yield* stopAbandonedTurn(threadId, registered.nativeTurnId);
+                  }
+                  abandonedTurnStarts.add(threadId);
+                }),
+              ),
+            );
+            abandonedTurnStarts.delete(threadId);
             yield* registerRootTurn({
               turnInput,
               nativeTurnId: started.turn.id,

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

@@ -29,6 +29,7 @@ import * as ProviderTurnStartService from "./ProviderTurnStartService.ts";
 import * as RuntimeRequestService from "./RuntimeRequestService.ts";
 import * as ThreadTitleRegenerationService from "./ThreadTitleRegenerationService.ts";
 import * as ThreadManagementService from "./ThreadManagementService.ts";
+import * as SqlitePersistence from "../persistence/Sqlite.ts";
 import * as ServerSettings from "../serverSettings.ts";
 
 const threadId = ThreadId.make("thread:effect-worker-restart");
@@ -837,3 +838,94 @@ it.effect("settles a delegated child once its restart continuation fails for goo
     }).pipe(Effect.provide(layer));
   }),
 );
+
+it.effect("frees a worker from an effect that outlives its lease", () =>
+  Effect.gen(function* () {
+    const outbox = yield* EffectOutbox.EffectOutboxV2;
+    const wedgedThreadIds = [1, 2, 3, 4].map((index) =>
+      ThreadId.make(`thread:effect-worker-wedged-${index}`),
+    );
+    const otherThreadId = ThreadId.make("thread:effect-worker-other-provider");
+    // Sorted ahead of the other thread's start, so the wedged starts take every worker first.
+    const wedgedEffectIds = wedgedThreadIds.map((_, index) => `effect:a-wedged-start-${index}`);
+    const followUpEffectId = "effect:c-wedged-follow-up";
+    yield* outbox.enqueue([
+      ...wedgedThreadIds.map((wedgedThreadId, index) => ({
+        id: wedgedEffectIds[index]!,
+        commandId: CommandId.make(`command:effect-worker-wedged-${index}`),
+        threadId: wedgedThreadId,
+        request: {
+          type: "provider-turn.start" as const,
+          runId: RunId.make(`run:effect-worker-wedged-${index}`),
+        },
+      })),
+      {
+        id: "effect:b-other-start",
+        commandId: CommandId.make("command:effect-worker-other-provider"),
+        threadId: otherThreadId,
+        request: { type: "provider-turn.start", runId: RunId.make("run:effect-worker-other") },
+      },
+      {
+        id: followUpEffectId,
+        commandId: CommandId.make("command:effect-worker-wedged-follow-up"),
+        threadId: wedgedThreadIds[0]!,
+        request: { type: "terminal.cleanup" },
+      },
+    ]);
+
+    const providerAnswers = yield* Deferred.make<void>();
+    const wedgedStarts = yield* Ref.make(0);
+    const executed = yield* Ref.make<ReadonlyArray<string>>([]);
+    const layerExecutor = Layer.succeed(
+      EffectWorker.OrchestrationEffectExecutorV2,
+      EffectWorker.OrchestrationEffectExecutorV2.of({
+        execute: (effect) =>
+          (wedgedEffectIds.includes(effect.id)
+            ? Ref.update(wedgedStarts, (count) => count + 1).pipe(
+                Effect.andThen(Deferred.await(providerAnswers)),
+              )
+            : Effect.void
+          ).pipe(Effect.andThen(Ref.update(executed, (ids) => [...ids, effect.id]))),
+      }),
+    );
+    const layerWorker = EffectWorker.layerWithOptions({ workerId: "wedged-provider-worker" }).pipe(
+      Layer.provide(Layer.merge(Layer.succeed(EffectOutbox.EffectOutboxV2, outbox), layerExecutor)),
+    );
+    yield* EffectWorker.runDaemonWithOptions({ concurrency: 4 }).pipe(
+      Effect.provide(layerWorker),
+      Effect.forkScoped,
+    );
+    const awaitCondition = Effect.fnUntraced(function* (condition: Effect.Effect<boolean>) {
+      while (!(yield* condition)) yield* Effect.yieldNow;
+    });
+    const statusOf = (effectId: string) =>
+      outbox.get(effectId).pipe(Effect.map(Option.map((effect) => effect.status)));
+
+    yield* awaitCondition(Ref.get(wedgedStarts).pipe(Effect.map((count) => count === 4)));
+    assert.deepEqual(yield* Ref.get(executed), []);
+
+    yield* TestClock.adjust("30 seconds");
+    yield* awaitCondition(
+      Ref.get(executed).pipe(Effect.map((ids) => ids.includes("effect:b-other-start"))),
+    );
+    // The wedged threads keep their order: nothing runs behind a start still in flight.
+    assert.deepEqual(yield* statusOf(wedgedEffectIds[0]!), Option.some("running"));
+    assert.deepEqual(yield* statusOf(followUpEffectId), Option.some("pending"));
+
+    yield* Deferred.succeed(providerAnswers, undefined);
+    yield* outbox.notifyAvailable();
+    yield* awaitCondition(
+      Ref.get(executed).pipe(Effect.map((ids) => ids.includes(followUpEffectId))),
+    );
+    for (const effectId of wedgedEffectIds) {
+      assert.deepEqual(yield* statusOf(effectId), Option.some("succeeded"));
+    }
+  }).pipe(
+    Effect.provide(
+      Layer.mergeAll(
+        EffectOutbox.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
+        TestClock.layer(),
+      ),
+    ),
+  ),
+);