From 8655ad806e89ac7ffb3c1775585e2bcb7f35f894 Mon Sep 17 00:00:00 2001 From: Lakshmi Tanmay Date: Wed, 7 Oct 2026 15:47:14 +0530 Subject: [PATCH 1/4] fix(server): one silent provider no longer stalls every thread An effect that outlives its 30s lease now gives its worker slot back while it keeps running, and provider control requests fail after 2 minutes without an answer, so a wedged provider fails its own runs instead of holding all four effect workers. Fixes #16636 Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/orchestration-v2/EffectWorker.test.ts | 92 ++++++++++++ .../src/orchestration-v2/EffectWorker.ts | 136 +++++++++++------- .../ProviderSessionManager.test.ts | 67 +++++++++ .../ProviderSessionManager.ts | 56 ++++++-- 4 files changed, 289 insertions(+), 62 deletions(-) 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..7b4aee7f9ee1 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -1316,6 +1316,73 @@ 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, + 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..22fea56e7af9 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,25 @@ 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`, + }), + ), + }), + ); return { ...runtime, subscribeEvents, @@ -1511,7 +1536,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 +1551,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 +1574,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 +1602,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 +1643,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( @@ -1633,18 +1664,27 @@ export const layerWithOptions = ( ), 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"), + ), ), ), }; From c38fa760ce3e9a5e8291a84fc27f972993b493be Mon Sep 17 00:00:00 2001 From: Lakshmi Tanmay Date: Wed, 7 Oct 2026 15:57:27 +0530 Subject: [PATCH 2/4] fix(server): bound compaction requests and fix test layer options Compaction now gets the same 2-minute deadline as the other provider control requests, and the new turn-start deadline test passes the idleTimeoutMs its layer requires. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/orchestration-v2/ProviderSessionManager.test.ts | 1 + apps/server/src/orchestration-v2/ProviderSessionManager.ts | 6 ++++++ 2 files changed, 7 insertions(+) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 7b4aee7f9ee1..41788c21aa78 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -1374,6 +1374,7 @@ it.effect("ProviderSessionManagerV2 fails a turn start the provider never answer Effect.provide( layerTest({ state, + idleTimeoutMs: 60_000, startTurn: Deferred.succeed(startTurnReached, undefined).pipe( Effect.andThen(Effect.never), ), diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 22fea56e7af9..84dbd2f0c224 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -1521,6 +1521,7 @@ export const layerWithOptions = ( ), }), ); + const compactThread = runtime.compactThread; return { ...runtime, subscribeEvents, @@ -1662,6 +1663,11 @@ export const layerWithOptions = ( ), ), ), + ...(compactThread === undefined + ? {} + : { + compactThread: (input) => compactThread(input).pipe(withinDeadline("compaction")), + }), steerTurn: (input) => observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe( Effect.andThen( From 2850a0eb82e6963af913a544e84cb3bc01772d40 Mon Sep 17 00:00:00 2001 From: Lakshmi Tanmay Date: Wed, 7 Oct 2026 16:18:37 +0530 Subject: [PATCH 3/4] fix(server): stop Codex turns accepted after their start was abandoned When the turn-start deadline (or a Stop) gives up on turn/start before Codex answers, the adapter forgets which run the start belonged to. If Codex starts the turn anyway, the late turn/started had no run and kept working out of sight. The adapter now remembers the abandoned start and interrupts the turn Codex starts for it. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/CodexAdapterV2.test.ts | 64 +++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 29 ++++++++- 2 files changed, 92 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 63f0cbdae66b..8d07510ac7b7 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2285,6 +2285,70 @@ 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("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..d57f05c1a7e9 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1778,6 +1778,10 @@ 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. A turn Codex starts for one later has no run + // to report to. + const abandonedTurnStarts = new Set(); const turnWaiters = yield* Ref.make(new Map>()); const subagentThreads = yield* Ref.make(new Map()); const subagentModels = new Map(); @@ -4116,6 +4120,26 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi yield* Deferred.succeed(goalHold.next, next); return; } + // Codex accepted a start this adapter already gave up on, and the + // run has settled without it: stop the turn rather than let it work + // out of sight. + if (abandonedTurnStarts.delete(payload.threadId)) { + yield* Effect.logWarning("orchestration-v2.codex-abandoned-turn-started", { + nativeThreadId: payload.threadId, + nativeTurnId: payload.turn.id, + }); + yield* client + .request("turn/interrupt", { threadId: payload.threadId, turnId: payload.turn.id }) + .pipe( + Effect.catch((cause) => + Effect.logWarning("orchestration-v2.codex-abandoned-turn-stop-failed", { + cause, + }), + ), + Effect.forkIn(scope), + ); + return; + } // A goal turn with no run to own it (Codex continued after the run // settled, or raced a Stop): stop it and pause the goal rather than // work out of sight. @@ -5887,7 +5911,10 @@ 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.sync(() => abandonedTurnStarts.add(threadId)))); + abandonedTurnStarts.delete(threadId); yield* registerRootTurn({ turnInput, nativeTurnId: started.turn.id, From 00fdf2857acfe88f069d1adf3bda310649f4e037 Mon Sep 17 00:00:00 2001 From: Lakshmi Tanmay Date: Wed, 7 Oct 2026 16:43:13 +0530 Subject: [PATCH 4/4] fix(server): match abandoned Codex starts to their own turns An abandoned turn/start whose turn Codex had already reported started is now interrupted right away instead of only being marked. The turn Codex starts for an abandoned start is now checked before a newer pending start on the same thread, so the retry no longer adopts it. Codex takes a thread's starts in order, so that turn belongs to the abandoned start. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/CodexAdapterV2.test.ts | 139 ++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 63 ++++---- 2 files changed, 177 insertions(+), 25 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 8d07510ac7b7..4a947353682a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2349,6 +2349,145 @@ describe("CodexAdapterV2 post-settle continuation", () => { }).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 d57f05c1a7e9..7b2dc8f19fdd 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1779,9 +1779,24 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi }; const pendingRootTurns = yield* Ref.make(new Map()); // Native threads whose turn/start was given up before Codex answered, - // by a deadline or a Stop. A turn Codex starts for one later has no run - // to report to. + // 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(); @@ -4080,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({ @@ -4120,26 +4141,6 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi yield* Deferred.succeed(goalHold.next, next); return; } - // Codex accepted a start this adapter already gave up on, and the - // run has settled without it: stop the turn rather than let it work - // out of sight. - if (abandonedTurnStarts.delete(payload.threadId)) { - yield* Effect.logWarning("orchestration-v2.codex-abandoned-turn-started", { - nativeThreadId: payload.threadId, - nativeTurnId: payload.turn.id, - }); - yield* client - .request("turn/interrupt", { threadId: payload.threadId, turnId: payload.turn.id }) - .pipe( - Effect.catch((cause) => - Effect.logWarning("orchestration-v2.codex-abandoned-turn-stop-failed", { - cause, - }), - ), - Effect.forkIn(scope), - ); - return; - } // A goal turn with no run to own it (Codex continued after the run // settled, or raced a Stop): stop it and pause the goal rather than // work out of sight. @@ -5911,9 +5912,21 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi else next.delete(threadId); return next; }); - const started = yield* client - .request("turn/start", turnStartParams) - .pipe(Effect.onInterrupt(() => Effect.sync(() => abandonedTurnStarts.add(threadId)))); + 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,