diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 3721f62762c6..ece7caf687b5 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -31,6 +31,8 @@ import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +// T3-CUSTOM(expbkt3): one bounded seam keeps the upstream worker model intact. +import { runProviderCommandWithinLaneDeadline } from "../providerCommandLane.expbkt3.ts"; import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts"; import { decideWorktreeRecovery, describeWorktreeRecreation } from "../threadWorktreeRecovery.ts"; import { @@ -2981,7 +2983,12 @@ const make = Effect.gen(function* () { }); const processDomainEventSafely = (event: ProviderIntentEvent) => - processDomainEvent(event).pipe( + // T3-CUSTOM(expbkt3): bound the one global lane without replacing it. + runProviderCommandWithinLaneDeadline(processDomainEvent(event), { + eventType: event.type, + threadId: event.payload.threadId, + commandId: event.commandId, + }).pipe( Effect.catchCause((cause) => { if (Cause.hasInterruptsOnly(cause)) { return Effect.interrupt; diff --git a/apps/server/src/orchestration/providerCommandLane.expbkt3.test.ts b/apps/server/src/orchestration/providerCommandLane.expbkt3.test.ts new file mode 100644 index 000000000000..5473d940e14c --- /dev/null +++ b/apps/server/src/orchestration/providerCommandLane.expbkt3.test.ts @@ -0,0 +1,39 @@ +/** T3-CUSTOM(expbkt3): regression for the 2026-08-20 global prompt-queue outage. */ +import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +import { assert, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Ref from "effect/Ref"; +import * as TestClock from "effect/testing/TestClock"; + +import { runProviderCommandWithinLaneDeadline } from "./providerCommandLane.expbkt3.ts"; + +it.effect("lets the next worker item run when a provider command is uninterruptible", () => + Effect.gen(function* () { + const firstStarted = yield* Deferred.make(); + const secondRan = yield* Ref.make(false); + const worker = yield* makeDrainableWorker((item: "wedged" | "next") => + runProviderCommandWithinLaneDeadline( + item === "wedged" + ? Deferred.succeed(firstStarted, undefined).pipe( + Effect.andThen(Effect.uninterruptible(Effect.never)), + ) + : Ref.set(secondRan, true), + { + eventType: `test.${item}`, + threadId: item, + commandId: item, + timeout: "50 millis", + }, + ), + ); + + yield* worker.enqueue("wedged"); + yield* Deferred.await(firstStarted); + yield* worker.enqueue("next"); + yield* TestClock.adjust("60 millis"); + yield* worker.drain; + + assert.isTrue(yield* Ref.get(secondRan)); + }).pipe(Effect.scoped), +); diff --git a/apps/server/src/orchestration/providerCommandLane.expbkt3.ts b/apps/server/src/orchestration/providerCommandLane.expbkt3.ts new file mode 100644 index 000000000000..20df3104aebd --- /dev/null +++ b/apps/server/src/orchestration/providerCommandLane.expbkt3.ts @@ -0,0 +1,43 @@ +/** + * T3-CUSTOM(expbkt3): Non-invasive deadline for the upstream provider-command lane. + * + * ProviderCommandReactor deliberately uses one ordered DrainableWorker. Keep + * that upstream model, but never join one provider operation forever: its + * finalizer may be uninterruptible while an SDK callback is pending. The + * command runs detached so timing out the join releases the lane immediately; + * cleanup is then requested without awaiting the same stuck finalizer. + */ +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Option from "effect/Option"; + +export const DEFAULT_PROVIDER_COMMAND_LANE_DEADLINE: Duration.Input = "30 seconds"; + +export interface ProviderCommandLaneDeadlineInput { + readonly eventType: string; + readonly threadId: string; + readonly commandId: string | null; + readonly timeout?: Duration.Input; +} + +export const runProviderCommandWithinLaneDeadline = ( + command: Effect.Effect, + input: ProviderCommandLaneDeadlineInput, +) => + Effect.gen(function* () { + // Start in the current scheduler turn so healthy commands retain the + // upstream worker's observable ordering; only their lifetime is detached. + const commandFiber = yield* command.pipe(Effect.forkDetach({ startImmediately: true })); + const timeout = input.timeout ?? DEFAULT_PROVIDER_COMMAND_LANE_DEADLINE; + const completed = yield* Fiber.join(commandFiber).pipe(Effect.timeoutOption(timeout)); + if (Option.isSome(completed)) return; + + yield* Fiber.interrupt(commandFiber).pipe(Effect.forkDetach, Effect.asVoid); + yield* Effect.logWarning("provider command timed out and released the reactor lane", { + eventType: input.eventType, + threadId: input.threadId, + commandId: input.commandId, + timeout: String(timeout), + }); + });