fix(server): a stuck provider call no longer freezes its thread until restart

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

@@ -13,6 +13,7 @@ import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
+import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as Queue from "effect/Queue";
@@ -751,3 +752,148 @@ it.effect("safely retries after replacement cleanup succeeds and start fails", (
     ]);
   }),
 );
+
+for (const { attemptCount, interruptible } of [
+  { attemptCount: 1, interruptible: true },
+  { attemptCount: 5, interruptible: true },
+  // A handler parked in an uninterruptible wait, like an ACP response waiting
+  // for its native acknowledgement, must not hold the lane past the timeout.
+  { attemptCount: 1, interruptible: false },
+]) {
+  it.effect(
+    `${attemptCount === 1 ? "retries" : "fails"} a hung ${interruptible ? "" : "uninterruptible "}effect once its execution timeout elapses (attempt ${attemptCount})`,
+    () =>
+      Effect.gen(function* () {
+        const now = DateTime.formatIso(yield* DateTime.now);
+        const effectId = `effect:worker-execution-timeout:${attemptCount}:${interruptible}`;
+        const workerId = "worker-execution-timeout";
+        const claimedEffect: OrchestrationEffectV2 = {
+          id: effectId,
+          commandId: CommandId.make("command:worker-execution-timeout"),
+          threadId: ThreadId.make("thread:worker-execution-timeout"),
+          request: {
+            type: "provider-turn.interrupt",
+            providerSessionId: oldSessionId,
+            providerThreadId,
+            providerTurnId,
+          },
+          status: "running",
+          attemptCount,
+          availableAt: now,
+          leaseOwner: workerId,
+          leaseExpiresAt: now,
+          createdAt: now,
+          updatedAt: now,
+          completedAt: null,
+          lastError: null,
+        };
+        const executionStarted = yield* Deferred.make<void>();
+        const retries = yield* Ref.make<ReadonlyArray<{ error: string; delayMs: number }>>([]);
+        const failures = yield* Ref.make<ReadonlyArray<string>>([]);
+        const outboxLayer = Layer.mock(EffectOutboxV2)({
+          claimNext: () => Effect.succeed(Option.some(claimedEffect)),
+          get: () => Effect.succeed(Option.some(claimedEffect)),
+          awaitCancellation: () => Effect.never,
+          clearCancellation: () => Effect.void,
+          retry: ({ error, delayMs }) =>
+            Ref.update(retries, (existing) => [...existing, { error, delayMs }]).pipe(
+              Effect.as(true),
+            ),
+          fail: ({ error }) =>
+            Ref.update(failures, (existing) => [...existing, error]).pipe(Effect.as(true)),
+        });
+        const executorLayer = Layer.succeed(
+          OrchestrationEffectExecutorV2,
+          OrchestrationEffectExecutorV2.of({
+            execute: () =>
+              Deferred.succeed(executionStarted, undefined).pipe(
+                Effect.andThen(interruptible ? Effect.never : Effect.uninterruptible(Effect.never)),
+              ),
+          }),
+        );
+        const workerLayer = effectWorkerLayerWithOptions({
+          workerId,
+          maxAttempts: 5,
+          effectTimeoutMs: 1_000,
+        }).pipe(Layer.provide(Layer.merge(outboxLayer, executorLayer)));
+
+        const fiber = yield* OrchestrationEffectWorkerV2.pipe(
+          Effect.flatMap((worker) => worker.runOnce),
+          Effect.provide(workerLayer),
+          Effect.forkChild,
+        );
+        yield* Deferred.await(executionStarted);
+        yield* TestClock.adjust("999 millis");
+        assert.isUndefined(fiber.pollUnsafe());
+
+        yield* TestClock.adjust("1 millis");
+        assert.isTrue(yield* Fiber.join(fiber));
+        if (attemptCount === 1) {
+          const retry = (yield* Ref.get(retries))[0];
+          assert.isDefined(retry);
+          assert.include(retry.error, "timed out after 1000ms");
+          assert.equal(retry.delayMs, 100);
+          assert.isEmpty(yield* Ref.get(failures));
+        } else {
+          assert.isEmpty(yield* Ref.get(retries));
+          assert.include((yield* Ref.get(failures))[0] ?? "", "timed out after 1000ms");
+        }
+      }).pipe(Effect.provide(TestClock.layer())),
+  );
+}
+
+it.effect("a cancellation stops the running handler before the worker moves on", () =>
+  Effect.gen(function* () {
+    const now = DateTime.formatIso(yield* DateTime.now);
+    const effectId = "effect:worker-cancellation-stops-handler";
+    const workerId = "worker-cancellation-stops-handler";
+    const claimedEffect: OrchestrationEffectV2 = {
+      id: effectId,
+      commandId: CommandId.make("command:worker-cancellation-stops-handler"),
+      threadId: ThreadId.make("thread:worker-cancellation-stops-handler"),
+      request: { type: "terminal.cleanup" },
+      status: "running",
+      attemptCount: 1,
+      availableAt: now,
+      leaseOwner: workerId,
+      leaseExpiresAt: now,
+      createdAt: now,
+      updatedAt: now,
+      completedAt: null,
+      lastError: null,
+    };
+    const executionStarted = yield* Deferred.make<void>();
+    const cancelled = yield* Deferred.make<void>();
+    const handlerStopped = yield* Ref.make(false);
+    const outboxLayer = Layer.mock(EffectOutboxV2)({
+      claimNext: () => Effect.succeed(Option.some(claimedEffect)),
+      get: () => Effect.succeed(Option.some(claimedEffect)),
+      awaitCancellation: () => Deferred.await(cancelled),
+      clearCancellation: () => Effect.void,
+    });
+    const executorLayer = Layer.succeed(
+      OrchestrationEffectExecutorV2,
+      OrchestrationEffectExecutorV2.of({
+        execute: () =>
+          Deferred.succeed(executionStarted, undefined).pipe(
+            Effect.andThen(Effect.never),
+            Effect.onInterrupt(() => Ref.set(handlerStopped, true)),
+          ),
+      }),
+    );
+    const workerLayer = effectWorkerLayerWithOptions({ workerId }).pipe(
+      Layer.provide(Layer.merge(outboxLayer, executorLayer)),
+    );
+
+    const fiber = yield* OrchestrationEffectWorkerV2.pipe(
+      Effect.flatMap((worker) => worker.runOnce),
+      Effect.provide(workerLayer),
+      Effect.forkChild,
+    );
+    yield* Deferred.await(executionStarted);
+    yield* Deferred.succeed(cancelled, undefined);
+
+    assert.isTrue(yield* Fiber.join(fiber));
+    assert.isTrue(yield* Ref.get(handlerStopped));
+  }),
+);

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

@@ -5,6 +5,7 @@ import * as DateTime from "effect/DateTime";
 import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
+import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
 import * as Metric from "effect/Metric";
 import * as Option from "effect/Option";
@@ -42,6 +43,31 @@ export class OrchestrationEffectExecutionError extends Schema.TaggedError<Orches
   },
 ) {}
 
+export class OrchestrationEffectTimeoutError extends Schema.TaggedError<OrchestrationEffectTimeoutError>()(
+  "OrchestrationEffectTimeoutError",
+  {
+    effectId: Schema.String,
+    effectType: Schema.String,
+    timeoutMs: Schema.Number,
+  },
+) {
+  override get message(): string {
+    return `Effect ${this.effectType} (${this.effectId}) timed out after ${this.timeoutMs}ms`;
+  }
+}
+
+/**
+ * Backstop for a handler that never returns. Only one non-title effect runs per
+ * thread, so a hung row would otherwise block that thread's interrupt, detach and
+ * cleanup until restart. Every handler returns once its provider or git call
+ * answers (a turn start returns when the provider accepts the turn, not when it
+ * finishes), and the slowest legitimate ones stay well under this: title
+ * generation caps each of its three attempts at 180s, worktree git operations at
+ * 5 minutes, and a provider start bounds each provider call at
+ * `ProviderStartTimeout` so it settles its run before this fires.
+ */
+const DEFAULT_EFFECT_TIMEOUT_MS = 15 * 60_000;
+
 /**
  * Pure interrupt races with hard process teardown or a dead session produce
  * "not active" protocol errors. Retrying those only delays recovery.
@@ -475,6 +501,7 @@ export interface OrchestrationEffectWorkerOptions {
   readonly workerId?: string;
   readonly leaseDurationMs?: number;
   readonly maxAttempts?: number;
+  readonly effectTimeoutMs?: number;
 }
 
 export const layerWithOptions = (
@@ -492,6 +519,7 @@ export const layerWithOptions = (
       const workerId = options.workerId ?? `orchestration-v2:${process.pid}`;
       const leaseDurationMs = Math.max(1, options.leaseDurationMs ?? 30_000);
       const maxAttempts = Math.max(1, options.maxAttempts ?? 5);
+      const effectTimeoutMs = Math.max(1, options.effectTimeoutMs ?? DEFAULT_EFFECT_TIMEOUT_MS);
       const wasCancelled = (effectId: string) =>
         outbox.get(effectId).pipe(
           Effect.map(
@@ -626,9 +654,31 @@ export const layerWithOptions = (
           }).pipe(Effect.onError((cause) => requeueClaim(effect, cause)));
           if (cancelledBeforeExecution) return true;
 
-          const execution = executor
-            .execute(effect, { willRetry: effect.attemptCount < maxAttempts })
-            .pipe(Effect.as("executed" as const));
+          // A hung handler fails through the normal retry/fail path below under
+          // this claim's lease. Nothing reclaims a still-running row instead.
+          // The handler runs in its own fiber because a plain timeout waits for
+          // its loser to exit, so a handler parked in an uninterruptible wait
+          // (an ACP response awaiting its native acknowledgement) would hold the
+          // lane anyway. On timeout the claim settles and the handler is left to
+          // finish its interruption in the background; a cancellation still
+          // waits for the handler to stop.
+          const execution = Effect.uninterruptibleMask((restore) =>
+            Effect.gen(function* () {
+              const handler = yield* executor
+                .execute(effect, { willRetry: effect.attemptCount < maxAttempts })
+                .pipe(Effect.forkDetach({ startImmediately: true }));
+              const handlerExit = yield* restore(
+                Fiber.await(handler).pipe(Effect.timeoutOption(Duration.millis(effectTimeoutMs))),
+              ).pipe(Effect.onInterrupt(() => Fiber.interrupt(handler)));
+              if (Option.isSome(handlerExit)) return yield* handlerExit.value;
+              yield* Fiber.interrupt(handler).pipe(Effect.forkDetach);
+              return yield* new OrchestrationEffectTimeoutError({
+                effectId: effect.id,
+                effectType: effect.request.type,
+                timeoutMs: effectTimeoutMs,
+              });
+            }),
+          ).pipe(Effect.as("executed" as const));
           const exit = yield* Effect.exit(Effect.raceFirst(execution, cancellation)).pipe(
             Effect.ensuring(outbox.clearCancellation(effect.id)),
           );

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

@@ -2526,6 +2526,69 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => {
     }),
   );
 
+  it.effect("frees a thread's lane when a running effect exceeds its execution timeout", () =>
+    Effect.gen(function* () {
+      const outbox = yield* EffectOutboxV2;
+      const commandId = CommandId.make("command:foundation-execution-timeout");
+      const threadId = ThreadId.make("thread:foundation-execution-timeout");
+      const hungEffectId = "effect:foundation-execution-timeout:hung";
+      const nextEffectId = "effect:foundation-execution-timeout:next";
+      const hungStarted = yield* Deferred.make<void>();
+      const executions = yield* Ref.make<ReadonlyArray<string>>([]);
+      yield* outbox.enqueue([
+        {
+          id: hungEffectId,
+          commandId,
+          threadId,
+          request: {
+            type: "provider-session.detach",
+            providerSessionId: ProviderSessionId.make("session:foundation-execution-timeout"),
+          },
+        },
+        { id: nextEffectId, commandId, threadId, request: { type: "terminal.cleanup" } },
+      ]);
+
+      const executorLayer = Layer.succeed(
+        OrchestrationEffectExecutorV2,
+        OrchestrationEffectExecutorV2.of({
+          execute: (effect) =>
+            Ref.update(executions, (current) => [...current, effect.id]).pipe(
+              Effect.andThen(
+                effect.id === hungEffectId
+                  ? Deferred.succeed(hungStarted, undefined).pipe(Effect.andThen(Effect.never))
+                  : Effect.void,
+              ),
+            ),
+        }),
+      );
+      const workerLayer = effectWorkerLayerWithOptions({
+        workerId: "execution-timeout-worker",
+        effectTimeoutMs: 1_000,
+      }).pipe(Layer.provide(Layer.merge(Layer.succeed(EffectOutboxV2, outbox), executorLayer)));
+
+      yield* Effect.gen(function* () {
+        const worker = yield* OrchestrationEffectWorkerV2;
+        const hungFiber = yield* worker.runOnce.pipe(Effect.forkChild);
+        yield* Deferred.await(hungStarted);
+        // The lane stays closed while the first effect is still running.
+        yield* TestClock.adjust("999 millis");
+        assert.isFalse(yield* worker.runOnce);
+        assert.deepEqual(yield* Ref.get(executions), [hungEffectId]);
+
+        yield* TestClock.adjust("1 millis");
+        assert.isTrue(yield* Fiber.join(hungFiber));
+        const hung = Option.getOrThrow(yield* outbox.get(hungEffectId));
+        assert.equal(hung.status, "pending");
+        assert.equal(hung.attemptCount, 1);
+        assert.include(hung.lastError ?? "", "timed out after 1000ms");
+
+        assert.isTrue(yield* worker.runOnce);
+        assert.deepEqual(yield* Ref.get(executions), [hungEffectId, nextEffectId]);
+        assert.equal(Option.getOrThrow(yield* outbox.get(nextEffectId)).status, "succeeded");
+      }).pipe(Effect.provide(workerLayer));
+    }).pipe(Effect.provide(TestClock.layer())),
+  );
+
   it.effect(
     "persists shutdown continuation intent through the real event sink without domain events",
     () =>