diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 63f0cbdae66b..4a947353682a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/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(); + 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(); + const turnRegistered = yield* Deferred.make(); + 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 = []; + const retriedTurnRegistered = yield* Deferred.make(); + 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"; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index d562f876df49..7b2dc8f19fdd 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/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()); + // 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(); + 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>()); const subagentThreads = yield* Ref.make(new Map()); const subagentModels = new Map(); @@ -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, diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index d9bd8689f092..9481d0a3bb59 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/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(); + const wedgedStarts = yield* Ref.make(0); + const executed = yield* Ref.make>([]); + 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) { + 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(), + ), + ), + ), +); diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index b3ee780334d3..f04597e5a622 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/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"; @@ -535,6 +536,7 @@ export const layerWithOptions = ( Effect.gen(function* () { const outbox = yield* EffectOutbox.EffectOutboxV2; const executor = yield* OrchestrationEffectExecutorV2; + const workerScope = yield* Effect.scope; 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); @@ -678,65 +680,91 @@ 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)); - const exit = yield* Effect.exit(Effect.raceFirst(execution, cancellation)).pipe( - Effect.ensuring(outbox.clearCancellation(effect.id)), - ); - if (Exit.isSuccess(exit) && exit.value === "cancelled") { - return true; - } - if (Exit.isSuccess(exit)) { - return yield* Effect.gen(function* () { - const completed = yield* outbox.succeed({ effectId: effect.id, workerId }); - if (!completed) { - if (yield* wasCancelled(effect.id)) return true; - return yield* new OrchestrationEffectWorkerError({ - operation: "complete", - effectId: effect.id, - cause: "The worker no longer owns the effect lease.", - }); - } + const settle = Effect.gen(function* () { + const execution = executor + .execute(effect, { willRetry: effect.attemptCount < maxAttempts }) + .pipe(Effect.as("executed" as const)); + const exit = yield* Effect.exit(Effect.raceFirst(execution, cancellation)).pipe( + Effect.ensuring(outbox.clearCancellation(effect.id)), + ); + if (Exit.isSuccess(exit) && exit.value === "cancelled") { return true; - }).pipe(Effect.onError((cause) => recoverPostSuccessSettlement(effect, cause))); - } + } + if (Exit.isSuccess(exit)) { + return yield* Effect.gen(function* () { + const completed = yield* outbox.succeed({ effectId: effect.id, workerId }); + if (!completed) { + if (yield* wasCancelled(effect.id)) return true; + return yield* new OrchestrationEffectWorkerError({ + operation: "complete", + effectId: effect.id, + cause: "The worker no longer owns the effect lease.", + }); + } + return true; + }).pipe(Effect.onError((cause) => recoverPostSuccessSettlement(effect, cause))); + } - const error = Cause.pretty(exit.cause); - const nonRetryable = isNonRetryableProviderTurnControlFailure(effect.request.type, error); - yield* Effect.logWarning("Orchestration effect execution failed", { + const error = Cause.pretty(exit.cause); + const nonRetryable = isNonRetryableProviderTurnControlFailure( + effect.request.type, + error, + ); + yield* Effect.logWarning("Orchestration effect execution failed", { + effectId: effect.id, + effectType: effect.request.type, + attemptCount: effect.attemptCount, + nonRetryable, + error, + }); + // Prefer succeed for terminal interrupt races so the outbox does not + // keep a failed interrupt around; fail only when we must not retry. + const updated = nonRetryable + ? yield* outbox + .succeed({ effectId: effect.id, workerId }) + .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause))) + : effect.attemptCount >= maxAttempts + ? yield* outbox + .fail({ effectId: effect.id, workerId, error }) + .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause))) + : yield* outbox + .retry({ + effectId: effect.id, + workerId, + error, + delayMs: Math.min(30_000, 100 * 2 ** Math.max(0, effect.attemptCount - 1)), + }) + .pipe(Effect.onError((cause) => requeueClaim(effect, cause))); + if (!updated) { + if (yield* wasCancelled(effect.id)) return true; + return yield* new OrchestrationEffectWorkerError({ + operation: "reschedule", + effectId: effect.id, + cause: "The worker no longer owns the effect lease.", + }); + } + return true; + }); + // An effect that outlives its lease keeps running and keeps its row + // `running`, so the thread's later effects still wait behind it. It + // stops holding this worker slot, though: a provider that never + // answers would otherwise take every slot and stall all threads. + const settlement = yield* Effect.forkIn(settle, workerScope); + const settled = yield* Fiber.await(settlement).pipe( + Effect.timeoutOption(Duration.millis(leaseDurationMs)), + Effect.onInterrupt(() => Fiber.interrupt(settlement)), + ); + if (Option.isSome(settled)) return yield* settled.value; + yield* Effect.logWarning("Orchestration effect outlived its lease; freeing its worker", { effectId: effect.id, effectType: effect.request.type, - attemptCount: effect.attemptCount, - nonRetryable, - error, }); - // Prefer succeed for terminal interrupt races so the outbox does not - // keep a failed interrupt around; fail only when we must not retry. - const updated = nonRetryable - ? yield* outbox - .succeed({ effectId: effect.id, workerId }) - .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause))) - : effect.attemptCount >= maxAttempts - ? yield* outbox - .fail({ effectId: effect.id, workerId, error }) - .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause))) - : yield* outbox - .retry({ - effectId: effect.id, - workerId, - error, - delayMs: Math.min(30_000, 100 * 2 ** Math.max(0, effect.attemptCount - 1)), - }) - .pipe(Effect.onError((cause) => requeueClaim(effect, cause))); - if (!updated) { - if (yield* wasCancelled(effect.id)) return true; - return yield* new OrchestrationEffectWorkerError({ - operation: "reschedule", - effectId: effect.id, - cause: "The worker no longer owns the effect lease.", - }); - } + yield* Fiber.join(settlement).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Orchestration effect worker failed", cause), + ), + Effect.forkIn(workerScope), + ); return true; }).pipe( Effect.mapError((cause) => diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index bfb076ee3bf3..41788c21aa78 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -1316,6 +1316,74 @@ it.effect("ProviderSessionManagerV2 releases an idle session whose turn start wa }), ); +it.effect("ProviderSessionManagerV2 fails a turn start the provider never answers", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const startTurnReached = yield* Deferred.make(); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-unanswered-turn-start"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + const runtime = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const runId = idAllocator.derive.run({ threadId, ordinal: 1 }); + const starting = yield* runtime + .startTurn({ + appThread: (yield* projectionStore.getThreadProjection(threadId)).thread, + threadId, + runId, + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }), + rootNodeId: idAllocator.derive.rootNode({ runId }), + providerThread: makeProviderThread({ idAllocator, threadId, providerSessionId, now }), + message: { + createdBy: "user", + creationSource: "web", + messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }), + text: "sent to a provider that stopped answering", + attachments: [], + }, + modelSelection, + runtimePolicy, + }) + .pipe(Effect.forkChild); + yield* Deferred.await(startTurnReached); + + yield* TestClock.adjust("2 minutes"); + const exit = yield* Fiber.await(starting); + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) { + assert.include(Cause.pretty(exit.cause), "did not answer the turn start within 2 minutes"); + } + }).pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 60_000, + startTurn: Deferred.succeed(startTurnReached, undefined).pipe( + Effect.andThen(Effect.never), + ), + }), + ), + ); + }), +); + it.effect( "ProviderSessionManagerV2 keeps a shared session busy when another thread's start is stopped early", () => diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index f84441129cc3..84dbd2f0c224 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -48,6 +48,7 @@ import * as IdAllocator from "./IdAllocator.ts"; import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; import { ProviderAdapterEventStreamError, + ProviderAdapterProtocolError, ProviderAdapterV2RuntimePolicy, type ProviderAdapterV2Error, type ProviderAdapterV2Event, @@ -60,6 +61,11 @@ import * as ProjectionStore from "./ProjectionStore.ts"; const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000; const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000; const RELEASE_SCOPE_CLOSE_TIMEOUT_MS = 30 * 1000; +/** + * How long a provider may take to accept a control request. Turn starts + * normally take seconds, and well under a minute at worst. + */ +const PROVIDER_REQUEST_TIMEOUT_MS = 2 * 60 * 1000; /** The identity a turn's start and its `turn.terminal` share. */ const busyTurnKey = (providerThreadId: ProviderThreadId, runOrdinal: number) => @@ -1496,6 +1502,26 @@ export const layerWithOptions = ( modelFamily: normalizeModelMetricLabel(model), }, }); + // A provider that stops answering fails its own requests instead of + // holding the effect that sent them. Every adapter returns once the + // provider accepts the request and streams the turn as events, so this + // bounds the acceptance, never a running turn. + const withinDeadline = + (request: string) => + (effect: Effect.Effect) => + effect.pipe( + Effect.timeoutOrElse({ + duration: Duration.millis(PROVIDER_REQUEST_TIMEOUT_MS), + orElse: () => + Effect.fail( + new ProviderAdapterProtocolError({ + driver: runtime.driver, + detail: `the provider did not answer the ${request} within 2 minutes`, + }), + ), + }), + ); + const compactThread = runtime.compactThread; return { ...runtime, subscribeEvents, @@ -1511,7 +1537,7 @@ export const layerWithOptions = ( providerInstanceId: runtime.instanceId, }), ).pipe( - Effect.andThen(runtime.ensureThread(input)), + Effect.andThen(runtime.ensureThread(input).pipe(withinDeadline("thread start"))), Effect.tap((providerThread) => markProviderThreadLoaded({ providerSessionId, @@ -1526,8 +1552,9 @@ export const layerWithOptions = ( ), resumeThread: (input) => { const threadId = input.threadId ?? input.providerThread.appThreadId; + const resume = () => runtime.resumeThread(input).pipe(withinDeadline("thread resume")); if (threadId === null || threadId === undefined) { - return runtime.resumeThread(input); + return resume(); } const providerThreadKey = providerThreadLoadKey({ providerThread: input.providerThread, @@ -1548,7 +1575,7 @@ export const layerWithOptions = ( isProviderThreadLoaded({ providerSessionId, threadId, providerThreadKey }), ), Effect.flatMap((loaded) => - loaded ? Effect.succeed(input.providerThread) : runtime.resumeThread(input), + loaded ? Effect.succeed(input.providerThread) : resume(), ), Effect.tap((providerThread) => markProviderThreadLoaded({ @@ -1576,7 +1603,7 @@ export const layerWithOptions = ( providerInstanceId: runtime.instanceId, }), ).pipe( - Effect.andThen(runtime.forkThread(input)), + Effect.andThen(runtime.forkThread(input).pipe(withinDeadline("thread fork"))), Effect.tap((providerThread) => markProviderThreadLoaded({ providerSessionId, @@ -1617,7 +1644,12 @@ export const layerWithOptions = ( ), ), () => - runtime.startTurn(input).pipe(turnMetrics("send", input.modelSelection.model)), + runtime + .startTurn(input) + .pipe( + withinDeadline("turn start"), + turnMetrics("send", input.modelSelection.model), + ), (_, exit) => Exit.isFailure(exit) ? observeActivity( @@ -1631,20 +1663,34 @@ export const layerWithOptions = ( ), ), ), + ...(compactThread === undefined + ? {} + : { + compactThread: (input) => compactThread(input).pipe(withinDeadline("compaction")), + }), steerTurn: (input) => observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe( - Effect.andThen(runtime.steerTurn(input).pipe(turnMetrics("steer"))), + Effect.andThen( + runtime.steerTurn(input).pipe(withinDeadline("steer"), turnMetrics("steer")), + ), ), interruptTurn: (input) => observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe( - Effect.andThen(runtime.interruptTurn(input).pipe(turnMetrics("interrupt"))), + Effect.andThen( + runtime + .interruptTurn(input) + .pipe(withinDeadline("interrupt"), turnMetrics("interrupt")), + ), ), respondToRuntimeRequest: (input) => observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe( Effect.andThen( runtime .respondToRuntimeRequest(input) - .pipe(turnMetrics("runtime-request-response")), + .pipe( + withinDeadline("request response"), + turnMetrics("runtime-request-response"), + ), ), ), };