From 91fb8eb8e44247f97004d0b6a2808e1931b2957c Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Sat, 17 Oct 2026 21:23:59 +0000 Subject: [PATCH 1/3] test(server): reproduce wedged thread after a missed run near a full window A Codex thread at 366k/475k tokens whose previous run failed to start refuses every later turn with ContextHandoffBudgetError. This test fails until the fix lands. Co-Authored-By: Claude Opus 5.5 --- .../ProviderSwitch.integration.test.ts | 109 ++++++++++++++++++ 1 file changed, 109 insertions(+) diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index e5bde378f5c7..27d75e718922 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1066,6 +1066,115 @@ describe("orchestration v2 provider switching", () => { ), ); + it.live("keeps a nearly full native thread usable after a run fails to start", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("handoff-retry-full-window"); + const capturedTurns = yield* Ref.make>([]); + const injectedHistory = yield* Ref.make>([]); + const failStart = yield* Ref.make(false); + const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ + makeTestAdapter({ + instanceId: CODEX_MODEL_SELECTION.instanceId, + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + injectedHistory, + failStartOnce: failStart, + tokenUsageByRunOrdinal: { 1: { usedTokens: 366_046, maxTokens: 475_000 } }, + }), + ]); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const send = (ordinal: number, text: string) => + orchestrator + .dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`full-window:${ordinal}`), + threadId, + messageId: MessageId.make(`full-window:${ordinal}`), + createdBy: "user", + creationSource: "web", + text, + attachments: [], + modelSelection: CODEX_MODEL_SELECTION, + dispatchMode: { type: "start_immediately" }, + }) + .pipe( + Effect.andThen( + orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.ordinal === ordinal && + (event.payload.status === "completed" || event.payload.status === "failed"), + ), + Stream.runHead, + ), + ), + Effect.andThen(worker.drain()), + Effect.andThen(orchestrator.getThreadProjection(threadId)), + Effect.map((projection) => projection.runs.at(-1)!), + ); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("full-window:create"), + threadId, + projectId, + createdBy: "user", + creationSource: "web", + title: "Full window retry", + modelSelection: CODEX_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + assert.equal((yield* send(1, "Long-running work")).status, "completed"); + yield* Ref.set(failStart, true); + assert.equal((yield* send(2, "Request the provider never saw")).status, "failed"); + + const retry = yield* send(3, "Try that again"); + assert.equal(retry.status, "completed"); + assert.equal((yield* Ref.get(capturedTurns)).at(-1)?.text, "Try that again"); + const retried = yield* orchestrator.getThreadProjection(threadId); + const missedRequest = retried.turnItems.find( + (item) => + item.type === "user_message" && item.text === "Request the provider never saw", + )!; + const settled = retried.contextHandoffs.at(-1)!; + assert.equal(settled.strategy, "delta_since_target_last_seen"); + assert.notEqual(settled.delivery?.status, "pending"); + assert.include(settled.delivery?.omittedItemIds ?? [], missedRequest.id); + + assert.equal((yield* send(4, "Next request")).status, "completed"); + assert.equal((yield* Ref.get(capturedTurns)).at(-1)?.text, "Next request"); + assert.lengthOf( + (yield* orchestrator.getThreadProjection(threadId)).contextHandoffs, + retried.contextHandoffs.length, + ); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { + name: "handoff-retry-full-window", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { type: "readOnly" }, + }, + }, + layerRegistry, + ), + ), + ); + }), + ), + ); + it.live.each(["native", "legacy"] as const)( "starts a replacement native thread as new after the resume fallback with %s attempts", (attempts) => From 91ae67f17e9e279f5a381d3e2dae54f333727244 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Sat, 17 Oct 2026 21:32:19 +0000 Subject: [PATCH 2/3] fix(server): a run that failed to start no longer wedges a nearly full thread When a run fails before the provider sees it, the next turn on the same native thread replays the missed items as a retry handoff. The handoff budget reserves a quarter of the window on top of measured usage, so a thread past about 75% full had no budget for even the recovery pointer. Every later turn failed with ContextHandoffBudgetError and added another missed run, until the user compacted. The missed items are optional context for a native thread that already holds the conversation. When they cannot fit, deliver the required handoffs alone and record the retry handoff as omitted once the provider accepts the turn, so it is not offered again. Required handoffs still fail with ContextHandoffBudgetError when they do not fit. Co-Authored-By: Claude Opus 5.5 --- .../ProviderTurnStartService.ts | 50 +++++++++++++++++-- 1 file changed, 46 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 5c1a27cb4737..192b15e56108 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -3,6 +3,7 @@ import { projectComposerContextForProvider } from "@t3tools/shared/composerConte import { CommandId, latestProviderTurnForAttempt, + type OrchestrationV2ContextHandoff, type OrchestrationV2DomainEvent, type OrchestrationV2ExecutionNode, type OrchestrationV2ProviderThread, @@ -1098,8 +1099,7 @@ export const layer: Layer.Layer< createdAt: yield* DateTime.now, }), ]; - const delivery = yield* deliverContextHandoffs({ - handoffs: [...effectiveHandoffs, ...retryHandoff], + const deliveryInput = { deferInline: compact, providerThread: runningProviderThread, budget: Effect.gen(function* () { @@ -1125,7 +1125,7 @@ export const layer: Layer.Layer< ...history, }), }), - persist: (handoff) => + persist: (handoff: OrchestrationV2ContextHandoff) => Effect.gen(function* () { const updatedAt = yield* DateTime.now; yield* eventSink.write({ @@ -1144,7 +1144,49 @@ export const layer: Layer.Layer< ], }); }), - }); + }; + const nativeThreadId = runningProviderThread.nativeThreadRef?.nativeId; + // Missed runs never reached this native thread, so replaying them is + // optional context. When not even their recovery pointer fits, record + // them as omitted instead of refusing every later request. + const delivery = yield* deliverContextHandoffs({ + ...deliveryInput, + handoffs: [...effectiveHandoffs, ...retryHandoff], + }).pipe( + Effect.catchTags({ + ContextHandoffBudgetError: (error) => + retryHandoff.length === 0 || nativeThreadId == null + ? Effect.fail(error) + : Effect.gen(function* () { + const required = yield* deliverContextHandoffs({ + ...deliveryInput, + handoffs: effectiveHandoffs, + }); + const omitted = retryHandoff.map( + (handoff): OrchestrationV2ContextHandoff => ({ + ...handoff, + delivery: { + nativeThreadId, + status: "inline", + itemIds: [], + omittedItemIds: [ + ...(handoff.history?.omittedItemIds ?? []), + ...(handoff.history?.messages ?? []).map((message) => message.itemId), + ], + }, + }), + ); + return { + ...required, + delivered: required.delivered.pipe( + Effect.andThen( + Effect.forEach(omitted, deliveryInput.persist, { discard: true }), + ), + ), + }; + }), + }), + ); if (!(yield* isCurrentAttemptInStatus("running"))) return; const start = compact ? session.compactThread! : session.startTurn; const context = [delivery.context, restartNote] From 8bc03f1a1dc6aa0d0fa0cdda97c96e7b8001b42e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:13:20 +0000 Subject: [PATCH 3/3] fix(server): keep failed-start retries usable in full threads --- .../ProviderTurnStartService.ts | 106 +++++++++--------- .../ProviderSwitch.integration.test.ts | 46 ++++++-- 2 files changed, 90 insertions(+), 62 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 192b15e56108..29a2aa3265cb 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1079,26 +1079,24 @@ export const layer: Layer.Layer< Effect.gen(function* () { const retryHandoff = missedItems.length === 0 - ? [] - : [ - yield* contextHandoffService.prepareProviderHandoff({ - threadId: projection.thread.id, - targetRunId: run.id, - transferId: null, - fromProviderThreadIds: [providerThread.id], - toProviderThreadId: providerThread.id, - fromProviderInstanceId: run.providerInstanceId, - toProviderInstanceId: run.providerInstanceId, - coveredRunOrdinals: { - from: missedRuns[0]!.ordinal, - to: missedRuns.at(-1)!.ordinal, - }, - strategy: "delta_since_target_last_seen", - items: missedItems, - runs: projection.runs, - createdAt: yield* DateTime.now, - }), - ]; + ? undefined + : yield* contextHandoffService.prepareProviderHandoff({ + threadId: projection.thread.id, + targetRunId: run.id, + transferId: null, + fromProviderThreadIds: [providerThread.id], + toProviderThreadId: providerThread.id, + fromProviderInstanceId: run.providerInstanceId, + toProviderInstanceId: run.providerInstanceId, + coveredRunOrdinals: { + from: missedRuns[0]!.ordinal, + to: missedRuns.at(-1)!.ordinal, + }, + strategy: "delta_since_target_last_seen", + items: missedItems, + runs: projection.runs, + createdAt: yield* DateTime.now, + }); const deliveryInput = { deferInline: compact, providerThread: runningProviderThread, @@ -1146,45 +1144,43 @@ export const layer: Layer.Layer< }), }; const nativeThreadId = runningProviderThread.nativeThreadRef?.nativeId; - // Missed runs never reached this native thread, so replaying them is - // optional context. When not even their recovery pointer fits, record - // them as omitted instead of refusing every later request. + // Missed-run recovery is optional in an existing native thread. Required + // handoffs must still fit, and omissions settle only after turn acceptance. const delivery = yield* deliverContextHandoffs({ ...deliveryInput, - handoffs: [...effectiveHandoffs, ...retryHandoff], + handoffs: + retryHandoff === undefined ? effectiveHandoffs : [...effectiveHandoffs, retryHandoff], }).pipe( Effect.catchTags({ - ContextHandoffBudgetError: (error) => - retryHandoff.length === 0 || nativeThreadId == null - ? Effect.fail(error) - : Effect.gen(function* () { - const required = yield* deliverContextHandoffs({ - ...deliveryInput, - handoffs: effectiveHandoffs, - }); - const omitted = retryHandoff.map( - (handoff): OrchestrationV2ContextHandoff => ({ - ...handoff, - delivery: { - nativeThreadId, - status: "inline", - itemIds: [], - omittedItemIds: [ - ...(handoff.history?.omittedItemIds ?? []), - ...(handoff.history?.messages ?? []).map((message) => message.itemId), - ], - }, - }), - ); - return { - ...required, - delivered: required.delivered.pipe( - Effect.andThen( - Effect.forEach(omitted, deliveryInput.persist, { discard: true }), - ), - ), - }; - }), + ContextHandoffBudgetError: Effect.fnUntraced(function* (error) { + if (retryHandoff === undefined || nativeThreadId == null || !sameNativeThread) + return yield* error; + const required = yield* deliverContextHandoffs({ + ...deliveryInput, + handoffs: effectiveHandoffs, + }); + return { + ...required, + delivered: required.delivered.pipe( + Effect.andThen( + deliveryInput.persist({ + ...retryHandoff, + delivery: { + nativeThreadId, + status: "inline", + itemIds: [], + omittedItemIds: [ + ...(retryHandoff.history?.omittedItemIds ?? []), + ...(retryHandoff.history?.messages ?? []).map( + (message) => message.itemId, + ), + ], + }, + }), + ), + ), + }; + }), }), ); if (!(yield* isCurrentAttemptInStatus("running"))) return; diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 27d75e718922..914983079c4f 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1066,13 +1066,14 @@ describe("orchestration v2 provider switching", () => { ), ); - it.live("keeps a nearly full native thread usable after a run fails to start", () => + it.live("omits failed-start recovery but preserves required handoffs", () => Effect.scoped( Effect.gen(function* () { const cwd = yield* checkpointWorkspace("handoff-retry-full-window"); const capturedTurns = yield* Ref.make>([]); const injectedHistory = yield* Ref.make>([]); const failStart = yield* Ref.make(false); + const generation = yield* Ref.make(0); const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ makeTestAdapter({ instanceId: CODEX_MODEL_SELECTION.instanceId, @@ -1083,13 +1084,22 @@ describe("orchestration v2 provider switching", () => { capturedTurns, injectedHistory, failStartOnce: failStart, + nativeThreadGeneration: generation, tokenUsageByRunOrdinal: { 1: { usedTokens: 366_046, maxTokens: 475_000 } }, }), + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + }), ]); yield* Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - const send = (ordinal: number, text: string) => + const send = (ordinal: number, text: string, selection = CODEX_MODEL_SELECTION) => orchestrator .dispatch({ type: "message.dispatch", @@ -1100,7 +1110,7 @@ describe("orchestration v2 provider switching", () => { creationSource: "web", text, attachments: [], - modelSelection: CODEX_MODEL_SELECTION, + modelSelection: selection, dispatchMode: { type: "start_immediately" }, }) .pipe( @@ -1137,9 +1147,7 @@ describe("orchestration v2 provider switching", () => { yield* Ref.set(failStart, true); assert.equal((yield* send(2, "Request the provider never saw")).status, "failed"); - const retry = yield* send(3, "Try that again"); - assert.equal(retry.status, "completed"); - assert.equal((yield* Ref.get(capturedTurns)).at(-1)?.text, "Try that again"); + assert.equal((yield* send(3, "Try that again")).status, "completed"); const retried = yield* orchestrator.getThreadProjection(threadId); const missedRequest = retried.turnItems.find( (item) => @@ -1151,11 +1159,35 @@ describe("orchestration v2 provider switching", () => { assert.include(settled.delivery?.omittedItemIds ?? [], missedRequest.id); assert.equal((yield* send(4, "Next request")).status, "completed"); - assert.equal((yield* Ref.get(capturedTurns)).at(-1)?.text, "Next request"); + assert.deepEqual( + (yield* Ref.get(capturedTurns)).map((turn) => turn.text), + ["Long-running work", "Try that again", "Next request"], + ); + assert.equal(yield* Ref.get(generation), 1); + assert.isEmpty(yield* Ref.get(injectedHistory)); assert.lengthOf( (yield* orchestrator.getThreadProjection(threadId)).contextHandoffs, retried.contextHandoffs.length, ); + + yield* Ref.set(failStart, true); + assert.equal((yield* send(5, "Another missed request")).status, "failed"); + assert.equal( + (yield* send(6, "Work on Claude", CLAUDE_MODEL_SELECTION)).status, + "completed", + ); + const required = yield* send(7, "Return to Codex"); + assert.equal(required.status, "failed"); + const rejected = yield* orchestrator.getThreadProjection(threadId); + const failure = rejected.turnItems.find( + (item) => item.type === "error" && item.runId === required.id, + ); + assert.include( + failure?.type === "error" ? failure.failure.message : "", + "Insufficient context allowance for the provider handoff", + ); + assert.lengthOf(yield* Ref.get(capturedTurns), 4); + assert.isEmpty(yield* Ref.get(injectedHistory)); }).pipe( Effect.provide( ProviderReplayHarness.layerWithRegistry(