From 5205ff518e0b325b2a041837e441289f98b503c7 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 12:29:18 +0000 Subject: [PATCH 1/4] test(server): reproduce #30136, rewound ACP conversation after Edit from here These tests reproduce #30136 and fail on main. - After an ACP rollback, which starts a fresh, empty agent session, the next turn resumes that session and sends only the new prompt, so the kept turn never reaches the agent. - When a resume after a rollback fails, the recovery handoff still carries the rewound run, because prepareRecoveryHandoff keeps rolled_back runs. The provider switch test adapter gains an opt-in rollback mode that mirrors the real adapters: truncate in place, or answer with a new native session like ACP. Co-Authored-By: Claude Opus 5.5 --- .../ProviderSwitch.integration.test.ts | 200 +++++++++++++++++- 1 file changed, 198 insertions(+), 2 deletions(-) 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 f36e463079f99..92563dc0bdee0 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -121,6 +121,8 @@ function makeTestAdapter(input: { readonly releaseFirstTurn?: Deferred.Deferred; readonly backgroundWork?: Ref.Ref; readonly closedSessions?: Ref.Ref>; + // ACP cannot truncate, so its rollback answers with a fresh, empty native session. + readonly rollback?: "truncates" | "replaces_native_thread"; }): ProviderAdapter.ProviderAdapterV2["Service"] { return { instanceId: input.instanceId, @@ -392,8 +394,25 @@ function makeTestAdapter(input: { respondToRuntimeRequest: () => Effect.void, readThreadSnapshot: () => unimplemented(input.driver, "readThreadSnapshot unused in provider switch test"), - rollbackThread: () => - unimplemented(input.driver, "rollbackThread unused in provider switch test"), + rollbackThread: ({ providerThread }) => + input.rollback === undefined + ? unimplemented(input.driver, "rollbackThread unused in provider switch test") + : Effect.succeed({ + providerThread: + input.rollback === "truncates" || providerThread.nativeThreadRef === null + ? providerThread + : { + ...providerThread, + nativeThreadRef: { + ...providerThread.nativeThreadRef, + nativeId: `${providerThread.nativeThreadRef.nativeId}:after-rollback`, + }, + nativeConversationHeadRef: null, + }, + providerTurns: [], + messages: [], + runtimeRequests: [], + }), forkThread: () => unimplemented(input.driver, "forkThread unused in provider switch test"), }; @@ -1477,6 +1496,183 @@ describe("orchestration v2 provider switching", () => { ), ); + it.live.each([ + { + name: "an ACP rollback that starts a fresh agent session", + instanceId: GROK_MODEL_SELECTION.instanceId, + driver: GROK_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + modelSelection: GROK_MODEL_SELECTION, + rollback: "replaces_native_thread", + failResumeAfterRollback: false, + }, + { + name: "a native rollback whose next resume fails", + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + rollback: "truncates", + failResumeAfterRollback: true, + }, + ] as const)( + "gives the agent the kept conversation and not the rewound turn after $name", + (scenario) => + Effect.scoped( + Effect.gen(function* () { + const name = `rollback-kept-context-${scenario.driver}`; + const cwd = yield* checkpointWorkspace(name); + const capturedTurns = yield* Ref.make>([]); + const injectedHistory = yield* Ref.make>([]); + const failResumeOnce = yield* Ref.make(false); + const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ + makeTestAdapter({ + instanceId: scenario.instanceId, + driver: scenario.driver, + capabilities: scenario.capabilities, + modelSelection: scenario.modelSelection, + responseByRunOrdinal: { 1: "noted", 2: "noted" }, + capturedTurns, + injectedHistory, + failResumeOnce, + nativeThreadGeneration: yield* Ref.make(0), + rollback: scenario.rollback, + }), + ]); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const runToCompletion = Effect.fn("runToCompletion")(function* ( + ordinal: number, + text: string, + ) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${name}:${ordinal}`), + threadId, + messageId: MessageId.make(`${name}:${ordinal}`), + createdBy: "user", + creationSource: "web", + text, + attachments: [], + modelSelection: scenario.modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.ordinal === ordinal && + (event.payload.status === "completed" || event.payload.status === "failed"), + ), + Stream.runHead, + ); + yield* worker.drain(); + }); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`${name}:create`), + threadId, + projectId, + createdBy: "user", + creationSource: "web", + title: "Edit from here", + modelSelection: scenario.modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + yield* runToCompletion( + 1, + "My favourite colour is teal. Reply with only the word: noted", + ); + yield* runToCompletion( + 2, + "My favourite animal is an otter. Reply with only the word: noted", + ); + const beforeRollback = yield* orchestrator.getThreadProjection(threadId); + const keptCheckpoint = beforeRollback.checkpoints.find( + (checkpoint) => checkpoint.appRunOrdinal === 1 && checkpoint.status === "ready", + ); + if (keptCheckpoint === undefined) { + return yield* Effect.die("Expected a checkpoint after the first turn."); + } + yield* orchestrator.dispatch({ + type: "checkpoint.rollback", + commandId: CommandId.make(`${name}:rollback`), + threadId, + scopeId: keptCheckpoint.scopeId, + checkpointId: keptCheckpoint.id, + restoreFiles: false, + }); + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + (event.type === "run.updated" && + event.payload.ordinal === 2 && + event.payload.status === "rolled_back") || + (event.type === "thread.metadata-updated" && + event.payload.rollbackFailure != null), + ), + Stream.runHead, + ); + yield* worker.drain(); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(threadId)).runs.map((run) => run.status), + ["completed", "rolled_back"], + ); + if (scenario.failResumeAfterRollback) { + const providerSessionId = (yield* orchestrator.getThreadProjection(threadId)) + .providerThreads[0]?.providerSessionId; + if (providerSessionId == null) { + return yield* Effect.die("Expected a provider session after the rollback."); + } + yield* orchestrator.dispatch({ + type: "provider-session.detach", + commandId: CommandId.make(`${name}:detach`), + threadId, + providerSessionId, + }); + yield* worker.drain(); + yield* Ref.set(failResumeOnce, true); + } + const historyBeforeRun3 = (yield* Ref.get(injectedHistory)).length; + yield* runToCompletion(3, "What did I say first?"); + + const run3 = (yield* Ref.get(capturedTurns)).at(-1); + const delivered = + (run3?.text ?? "") + + (yield* encodeJson((yield* Ref.get(injectedHistory)).slice(historyBeforeRun3))); + assert.include( + delivered, + "favourite colour is teal", + "the agent's session no longer holds the kept turn, so the next turn must carry it", + ); + assert.notInclude( + delivered, + "favourite animal is an otter", + "the rewound turn must not reach the agent", + ); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { + name, + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { type: "readOnly" }, + }, + }, + layerRegistry, + ), + ), + ); + }), + ), + ); + it.live.each([ ["background work", new ClaudeBackgroundWorkBlocksQueryReplacementError()], ["an unreadable append file", new ClaudeAppendSystemPromptFileError({ path: "missing.md" })], From c2370d9ad624fd2fb769ffad9476a06c4da40447 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 12:43:12 +0000 Subject: [PATCH 2/4] fix(server): ACP agents keep the kept conversation after Edit from here ACP agents cannot truncate their conversation, so rollback starts a fresh agent session. The capability still claimed native rollback, so the next turn resumed that empty session and sent only the new prompt. The resume fallback summary also replayed rewound turns for every provider. ACP now reports providerCanRollbackConversation: false, and rollback is allowed for adapters that can reset their thread and consume handoffs. Turn start takes the existing fresh-session recovery path when a non-truncating provider's native session holds none of the kept turns, and the recovery summary skips rolled-back runs. Fixes pingdotgg/t3code#30136 Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/CommandPolicy.ts | 4 +- .../ProviderTurnStartService.ts | 48 ++++++++++++++++--- .../ProviderSwitch.integration.test.ts | 31 ++++++++++-- packages/provider-acp/src/server/adapter.ts | 4 +- 4 files changed, 73 insertions(+), 14 deletions(-) diff --git a/apps/server/src/orchestration-v2/CommandPolicy.ts b/apps/server/src/orchestration-v2/CommandPolicy.ts index 02f329e536feb..d9bb46efbf414 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.ts @@ -312,9 +312,11 @@ const ensureNativeFork: CommandPolicyV2Shape["ensureNativeFork"] = (input) => { }; const ensureRollback: CommandPolicyV2Shape["ensureRollback"] = (input) => { + // Without native truncation, the next turn replays the kept conversation as a handoff. if ( !input.capabilities.threads.canRollbackThread || - !input.capabilities.checkpointing.providerCanRollbackConversation + (!input.capabilities.checkpointing.providerCanRollbackConversation && + !input.capabilities.context.canConsumeHandoffSummaries) ) { return Effect.fail( unsupported(input, "rollback", "providerInstanceId conversation rollback is unavailable"), diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 62a9e8b88e96d..471952a1bda89 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1096,7 +1096,10 @@ export const layer: Layer.Layer< (item) => item.runId === null || projection.runs.some( - (source) => source.id === item.runId && source.ordinal < run.ordinal, + (source) => + source.id === item.runId && + source.ordinal < run.ordinal && + source.status !== "rolled_back", ), ), createdAt, @@ -1249,15 +1252,46 @@ export const layer: Layer.Layer< handoff.delivery?.nativeThreadId === providerThread.nativeThreadRef?.nativeId && handoff.delivery?.status === "pending", ); + // A provider that cannot truncate rolls back into a fresh native session holding none of the kept turns. + const nativeId = providerThread.nativeThreadRef.nativeId; + const providerTurnAttemptIds = new Set( + projection.providerTurns.map((turn) => turn.runAttemptId), + ); + const keptAttempts = projection.attempts.filter( + (attempt) => + attempt.providerThreadId === providerThread.id && + providerTurnAttemptIds.has(attempt.id) && + projection.runs.some( + (source) => + source.id === attempt.runId && + source.ordinal < run.ordinal && + source.status !== "rolled_back", + ), + ); + const nativeSessionLacksKeptTurns = + !session.providerSession.capabilities.checkpointing.providerCanRollbackConversation && + keptAttempts.length > 0 && + !keptAttempts.some( + (attempt) => + attempt.nativeThreadId === undefined || attempt.nativeThreadId === nativeId, + ) && + !projection.contextHandoffs.some( + (handoff) => + handoff.toProviderThreadId === providerThread.id && + handoff.delivery?.nativeThreadId === nativeId && + handoff.delivery.status !== "pending", + ); const resumed = yield* Effect.result( - uncertainDelivery + uncertainDelivery || nativeSessionLacksKeptTurns ? Effect.fail( new ProviderAdapter.ProviderAdapterTurnStartError({ driver: session.driver, threadId: projection.thread.id, providerThreadId: providerThread.id, runId, - cause: "Uncertain native history injection", + cause: uncertainDelivery + ? "Uncertain native history injection" + : "Native session holds none of the kept conversation", }), ) : session.resumeThread({ @@ -1329,9 +1363,11 @@ export const layer: Layer.Layer< runId, reason: uncertainDelivery ? "uncertain_history_delivery" - : resumed._tag === "Success" - ? "handoff_budget_exhausted" - : "resume_failed", + : nativeSessionLacksKeptTurns + ? "native_session_lacks_kept_turns" + : resumed._tag === "Success" + ? "handoff_budget_exhausted" + : "resume_failed", errorTag: resumed._tag === "Failure" ? resumed.failure._tag : "ContextHandoffBudgetError", }); const replacement = yield* loadFromProvider( 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 92563dc0bdee0..30997f9ef9d00 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1505,6 +1505,7 @@ describe("orchestration v2 provider switching", () => { modelSelection: GROK_MODEL_SELECTION, rollback: "replaces_native_thread", failResumeAfterRollback: false, + replaysKeptTurn: true, }, { name: "a native rollback whose next resume fails", @@ -1514,6 +1515,18 @@ describe("orchestration v2 provider switching", () => { modelSelection: CLAUDE_MODEL_SELECTION, rollback: "truncates", failResumeAfterRollback: true, + replaysKeptTurn: true, + }, + { + // Pi forks into a new native thread that still holds the kept turns. + name: "a native rollback that forks a new native thread", + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + rollback: "replaces_native_thread", + failResumeAfterRollback: false, + replaysKeptTurn: false, }, ] as const)( "gives the agent the kept conversation and not the rewound turn after $name", @@ -1644,11 +1657,19 @@ describe("orchestration v2 provider switching", () => { const delivered = (run3?.text ?? "") + (yield* encodeJson((yield* Ref.get(injectedHistory)).slice(historyBeforeRun3))); - assert.include( - delivered, - "favourite colour is teal", - "the agent's session no longer holds the kept turn, so the next turn must carry it", - ); + if (scenario.replaysKeptTurn) { + assert.include( + delivered, + "favourite colour is teal", + "the agent's session no longer holds the kept turn, so the next turn must carry it", + ); + } else { + assert.notInclude( + delivered, + "favourite colour is teal", + "the agent's session still holds the kept turn, so no handoff replays it", + ); + } assert.notInclude( delivered, "favourite animal is an otter", diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 985f6c8ee9887..26d44aa3baece 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -704,8 +704,8 @@ export const AcpProviderCapabilitiesV2 = { checkpointing: { appCanCheckpointFilesystem: true, supportsNestedCheckpointScopes: true, - // ACP lacks truncation; rollback restores T3 checkpoints and starts a fresh agent session. - providerCanRollbackConversation: true, + // ACP lacks truncation; rollback starts a fresh agent session and turn start replays the kept history. + providerCanRollbackConversation: false, providerRollbackReturnsSnapshot: true, providerCanReadConversationSnapshot: false, }, From 4a8d6324938f908be869efadb797ce724b73ec99 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 13:07:37 +0000 Subject: [PATCH 3/4] fix(server): ACP rollback recovery survives legacy attempts and forks A kept attempt recorded before attempts carried their native session id counted as held by the current session, so rolling back past a newer turn gave the fresh ACP session only the new prompt. The reset is now recognised from attempts that did record a native id, and the check only runs for providers that cannot truncate natively. The recovery handoff also carries a fork's inherited conversation, which a fresh native session lacks just as it lacks the local kept turns. Co-Authored-By: Claude Opus 5.5 --- .../ProviderTurnStartService.ts | 92 +++++++++------ .../ProviderSwitch.integration.test.ts | 108 +++++++++++++++--- 2 files changed, 146 insertions(+), 54 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 471952a1bda89..3854ba08f6767 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -9,6 +9,7 @@ import { type OrchestrationV2ProviderThread, type OrchestrationV2Run, type OrchestrationV2RunAttempt, + type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, isCompactionCommandMessage, parseGoalCommand, @@ -115,6 +116,44 @@ const refusedBeforePrompt = (error: unknown): boolean => PROMPT_REFUSAL_TAGS.some((tag) => Predicate.isTagged(error, tag)) || (Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause)); +/** + * Whether `providerThread`'s native session holds none of the turns kept before `run`. Legacy + * attempts lack a native id, so a reset is only visible once some attempt here recorded one. + */ +const nativeSessionMissesKeptTurns = ( + projection: Pick< + OrchestrationV2ThreadProjection, + "runs" | "attempts" | "providerTurns" | "contextHandoffs" + >, + providerThread: OrchestrationV2ProviderThread, + run: OrchestrationV2Run, +): boolean => { + const nativeId = providerThread.nativeThreadRef?.nativeId; + if (nativeId === undefined) return false; + const acceptedAttemptIds = new Set(projection.providerTurns.map((turn) => turn.runAttemptId)); + const keptRunIds = new Set( + projection.runs + .filter((source) => source.ordinal < run.ordinal && source.status !== "rolled_back") + .map((source) => source.id), + ); + const accepted = projection.attempts.filter( + (attempt) => + attempt.providerThreadId === providerThread.id && acceptedAttemptIds.has(attempt.id), + ); + const kept = accepted.filter((attempt) => keptRunIds.has(attempt.runId)); + return ( + kept.length > 0 && + accepted.some((attempt) => attempt.nativeThreadId !== undefined) && + !kept.some((attempt) => attempt.nativeThreadId === nativeId) && + !projection.contextHandoffs.some( + (handoff) => + handoff.toProviderThreadId === providerThread.id && + handoff.delivery?.nativeThreadId === nativeId && + handoff.delivery.status !== "pending", + ) + ); +}; + export interface ProviderTurnStartServiceV2Shape { readonly start: (input: { readonly threadId: ThreadId; @@ -1092,16 +1131,22 @@ export const layer: Layer.Layer< 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 && - source.status !== "rolled_back", - ), - ), + items: [ + ...(yield* ContextHandoffService.readInheritedHandoffItems( + projectionStore, + projection.thread, + )), + ...(yield* projectionStore.getTurnStartHistory(input.threadId)).filter( + (item) => + item.runId === null || + projection.runs.some( + (source) => + source.id === item.runId && + source.ordinal < run.ordinal && + source.status !== "rolled_back", + ), + ), + ], createdAt, }); return { handoff, transferId, createdAt }; @@ -1253,34 +1298,9 @@ export const layer: Layer.Layer< handoff.delivery?.status === "pending", ); // A provider that cannot truncate rolls back into a fresh native session holding none of the kept turns. - const nativeId = providerThread.nativeThreadRef.nativeId; - const providerTurnAttemptIds = new Set( - projection.providerTurns.map((turn) => turn.runAttemptId), - ); - const keptAttempts = projection.attempts.filter( - (attempt) => - attempt.providerThreadId === providerThread.id && - providerTurnAttemptIds.has(attempt.id) && - projection.runs.some( - (source) => - source.id === attempt.runId && - source.ordinal < run.ordinal && - source.status !== "rolled_back", - ), - ); const nativeSessionLacksKeptTurns = !session.providerSession.capabilities.checkpointing.providerCanRollbackConversation && - keptAttempts.length > 0 && - !keptAttempts.some( - (attempt) => - attempt.nativeThreadId === undefined || attempt.nativeThreadId === nativeId, - ) && - !projection.contextHandoffs.some( - (handoff) => - handoff.toProviderThreadId === providerThread.id && - handoff.delivery?.nativeThreadId === nativeId && - handoff.delivery.status !== "pending", - ); + nativeSessionMissesKeptTurns(projection, providerThread, run); const resumed = yield* Effect.result( uncertainDelivery || nativeSessionLacksKeptTurns ? Effect.fail( 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 30997f9ef9d00..c35e9b27fe10d 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1506,6 +1506,7 @@ describe("orchestration v2 provider switching", () => { rollback: "replaces_native_thread", failResumeAfterRollback: false, replaysKeptTurn: true, + history: "plain", }, { name: "a native rollback whose next resume fails", @@ -1516,6 +1517,7 @@ describe("orchestration v2 provider switching", () => { rollback: "truncates", failResumeAfterRollback: true, replaysKeptTurn: true, + history: "plain", }, { // Pi forks into a new native thread that still holds the kept turns. @@ -1527,13 +1529,36 @@ describe("orchestration v2 provider switching", () => { rollback: "replaces_native_thread", failResumeAfterRollback: false, replaysKeptTurn: false, + history: "plain", + }, + { + name: "an ACP rollback to a turn recorded before attempts kept their native session", + instanceId: GROK_MODEL_SELECTION.instanceId, + driver: GROK_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + modelSelection: GROK_MODEL_SELECTION, + rollback: "replaces_native_thread", + failResumeAfterRollback: false, + replaysKeptTurn: true, + history: "legacy_first_turn", + }, + { + name: "an ACP rollback in a forked thread", + instanceId: GROK_MODEL_SELECTION.instanceId, + driver: GROK_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + modelSelection: GROK_MODEL_SELECTION, + rollback: "replaces_native_thread", + failResumeAfterRollback: false, + replaysKeptTurn: true, + history: "forked", }, ] as const)( "gives the agent the kept conversation and not the rewound turn after $name", (scenario) => Effect.scoped( Effect.gen(function* () { - const name = `rollback-kept-context-${scenario.driver}`; + const name = `rollback-kept-context-${scenario.driver}-${scenario.history}`; const cwd = yield* checkpointWorkspace(name); const capturedTurns = yield* Ref.make>([]); const injectedHistory = yield* Ref.make>([]); @@ -1558,12 +1583,13 @@ describe("orchestration v2 provider switching", () => { const runToCompletion = Effect.fn("runToCompletion")(function* ( ordinal: number, text: string, + thread: ThreadId = threadId, ) { yield* orchestrator.dispatch({ type: "message.dispatch", - commandId: CommandId.make(`${name}:${ordinal}`), - threadId, - messageId: MessageId.make(`${name}:${ordinal}`), + commandId: CommandId.make(`${thread}:${ordinal}`), + threadId: thread, + messageId: MessageId.make(`${thread}:${ordinal}`), createdBy: "user", creationSource: "web", text, @@ -1575,6 +1601,7 @@ describe("orchestration v2 provider switching", () => { Stream.filter( ({ event }) => event.type === "run.updated" && + event.threadId === thread && event.payload.ordinal === ordinal && (event.payload.status === "completed" || event.payload.status === "failed"), ), @@ -1582,20 +1609,44 @@ describe("orchestration v2 provider switching", () => { ); yield* worker.drain(); }); - yield* orchestrator.dispatch({ - type: "thread.create", - commandId: CommandId.make(`${name}:create`), - threadId, - projectId, - createdBy: "user", - creationSource: "web", - title: "Edit from here", - modelSelection: scenario.modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - }); + const createThread = (thread: ThreadId) => + orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`${thread}:create`), + threadId: thread, + projectId, + createdBy: "user", + creationSource: "web", + title: "Edit from here", + modelSelection: scenario.modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + if (scenario.history === "forked") { + const sourceThreadId = ThreadId.make(`${name}:source`); + yield* createThread(sourceThreadId); + yield* runToCompletion( + 1, + "My home town is Oslo. Reply with only the word: noted", + sourceThreadId, + ); + yield* orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make(`${name}:fork`), + sourceThreadId, + targetThreadId: threadId, + sourcePoint: { + type: "run", + runId: (yield* orchestrator.getThreadProjection(sourceThreadId)).runs[0]!.id, + }, + createdBy: "user", + creationSource: "web", + }); + } else { + yield* createThread(threadId); + } yield* runToCompletion( 1, "My favourite colour is teal. Reply with only the word: noted", @@ -1605,6 +1656,20 @@ describe("orchestration v2 provider switching", () => { "My favourite animal is an otter. Reply with only the word: noted", ); const beforeRollback = yield* orchestrator.getThreadProjection(threadId); + if (scenario.history === "legacy_first_turn") { + const firstRunId = beforeRollback.runs.find((run) => run.ordinal === 1)?.id; + yield* (yield* EventSink.EventSinkV2).write({ + events: beforeRollback.attempts + .filter((attempt) => attempt.runId === firstRunId) + .map(({ nativeThreadId: _nativeThreadId, ...legacy }, index) => ({ + id: EventId.make(`${name}:legacy-attempt:${index}`), + type: "run-attempt.updated" as const, + threadId, + occurredAt: beforeRollback.thread.createdAt, + payload: legacy, + })), + }); + } const keptCheckpoint = beforeRollback.checkpoints.find( (checkpoint) => checkpoint.appRunOrdinal === 1 && checkpoint.status === "ready", ); @@ -1670,6 +1735,13 @@ describe("orchestration v2 provider switching", () => { "the agent's session still holds the kept turn, so no handoff replays it", ); } + if (scenario.history === "forked") { + assert.include( + delivered, + "home town is Oslo", + "the fork's inherited turn is kept too", + ); + } assert.notInclude( delivered, "favourite animal is an otter", From c4fa57c29c03eba6694663a7de7031a4c54e2172 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 13:33:13 +0000 Subject: [PATCH 4/4] fix(server): ACP rollback recovery covers fork baselines and id-less attempts A non-truncating provider's session is stale when the newest turn it accepted was rolled back, or ran under another native id. The first signal needs no native ids, so threads whose attempts predate them recover too. Inherited fork history now counts as kept, so rewinding a fork to its baseline replays it. Co-Authored-By: Claude Opus 5.5 --- .../ProviderTurnStartService.ts | 41 ++++--- .../ProviderSwitch.integration.test.ts | 106 ++++++++++++------ 2 files changed, 98 insertions(+), 49 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 3854ba08f6767..4ef9ede9cd27c 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -117,13 +117,14 @@ const refusedBeforePrompt = (error: unknown): boolean => (Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause)); /** - * Whether `providerThread`'s native session holds none of the turns kept before `run`. Legacy - * attempts lack a native id, so a reset is only visible once some attempt here recorded one. + * Whether a non-truncating provider's native session lost the conversation kept before `run`. + * Rollback resets it: the newest accepted turn was rolled back (works without legacy-missing + * native ids) or ran under another native id (rollback of runs the provider never accepted). */ const nativeSessionMissesKeptTurns = ( projection: Pick< OrchestrationV2ThreadProjection, - "runs" | "attempts" | "providerTurns" | "contextHandoffs" + "thread" | "runs" | "attempts" | "providerTurns" | "contextHandoffs" >, providerThread: OrchestrationV2ProviderThread, run: OrchestrationV2Run, @@ -131,20 +132,28 @@ const nativeSessionMissesKeptTurns = ( const nativeId = providerThread.nativeThreadRef?.nativeId; if (nativeId === undefined) return false; const acceptedAttemptIds = new Set(projection.providerTurns.map((turn) => turn.runAttemptId)); - const keptRunIds = new Set( - projection.runs - .filter((source) => source.ordinal < run.ordinal && source.status !== "rolled_back") - .map((source) => source.id), - ); - const accepted = projection.attempts.filter( - (attempt) => - attempt.providerThreadId === providerThread.id && acceptedAttemptIds.has(attempt.id), - ); - const kept = accepted.filter((attempt) => keptRunIds.has(attempt.runId)); + const accepted = projection.attempts.filter((attempt) => acceptedAttemptIds.has(attempt.id)); + const runsById = new Map(projection.runs.map((source) => [source.id, source])); + const ordinalOf = (attempt: OrchestrationV2RunAttempt) => + runsById.get(attempt.runId)?.ordinal ?? 0; + const newest = accepted + .filter((attempt) => attempt.providerThreadId === providerThread.id) + .toSorted((a, b) => ordinalOf(a) - ordinalOf(b) || a.attemptOrdinal - b.attemptOrdinal) + .at(-1); + if (newest === undefined) return false; + // Inherited fork history is kept even when every local run was rolled back. + const keepsHistory = + (projection.thread.forkedFrom?.type === "run" && projection.thread.lineage.sideChat !== true) || + accepted.some((attempt) => { + const source = runsById.get(attempt.runId); + return ( + source !== undefined && source.ordinal < run.ordinal && source.status !== "rolled_back" + ); + }); return ( - kept.length > 0 && - accepted.some((attempt) => attempt.nativeThreadId !== undefined) && - !kept.some((attempt) => attempt.nativeThreadId === nativeId) && + keepsHistory && + (runsById.get(newest.runId)?.status === "rolled_back" || + (newest.nativeThreadId !== undefined && newest.nativeThreadId !== nativeId)) && !projection.contextHandoffs.some( (handoff) => handoff.toProviderThreadId === providerThread.id && 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 c35e9b27fe10d..aa45a633cd40a 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1505,8 +1505,9 @@ describe("orchestration v2 provider switching", () => { modelSelection: GROK_MODEL_SELECTION, rollback: "replaces_native_thread", failResumeAfterRollback: false, - replaysKeptTurn: true, history: "plain", + delivers: ["favourite colour is teal"], + withholds: ["favourite animal is an otter"], }, { name: "a native rollback whose next resume fails", @@ -1516,8 +1517,9 @@ describe("orchestration v2 provider switching", () => { modelSelection: CLAUDE_MODEL_SELECTION, rollback: "truncates", failResumeAfterRollback: true, - replaysKeptTurn: true, history: "plain", + delivers: ["favourite colour is teal"], + withholds: ["favourite animal is an otter"], }, { // Pi forks into a new native thread that still holds the kept turns. @@ -1528,8 +1530,9 @@ describe("orchestration v2 provider switching", () => { modelSelection: CLAUDE_MODEL_SELECTION, rollback: "replaces_native_thread", failResumeAfterRollback: false, - replaysKeptTurn: false, history: "plain", + delivers: [], + withholds: ["favourite colour is teal", "favourite animal is an otter"], }, { name: "an ACP rollback to a turn recorded before attempts kept their native session", @@ -1539,8 +1542,9 @@ describe("orchestration v2 provider switching", () => { modelSelection: GROK_MODEL_SELECTION, rollback: "replaces_native_thread", failResumeAfterRollback: false, - replaysKeptTurn: true, history: "legacy_first_turn", + delivers: ["favourite colour is teal"], + withholds: ["favourite animal is an otter"], }, { name: "an ACP rollback in a forked thread", @@ -1550,8 +1554,33 @@ describe("orchestration v2 provider switching", () => { modelSelection: GROK_MODEL_SELECTION, rollback: "replaces_native_thread", failResumeAfterRollback: false, - replaysKeptTurn: true, history: "forked", + delivers: ["favourite colour is teal", "home town is Oslo"], + withholds: ["favourite animal is an otter"], + }, + { + name: "an ACP rollback where no attempt recorded its native session", + instanceId: GROK_MODEL_SELECTION.instanceId, + driver: GROK_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + modelSelection: GROK_MODEL_SELECTION, + rollback: "replaces_native_thread", + failResumeAfterRollback: false, + history: "legacy_all_turns", + delivers: ["favourite colour is teal"], + withholds: ["favourite animal is an otter"], + }, + { + name: "an ACP rollback of a forked thread to its baseline", + instanceId: GROK_MODEL_SELECTION.instanceId, + driver: GROK_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + modelSelection: GROK_MODEL_SELECTION, + rollback: "replaces_native_thread", + failResumeAfterRollback: false, + history: "forked_to_baseline", + delivers: ["home town is Oslo"], + withholds: ["favourite colour is teal", "favourite animal is an otter"], }, ] as const)( "gives the agent the kept conversation and not the rewound turn after $name", @@ -1563,6 +1592,7 @@ describe("orchestration v2 provider switching", () => { const capturedTurns = yield* Ref.make>([]); const injectedHistory = yield* Ref.make>([]); const failResumeOnce = yield* Ref.make(false); + const nativeThreadGeneration = yield* Ref.make(0); const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ makeTestAdapter({ instanceId: scenario.instanceId, @@ -1573,7 +1603,7 @@ describe("orchestration v2 provider switching", () => { capturedTurns, injectedHistory, failResumeOnce, - nativeThreadGeneration: yield* Ref.make(0), + nativeThreadGeneration, rollback: scenario.rollback, }), ]); @@ -1624,7 +1654,9 @@ describe("orchestration v2 provider switching", () => { branch: null, worktreePath: null, }); - if (scenario.history === "forked") { + const forked = + scenario.history === "forked" || scenario.history === "forked_to_baseline"; + if (forked) { const sourceThreadId = ThreadId.make(`${name}:source`); yield* createThread(sourceThreadId); yield* runToCompletion( @@ -1656,11 +1688,17 @@ describe("orchestration v2 provider switching", () => { "My favourite animal is an otter. Reply with only the word: noted", ); const beforeRollback = yield* orchestrator.getThreadProjection(threadId); - if (scenario.history === "legacy_first_turn") { + if ( + scenario.history === "legacy_first_turn" || + scenario.history === "legacy_all_turns" + ) { const firstRunId = beforeRollback.runs.find((run) => run.ordinal === 1)?.id; yield* (yield* EventSink.EventSinkV2).write({ events: beforeRollback.attempts - .filter((attempt) => attempt.runId === firstRunId) + .filter( + (attempt) => + scenario.history === "legacy_all_turns" || attempt.runId === firstRunId, + ) .map(({ nativeThreadId: _nativeThreadId, ...legacy }, index) => ({ id: EventId.make(`${name}:legacy-attempt:${index}`), type: "run-attempt.updated" as const, @@ -1670,8 +1708,10 @@ describe("orchestration v2 provider switching", () => { })), }); } + const keptOrdinal = scenario.history === "forked_to_baseline" ? 0 : 1; const keptCheckpoint = beforeRollback.checkpoints.find( - (checkpoint) => checkpoint.appRunOrdinal === 1 && checkpoint.status === "ready", + (checkpoint) => + (checkpoint.appRunOrdinal ?? 0) === keptOrdinal && checkpoint.status === "ready", ); if (keptCheckpoint === undefined) { return yield* Effect.die("Expected a checkpoint after the first turn."); @@ -1698,7 +1738,9 @@ describe("orchestration v2 provider switching", () => { yield* worker.drain(); assert.deepEqual( (yield* orchestrator.getThreadProjection(threadId)).runs.map((run) => run.status), - ["completed", "rolled_back"], + scenario.history === "forked_to_baseline" + ? ["rolled_back", "rolled_back"] + : ["completed", "rolled_back"], ); if (scenario.failResumeAfterRollback) { const providerSessionId = (yield* orchestrator.getThreadProjection(threadId)) @@ -1722,31 +1764,29 @@ describe("orchestration v2 provider switching", () => { const delivered = (run3?.text ?? "") + (yield* encodeJson((yield* Ref.get(injectedHistory)).slice(historyBeforeRun3))); - if (scenario.replaysKeptTurn) { - assert.include( - delivered, - "favourite colour is teal", - "the agent's session no longer holds the kept turn, so the next turn must carry it", - ); - } else { - assert.notInclude( - delivered, - "favourite colour is teal", - "the agent's session still holds the kept turn, so no handoff replays it", - ); + for (const kept of scenario.delivers) { + assert.include(delivered, kept, "the agent's session lacks this kept turn"); } - if (scenario.history === "forked") { - assert.include( - delivered, - "home town is Oslo", - "the fork's inherited turn is kept too", + for (const withheld of scenario.withholds) { + assert.notInclude(delivered, withheld, "the session holds it or it was rewound"); + } + if (scenario.name === "an ACP rollback that starts a fresh agent session") { + const recoveries = (projection: OrchestrationV2ThreadProjection) => + projection.contextHandoffs.filter( + (handoff) => handoff.strategy === "full_thread_summary", + ).length; + const recoveriesBeforeRun4 = recoveries( + yield* orchestrator.getThreadProjection(threadId), + ); + const sessionsBeforeRun4 = yield* Ref.get(nativeThreadGeneration); + yield* runToCompletion(4, "And what did I say second?"); + assert.strictEqual( + recoveries(yield* orchestrator.getThreadProjection(threadId)), + recoveriesBeforeRun4, + "the recovered session holds the kept turns, so the next turn replays nothing", ); + assert.strictEqual(yield* Ref.get(nativeThreadGeneration), sessionsBeforeRun4); } - assert.notInclude( - delivered, - "favourite animal is an otter", - "the rewound turn must not reach the agent", - ); }).pipe( Effect.provide( ProviderReplayHarness.layerWithRegistry(