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.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index cdbd259440ab2..18f58572ca340 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -757,8 +757,21 @@ function makeLocalCommandHarness(input: { ), ); const fallbackEnsureThread = vi.fn(() => Effect.succeed(providerThread)); + const providerSession = { + id: providerSessionId, + driver: providerThread.driver, + providerInstanceId: newInstanceId, + status: "ready", + cwd: "/tmp/native-account-command", + model: null, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }; const resumeFallbackSession = { driver: providerThread.driver, + providerSession, resumeThread: () => Effect.fail( input.resumeFailure ?? @@ -770,18 +783,6 @@ function makeLocalCommandHarness(input: { ), ensureThread: fallbackEnsureThread, }; - const providerSession = { - id: providerSessionId, - driver: providerThread.driver, - providerInstanceId: newInstanceId, - status: "ready", - cwd: "/tmp/native-account-command", - model: null, - capabilities: CodexProviderCapabilitiesV2, - createdAt: now, - updatedAt: now, - lastError: null, - }; const startTurn = vi.fn((_turn: ProviderAdapter.ProviderAdapterV2TurnInput) => Effect.void); const sideChatThread = { ...providerThread, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 1d6e34b4a8652..a7b49dc11523a 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,53 @@ const refusedBeforePrompt = (error: unknown): boolean => PROMPT_REFUSAL_TAGS.some((tag) => Predicate.isTagged(error, tag)) || (Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause)); +/** + * 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, + "thread" | "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 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 ( + keepsHistory && + (runsById.get(newest.runId)?.status === "rolled_back" || + (newest.nativeThreadId !== undefined && newest.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; @@ -1091,13 +1139,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, - ), - ), + 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 }; @@ -1245,7 +1302,7 @@ export const layer: Layer.Layer< } if (providerThread.nativeThreadRef === null) { // Pass the stored provider row for adoption; adapter-minted duplicates make activeProviderThreadId flap. - return yield* loadFromProvider( + const created = yield* loadFromProvider( session.ensureThread({ threadId: projection.thread.id, title: projection.thread.title, @@ -1255,6 +1312,57 @@ export const layer: Layer.Layer< existingProviderThread: providerThread, }), ); + // Non-truncating providers clear the native ref on rollback; hand the new session the kept turns. + // Cursor ensureThread may mint a new row id from the new native agent; match kept turns by the stored id. + if ( + created !== undefined && + !session.providerSession.capabilities.checkpointing.providerCanRollbackConversation && + nativeSessionMissesKeptTurns(projection, { ...created, id: providerThread.id }, run) + ) { + const { handoff, transferId, createdAt } = yield* recoveryHandoff; + effectiveHandoffs = withRecoveryHandoff(handoff); + 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 created; } const uncertainDelivery = projection.contextHandoffs.some( (handoff) => @@ -1262,15 +1370,21 @@ 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 nativeSessionLacksKeptTurns = + !session.providerSession.capabilities.checkpointing.providerCanRollbackConversation && + nativeSessionMissesKeptTurns(projection, providerThread, run); 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({ @@ -1350,9 +1464,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 629b3902e763d..ebba888f6b990 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,9 @@ 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. + // Cursor clears the native ref so the next turn's ensureThread opens a new agent. + readonly rollback?: "truncates" | "replaces_native_thread" | "clears_native_thread"; }): ProviderAdapter.ProviderAdapterV2["Service"] { return { instanceId: input.instanceId, @@ -392,8 +395,31 @@ 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 === "clears_native_thread" + ? { + ...providerThread, + nativeThreadRef: null, + nativeConversationHeadRef: null, + } + : 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 +1503,329 @@ 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, + history: "plain", + delivers: ["favourite colour is teal"], + withholds: ["favourite animal is an otter"], + }, + { + // Cursor-shaped: rollback clears nativeThreadRef; ensureThread mints a new row id. + name: "a rollback that clears the native agent then ensureThread mints a new id", + instanceId: GROK_MODEL_SELECTION.instanceId, + driver: GROK_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + modelSelection: GROK_MODEL_SELECTION, + rollback: "clears_native_thread", + failResumeAfterRollback: false, + history: "plain", + delivers: ["favourite colour is teal"], + withholds: ["favourite animal is an otter"], + }, + { + 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, + 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. + 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, + 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", + instanceId: GROK_MODEL_SELECTION.instanceId, + driver: GROK_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + modelSelection: GROK_MODEL_SELECTION, + rollback: "replaces_native_thread", + failResumeAfterRollback: false, + history: "legacy_first_turn", + delivers: ["favourite colour is teal"], + withholds: ["favourite animal is an otter"], + }, + { + 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, + 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", + (scenario) => + Effect.scoped( + Effect.gen(function* () { + const name = `rollback-kept-context-${scenario.driver}-${scenario.history}`; + const cwd = yield* checkpointWorkspace(name); + 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, + driver: scenario.driver, + capabilities: scenario.capabilities, + modelSelection: scenario.modelSelection, + responseByRunOrdinal: { 1: "noted", 2: "noted" }, + capturedTurns, + injectedHistory, + failResumeOnce, + nativeThreadGeneration, + 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, + thread: ThreadId = threadId, + ) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${thread}:${ordinal}`), + threadId: thread, + messageId: MessageId.make(`${thread}:${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.threadId === thread && + event.payload.ordinal === ordinal && + (event.payload.status === "completed" || event.payload.status === "failed"), + ), + Stream.runHead, + ); + yield* worker.drain(); + }); + 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, + }); + const forked = + scenario.history === "forked" || scenario.history === "forked_to_baseline"; + if (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", + ); + yield* runToCompletion( + 2, + "My favourite animal is an otter. Reply with only the word: noted", + ); + const beforeRollback = yield* orchestrator.getThreadProjection(threadId); + 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) => + 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, + threadId, + occurredAt: beforeRollback.thread.createdAt, + payload: legacy, + })), + }); + } + const keptOrdinal = scenario.history === "forked_to_baseline" ? 0 : 1; + const keptCheckpoint = beforeRollback.checkpoints.find( + (checkpoint) => + (checkpoint.appRunOrdinal ?? 0) === keptOrdinal && 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), + scenario.history === "forked_to_baseline" + ? ["rolled_back", "rolled_back"] + : ["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))); + for (const kept of scenario.delivers) { + assert.include(delivered, kept, "the agent's session lacks this kept turn"); + } + 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); + } + }).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" })], diff --git a/docs/user/cursor.md b/docs/user/cursor.md index 4263ceb14fd8d..be00a7c9925e9 100644 --- a/docs/user/cursor.md +++ b/docs/user/cursor.md @@ -51,15 +51,19 @@ The adapter supports: - image attachments; - interruption, queued app messages, and orchestrator-owned interrupt/restart steering; - provider conversation snapshots through `Agent.messages.list()`; +- Edit from here / checkpoint rollback (the SDK cannot truncate history; the next turn starts a fresh + agent and receives the kept conversation as a portable handoff); - Cursor `task` subagents, projected as read-only child app threads with their tool activity and final result. -The public SDK does not currently expose native agent fork, conversation rollback, active steering, -or interactive approval callbacks. Direct active steering is advertised as unsupported, while V2 -steering uses the orchestrator's interrupt-and-restart path and preserves the app run identity across -provider turns. Same-provider Cursor forks use the orchestrator's portable full-thread context -handoff into a fresh Cursor agent. The SDK has in-process custom callback tools, but the V2 adapter -intentionally uses the authenticated, thread-scoped MCP server instead. +The public SDK does not currently expose native agent fork, conversation truncation, active steering, +or interactive approval callbacks. Checkpoint rollback restores your files as usual, clears the Cursor +agent binding, and the next turn creates a new agent that is handed the conversation kept before the +checkpoint. Direct active steering is advertised as unsupported, while V2 steering uses the +orchestrator's interrupt-and-restart path and preserves the app run identity across provider turns. +Same-provider Cursor forks use the orchestrator's portable full-thread context handoff into a fresh +Cursor agent. The SDK has in-process custom callback tools, but the V2 adapter intentionally uses the +authenticated, thread-scoped MCP server instead. Portable handoffs summarize eligible timeline items and can omit the end of long messages. Read [Context in portable handoffs](./portable-handoffs.md) before using a fork for work whose exact diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 678cd7edf2296..42b84bf2dab51 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -732,8 +732,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, }, diff --git a/packages/provider-cursor/src/server/adapter.test.ts b/packages/provider-cursor/src/server/adapter.test.ts index cc8afd31311bb..6124c70fa4f31 100644 --- a/packages/provider-cursor/src/server/adapter.test.ts +++ b/packages/provider-cursor/src/server/adapter.test.ts @@ -3,6 +3,7 @@ import { assert, describe, it } from "@effect/vitest"; import type { InteractionUpdate } from "@cursor/sdk"; import { toolOutputImages } from "@t3tools/shared/toolOutput"; import { + CheckpointId, EnvironmentId, MessageId, NodeId, @@ -14,7 +15,9 @@ import { ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; @@ -1602,6 +1605,231 @@ describe("CursorAdapterV2", () => { ), ); + it.effect("advertises Cursor conversation rollback and clears the native agent", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const workspace = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "cursor-v2-rollback-", + }); + const instanceId = ProviderInstanceId.make("cursor"); + const threadId = ThreadId.make("cursor-rollback-thread"); + const modelSelection = { instanceId, model: "composer-2.5" }; + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: workspace, + }); + let createCount = 0; + const closedAgents: Array = []; + const adapter = yield* makeCursorAdapterV2({ + instanceId, + environment: { HOME: workspace }, + }).pipe( + Effect.provideService(CursorAgentSdk.CursorAgentSdkRunner, { + assertComplete: Effect.void, + open: () => + Effect.sync(() => { + createCount += 1; + const agentId = `native-cursor-rollback-${createCount}`; + return { + agentId, + listMessages: Effect.succeed([]), + close: Effect.sync(() => { + closedAgents.push(agentId); + }), + send: () => Effect.die("unexpected Cursor run during rollback"), + }; + }), + }), + Effect.provide(TestProviderHost.layer({ cwd: workspace })), + ); + const capabilities = yield* adapter.getCapabilities(); + assert.isTrue(capabilities.threads.canRollbackThread); + assert.isFalse(capabilities.checkpointing.providerCanRollbackConversation); + assert.isTrue(capabilities.checkpointing.providerRollbackReturnsSnapshot); + + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("cursor-rollback-session"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + assert.equal(providerThread.nativeThreadRef?.nativeId, "native-cursor-rollback-1"); + + const snapshot = yield* runtime.rollbackThread({ + providerThread, + target: { + type: "thread_start", + checkpointId: CheckpointId.make("cursor-rollback-checkpoint"), + appRunOrdinal: 0, + }, + providerThreadTurns: [], + }); + assert.equal(snapshot.providerThread.id, providerThread.id); + assert.isNull(snapshot.providerThread.nativeThreadRef); + assert.isNull(snapshot.providerThread.nativeConversationHeadRef); + assert.deepEqual(snapshot.providerTurns, []); + assert.deepEqual(snapshot.messages, []); + assert.equal(createCount, 1); + assert.deepEqual(closedAgents, ["native-cursor-rollback-1"]); + + const recovered = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + existingProviderThread: snapshot.providerThread, + }); + assert.equal(recovered.id, providerThread.id); + assert.equal(recovered.nativeThreadRef?.nativeId, "native-cursor-rollback-2"); + assert.equal(createCount, 2); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(NodeServices.layer, IdAllocator.layer, McpProviderSessions.layer), + ), + ), + ); + + it.effect("rejects Cursor rollback while a turn is active", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const workspace = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "cursor-v2-rollback-busy-", + }); + const instanceId = ProviderInstanceId.make("cursor"); + const threadId = ThreadId.make("cursor-rollback-busy-thread"); + const modelSelection = { instanceId, model: "composer-2.5" }; + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: workspace, + }); + const holdTurn = yield* Deferred.make(); + const adapter = yield* makeCursorAdapterV2({ + instanceId, + environment: { HOME: workspace }, + }).pipe( + Effect.provideService(CursorAgentSdk.CursorAgentSdkRunner, { + assertComplete: Effect.void, + open: () => + Effect.succeed({ + agentId: "native-cursor-rollback-busy", + listMessages: Effect.succeed([]), + close: Effect.void, + send: () => + Effect.succeed({ + agentId: "native-cursor-rollback-busy", + runId: "native-cursor-rollback-busy-run", + wait: Deferred.await(holdTurn).pipe( + Effect.as({ + id: "native-cursor-rollback-busy-run", + requestId: "native-request", + status: "finished" as const, + model: { id: "composer-2.5" }, + durationMs: 1, + }), + ), + cancel: Deferred.succeed(holdTurn, undefined).pipe(Effect.asVoid), + }), + }), + }), + Effect.provide(TestProviderHost.layer({ cwd: workspace })), + ); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("cursor-rollback-busy-session"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const now = yield* DateTime.now; + const turnFiber = yield* runtime + .startTurn({ + threadId, + providerThread, + modelSelection, + runtimePolicy, + runId: RunId.make("cursor-rollback-busy-run"), + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: RunAttemptId.make("cursor-rollback-busy-attempt"), + rootNodeId: NodeId.make("cursor-rollback-busy-root"), + appThread: { + id: threadId, + projectId: ProjectId.make("cursor-rollback-busy-project"), + createdBy: "user", + creationSource: "web", + title: "Cursor rollback busy", + providerInstanceId: instanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: providerThread.id, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: threadId, + }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + message: { + messageId: MessageId.make("cursor-rollback-busy-message"), + createdBy: "user", + creationSource: "web", + text: "Busy turn.", + attachments: [], + }, + }) + .pipe(Effect.forkScoped); + yield* runtime.events.pipe( + Stream.takeUntil( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running", + ), + Stream.runCollect, + ); + + const activeFailure = yield* runtime + .rollbackThread({ + providerThread, + target: { + type: "thread_start", + checkpointId: CheckpointId.make("cursor-rollback-busy-checkpoint"), + appRunOrdinal: 0, + }, + providerThreadTurns: [], + }) + .pipe(Effect.flip); + assert.equal(activeFailure._tag, "ProviderAdapterRollbackThreadError"); + + yield* Deferred.succeed(holdTurn, undefined); + yield* Fiber.join(turnFiber); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(NodeServices.layer, IdAllocator.layer, McpProviderSessions.layer), + ), + ), + ); + it("maps runtime modes to the SDK sandbox and auto-review controls", () => { const base = { interactionMode: "default" as const, diff --git a/packages/provider-cursor/src/server/adapter.ts b/packages/provider-cursor/src/server/adapter.ts index e5b77e1041d4a..f036131212efd 100644 --- a/packages/provider-cursor/src/server/adapter.ts +++ b/packages/provider-cursor/src/server/adapter.ts @@ -85,7 +85,7 @@ export const CursorProviderCapabilitiesV2 = { threads: { canCreateEmptyThread: true, canReadThreadSnapshot: true, - canRollbackThread: false, + canRollbackThread: true, canForkThread: false, canForkFromTurn: false, canForkFromSubagentThread: false, @@ -153,8 +153,9 @@ export const CursorProviderCapabilitiesV2 = { checkpointing: { appCanCheckpointFilesystem: true, supportsNestedCheckpointScopes: true, + // SDK cannot truncate; rollback clears the native agent and the next turn hands off kept turns. providerCanRollbackConversation: false, - providerRollbackReturnsSnapshot: false, + providerRollbackReturnsSnapshot: true, providerCanReadConversationSnapshot: true, }, identity: { @@ -2619,7 +2620,7 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( runtimePolicy: threadInput.runtimePolicy, }); const now = yield* DateTime.now; - return makeProviderThread({ + const created = makeProviderThread({ idAllocator, providerInstanceId: adapterOptions.instanceId, appThreadId: threadInput.threadId, @@ -2627,6 +2628,10 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( nativeThreadId: opened.nativeThreadId, now, }); + // Adoption after rollback clears the native ref; keep the stored row id. + return threadInput.existingProviderThread === undefined + ? created + : { ...created, id: threadInput.existingProviderThread.id }; }, (effect, threadInput) => effect.pipe( @@ -2801,15 +2806,49 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( ), ), ), - rollbackThread: (rollbackInput) => - Effect.fail( - new ProviderAdapter.ProviderAdapterRollbackThreadError({ - driver: CursorAgentSdk.CURSOR_PROVIDER, - providerThreadId: rollbackInput.providerThread.id, - checkpointId: rollbackInput.target.checkpointId, - cause: "Cursor Agent SDK does not expose conversation rollback.", - }), - ), + // The SDK cannot truncate; the next turn creates an agent and hands it the kept turns. + rollbackThread: Effect.fn("CursorAdapterV2.rollbackThread")( + function* (rollbackInput: ProviderAdapter.ProviderAdapterV2RollbackThreadInput) { + const currentTurn = yield* Ref.get(activeTurn); + if (currentTurn !== null) { + return yield* new ProviderAdapter.ProviderAdapterProtocolError({ + driver: CursorAgentSdk.CURSOR_PROVIDER, + detail: `Cannot roll back Cursor provider thread ${rollbackInput.providerThread.id} while turn ${currentTurn.providerTurnId} is active`, + }); + } + const existing = yield* Ref.get(liveAgent); + if (existing !== null) { + yield* existing.session.close.pipe(Effect.ignore); + yield* Ref.set(liveAgent, null); + } + yield* Ref.set(told, null); + const now = yield* DateTime.now; + return { + providerThread: { + ...rollbackInput.providerThread, + nativeThreadRef: null, + nativeConversationHeadRef: null, + status: "idle" as const, + updatedAt: now, + }, + providerTurns: [], + messages: [], + runtimeRequests: [], + }; + }, + (effect, rollbackInput) => + effect.pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterRollbackThreadError({ + driver: CursorAgentSdk.CURSOR_PROVIDER, + providerThreadId: rollbackInput.providerThread.id, + checkpointId: rollbackInput.target.checkpointId, + cause, + }), + ), + ), + ), forkThread: (forkInput) => Effect.fail( new ProviderAdapter.ProviderAdapterForkThreadError({ diff --git a/packages/provider-cursor/src/server/status.test.ts b/packages/provider-cursor/src/server/status.test.ts index c3dc7deb120ff..20331ad918d44 100644 --- a/packages/provider-cursor/src/server/status.test.ts +++ b/packages/provider-cursor/src/server/status.test.ts @@ -111,7 +111,7 @@ describe("buildCursorProviderSnapshot", () => { status: "warning", message: "Cursor SDK model discovery returned no built-in models.", models: [], - supportsConversationRollback: false, + supportsConversationRollback: true, }); }); }); diff --git a/packages/provider-cursor/src/server/status.ts b/packages/provider-cursor/src/server/status.ts index ffeca7a1da0b7..3d637291827dd 100644 --- a/packages/provider-cursor/src/server/status.ts +++ b/packages/provider-cursor/src/server/status.ts @@ -25,7 +25,7 @@ import * as CursorSdkCatalog from "./CursorSdkCatalog.ts"; const CURSOR_PRESENTATION = { displayName: "Cursor", - supportsConversationRollback: false, + supportsConversationRollback: true, showInteractionModeToggle: true, } as const; const EMPTY_CAPABILITIES: ModelCapabilities = createModelCapabilities({