From 7b97444b04263a3c444becb64d0dfc2d3f0ae637 Mon Sep 17 00:00:00 2001 From: TinBane <363069+TinBane@users.noreply.github.com> Date: Sun, 25 Oct 2026 09:58:12 +1100 Subject: [PATCH 1/8] fix(server): restart oversized provider handoffs safely Preflight provider-switch handoffs against resumed and fresh native sessions. Replace only when fresh capacity can deliver the full app history, and preserve compact, retry, and oversized-input behavior. Co-Authored-By: GPT Daybreak Blue --- .../ContextHandoffDelivery.test.ts | 55 +++ .../ContextHandoffDelivery.ts | 116 +++-- .../ProviderTurnStartService.ts | 395 +++++++++++------- .../ProviderSwitch.integration.test.ts | 77 +++- 4 files changed, 431 insertions(+), 212 deletions(-) create mode 100644 apps/server/src/orchestration-v2/ContextHandoffDelivery.test.ts diff --git a/apps/server/src/orchestration-v2/ContextHandoffDelivery.test.ts b/apps/server/src/orchestration-v2/ContextHandoffDelivery.test.ts new file mode 100644 index 0000000000000..00c7f727853df --- /dev/null +++ b/apps/server/src/orchestration-v2/ContextHandoffDelivery.test.ts @@ -0,0 +1,55 @@ +import { assert, it } from "@effect/vitest"; +import { + ProviderDriverKind, + ProviderInstanceId, + ProviderSessionId, + ProviderThreadId, + ThreadId, + type OrchestrationV2ProviderThread, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import { deliverContextHandoffs } from "./ContextHandoffDelivery.ts"; + +it.effect("does not evaluate the budget when no handoff needs delivery", () => + Effect.gen(function* () { + let budgetEvaluated = false; + const now = DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"); + const providerThread: OrchestrationV2ProviderThread = { + id: ProviderThreadId.make("provider-thread:settled"), + driver: ProviderDriverKind.make("acp"), + providerInstanceId: ProviderInstanceId.make("kiro"), + providerSessionId: ProviderSessionId.make("provider-session:settled"), + appThreadId: ThreadId.make("thread:settled"), + ownerNodeId: null, + nativeThreadRef: { + driver: ProviderDriverKind.make("acp"), + nativeId: "native:settled", + strength: "strong", + }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + contextUsage: null, + forkedFrom: null, + createdAt: now, + updatedAt: now, + }; + + yield* deliverContextHandoffs({ + handoffs: [], + providerThread, + budget: Effect.sync(() => { + budgetEvaluated = true; + return 0; + }), + alreadyDeliveredItemIds: new Set(), + persist: () => Effect.void, + }); + + assert.isFalse(budgetEvaluated); + }), +); diff --git a/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts b/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts index ff2ca1fbec441..232fe527d6527 100644 --- a/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts +++ b/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts @@ -12,6 +12,68 @@ import { selectHistory, } from "@t3tools/provider-core/server/handoffBudget"; +function pendingContextHandoffs(input: { + readonly handoffs: ReadonlyArray; + readonly providerThread: OrchestrationV2ProviderThread; +}) { + const nativeThreadId = input.providerThread.nativeThreadRef?.nativeId ?? undefined; + return input.handoffs.filter( + (handoff) => + nativeThreadId === undefined || + handoff.delivery?.nativeThreadId !== nativeThreadId || + handoff.delivery.status === "pending", + ); +} + +export function planContextHandoffDelivery(input: { + readonly handoffs: ReadonlyArray; + readonly providerThread: OrchestrationV2ProviderThread; + readonly budget: number; + readonly alreadyDeliveredItemIds: ReadonlySet; +}) { + const nativeThreadId = input.providerThread.nativeThreadRef?.nativeId ?? undefined; + const pending = pendingContextHandoffs(input); + if (pending.length === 0) return { status: "empty" as const }; + let coverage = pending + .map( + (handoff) => + `Context handoff (${handoff.strategy === "fork_delta_summary" ? "merge_back / fork_delta_summary" : handoff.strategy}):\n${ + handoff.history?.coverage ?? + `From thread ${handoff.threadId}, runs ${handoff.coveredRunOrdinals.from}-${handoff.coveredRunOrdinals.to}. Recover history with t3_thread_read, view=activity; paginate with afterPosition, and use itemId/textOffset for long items.` + }`, + ) + .join("\n"); + if (historyCost([], coverage) > Math.min(4_000, input.budget / 2)) { + const strategies = Array.from(new Set(pending.map((handoff) => handoff.strategy))); + coverage = `Context handoff (${strategies.join(", ")}). ${pending.length} handoff records; detailed coverage references omitted. Recover history with t3_thread_read({threadId:"${input.providerThread.appThreadId ?? pending[0]!.threadId}",view:"activity",limit:20,maxCharsPerItem:4000}); paginate with afterPosition=nextPosition. Follow fork/handoff source references in activity. For long items use itemId and textOffset=nextTextOffset until null.`; + } + const seen = new Set(input.alreadyDeliveredItemIds); + const messages = pending + .flatMap((handoff) => handoff.history?.messages ?? []) + .filter((message) => { + if (seen.has(message.itemId)) return false; + seen.add(message.itemId); + return true; + }); + const oldContext = pending + .filter((handoff) => handoff.history === undefined) + .map((handoff) => handoff.summaryText) + .join("\n\n"); + const fullCoverage = + oldContext && historyCost([], `${coverage}\n${oldContext}`) + 512 <= input.budget + ? `${coverage}\n${oldContext}` + : coverage; + const selected = selectHistory({ + messages, + coverage: fullCoverage, + omittedItems: pending.reduce((sum, handoff) => sum + (handoff.history?.omittedItems ?? 0), 0), + budget: input.budget, + }); + return historyCost(selected.messages, selected.context) > input.budget + ? { status: "overflow" as const } + : { status: "ready" as const, pending, selected, nativeThreadId }; +} + /** Persist around injection; ambiguous delivery requires a fresh native thread. Call unsent if refused before prompt consumption. */ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextHandoffs")( function* (input: { @@ -25,55 +87,25 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH ) => Effect.Effect; readonly persist: (handoff: OrchestrationV2ContextHandoff) => Effect.Effect; }) { - const nativeThreadId = input.providerThread.nativeThreadRef?.nativeId ?? undefined; - const pending = input.handoffs.filter( - (handoff) => - nativeThreadId === undefined || - handoff.delivery?.nativeThreadId !== nativeThreadId || - handoff.delivery.status === "pending", - ); - if (pending.length === 0 || (input.deferInline && input.inject === undefined)) + if ( + pendingContextHandoffs(input).length === 0 || + (input.deferInline && input.inject === undefined) + ) return { context: "", delivered: Effect.void, unsent: Effect.void }; const budget = typeof input.budget === "number" ? input.budget : yield* input.budget; - let coverage = pending - .map( - (handoff) => - `Context handoff (${handoff.strategy === "fork_delta_summary" ? "merge_back / fork_delta_summary" : handoff.strategy}):\n${ - handoff.history?.coverage ?? - `From thread ${handoff.threadId}, runs ${handoff.coveredRunOrdinals.from}-${handoff.coveredRunOrdinals.to}. Recover history with t3_thread_read, view=activity; paginate with afterPosition, and use itemId/textOffset for long items.` - }`, - ) - .join("\n"); - if (historyCost([], coverage) > Math.min(4_000, budget / 2)) { - const strategies = Array.from(new Set(pending.map((handoff) => handoff.strategy))); - coverage = `Context handoff (${strategies.join(", ")}). ${pending.length} handoff records; detailed coverage references omitted. Recover history with t3_thread_read({threadId:"${input.providerThread.appThreadId ?? pending[0]!.threadId}",view:"activity",limit:20,maxCharsPerItem:4000}); paginate with afterPosition=nextPosition. Follow fork/handoff source references in activity. For long items use itemId and textOffset=nextTextOffset until null.`; - } - const seen = new Set(input.alreadyDeliveredItemIds); - const messages = pending - .flatMap((handoff) => handoff.history?.messages ?? []) - .filter((message) => { - if (seen.has(message.itemId)) return false; - seen.add(message.itemId); - return true; - }); - const oldContext = pending - .filter((handoff) => handoff.history === undefined) - .map((handoff) => handoff.summaryText) - .join("\n\n"); - const fullCoverage = - oldContext && historyCost([], `${coverage}\n${oldContext}`) + 512 <= budget - ? `${coverage}\n${oldContext}` - : coverage; - const selected = selectHistory({ - messages, - coverage: fullCoverage, - omittedItems: pending.reduce((sum, handoff) => sum + (handoff.history?.omittedItems ?? 0), 0), + const plan = planContextHandoffDelivery({ + handoffs: input.handoffs, + providerThread: input.providerThread, budget, + alreadyDeliveredItemIds: input.alreadyDeliveredItemIds, }); - if (historyCost(selected.messages, selected.context) > budget) { + if (plan.status === "empty") + return { context: "", delivered: Effect.void, unsent: Effect.void }; + if (plan.status === "overflow") { if (input.deferInline) return { context: "", delivered: Effect.void, unsent: Effect.void }; return yield* new ContextHandoffBudgetError(); } + const { nativeThreadId, pending, selected } = plan; const omittedItemIds = new Set(selected.omittedItemIds); const persist = (status: "pending" | "injected" | "inline") => Effect.forEach( diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 97fdb7c85ca71..80f06d64b09c0 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -43,7 +43,7 @@ import { historicalMessage, latestNativeContextUsage, } from "@t3tools/provider-core/server/handoffBudget"; -import { deliverContextHandoffs } from "./ContextHandoffDelivery.ts"; +import { deliverContextHandoffs, planContextHandoffDelivery } from "./ContextHandoffDelivery.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; @@ -695,7 +695,116 @@ export const layer: Layer.Layer< }); let effectiveHandoffs = handoffs; let resumeFellBack = false; - const loadedProviderThread = yield* Effect.gen(function* () { + const replaceNativeThread = Effect.fn("ProviderTurnStartService.replaceNativeThread")( + function* (replacementInput: { + readonly reason: "resume_failed" | "uncertain_history_delivery" | "handoff_overflow"; + readonly errorTag?: string; + }) { + yield* Effect.logWarning("Replacing the native provider thread with portable context", { + driver: session.driver, + providerThreadId: providerThread.id, + runId, + reason: replacementInput.reason, + ...(replacementInput.errorTag === undefined + ? {} + : { errorTag: replacementInput.errorTag }), + }); + const replacement = yield* loadFromProvider( + session.ensureThread({ + threadId: projection.thread.id, + title: projection.thread.title, + modelSelection: run.modelSelection, + runtimePolicy: resolvedRuntimePolicy, + providerSessionId, + // Keep the durable row but force the adapter to bind fresh native state. + existingProviderThread: { ...providerThread, nativeThreadRef: null }, + }), + ); + if (replacement === undefined) return undefined; + if (!(yield* isCurrentAttemptInStatus("starting"))) return undefined; + const transferId = yield* idAllocator.allocate.contextTransfer({ + sourceThreadId: projection.thread.id, + targetThreadId: projection.thread.id, + type: "provider_resume_fallback", + }); + const createdAt = yield* DateTime.now; + const handoff = yield* contextHandoffService.prepareProviderHandoff({ + threadId: projection.thread.id, + targetRunId: run.id, + transferId, + fromProviderThreadIds: [providerThread.id], + toProviderThreadId: providerThread.id, + fromProviderInstanceId: providerThread.providerInstanceId, + toProviderInstanceId: run.providerInstanceId, + coveredRunOrdinals: { from: 1, to: Math.max(1, run.ordinal - 1) }, + strategy: "full_thread_summary", + runs: projection.runs, + items: (yield* projectionStore.getTurnStartHistory(input.threadId)).filter( + (item) => + item.runId === null || + projection.runs.some( + (source) => source.id === item.runId && source.ordinal < run.ordinal, + ), + ), + createdAt, + }); + const replacementWrite = yield* eventSink.writeIfRunCurrent({ + threadId: projection.thread.id, + runId: run.id, + activeAttemptId: attempt.id, + expectedStatus: "starting", + events: [ + { + id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), + type: "context-handoff.updated", + threadId: projection.thread.id, + runId: run.id, + providerInstanceId: run.providerInstanceId, + occurredAt: createdAt, + payload: handoff, + }, + { + id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), + type: "context-transfer.updated", + threadId: projection.thread.id, + runId: run.id, + providerInstanceId: run.providerInstanceId, + occurredAt: createdAt, + payload: { + id: transferId, + type: "provider_handoff", + sourceThreadId: projection.thread.id, + targetThreadId: projection.thread.id, + sourcePoint: { threadId: projection.thread.id }, + basePoint: null, + sourceProviderInstanceId: providerThread.providerInstanceId, + targetProviderInstanceId: run.providerInstanceId, + targetRunId: run.id, + status: "resolved_portable", + resolution: { strategy: "portable_context", contextHandoffId: handoff.id }, + createdBy: "system", + error: null, + createdAt, + updatedAt: createdAt, + consumedAt: null, + }, + }, + ], + }); + if (!replacementWrite.committed) return undefined; + // Delivery ranks later handoffs as newer history. Pending handoffs from + // earlier runs predate this summary; handoffs for this run (provider-switch + // delta, merge-back) are newer and must stay last. + effectiveHandoffs = [ + ...effectiveHandoffs.filter((candidate) => candidate.targetRunId !== run.id), + handoff, + ...effectiveHandoffs.filter((candidate) => candidate.targetRunId === run.id), + ]; + resumeFellBack = true; + return replacement; + }, + ); + let loadedProviderThread = yield* Effect.gen(function* () { if (nativeForkTransfer !== undefined) { const sourceProjection = yield* projectionStore.getThreadRecords( nativeForkTransfer.sourceThreadId, @@ -775,100 +884,10 @@ export const layer: Layer.Layer< return resumed.success; } - yield* Effect.logWarning("Provider resume failed; attempting a fresh native session", { - driver: session.driver, - providerThreadId: providerThread.id, - runId, + return yield* replaceNativeThread({ reason: uncertainDelivery ? "uncertain_history_delivery" : "resume_failed", errorTag: resumed.failure._tag, }); - const replacement = yield* loadFromProvider( - session.ensureThread({ - threadId: projection.thread.id, - title: projection.thread.title, - modelSelection: run.modelSelection, - runtimePolicy: resolvedRuntimePolicy, - providerSessionId, - // Clear the native ref to force a fresh session while preserving row identity. - existingProviderThread: { ...providerThread, nativeThreadRef: null }, - }), - ); - if (replacement === undefined) return undefined; - const transferId = yield* idAllocator.allocate.contextTransfer({ - sourceThreadId: projection.thread.id, - targetThreadId: projection.thread.id, - type: "provider_resume_fallback", - }); - const createdAt = yield* DateTime.now; - const handoff = yield* contextHandoffService.prepareProviderHandoff({ - threadId: projection.thread.id, - targetRunId: run.id, - transferId, - fromProviderThreadIds: [providerThread.id], - toProviderThreadId: providerThread.id, - fromProviderInstanceId: providerThread.providerInstanceId, - toProviderInstanceId: run.providerInstanceId, - coveredRunOrdinals: { from: 1, to: Math.max(1, run.ordinal - 1) }, - strategy: "full_thread_summary", - runs: projection.runs, - items: (yield* projectionStore.getTurnStartHistory(input.threadId)).filter( - (item) => - item.runId === null || - projection.runs.some( - (source) => source.id === item.runId && source.ordinal < run.ordinal, - ), - ), - createdAt, - }); - // Delivery ranks later handoffs as newer history. Pending handoffs from - // earlier runs predate this summary; handoffs for this run (provider-switch - // delta, merge-back) are newer and must stay last. - effectiveHandoffs = [ - ...effectiveHandoffs.filter((candidate) => candidate.targetRunId !== run.id), - handoff, - ...effectiveHandoffs.filter((candidate) => candidate.targetRunId === run.id), - ]; - resumeFellBack = true; - yield* eventSink.write({ - events: [ - { - id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), - type: "context-handoff.updated", - threadId: projection.thread.id, - runId: run.id, - providerInstanceId: run.providerInstanceId, - occurredAt: createdAt, - payload: handoff, - }, - { - id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), - type: "context-transfer.updated", - threadId: projection.thread.id, - runId: run.id, - providerInstanceId: run.providerInstanceId, - occurredAt: createdAt, - payload: { - id: transferId, - type: "provider_handoff", - sourceThreadId: projection.thread.id, - targetThreadId: projection.thread.id, - sourcePoint: { threadId: projection.thread.id }, - basePoint: null, - sourceProviderInstanceId: providerThread.providerInstanceId, - targetProviderInstanceId: run.providerInstanceId, - targetRunId: run.id, - status: "resolved_portable", - resolution: { strategy: "portable_context", contextHandoffId: handoff.id }, - createdBy: "system", - error: null, - createdAt, - updatedAt: createdAt, - consumedAt: null, - }, - }, - ], - }); - return replacement; }); if (loadedProviderThread === undefined) return; if (!(yield* isCurrentAttemptInStatus("starting"))) { @@ -884,30 +903,137 @@ export const layer: Layer.Layer< source.startedAt !== null && source.providerThreadId === providerThread.id, )?.modelSelection; - const sameSelection = - previousSelection === undefined || - modelSelectionsEqual(previousSelection, run.modelSelection); - const sameNativeThread = - loadedProviderThread.nativeThreadRef?.nativeId === providerThread.nativeThreadRef?.nativeId; - const threadUsage = loadedProviderThread.contextUsage ?? providerThread.contextUsage; - const previousUsage = measuredContext - ? { ...threadUsage, ...measuredContext.usage } - : threadUsage; - const reuseTelemetry = - sameSelection || - (previousSelection !== undefined && - session.canReuseContextUsage?.(previousSelection, run.modelSelection) === true); - const knownModelWindow = session.getModelContextWindow?.( - run.modelSelection, - resolvedRuntimePolicy.cwd, + const usageStateFor = (loaded: OrchestrationV2ProviderThread) => { + const sameSelection = + previousSelection === undefined || + modelSelectionsEqual(previousSelection, run.modelSelection); + const sameNativeThread = + loaded.nativeThreadRef?.nativeId === providerThread.nativeThreadRef?.nativeId; + const threadUsage = loaded.contextUsage ?? providerThread.contextUsage; + const previousUsage = measuredContext + ? { ...threadUsage, ...measuredContext.usage } + : threadUsage; + const reuseTelemetry = + sameSelection || + (previousSelection !== undefined && + session.canReuseContextUsage?.(previousSelection, run.modelSelection) === true); + const knownModelWindow = session.getModelContextWindow?.( + run.modelSelection, + resolvedRuntimePolicy.cwd, + ); + const handoffUsage = contextUsageForHandoff({ + sameNativeThread, + sameSelection, + reuseTelemetry, + previousUsage, + knownModelWindow, + }); + return { + sameNativeThread, + handoffUsage, + modelContextWindow: + knownModelWindow ?? + (handoffUsage !== null || reuseTelemetry ? previousUsage?.maxTokens : undefined), + }; + }; + let usageState = usageStateFor(loadedProviderThread); + const userText = projectComposerContextForProvider({ + text: message.text, + records: message.context?.records ?? [], + }); + // Codex and Claude parse /goal from the prompt, so it goes verbatim and history waits a turn. + const goalDriver = session.driver === "codex" || session.driver === "claudeAgent"; + const goalCommand = goalDriver && parseGoalCommand(message) !== null; + // The provider turn marks cancellation work delivered; restart continuations use their own text or resume natively. + const noteContinuation = isRestartNoteContinuation( + run, + projection.runs, + projection.providerTurns, + projection.attempts, ); - const handoffUsage = contextUsageForHandoff({ - sameNativeThread, - sameSelection, - reuseTelemetry, - previousUsage, - knownModelWindow, + const restartCancelledWork = pendingRestartCancelledBackgroundWork({ + runs: projection.runs, + providerTurns: projection.providerTurns, + compactionMessageIds: new Set( + projection.messages + .filter( + (candidate) => + isCompactionCommandMessage(candidate) || + (goalDriver && parseGoalCommand(candidate) !== null), + ) + .map((candidate) => candidate.id), + ), + run, + attempts: projection.attempts, }); + const restartNote = + restartCancelledWork.length === 0 + ? "" + : restartCancelledBackgroundWorkNote(restartCancelledWork); + const tokenCap = yield* handoffTokenCapConfig.pipe( + Effect.orElseSucceed(() => DEFAULT_HANDOFF_TOKEN_CAP), + ); + const providerSwitchHandoff = effectiveHandoffs.some( + (handoff) => + handoff.targetRunId === run.id && + handoff.transferId != null && + (handoff.strategy === "delta_since_target_last_seen" || + handoff.strategy === "full_thread_summary"), + ); + const compact = isCompactionCommandMessage(message); + const deliveredAttemptIdsAtStart = new Set( + projection.providerTurns.map((turn) => turn.runAttemptId), + ); + const hasMissedProviderInput = projection.runs.some( + (source) => + source.ordinal < run.ordinal && + source.providerThreadId === providerThread.id && + (source.status === "failed" || source.status === "interrupted") && + !deliveredAttemptIdsAtStart.has(source.activeAttemptId), + ); + if ( + providerSwitchHandoff && + usageState.sameNativeThread && + usageState.handoffUsage?.usedTokens !== undefined && + !compact && + !goalCommand && + !noteContinuation && + restartNote === "" && + !hasMissedProviderInput + ) { + const budget = handoffBudget({ + tokenCap, + modelContextWindow: usageState.modelContextWindow, + userText, + attachments: message.attachments, + providerThread: { ...loadedProviderThread, contextUsage: usageState.handoffUsage }, + nativeContextEstimate: 0, + }); + const deliveryInput = { + handoffs: effectiveHandoffs, + providerThread: loadedProviderThread, + alreadyDeliveredItemIds: new Set(), + }; + if (planContextHandoffDelivery({ ...deliveryInput, budget }).status === "overflow") { + const freshBudget = handoffBudget({ + tokenCap, + modelContextWindow: usageState.modelContextWindow, + userText, + attachments: message.attachments, + providerThread: { ...loadedProviderThread, contextUsage: null }, + nativeContextEstimate: 0, + }); + if ( + planContextHandoffDelivery({ ...deliveryInput, budget: freshBudget }).status === "ready" + ) { + const replacement = yield* replaceNativeThread({ reason: "handoff_overflow" }); + if (replacement === undefined) return; + loadedProviderThread = replacement; + usageState = usageStateFor(replacement); + } + } + } + const { sameNativeThread, handoffUsage, modelContextWindow } = usageState; const runningProviderThread: OrchestrationV2ProviderThread = { ...loadedProviderThread, contextUsage: handoffUsage, @@ -1035,42 +1161,6 @@ export const layer: Layer.Layer< const routableSubagents = projection.subagents.filter((subagent) => RunExecutionService.canRouteRelatedSubagent(subagent.status), ); - const userText = projectComposerContextForProvider({ - text: message.text, - records: message.context?.records ?? [], - }); - // Codex and Claude parse /goal from the prompt, so it goes verbatim and history waits a turn. - const goalDriver = session.driver === "codex" || session.driver === "claudeAgent"; - const goalCommand = goalDriver && parseGoalCommand(message) !== null; - // The provider turn marks cancellation work delivered; restart continuations use their own text or resume natively. - const noteContinuation = isRestartNoteContinuation( - run, - projection.runs, - projection.providerTurns, - projection.attempts, - ); - const restartCancelledWork = pendingRestartCancelledBackgroundWork({ - runs: projection.runs, - providerTurns: projection.providerTurns, - compactionMessageIds: new Set( - projection.messages - .filter( - (candidate) => - isCompactionCommandMessage(candidate) || - (goalDriver && parseGoalCommand(candidate) !== null), - ) - .map((candidate) => candidate.id), - ), - run, - attempts: projection.attempts, - }); - const restartNote = - restartCancelledWork.length === 0 - ? "" - : restartCancelledBackgroundWorkNote(restartCancelledWork); - const tokenCap = yield* handoffTokenCapConfig.pipe( - Effect.orElseSucceed(() => DEFAULT_HANDOFF_TOKEN_CAP), - ); const settledHandoffs = projection.contextHandoffs.filter( (handoff) => handoff.toProviderThreadId === providerThread.id && @@ -1151,9 +1241,6 @@ export const layer: Layer.Layer< }, 0) : 0; }); - const modelContextWindow = - knownModelWindow ?? - (handoffUsage !== null || reuseTelemetry ? previousUsage?.maxTokens : undefined); const budgetProviderThread = { ...runningProviderThread, contextUsage: handoffUsage, 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 452509bb97b0d..d38f55e08e26e 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -412,6 +412,9 @@ describe("orchestration v2 provider switching", () => { "screenshot-fallback", "screenshot-reported-capacity-native", "screenshot-option-change-retry-native", + "prior-images-reasoning-change-overflow-native", + "prior-images-reasoning-change-overflow-oversized-native", + "prior-images-reasoning-change-overflow-compact-native", "prior-images-reasoning-change-replacement-small-native", "imported-prior-images-fallback", "prior-images-telemetry-native", @@ -433,6 +436,7 @@ describe("orchestration v2 provider switching", () => { const generation = yield* Ref.make(0); const priorImages = scenario.includes("prior-images"); const replaceNative = scenario.includes("replacement"); + const replaceForOverflow = scenario.includes("overflow"); const capacityScenario = modelScenario || scenario.includes("reported-capacity"); const returning = scenario === "large-current-input" || scenario.includes("-change-") || priorImages; @@ -463,7 +467,11 @@ describe("orchestration v2 provider switching", () => { responseByRunOrdinal: {}, capturedTurns, ...(scenario.includes("retry") || scenario.includes("unsent") ? { failStartOnce } : {}), - ...(replaceNative ? { failResumeOnce, nativeThreadGeneration: generation } : {}), + ...(replaceNative + ? { failResumeOnce, nativeThreadGeneration: generation } + : replaceForOverflow + ? { nativeThreadGeneration: generation } + : {}), ...(scenario.includes("reported-capacity") ? { initialContextUsage: { usedTokens: 999_999, maxTokens: 1_000_000 } } : {}), @@ -472,7 +480,7 @@ describe("orchestration v2 provider switching", () => { canReuseContextUsage: canReuseCodexContextUsage, tokenUsageByRunOrdinal: { 2: { - usedTokens: replaceNative ? 30_000 : 37_321, + usedTokens: replaceNative ? 30_000 : replaceForOverflow ? 250_000 : 37_321, maxTokens: replaceNative ? scenario.includes("small") ? 32_000 @@ -537,13 +545,15 @@ describe("orchestration v2 provider switching", () => { ...screenshot, id: `prior-${index}`, })) - : turnUsageScenario && ordinal >= targetOrdinal - ? [screenshot, { ...screenshot, id: "current-2" }] - : scenario.startsWith("screenshot") && ordinal >= targetOrdinal - ? screenshots - : priorImages && ordinal === (scenario.startsWith("imported") ? 1 : 2) - ? [screenshot] - : [], + : scenario.includes("overflow-compact") && ordinal >= targetOrdinal + ? [] + : turnUsageScenario && ordinal >= targetOrdinal + ? [screenshot, { ...screenshot, id: "current-2" }] + : scenario.startsWith("screenshot") && ordinal >= targetOrdinal + ? screenshots + : priorImages && ordinal === (scenario.startsWith("imported") ? 1 : 2) + ? [screenshot] + : [], modelSelection: selection, dispatchMode: { type: "start_immediately" }, }); @@ -593,13 +603,17 @@ describe("orchestration v2 provider switching", () => { ) : undefined; yield* Effect.addFinalizer(() => Effect.sync(() => spy?.mockRestore())); - const current = scenario.startsWith("compact") - ? "/compact" - : capacityScenario && !turnUsageScenario - ? "x".repeat(70_000) - : scenario === "large-current-input" - ? "x".repeat(9_000) - : "Continue work"; + const current = scenario.includes("overflow-oversized") + ? "x".repeat(300_000) + : scenario.includes("overflow-compact") + ? "/compact" + : scenario.startsWith("compact") + ? "/compact" + : capacityScenario && !turnUsageScenario + ? "x".repeat(70_000) + : scenario === "large-current-input" + ? "x".repeat(9_000) + : "Continue work"; if (returning) { if (scenario.includes("unsent")) yield* Ref.set(failStartOnce, true); yield* dispatch( @@ -733,6 +747,37 @@ describe("orchestration v2 provider switching", () => { ); return; } + if (replaceForOverflow) { + const replaced = yield* orchestrator.getThreadProjection(threadId); + if (scenario.includes("overflow-oversized")) { + assert.equal(replaced.runs.at(-1)?.status, "failed"); + const target = replaced.providerThreads.find( + (thread) => thread.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, + )!; + assert.equal(yield* Ref.get(generation), 1); + assert.equal(target.nativeThreadRef?.nativeId?.endsWith(":0"), true); + return; + } + assert.equal(replaced.runs.at(-1)?.status, "completed"); + const target = replaced.providerThreads.find( + (thread) => thread.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, + )!; + if (scenario.includes("overflow-compact")) { + assert.equal(yield* Ref.get(generation), 1); + assert.equal(target.nativeThreadRef?.nativeId?.endsWith(":0"), true); + return; + } + assert.isNull(target.contextUsage); + assert.equal(yield* Ref.get(generation), 2); + const accepted = (yield* Ref.get(capturedTurns)).at(-1)!; + assert.equal(accepted.driver, CLAUDE_DRIVER); + assert.equal(accepted.nativeThreadId?.endsWith(":1"), true); + assert.equal(accepted.text, "Continue work"); + const history = yield* encodeJson(yield* Ref.get(injectedHistory)); + assert.include(history, "Original request with constraints"); + assert.include(history, "New source constraint"); + return; + } if (replaceNative) { if (scenario.includes("legacy-replacement")) { const recovered = yield* orchestrator.getThreadProjection(threadId); From c26e1c50951430b0dd765ee9c5588ad521761fee Mon Sep 17 00:00:00 2001 From: TinBane <363069+TinBane@users.noreply.github.com> Date: Sun, 25 Oct 2026 10:55:23 +1100 Subject: [PATCH 2/8] fix(server): preflight exact replacement handoffs Prepare and budget the full replacement payload before creating a fresh native thread. Use fresh-thread context-window semantics so model changes cannot inherit stale capacity. Co-Authored-By: GPT Daybreak Blue --- .../ProviderTurnStartService.ts | 95 ++++++++++++------- .../ProviderSwitch.integration.test.ts | 44 ++++++--- 2 files changed, 94 insertions(+), 45 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 80f06d64b09c0..465b79da9c7f6 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -695,10 +695,45 @@ export const layer: Layer.Layer< }); let effectiveHandoffs = handoffs; let resumeFellBack = false; + const prepareReplacementHandoff = Effect.fn( + "ProviderTurnStartService.prepareReplacementHandoff", + )(function* () { + const transferId = yield* idAllocator.allocate.contextTransfer({ + sourceThreadId: projection.thread.id, + targetThreadId: projection.thread.id, + type: "provider_resume_fallback", + }); + const createdAt = yield* DateTime.now; + const handoff = yield* contextHandoffService.prepareProviderHandoff({ + threadId: projection.thread.id, + targetRunId: run.id, + transferId, + fromProviderThreadIds: [providerThread.id], + toProviderThreadId: providerThread.id, + fromProviderInstanceId: providerThread.providerInstanceId, + toProviderInstanceId: run.providerInstanceId, + coveredRunOrdinals: { from: 1, to: Math.max(1, run.ordinal - 1) }, + strategy: "full_thread_summary", + runs: projection.runs, + items: (yield* projectionStore.getTurnStartHistory(input.threadId)).filter( + (item) => + item.runId === null || + projection.runs.some( + (source) => source.id === item.runId && source.ordinal < run.ordinal, + ), + ), + createdAt, + }); + return { transferId, createdAt, handoff }; + }); + type PreparedReplacementHandoff = Effect.Success< + ReturnType + >; const replaceNativeThread = Effect.fn("ProviderTurnStartService.replaceNativeThread")( function* (replacementInput: { readonly reason: "resume_failed" | "uncertain_history_delivery" | "handoff_overflow"; readonly errorTag?: string; + readonly prepared?: PreparedReplacementHandoff; }) { yield* Effect.logWarning("Replacing the native provider thread with portable context", { driver: session.driver, @@ -722,32 +757,8 @@ export const layer: Layer.Layer< ); if (replacement === undefined) return undefined; if (!(yield* isCurrentAttemptInStatus("starting"))) return undefined; - const transferId = yield* idAllocator.allocate.contextTransfer({ - sourceThreadId: projection.thread.id, - targetThreadId: projection.thread.id, - type: "provider_resume_fallback", - }); - const createdAt = yield* DateTime.now; - const handoff = yield* contextHandoffService.prepareProviderHandoff({ - threadId: projection.thread.id, - targetRunId: run.id, - transferId, - fromProviderThreadIds: [providerThread.id], - toProviderThreadId: providerThread.id, - fromProviderInstanceId: providerThread.providerInstanceId, - toProviderInstanceId: run.providerInstanceId, - coveredRunOrdinals: { from: 1, to: Math.max(1, run.ordinal - 1) }, - strategy: "full_thread_summary", - runs: projection.runs, - items: (yield* projectionStore.getTurnStartHistory(input.threadId)).filter( - (item) => - item.runId === null || - projection.runs.some( - (source) => source.id === item.runId && source.ordinal < run.ordinal, - ), - ), - createdAt, - }); + const { transferId, createdAt, handoff } = + replacementInput.prepared ?? (yield* prepareReplacementHandoff()); const replacementWrite = yield* eventSink.writeIfRunCurrent({ threadId: projection.thread.id, runId: run.id, @@ -1015,18 +1026,38 @@ export const layer: Layer.Layer< alreadyDeliveredItemIds: new Set(), }; if (planContextHandoffDelivery({ ...deliveryInput, budget }).status === "overflow") { + const prospectiveUsageState = usageStateFor({ + ...loadedProviderThread, + nativeThreadRef: null, + contextUsage: null, + }); const freshBudget = handoffBudget({ tokenCap, - modelContextWindow: usageState.modelContextWindow, + modelContextWindow: prospectiveUsageState.modelContextWindow, userText, attachments: message.attachments, - providerThread: { ...loadedProviderThread, contextUsage: null }, + providerThread: { + ...loadedProviderThread, + contextUsage: prospectiveUsageState.handoffUsage, + }, nativeContextEstimate: 0, }); - if ( - planContextHandoffDelivery({ ...deliveryInput, budget: freshBudget }).status === "ready" - ) { - const replacement = yield* replaceNativeThread({ reason: "handoff_overflow" }); + const prepared = yield* prepareReplacementHandoff(); + const prospectivePlan = planContextHandoffDelivery({ + handoffs: [prepared.handoff, ...effectiveHandoffs], + providerThread: { + ...loadedProviderThread, + nativeThreadRef: null, + contextUsage: null, + }, + budget: freshBudget, + alreadyDeliveredItemIds: new Set(), + }); + if (prospectivePlan.status === "ready") { + const replacement = yield* replaceNativeThread({ + reason: "handoff_overflow", + prepared, + }); if (replacement === undefined) return; loadedProviderThread = replacement; usageState = usageStateFor(replacement); 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 d38f55e08e26e..ce40640ef7691 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -414,6 +414,8 @@ describe("orchestration v2 provider switching", () => { "screenshot-option-change-retry-native", "prior-images-reasoning-change-overflow-native", "prior-images-reasoning-change-overflow-oversized-native", + "reasoning-change-overflow-prospective-payload-native", + "turn-usage-model-change-overflow-unknown-window-native", "prior-images-reasoning-change-overflow-compact-native", "prior-images-reasoning-change-replacement-small-native", "imported-prior-images-fallback", @@ -480,12 +482,20 @@ describe("orchestration v2 provider switching", () => { canReuseContextUsage: canReuseCodexContextUsage, tokenUsageByRunOrdinal: { 2: { - usedTokens: replaceNative ? 30_000 : replaceForOverflow ? 250_000 : 37_321, + usedTokens: scenario.includes("prospective-payload") + ? 16_850 + : replaceNative + ? 30_000 + : replaceForOverflow + ? 250_000 + : 37_321, maxTokens: replaceNative ? scenario.includes("small") ? 32_000 : 64_000 - : 258_400, + : scenario.includes("prospective-payload") + ? 16_935 + : 258_400, }, }, } @@ -540,14 +550,16 @@ describe("orchestration v2 provider switching", () => { creationSource: "web", text, attachments: - turnUsageScenario && ordinal === 2 + turnUsageScenario && !scenario.includes("prospective-payload") && ordinal === 2 ? Array.from({ length: 8 }, (_, index) => ({ ...screenshot, id: `prior-${index}`, })) : scenario.includes("overflow-compact") && ordinal >= targetOrdinal ? [] - : turnUsageScenario && ordinal >= targetOrdinal + : turnUsageScenario && + !scenario.includes("prospective-payload") && + ordinal >= targetOrdinal ? [screenshot, { ...screenshot, id: "current-2" }] : scenario.startsWith("screenshot") && ordinal >= targetOrdinal ? screenshots @@ -605,15 +617,17 @@ describe("orchestration v2 provider switching", () => { yield* Effect.addFinalizer(() => Effect.sync(() => spy?.mockRestore())); const current = scenario.includes("overflow-oversized") ? "x".repeat(300_000) - : scenario.includes("overflow-compact") - ? "/compact" - : scenario.startsWith("compact") + : scenario.includes("unknown-window") + ? "x".repeat(100_000) + : scenario.includes("overflow-compact") ? "/compact" - : capacityScenario && !turnUsageScenario - ? "x".repeat(70_000) - : scenario === "large-current-input" - ? "x".repeat(9_000) - : "Continue work"; + : scenario.startsWith("compact") + ? "/compact" + : capacityScenario && !turnUsageScenario + ? "x".repeat(70_000) + : scenario === "large-current-input" + ? "x".repeat(9_000) + : "Continue work"; if (returning) { if (scenario.includes("unsent")) yield* Ref.set(failStartOnce, true); yield* dispatch( @@ -749,7 +763,11 @@ describe("orchestration v2 provider switching", () => { } if (replaceForOverflow) { const replaced = yield* orchestrator.getThreadProjection(threadId); - if (scenario.includes("overflow-oversized")) { + if ( + scenario.includes("overflow-oversized") || + scenario.includes("prospective-payload") || + scenario.includes("unknown-window") + ) { assert.equal(replaced.runs.at(-1)?.status, "failed"); const target = replaced.providerThreads.find( (thread) => thread.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, From f17eb391abcf6bf6d0f1f1e59ff181d09c0554f2 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 17:48:43 +0000 Subject: [PATCH 3/8] fix(server): create fresh ACP sessions for oversized handoffs --- .../Adapters/AcpAdapterV2.test.ts | 312 +++++++++--------- .../ContextHandoffBudget.test.ts | 11 + .../ContextHandoffDelivery.test.ts | 55 --- .../ProviderTurnStartService.ts | 72 ++-- .../ProviderSwitch.integration.test.ts | 12 +- packages/provider-acp/src/server/adapter.ts | 149 +++++---- .../src/server/ProviderAdapter.ts | 1 + 7 files changed, 311 insertions(+), 301 deletions(-) delete mode 100644 apps/server/src/orchestration-v2/ContextHandoffDelivery.test.ts diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 1cb40e1fab2ed..8c9e0eb9cfc95 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -2918,162 +2918,172 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); - it.effect("replaces the ACP session and clears conversation state on rollback", () => - Effect.gen(function* () { - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const idAllocator = yield* IdAllocator.IdAllocatorV2; - const path = yield* Path.Path; - const selfInvocation = yield* resolveSelfInvocation(); - const mockAgentPath = yield* path.fromFileUrl( - new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), - ); - const instanceId = ProviderInstanceId.make("acp-test-rollback-session"); - const rollbackThreadId = ThreadId.make("thread-acp-rollback-session-target"); - yield* (yield* McpProviderSessions.McpProviderSessions).set({ - environmentId: EnvironmentId.make("environment-acp-rollback-session-target"), - threadId: rollbackThreadId, - providerSessionId: "mcp-session-acp-rollback-session-target", - providerInstanceId: instanceId, - endpoint: "http://127.0.0.1:43124/mcp", - authorizationHeader: "Bearer rollback-target-token", - browserToolsAvailable: false, - }); - const runtimeInputs: Array = []; - const makeRuntime = makeMockRuntime({ - childProcessSpawner, - mockAgentPath, - environment: { T3_ACP_PROMPT_DELAY_MS: "100" }, - }); - const adapter = yield* makeAcpAdapterV2({ - instanceId, - flavor: { - driver: ACP_TEST_DRIVER, - capabilities: AcpProviderCapabilitiesV2, - makeRuntime: (runtimeInput) => - Effect.sync(() => runtimeInputs.push(runtimeInput)).pipe( - Effect.andThen(makeRuntime(runtimeInput)), - ), - }, - selfInvocation, - }); - const threadId = ThreadId.make("thread-acp-rollback-session"); - const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ - runtimeMode: "full-access", - interactionMode: "default", - cwd: process.cwd(), - }); - const modelSelection = { instanceId, model: "default" } as const; - const runtime = yield* adapter.openSession({ - threadId, - providerSessionId: ProviderSessionId.make("provider-session-acp-rollback-session"), - modelSelection, - runtimePolicy, - }); - assert.isTrue(runtime.providerSession.capabilities.threads.canRollbackThread); - const events = yield* Queue.unbounded(); - yield* runtime.events.pipe( - Stream.runForEach((event) => Queue.offer(events, event)), - Effect.forkScoped, - ); - const providerThread = yield* runtime.ensureThread({ - threadId, - modelSelection, - runtimePolicy, - }); - const firstSessionId = providerThread.nativeThreadRef?.nativeId; - yield* runtime.startTurn( - makeTurnInput({ - threadId, - providerThread, + it.effect.each(["rollback", "fresh"] as const)( + "replaces the ACP session and clears conversation state on %s", + (operation) => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const path = yield* Path.Path; + const selfInvocation = yield* resolveSelfInvocation(); + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const instanceId = ProviderInstanceId.make("acp-test-rollback-session"); + const rollbackThreadId = ThreadId.make("thread-acp-rollback-session-target"); + yield* (yield* McpProviderSessions.McpProviderSessions).set({ + environmentId: EnvironmentId.make("environment-acp-rollback-session-target"), + threadId: rollbackThreadId, + providerSessionId: "mcp-session-acp-rollback-session-target", + providerInstanceId: instanceId, + endpoint: "http://127.0.0.1:43124/mcp", + authorizationHeader: "Bearer rollback-target-token", + browserToolsAvailable: false, + }); + const runtimeInputs: Array = []; + const makeRuntime = makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + environment: { T3_ACP_PROMPT_DELAY_MS: "100" }, + }); + const adapter = yield* makeAcpAdapterV2({ instanceId, - runtimePolicy, - now: yield* DateTime.now, - }), - ); - const activeRollback = yield* runtime - .rollbackThread({ - providerThread, - providerThreadTurns: [], - target: { - type: "thread_start", - checkpointId: CheckpointId.make("checkpoint-acp-active-rollback"), - appRunOrdinal: 0, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + makeRuntime: (runtimeInput) => + Effect.sync(() => runtimeInputs.push(runtimeInput)).pipe( + Effect.andThen(makeRuntime(runtimeInput)), + ), }, - }) - .pipe(Effect.result); - assert.equal(activeRollback._tag, "Failure"); - if (activeRollback._tag === "Failure") { - assert.include(String(activeRollback.failure.cause), "while turn"); - } - const firstProviderTurnId = idAllocator.derive.providerTurn({ - driver: ACP_TEST_DRIVER, - nativeTurnId: acpScopedNativeId(instanceId, `${firstSessionId}:turn:1`), - }); - while (true) { - const event = yield* Queue.take(events); - if (event.type === "turn.terminal" && event.providerTurnId === firstProviderTurnId) break; - } - const snapshotBeforeRollback = yield* runtime.readThreadSnapshot({ providerThread }); - - const rolledBack = yield* runtime.rollbackThread({ - providerThread: { ...providerThread, appThreadId: rollbackThreadId }, - providerThreadTurns: snapshotBeforeRollback.providerTurns, - target: { - type: "thread_start", - checkpointId: CheckpointId.make("checkpoint-acp-rollback-session"), - appRunOrdinal: 0, - }, - }); - assert.isString(rolledBack.providerThread.nativeThreadRef?.nativeId); - assert.deepEqual(rolledBack.providerTurns, []); - assert.equal( - runtimeInputs[1]?.processEnvironment?.T3_ACP_MCP_AUTHORIZATION, - "Bearer rollback-target-token", - ); - const replacementMcpServer = runtimeInputs[1]?.mcpServers[0]; - const replacementMcpEnvironment = - replacementMcpServer !== undefined && - "env" in replacementMcpServer && - Array.isArray(replacementMcpServer.env) - ? replacementMcpServer.env - : undefined; - assert.equal( - replacementMcpEnvironment?.find((variable) => variable.name === "T3_ACP_MCP_AUTHORIZATION") - ?.value, - "Bearer rollback-target-token", - ); - const snapshotAfterRollback = yield* runtime.readThreadSnapshot({ - providerThread: rolledBack.providerThread, - }); - assert.deepEqual(snapshotAfterRollback.providerTurns, []); - assert.deepEqual(snapshotAfterRollback.messages, []); + selfInvocation, + }); + const threadId = ThreadId.make("thread-acp-rollback-session"); + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const modelSelection = { instanceId, model: "default" } as const; + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-acp-rollback-session"), + initialNativeThreadId: "persisted-session", + initialProviderItemIdentityVersion: 2, + modelSelection, + runtimePolicy, + }); + assert.isTrue(runtime.providerSession.capabilities.threads.canRollbackThread); + const events = yield* Queue.unbounded(); + yield* runtime.events.pipe( + Stream.runForEach((event) => Queue.offer(events, event)), + Effect.forkScoped, + ); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const firstSessionId = providerThread.nativeThreadRef?.nativeId; + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + }), + ); + const replace = () => + operation === "fresh" + ? runtime + .ensureThread({ + threadId: rollbackThreadId, + modelSelection, + runtimePolicy, + existingProviderThread: { ...providerThread, nativeThreadRef: null }, + forceNewNativeThread: true, + }) + .pipe( + Effect.map((replacement) => ({ providerThread: replacement, providerTurns: [] })), + ) + : runtime.rollbackThread({ + providerThread: { ...providerThread, appThreadId: rollbackThreadId }, + providerThreadTurns: [], + target: { + type: "thread_start", + checkpointId: CheckpointId.make("checkpoint-acp-rollback"), + appRunOrdinal: 0, + }, + }); + const activeRollback = yield* replace().pipe(Effect.result); + assert.equal(activeRollback._tag, "Failure"); + if (activeRollback._tag === "Failure") { + assert.include(String(activeRollback.failure.cause), "while turn"); + } + const firstProviderTurnId = idAllocator.derive.providerTurn({ + driver: ACP_TEST_DRIVER, + nativeTurnId: acpScopedNativeId(instanceId, `${firstSessionId}:turn:1`), + }); + while (true) { + const event = yield* Queue.take(events); + if (event.type === "turn.terminal" && event.providerTurnId === firstProviderTurnId) break; + } - yield* runtime.startTurn( - makeTurnInput({ - threadId: rollbackThreadId, + const rolledBack = yield* replace(); + assert.isString(rolledBack.providerThread.nativeThreadRef?.nativeId); + assert.notEqual(rolledBack.providerThread.nativeThreadRef?.nativeId, firstSessionId); + assert.equal(rolledBack.providerThread.id, providerThread.id); + assert.deepEqual(rolledBack.providerTurns, []); + assert.equal( + runtimeInputs[1]?.processEnvironment?.T3_ACP_MCP_AUTHORIZATION, + "Bearer rollback-target-token", + ); + const replacementMcpServer = runtimeInputs[1]?.mcpServers[0]; + const replacementMcpEnvironment = + replacementMcpServer !== undefined && + "env" in replacementMcpServer && + Array.isArray(replacementMcpServer.env) + ? replacementMcpServer.env + : undefined; + assert.equal( + replacementMcpEnvironment?.find( + (variable) => variable.name === "T3_ACP_MCP_AUTHORIZATION", + )?.value, + "Bearer rollback-target-token", + ); + const snapshotAfterRollback = yield* runtime.readThreadSnapshot({ providerThread: rolledBack.providerThread, - instanceId, - runtimePolicy, - now: yield* DateTime.now, - ordinal: 2, - }), - ); - const secondProviderTurnId = idAllocator.derive.providerTurn({ - driver: ACP_TEST_DRIVER, - nativeTurnId: acpScopedNativeId( - instanceId, - `${rolledBack.providerThread.nativeThreadRef?.nativeId}:turn:2`, - ), - }); - let secondStatus: string | null = null; - while (secondStatus === null) { - const event = yield* Queue.take(events); - if (event.type === "turn.terminal" && event.providerTurnId === secondProviderTurnId) { - secondStatus = event.status; + }); + assert.deepEqual(snapshotAfterRollback.providerTurns, []); + assert.deepEqual(snapshotAfterRollback.messages, []); + + yield* runtime.startTurn( + makeTurnInput({ + threadId: rollbackThreadId, + providerThread: rolledBack.providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + ordinal: 2, + }), + ); + const secondProviderTurnId = idAllocator.derive.providerTurn({ + driver: ACP_TEST_DRIVER, + nativeTurnId: acpScopedNativeId( + instanceId, + `${rolledBack.providerThread.nativeThreadRef?.nativeId}:turn:2`, + ), + }); + let secondStatus: string | null = null; + while (secondStatus === null) { + const event = yield* Queue.take(events); + if (event.type === "turn.terminal" && event.providerTurnId === secondProviderTurnId) { + secondStatus = event.status; + } } - } - assert.equal(secondStatus, "completed"); - }).pipe(Effect.provide(layerTest), Effect.scoped), + assert.equal(secondStatus, "completed"); + }).pipe(Effect.provide(layerTest), Effect.scoped), ); it.effect("quarantines callbacks from a failed rollback replacement before retrying", () => diff --git a/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts b/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts index c4068702e3a57..4bc0340c63ea7 100644 --- a/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts +++ b/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts @@ -399,6 +399,17 @@ describe("handoff budget", () => { }); describe("handoff delivery", () => { + it.effect("does not evaluate the budget when no handoff needs delivery", () => + Effect.gen(function* () { + yield* deliverContextHandoffs({ + handoffs: [], + providerThread, + budget: Effect.die("settled handoffs need no budget"), + alreadyDeliveredItemIds: new Set(), + persist: () => Effect.void, + }); + }), + ); it.effect.each([ { native: true, label: "injected" }, { native: false, label: "inline" }, diff --git a/apps/server/src/orchestration-v2/ContextHandoffDelivery.test.ts b/apps/server/src/orchestration-v2/ContextHandoffDelivery.test.ts deleted file mode 100644 index 00c7f727853df..0000000000000 --- a/apps/server/src/orchestration-v2/ContextHandoffDelivery.test.ts +++ /dev/null @@ -1,55 +0,0 @@ -import { assert, it } from "@effect/vitest"; -import { - ProviderDriverKind, - ProviderInstanceId, - ProviderSessionId, - ProviderThreadId, - ThreadId, - type OrchestrationV2ProviderThread, -} from "@t3tools/contracts"; -import * as DateTime from "effect/DateTime"; -import * as Effect from "effect/Effect"; - -import { deliverContextHandoffs } from "./ContextHandoffDelivery.ts"; - -it.effect("does not evaluate the budget when no handoff needs delivery", () => - Effect.gen(function* () { - let budgetEvaluated = false; - const now = DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"); - const providerThread: OrchestrationV2ProviderThread = { - id: ProviderThreadId.make("provider-thread:settled"), - driver: ProviderDriverKind.make("acp"), - providerInstanceId: ProviderInstanceId.make("kiro"), - providerSessionId: ProviderSessionId.make("provider-session:settled"), - appThreadId: ThreadId.make("thread:settled"), - ownerNodeId: null, - nativeThreadRef: { - driver: ProviderDriverKind.make("acp"), - nativeId: "native:settled", - strength: "strong", - }, - nativeConversationHeadRef: null, - status: "idle", - firstRunOrdinal: 1, - lastRunOrdinal: 1, - handoffIds: [], - contextUsage: null, - forkedFrom: null, - createdAt: now, - updatedAt: now, - }; - - yield* deliverContextHandoffs({ - handoffs: [], - providerThread, - budget: Effect.sync(() => { - budgetEvaluated = true; - return 0; - }), - alreadyDeliveredItemIds: new Set(), - persist: () => Effect.void, - }); - - assert.isFalse(budgetEvaluated); - }), -); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 465b79da9c7f6..86aa3971be9f2 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -724,7 +724,12 @@ export const layer: Layer.Layer< ), createdAt, }); - return { transferId, createdAt, handoff }; + const handoffs = [ + ...effectiveHandoffs.filter((candidate) => candidate.targetRunId !== run.id), + handoff, + ...effectiveHandoffs.filter((candidate) => candidate.targetRunId === run.id), + ]; + return { transferId, createdAt, handoff, handoffs }; }); type PreparedReplacementHandoff = Effect.Success< ReturnType @@ -744,6 +749,7 @@ export const layer: Layer.Layer< ? {} : { errorTag: replacementInput.errorTag }), }); + if (!(yield* isCurrentAttemptInStatus("starting"))) return undefined; const replacement = yield* loadFromProvider( session.ensureThread({ threadId: projection.thread.id, @@ -751,13 +757,14 @@ export const layer: Layer.Layer< modelSelection: run.modelSelection, runtimePolicy: resolvedRuntimePolicy, providerSessionId, + forceNewNativeThread: true, // Keep the durable row but force the adapter to bind fresh native state. existingProviderThread: { ...providerThread, nativeThreadRef: null }, }), ); if (replacement === undefined) return undefined; if (!(yield* isCurrentAttemptInStatus("starting"))) return undefined; - const { transferId, createdAt, handoff } = + const { transferId, createdAt, handoff, handoffs } = replacementInput.prepared ?? (yield* prepareReplacementHandoff()); const replacementWrite = yield* eventSink.writeIfRunCurrent({ threadId: projection.thread.id, @@ -803,14 +810,7 @@ export const layer: Layer.Layer< ], }); if (!replacementWrite.committed) return undefined; - // Delivery ranks later handoffs as newer history. Pending handoffs from - // earlier runs predate this summary; handoffs for this run (provider-switch - // delta, merge-back) are newer and must stay last. - effectiveHandoffs = [ - ...effectiveHandoffs.filter((candidate) => candidate.targetRunId !== run.id), - handoff, - ...effectiveHandoffs.filter((candidate) => candidate.targetRunId === run.id), - ]; + effectiveHandoffs = handoffs; resumeFellBack = true; return replacement; }, @@ -984,6 +984,13 @@ export const layer: Layer.Layer< const tokenCap = yield* handoffTokenCapConfig.pipe( Effect.orElseSucceed(() => DEFAULT_HANDOFF_TOKEN_CAP), ); + const settledHandoffsFor = (nativeThreadId: string | undefined) => + projection.contextHandoffs.filter( + (handoff) => + handoff.toProviderThreadId === providerThread.id && + handoff.delivery?.nativeThreadId === nativeThreadId && + handoff.delivery?.status !== "pending", + ); const providerSwitchHandoff = effectiveHandoffs.some( (handoff) => handoff.targetRunId === run.id && @@ -992,15 +999,15 @@ export const layer: Layer.Layer< handoff.strategy === "full_thread_summary"), ); const compact = isCompactionCommandMessage(message); - const deliveredAttemptIdsAtStart = new Set( + const deliveredAttemptIds = new Set( projection.providerTurns.map((turn) => turn.runAttemptId), ); - const hasMissedProviderInput = projection.runs.some( + const missedRuns = projection.runs.filter( (source) => source.ordinal < run.ordinal && source.providerThreadId === providerThread.id && (source.status === "failed" || source.status === "interrupted") && - !deliveredAttemptIdsAtStart.has(source.activeAttemptId), + !deliveredAttemptIds.has(source.activeAttemptId), ); if ( providerSwitchHandoff && @@ -1010,7 +1017,7 @@ export const layer: Layer.Layer< !goalCommand && !noteContinuation && restartNote === "" && - !hasMissedProviderInput + missedRuns.length === 0 ) { const budget = handoffBudget({ tokenCap, @@ -1023,7 +1030,11 @@ export const layer: Layer.Layer< const deliveryInput = { handoffs: effectiveHandoffs, providerThread: loadedProviderThread, - alreadyDeliveredItemIds: new Set(), + alreadyDeliveredItemIds: new Set( + settledHandoffsFor(loadedProviderThread.nativeThreadRef?.nativeId).flatMap( + (handoff) => handoff.delivery?.itemIds ?? [], + ), + ), }; if (planContextHandoffDelivery({ ...deliveryInput, budget }).status === "overflow") { const prospectiveUsageState = usageStateFor({ @@ -1044,7 +1055,7 @@ export const layer: Layer.Layer< }); const prepared = yield* prepareReplacementHandoff(); const prospectivePlan = planContextHandoffDelivery({ - handoffs: [prepared.handoff, ...effectiveHandoffs], + handoffs: prepared.handoffs, providerThread: { ...loadedProviderThread, nativeThreadRef: null, @@ -1053,7 +1064,14 @@ export const layer: Layer.Layer< budget: freshBudget, alreadyDeliveredItemIds: new Set(), }); - if (prospectivePlan.status === "ready") { + if ( + prospectivePlan.status === "ready" && + !(yield* ( + session.hasPendingBackgroundWorkForThread?.(loadedProviderThread) ?? + session.hasPendingBackgroundWork ?? + Effect.succeed(false) + )) + ) { const replacement = yield* replaceNativeThread({ reason: "handoff_overflow", prepared, @@ -1192,12 +1210,7 @@ export const layer: Layer.Layer< const routableSubagents = projection.subagents.filter((subagent) => RunExecutionService.canRouteRelatedSubagent(subagent.status), ); - const settledHandoffs = projection.contextHandoffs.filter( - (handoff) => - handoff.toProviderThreadId === providerThread.id && - handoff.delivery?.nativeThreadId === runningProviderThread.nativeThreadRef?.nativeId && - handoff.delivery?.status !== "pending", - ); + const settledHandoffs = settledHandoffsFor(runningProviderThread.nativeThreadRef?.nativeId); const deliveredItemIds = new Set( settledHandoffs.flatMap((handoff) => handoff.delivery?.itemIds ?? []), ); @@ -1205,9 +1218,6 @@ export const layer: Layer.Layer< ...deliveredItemIds, ...settledHandoffs.flatMap((handoff) => handoff.delivery?.omittedItemIds ?? []), ]); - const deliveredAttemptIds = new Set( - projection.providerTurns.map((turn) => turn.runAttemptId), - ); const acceptedAttempts = projection.attempts.filter( (source) => source.providerThreadId === providerThread.id && deliveredAttemptIds.has(source.id), @@ -1278,15 +1288,7 @@ export const layer: Layer.Layer< }; // The resume fallback summary already covers every earlier run in order. A // retry copy after it would rank older missed requests as the newest history. - const missedRuns = projection.runs.filter( - (source) => - !resumeFellBack && - source.ordinal < run.ordinal && - source.providerThreadId === providerThread.id && - (source.status === "failed" || source.status === "interrupted") && - !deliveredAttemptIds.has(source.activeAttemptId), - ); - const missedRunIds = new Set(missedRuns.map((source) => source.id)); + const missedRunIds = new Set(resumeFellBack ? [] : missedRuns.map((source) => source.id)); const missedItems = missedRunIds.size === 0 ? [] 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 ce40640ef7691..c3ee0d9218215 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -103,6 +103,7 @@ function makeTestAdapter(input: { readonly failInjectionOnce?: Ref.Ref; readonly nativeThreadGeneration?: Ref.Ref; readonly failResume?: boolean; + readonly hasPendingBackgroundWork?: boolean; readonly failResumeOnce?: Ref.Ref; readonly initialContextUsage?: OrchestrationV2ProviderThread["contextUsage"]; readonly getModelContextWindow?: (selection: ModelSelection) => number | undefined; @@ -145,6 +146,7 @@ function makeTestAdapter(input: { providerSessionId: sessionInput.providerSessionId, providerSession, events: Stream.fromPubSub(events), + hasPendingBackgroundWork: Effect.succeed(input.hasPendingBackgroundWork ?? false), ...(input.getModelContextWindow === undefined ? {} : { getModelContextWindow: input.getModelContextWindow }), @@ -413,6 +415,8 @@ describe("orchestration v2 provider switching", () => { "screenshot-reported-capacity-native", "screenshot-option-change-retry-native", "prior-images-reasoning-change-overflow-native", + "prior-images-reasoning-change-overflow-inline", + "prior-images-reasoning-change-overflow-background-native", "prior-images-reasoning-change-overflow-oversized-native", "reasoning-change-overflow-prospective-payload-native", "turn-usage-model-change-overflow-unknown-window-native", @@ -467,6 +471,7 @@ describe("orchestration v2 provider switching", () => { capabilities: ClaudeProviderCapabilitiesV2, modelSelection: CLAUDE_MODEL_SELECTION, responseByRunOrdinal: {}, + hasPendingBackgroundWork: scenario.includes("overflow-background"), capturedTurns, ...(scenario.includes("retry") || scenario.includes("unsent") ? { failStartOnce } : {}), ...(replaceNative @@ -765,6 +770,7 @@ describe("orchestration v2 provider switching", () => { const replaced = yield* orchestrator.getThreadProjection(threadId); if ( scenario.includes("overflow-oversized") || + scenario.includes("overflow-background") || scenario.includes("prospective-payload") || scenario.includes("unknown-window") ) { @@ -790,8 +796,10 @@ describe("orchestration v2 provider switching", () => { const accepted = (yield* Ref.get(capturedTurns)).at(-1)!; assert.equal(accepted.driver, CLAUDE_DRIVER); assert.equal(accepted.nativeThreadId?.endsWith(":1"), true); - assert.equal(accepted.text, "Continue work"); - const history = yield* encodeJson(yield* Ref.get(injectedHistory)); + assert.include(accepted.text, "Continue work"); + const history = scenario.endsWith("-native") + ? yield* encodeJson(yield* Ref.get(injectedHistory)) + : accepted.text; assert.include(history, "Original request with constraints"); assert.include(history, "New source constraint"); return; diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index b3ce13af0a705..91a82b2de4666 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -6511,6 +6511,46 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return replacementExit.value.started; }); + const commitReplacementSession = Effect.fn("AcpAdapterV2.commitReplacementSession")( + function* ( + threadId: ThreadId | null, + candidate: AcpSessionRuntime.AcpSessionRuntimeStartResult, + ) { + rememberTerminalEnvironment(candidate.sessionId, yield* readMcpContext(threadId)); + yield* Ref.set(runtimeRestartRequired, false); + yield* Ref.set(activeSessionId, candidate.sessionId); + yield* setActiveSessionSetup(candidate); + yield* Ref.set(activeSelection, null); + yield* Ref.set(activeInteractionMode, null); + yield* Ref.set(promptInstructionStates, new Map()); + yield* Ref.set(compactionMarkers, new Map()); + yield* Ref.set(providerTurns, new Map()); + yield* Ref.set(snapshot, { + order: [], + messages: new Map(), + loadingRole: null, + loadingMessageId: null, + loadingIndex: 0, + }); + yield* continuationPermit.withPermit( + Effect.gen(function* () { + yield* Ref.update(continuationGeneration, (value) => value + 1); + yield* Ref.set(stoppedRunQuarantine, false); + yield* Ref.set(wakeBuffer, []); + yield* Ref.set(wakeReports, noWakeReports); + yield* Ref.set(continuationRequested, false); + yield* Ref.set(runningBackgroundTaskIds, new Set()); + yield* Ref.set(endedBackgroundTaskIds, new Set()); + yield* Ref.set(midTurnUnreportedCompletedTaskIds, new Set()); + yield* Ref.set(handledBackgroundTaskIdsInActiveTurn, new Set()); + yield* Ref.set(carryoverSubagents, null); + yield* Ref.set(suppressPostSettleMonitorPrompt, false); + yield* Ref.set(lastTurnRoute, null); + }), + ); + }, + ); + const initialStart = yield* Effect.result( startAcpRuntime(input.threadId, input.initialNativeThreadId), ); @@ -7745,28 +7785,57 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( : {}), ensureThread: Effect.fn("AcpAdapterV2.ensureThread")( function* (threadInput: ProviderAdapter.ProviderAdapterV2EnsureThreadInput) { - const now = yield* DateTime.now; - const sessionId = yield* Ref.get(activeSessionId); - if (sessionId === null) { - return yield* new ProviderAdapter.ProviderAdapterProtocolError({ - driver, - detail: "ACP runtime did not produce a session id", - }); - } - const providerThread = makeProviderThread({ - driver, - providerInstanceId: options.instanceId, - idAllocator, - appThreadId: threadInput.threadId, - providerSessionId: input.providerSessionId, - nativeThreadId: sessionId, - ...(itemIdentityVersion === undefined ? {} : { itemIdentityVersion }), - now, - }); - yield* Ref.update(providerThreadByNativeSessionId, (current) => - new Map(current).set(sessionId, providerThread), + return yield* runtimeTransitionPermit.withPermit( + Effect.gen(function* () { + if (threadInput.forceNewNativeThread) { + const currentTurn = yield* Ref.get(activeTurn); + if (currentTurn !== null) + return yield* new ProviderAdapter.ProviderAdapterProtocolError({ + driver, + detail: `Cannot replace ACP session while turn ${currentTurn.providerTurnId} is active`, + }); + yield* awaitRuntimeTeardown(); + const previousItemIdentityVersion = itemIdentityVersion; + itemIdentityVersion = 2; + prepareTerminalEnvironment(yield* readMcpContext(threadInput.threadId)); + yield* startReplacementAcpRuntime(threadInput.threadId, (candidate) => + commitReplacementSession(threadInput.threadId, candidate), + ).pipe( + Effect.tapError(() => + Effect.sync(() => { + itemIdentityVersion = previousItemIdentityVersion; + }), + ), + ); + } + const now = yield* DateTime.now; + const sessionId = yield* Ref.get(activeSessionId); + if (sessionId === null) { + return yield* new ProviderAdapter.ProviderAdapterProtocolError({ + driver, + detail: "ACP runtime did not produce a session id", + }); + } + const created = makeProviderThread({ + driver, + providerInstanceId: options.instanceId, + idAllocator, + appThreadId: threadInput.threadId, + providerSessionId: input.providerSessionId, + nativeThreadId: sessionId, + ...(itemIdentityVersion === undefined ? {} : { itemIdentityVersion }), + now, + }); + const providerThread = + threadInput.forceNewNativeThread && threadInput.existingProviderThread + ? { ...created, id: threadInput.existingProviderThread.id } + : created; + yield* Ref.update(providerThreadByNativeSessionId, (current) => + new Map(current).set(sessionId, providerThread), + ); + return providerThread; + }), ); - return providerThread; }, (effect, threadInput) => effect.pipe( @@ -8309,43 +8378,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const replacement = yield* startReplacementAcpRuntime( rollbackInput.providerThread.appThreadId, (candidate) => - Effect.gen(function* () { - rememberTerminalEnvironment( - candidate.sessionId, - yield* readMcpContext(rollbackInput.providerThread.appThreadId), - ); - yield* Ref.set(runtimeRestartRequired, false); - yield* Ref.set(activeSessionId, candidate.sessionId); - yield* setActiveSessionSetup(candidate); - yield* Ref.set(activeSelection, null); - yield* Ref.set(activeInteractionMode, null); - yield* Ref.set(promptInstructionStates, new Map()); - yield* Ref.set(compactionMarkers, new Map()); - yield* Ref.set(providerTurns, new Map()); - yield* Ref.set(snapshot, { - order: [], - messages: new Map(), - loadingRole: null, - loadingMessageId: null, - loadingIndex: 0, - }); - yield* continuationPermit.withPermit( - Effect.gen(function* () { - yield* Ref.update(continuationGeneration, (value) => value + 1); - yield* Ref.set(stoppedRunQuarantine, false); - yield* Ref.set(wakeBuffer, []); - yield* Ref.set(wakeReports, noWakeReports); - yield* Ref.set(continuationRequested, false); - yield* Ref.set(runningBackgroundTaskIds, new Set()); - yield* Ref.set(endedBackgroundTaskIds, new Set()); - yield* Ref.set(midTurnUnreportedCompletedTaskIds, new Set()); - yield* Ref.set(handledBackgroundTaskIdsInActiveTurn, new Set()); - yield* Ref.set(carryoverSubagents, null); - yield* Ref.set(suppressPostSettleMonitorPrompt, false); - yield* Ref.set(lastTurnRoute, null); - }), - ); - }), + commitReplacementSession(rollbackInput.providerThread.appThreadId, candidate), ).pipe(Effect.retry({ times: 1 })); const now = yield* DateTime.now; return { diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 7c77b7bb0ac6f..8c34cee714db8 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -411,6 +411,7 @@ export interface ProviderAdapterV2EnsureThreadInput { readonly runtimePolicy: ProviderAdapterV2RuntimePolicy; readonly providerSessionId?: ProviderSessionId; readonly existingProviderThread?: OrchestrationV2ProviderThread; + readonly forceNewNativeThread?: boolean; } export interface ProviderAdapterV2TurnInput { From 715706d14bf7ecf96a087d766472271dfb7a6158 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 17:50:16 +0000 Subject: [PATCH 4/8] fix(server): type handoff verification inputs precisely --- .../Adapters/AcpAdapterV2.test.ts | 7 ++++++- .../ContextHandoffBudget.test.ts | 16 +++++++--------- .../orchestration-v2/ProviderTurnStartService.ts | 2 +- 3 files changed, 14 insertions(+), 11 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 8c9e0eb9cfc95..276c70de8fd71 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -3005,7 +3005,12 @@ describe("AcpAdapterV2", () => { forceNewNativeThread: true, }) .pipe( - Effect.map((replacement) => ({ providerThread: replacement, providerTurns: [] })), + Effect.map((replacement): ProviderAdapter.ProviderAdapterV2ThreadSnapshot => ({ + providerThread: replacement, + providerTurns: [], + messages: [], + runtimeRequests: [], + })), ) : runtime.rollbackThread({ providerThread: { ...providerThread, appThreadId: rollbackThreadId }, diff --git a/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts b/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts index 4bc0340c63ea7..de5019657d6f0 100644 --- a/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts +++ b/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts @@ -400,15 +400,13 @@ describe("handoff budget", () => { describe("handoff delivery", () => { it.effect("does not evaluate the budget when no handoff needs delivery", () => - Effect.gen(function* () { - yield* deliverContextHandoffs({ - handoffs: [], - providerThread, - budget: Effect.die("settled handoffs need no budget"), - alreadyDeliveredItemIds: new Set(), - persist: () => Effect.void, - }); - }), + deliverContextHandoffs({ + handoffs: [], + providerThread, + budget: Effect.die("settled handoffs need no budget"), + alreadyDeliveredItemIds: new Set(), + persist: () => Effect.void, + }).pipe(Effect.asVoid), ); it.effect.each([ { native: true, label: "injected" }, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 86aa3971be9f2..ad904d1e92b6e 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -984,7 +984,7 @@ export const layer: Layer.Layer< const tokenCap = yield* handoffTokenCapConfig.pipe( Effect.orElseSucceed(() => DEFAULT_HANDOFF_TOKEN_CAP), ); - const settledHandoffsFor = (nativeThreadId: string | undefined) => + const settledHandoffsFor = (nativeThreadId: string | null | undefined) => projection.contextHandoffs.filter( (handoff) => handoff.toProviderThreadId === providerThread.id && From 8ba89bddf38981470e71a9f88e1762bb3793f0d7 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 17:57:49 +0000 Subject: [PATCH 5/8] fix(server): preserve native state during unsafe handoff replacement --- .../ProviderTurnStartService.ts | 41 +++++++++++++------ .../ProviderSwitch.integration.test.ts | 16 +++++++- 2 files changed, 44 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index ad904d1e92b6e..4bbe09eefef31 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -751,16 +751,33 @@ export const layer: Layer.Layer< }); if (!(yield* isCurrentAttemptInStatus("starting"))) return undefined; const replacement = yield* loadFromProvider( - session.ensureThread({ - threadId: projection.thread.id, - title: projection.thread.title, - modelSelection: run.modelSelection, - runtimePolicy: resolvedRuntimePolicy, - providerSessionId, - forceNewNativeThread: true, - // Keep the durable row but force the adapter to bind fresh native state. - existingProviderThread: { ...providerThread, nativeThreadRef: null }, - }), + session + .ensureThread({ + threadId: projection.thread.id, + title: projection.thread.title, + modelSelection: run.modelSelection, + runtimePolicy: resolvedRuntimePolicy, + providerSessionId, + forceNewNativeThread: true, + // Keep the durable row but force the adapter to bind fresh native state. + existingProviderThread: { ...providerThread, nativeThreadRef: null }, + }) + .pipe( + Effect.flatMap((replacement) => + replacementInput.reason === "handoff_overflow" && + (replacement.nativeThreadRef?.nativeId == null || + replacement.nativeThreadRef.nativeId === + providerThread.nativeThreadRef?.nativeId) + ? Effect.fail( + new ProviderAdapter.ProviderAdapterProtocolError({ + driver: session.driver, + detail: + "Provider did not create a fresh native thread for the context handoff", + }), + ) + : Effect.succeed(replacement), + ), + ), ); if (replacement === undefined) return undefined; if (!(yield* isCurrentAttemptInStatus("starting"))) return undefined; @@ -1067,8 +1084,8 @@ export const layer: Layer.Layer< if ( prospectivePlan.status === "ready" && !(yield* ( - session.hasPendingBackgroundWorkForThread?.(loadedProviderThread) ?? - session.hasPendingBackgroundWork ?? + session.hasPendingBackgroundWork ?? + session.hasPendingBackgroundWorkForThread?.(loadedProviderThread) ?? Effect.succeed(false) )) ) { 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 c3ee0d9218215..1ff57de57110d 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -104,6 +104,7 @@ function makeTestAdapter(input: { readonly nativeThreadGeneration?: Ref.Ref; readonly failResume?: boolean; readonly hasPendingBackgroundWork?: boolean; + readonly reuseNativeThread?: boolean; readonly failResumeOnce?: Ref.Ref; readonly initialContextUsage?: OrchestrationV2ProviderThread["contextUsage"]; readonly getModelContextWindow?: (selection: ModelSelection) => number | undefined; @@ -147,6 +148,7 @@ function makeTestAdapter(input: { providerSession, events: Stream.fromPubSub(events), hasPendingBackgroundWork: Effect.succeed(input.hasPendingBackgroundWork ?? false), + hasPendingBackgroundWorkForThread: () => Effect.succeed(false), ...(input.getModelContextWindow === undefined ? {} : { getModelContextWindow: input.getModelContextWindow }), @@ -159,7 +161,7 @@ function makeTestAdapter(input: { const generation = input.nativeThreadGeneration === undefined ? "" - : `:${yield* Ref.getAndUpdate(input.nativeThreadGeneration, (value) => value + 1)}`; + : `:${input.reuseNativeThread && threadInput.forceNewNativeThread ? 0 : yield* Ref.getAndUpdate(input.nativeThreadGeneration, (value) => value + 1)}`; const nativeThreadId = `${input.driver}:${threadInput.threadId}${generation}`; return { id: ProviderThreadId.make(`provider-thread:${nativeThreadId}`), @@ -417,6 +419,7 @@ describe("orchestration v2 provider switching", () => { "prior-images-reasoning-change-overflow-native", "prior-images-reasoning-change-overflow-inline", "prior-images-reasoning-change-overflow-background-native", + "prior-images-reasoning-change-overflow-reused-native", "prior-images-reasoning-change-overflow-oversized-native", "reasoning-change-overflow-prospective-payload-native", "turn-usage-model-change-overflow-unknown-window-native", @@ -472,6 +475,7 @@ describe("orchestration v2 provider switching", () => { modelSelection: CLAUDE_MODEL_SELECTION, responseByRunOrdinal: {}, hasPendingBackgroundWork: scenario.includes("overflow-background"), + reuseNativeThread: scenario.includes("overflow-reused"), capturedTurns, ...(scenario.includes("retry") || scenario.includes("unsent") ? { failStartOnce } : {}), ...(replaceNative @@ -771,6 +775,7 @@ describe("orchestration v2 provider switching", () => { if ( scenario.includes("overflow-oversized") || scenario.includes("overflow-background") || + scenario.includes("overflow-reused") || scenario.includes("prospective-payload") || scenario.includes("unknown-window") ) { @@ -780,6 +785,15 @@ describe("orchestration v2 provider switching", () => { )!; assert.equal(yield* Ref.get(generation), 1); assert.equal(target.nativeThreadRef?.nativeId?.endsWith(":0"), true); + if (scenario.includes("overflow-reused")) { + assert.isFalse( + replaced.contextHandoffs.some( + (handoff) => + handoff.targetRunId === replaced.runs.at(-1)?.id && + handoff.fromProviderThreadIds.includes(target.id), + ), + ); + } return; } assert.equal(replaced.runs.at(-1)?.status, "completed"); From 292aad9f3edc8e1c60850a6519583faf5dd3f7c2 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 18:00:51 +0000 Subject: [PATCH 6/8] refactor(server): validate native replacement with filterOrFail --- .../ProviderTurnStartService.ts | 25 +++++++++---------- 1 file changed, 12 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 4bbe09eefef31..d51e76ac94692 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -763,19 +763,18 @@ export const layer: Layer.Layer< existingProviderThread: { ...providerThread, nativeThreadRef: null }, }) .pipe( - Effect.flatMap((replacement) => - replacementInput.reason === "handoff_overflow" && - (replacement.nativeThreadRef?.nativeId == null || - replacement.nativeThreadRef.nativeId === - providerThread.nativeThreadRef?.nativeId) - ? Effect.fail( - new ProviderAdapter.ProviderAdapterProtocolError({ - driver: session.driver, - detail: - "Provider did not create a fresh native thread for the context handoff", - }), - ) - : Effect.succeed(replacement), + Effect.filterOrFail( + (replacement) => + replacementInput.reason !== "handoff_overflow" || + (replacement.nativeThreadRef?.nativeId != null && + replacement.nativeThreadRef.nativeId !== + providerThread.nativeThreadRef?.nativeId), + () => + new ProviderAdapter.ProviderAdapterProtocolError({ + driver: session.driver, + detail: + "Provider did not create a fresh native thread for the context handoff", + }), ), ), ); From 7cf961eefca5804468523af048516cf0b95c5fdc Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 18:07:40 +0000 Subject: [PATCH 7/8] fix(server): preserve native state when fresh capacity is unknown --- apps/server/src/orchestration-v2/ProviderTurnStartService.ts | 1 + .../orchestration-v2/testkit/ProviderSwitch.integration.test.ts | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index d51e76ac94692..8b7cad3c6c821 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1081,6 +1081,7 @@ export const layer: Layer.Layer< alreadyDeliveredItemIds: new Set(), }); if ( + prospectiveUsageState.modelContextWindow !== undefined && prospectivePlan.status === "ready" && !(yield* ( session.hasPendingBackgroundWork ?? 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 1ff57de57110d..a63cfa7e88764 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -627,7 +627,7 @@ describe("orchestration v2 provider switching", () => { const current = scenario.includes("overflow-oversized") ? "x".repeat(300_000) : scenario.includes("unknown-window") - ? "x".repeat(100_000) + ? "Continue work" : scenario.includes("overflow-compact") ? "/compact" : scenario.startsWith("compact") From 0e73689c9fa07a50651c69ac85dd5a0a54c4f36f Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 18:09:11 +0000 Subject: [PATCH 8/8] fix(server): require prepared history to fit before replacement --- apps/server/src/orchestration-v2/ProviderTurnStartService.ts | 1 + .../testkit/ProviderSwitch.integration.test.ts | 5 ++++- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 8b7cad3c6c821..e67b3fafa468c 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1083,6 +1083,7 @@ export const layer: Layer.Layer< if ( prospectiveUsageState.modelContextWindow !== undefined && prospectivePlan.status === "ready" && + prospectivePlan.selected.omittedItemIds.length === 0 && !(yield* ( session.hasPendingBackgroundWork ?? session.hasPendingBackgroundWorkForThread?.(loadedProviderThread) ?? 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 a63cfa7e88764..e916bab5d6da8 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -422,6 +422,7 @@ describe("orchestration v2 provider switching", () => { "prior-images-reasoning-change-overflow-reused-native", "prior-images-reasoning-change-overflow-oversized-native", "reasoning-change-overflow-prospective-payload-native", + "prior-images-reasoning-change-overflow-prospective-payload-trimmed-native", "turn-usage-model-change-overflow-unknown-window-native", "prior-images-reasoning-change-overflow-compact-native", "prior-images-reasoning-change-replacement-small-native", @@ -503,7 +504,9 @@ describe("orchestration v2 provider switching", () => { ? 32_000 : 64_000 : scenario.includes("prospective-payload") - ? 16_935 + ? scenario.includes("trimmed") + ? 19_000 + : 16_935 : 258_400, }, },