diff --git a/apps/server/src/orchestration-v2/CheckpointRestoreSafety.test.ts b/apps/server/src/orchestration-v2/CheckpointRestoreSafety.test.ts index 5f79aefdd0b7b..2164e3862847c 100644 --- a/apps/server/src/orchestration-v2/CheckpointRestoreSafety.test.ts +++ b/apps/server/src/orchestration-v2/CheckpointRestoreSafety.test.ts @@ -24,6 +24,7 @@ import { ProjectionStoreV2 } from "./ProjectionStore.ts"; import * as ProjectStore from "./ProjectStore.ts"; import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; import { RuntimePolicyV2 } from "./RuntimePolicy.ts"; +import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts"; it.effect.each([ "nested", @@ -102,6 +103,7 @@ it.effect.each([ Layer.provide( Layer.mergeAll( NodeServices.layer, + Layer.mock(WorkspaceEntries.WorkspaceEntries)({ refresh: () => Effect.void }), IdAllocator.layer, Layer.mock(ProjectStore.ProjectStoreV2)({ get: () => Effect.succeed(Option.some({ workspaceRoot: parent } as never)), @@ -190,7 +192,7 @@ it.effect.each([ assert.equal(yield* fs.readFileString(otherFile), "other thread's uncommitted work"); } else { yield* service.execute({ threadId, providerThreadId, checkpointId, scopeId, restoreFiles }); - assert.deepEqual(calls, restoreFiles ? ["provider", "files"] : ["provider"]); + assert.deepEqual(calls, restoreFiles ? ["files", "provider"] : ["provider"]); assert.equal(yield* fs.exists(otherFile), !restoreFiles); } }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), diff --git a/apps/server/src/orchestration-v2/CheckpointRollbackService.test.ts b/apps/server/src/orchestration-v2/CheckpointRollbackService.test.ts index 433a3be43ac3c..f53b984247eba 100644 --- a/apps/server/src/orchestration-v2/CheckpointRollbackService.test.ts +++ b/apps/server/src/orchestration-v2/CheckpointRollbackService.test.ts @@ -16,6 +16,10 @@ import * as Option from "effect/Option"; import { symlinksSupported } from "@t3tools/shared/testing/symlinks"; import * as Layer from "effect/Layer"; +import * as ServerConfig from "../config.ts"; +import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts"; +import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; + import { resolveCodexRollbackTurnCount } from "./Adapters/CodexAdapterV2.ts"; import { isCheckpointRestoreIsolated } from "./CheckpointRestoreSafety.ts"; import * as CheckpointService from "./CheckpointService.ts"; @@ -36,6 +40,7 @@ const layerCheckpointRollbackService = CheckpointRollbackService.layer.pipe( Layer.provide( Layer.mergeAll( NodeServices.layer, + Layer.mock(WorkspaceEntries.WorkspaceEntries)({ refresh: () => Effect.void }), Layer.mock(ProjectStore.ProjectStoreV2)({ get: () => Effect.succeed(unrelatedProject) }), ), ), @@ -464,11 +469,154 @@ it.effect.each([ yield* service.execute({ threadId, providerThreadId, checkpointId, scopeId, restoreFiles }); assert.deepEqual( calls, - restoreFiles ? ["provider", "files", "projection"] : ["provider", "projection"], + restoreFiles ? ["files", "provider", "projection"] : ["provider", "projection"], ); }).pipe(Effect.provide(layerTest)); }); +it.effect.each([false, true])( + "keeps provider conversation and indexed files consistent, restoreFails=%s", + (restoreFails) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; + const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "t3-rollback-index-" }); + const currentFile = path.join(cwd, "rollback-current.txt"); + const restoredFile = path.join(cwd, "rollback-restored.txt"); + yield* fs.writeFileString(currentFile, "current turn"); + const search = () => workspaceEntries.search({ cwd, query: "rollback", limit: 10 }); + assert.deepEqual((yield* search()).entries, [{ path: "rollback-current.txt", kind: "file" }]); + + const threadId = ThreadId.make("rollback-restore-state"); + const providerThreadId = ProviderThreadId.make("rollback-restore-provider"); + const providerSessionId = ProviderSessionId.make("rollback-restore-session"); + const instanceId = ProviderInstanceId.make("rollback-restore-instance"); + const checkpointId = CheckpointId.make("rollback-restore-checkpoint"); + const scopeId = CheckpointScopeId.make("rollback-restore-scope"); + const providerThread = { + id: providerThreadId, + providerSessionId, + providerInstanceId: instanceId, + }; + const projection = { + thread: { + worktreePath: cwd, + activeProviderThreadId: providerThreadId, + modelSelection: { instanceId, model: "test" }, + }, + providerThreads: [providerThread], + providerSessions: [], + providerTurns: [ + { + id: "turn-1", + providerThreadId, + runAttemptId: "attempt-1", + ordinal: 1, + status: "completed", + }, + ], + nodes: [], + attempts: [{ id: "attempt-1", runId: "run-1" }], + checkpoints: [{ id: checkpointId, scopeId, status: "ready", appRunOrdinal: null }], + checkpointScopes: [{ id: scopeId, cwd }], + runs: [{ id: "run-1", ordinal: 1, status: "completed", rootNodeId: null }], + } as unknown as OrchestrationV2ThreadProjection; + const restoreFailure = new CheckpointService.CheckpointRestoreError({ + scopeId, + checkpointId, + cause: new Error("git restore failed"), + }); + const providerConversation = ["turn-1"]; + const persistedEvents: string[] = []; + const testLayer = CheckpointRollbackService.layer.pipe( + Layer.provide( + Layer.mergeAll( + IdAllocator.layer, + Layer.succeed(WorkspaceEntries.WorkspaceEntries, workspaceEntries), + Layer.mock(ProjectStore.ProjectStoreV2)({ + get: () => Effect.succeed(unrelatedProject), + }), + Layer.mock(CheckpointService.CheckpointServiceV2)({ + restore: () => + restoreFails + ? Effect.fail(restoreFailure) + : Effect.gen(function* () { + yield* fs.remove(currentFile); + yield* fs.writeFileString(restoredFile, "checkpoint turn"); + }).pipe(Effect.orDie), + }), + Layer.mock(EventSink.EventSinkV2)({ + write: ({ events }) => + Effect.sync(() => { + persistedEvents.push(...events.map((event) => event.type)); + return []; + }), + }), + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getThreadRecords: () => Effect.succeed(projection), + getShellSnapshot: () => + Effect.succeed({ + schemaVersion: 1, + snapshotSequence: 0, + threads: [], + archivedThreads: [], + }), + }), + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ + open: () => + Effect.succeed({ + rollbackThread: (input: ProviderAdapterV2RollbackThreadInput) => + Effect.gen(function* () { + const count = yield* resolveCodexRollbackTurnCount(input); + assert.equal(count, 1); + providerConversation.splice(providerConversation.length - count, count); + return { providerThread }; + }), + } as never), + }), + Layer.mock(RuntimePolicy.RuntimePolicyV2)({ + resolve: () => Effect.succeed({} as never), + }), + ), + ), + ); + const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2.pipe( + Effect.provide(testLayer), + ); + const rollback = service.execute({ threadId, providerThreadId, checkpointId, scopeId }); + if (restoreFails) { + const error = yield* Effect.flip(rollback); + assert.equal(error.reason, "unexpected-failure"); + assert.equal(error.cause, restoreFailure); + assert.deepEqual(providerConversation, ["turn-1"]); + assert.deepEqual(persistedEvents, []); + assert.equal(yield* fs.readFileString(currentFile), "current turn"); + assert.deepEqual((yield* search()).entries, [ + { path: "rollback-current.txt", kind: "file" }, + ]); + } else { + yield* rollback; + assert.deepEqual(providerConversation, []); + assert.include(persistedEvents, "run.updated"); + assert.equal(yield* fs.exists(currentFile), false); + assert.equal(yield* fs.readFileString(restoredFile), "checkpoint turn"); + assert.deepEqual((yield* search()).entries, [ + { path: "rollback-restored.txt", kind: "file" }, + ]); + } + }).pipe( + Effect.scoped, + Effect.provide( + WorkspaceEntries.layer.pipe( + Layer.provide(WorkspacePaths.layer), + Layer.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-rollback-config-" })), + Layer.provideMerge(NodeServices.layer), + ), + ), + ), +); + it.effect.skipIf(!symlinksSupported)( "rejects an archived thread sharing a worktree through a symlink", () => diff --git a/apps/server/src/orchestration-v2/CheckpointRollbackService.ts b/apps/server/src/orchestration-v2/CheckpointRollbackService.ts index 1710e7a8cd844..006e92ae0101d 100644 --- a/apps/server/src/orchestration-v2/CheckpointRollbackService.ts +++ b/apps/server/src/orchestration-v2/CheckpointRollbackService.ts @@ -14,6 +14,7 @@ import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; +import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts"; import { isCheckpointRestoreIsolated, SHARED_WORKSPACE_RESTORE_MESSAGE, @@ -91,6 +92,7 @@ export const layer: Layer.Layer< | FileSystem.FileSystem | Path.Path | ProjectStore.ProjectStoreV2 + | WorkspaceEntries.WorkspaceEntries > = Layer.effect( CheckpointRollbackServiceV2, Effect.gen(function* () { @@ -103,6 +105,7 @@ export const layer: Layer.Layer< const fileSystem = yield* FileSystem.FileSystem; const projects = yield* ProjectStore.ProjectStoreV2; const path = yield* Path.Path; + const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; const execute = Effect.fn("orchestrationV2.checkpointRollback.execute")(function* (input: { readonly threadId: ThreadId; @@ -197,31 +200,6 @@ export const layer: Layer.Layer< }); const targetOrdinal = checkpoint.appRunOrdinal ?? 0; - // Stopped and failed runs after the target leave the provider - // conversation too, so they must not stay visible. - const runsToRollback = projection.runs.filter( - (run) => - run.ordinal > targetOrdinal && - (run.status === "completed" || - run.status === "interrupted" || - run.status === "failed" || - run.status === "cancelled"), - ); - // Rolled-back turns stay in the audit history, but no longer exist in - // the provider conversation and must not be counted by a later rewind. - const rolledBackRunIds = new Set( - projection.runs.filter((run) => run.status === "rolled_back").map((run) => run.id), - ); - const rolledBackAttemptIds = new Set( - projection.attempts - .filter((attempt) => rolledBackRunIds.has(attempt.runId)) - .map((attempt) => attempt.id), - ); - const providerThreadTurns = projection.providerTurns.filter( - (turn) => - turn.providerThreadId === providerThread.id && - (turn.runAttemptId === null || !rolledBackAttemptIds.has(turn.runAttemptId)), - ); const rollbackTarget: ProviderAdapterV2RollbackTarget = targetOrdinal === 0 ? { @@ -254,6 +232,36 @@ export const layer: Layer.Layer< }; }); + if (input.restoreFiles !== false) { + yield* checkpoints.restore({ scope, checkpoint }); + yield* workspaceEntries.refresh(scope.cwd); + } + + // Stopped and failed runs after the target leave the provider + // conversation too, so they must not stay visible. + const runsToRollback = projection.runs.filter( + (run) => + run.ordinal > targetOrdinal && + (run.status === "completed" || + run.status === "interrupted" || + run.status === "failed" || + run.status === "cancelled"), + ); + // Rolled-back turns stay in the audit history, but no longer exist in + // the provider conversation and must not be counted by a later rewind. + const rolledBackRunIds = new Set( + projection.runs.filter((run) => run.status === "rolled_back").map((run) => run.id), + ); + const rolledBackAttemptIds = new Set( + projection.attempts + .filter((attempt) => rolledBackRunIds.has(attempt.runId)) + .map((attempt) => attempt.id), + ); + const providerThreadTurns = projection.providerTurns.filter( + (turn) => + turn.providerThreadId === providerThread.id && + (turn.runAttemptId === null || !rolledBackAttemptIds.has(turn.runAttemptId)), + ); const snapshot = runsToRollback.length === 0 ? { providerThread } @@ -262,7 +270,6 @@ export const layer: Layer.Layer< target: rollbackTarget, providerThreadTurns, }); - if (input.restoreFiles !== false) yield* checkpoints.restore({ scope, checkpoint }); const staleCheckpoints = projection.checkpoints.filter( (candidate) => candidate.scopeId === scope.id && diff --git a/apps/server/src/orchestration-v2/CheckpointService.test.ts b/apps/server/src/orchestration-v2/CheckpointService.test.ts index cef73dd1cbbc0..d50685ba186ee 100644 --- a/apps/server/src/orchestration-v2/CheckpointService.test.ts +++ b/apps/server/src/orchestration-v2/CheckpointService.test.ts @@ -18,6 +18,81 @@ import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; import * as CheckpointService from "./CheckpointService.ts"; import * as IdAllocator from "./IdAllocator.ts"; +it.effect.each(["not-repo", "git-failure", "interrupt"] as const)( + "distinguishes repository detection outcomes across checkpoint capture, outcome=%s", + (outcome) => { + const scope: OrchestrationV2CheckpointScope = { + id: CheckpointScopeId.make("checkpoint-scope:detection"), + threadId: ThreadId.make("thread:detection"), + runId: RunId.make("run:detection"), + nodeId: NodeId.make("node:detection"), + parentScopeId: null, + providerThreadId: ProviderThreadId.make("provider-thread:detection"), + kind: "root_run", + ordinalWithinParent: 0, + advancesAppRunCount: true, + cwd: "/repo", + createdAt: DateTime.makeUnsafe("2026-07-28T00:00:00.000Z"), + }; + const gitFailure = new VcsProcessTimeoutError({ + operation: "test.detect", + command: "git", + cwd: scope.cwd, + timeoutMs: 5000, + }); + const testLayer = CheckpointService.layer.pipe( + Layer.provide( + Layer.mergeAll( + IdAllocator.layer, + Layer.mock(CheckpointStore.CheckpointStore)({ + isGitRepository: () => + outcome === "interrupt" + ? Effect.interrupt + : outcome === "git-failure" + ? Effect.fail(gitFailure) + : Effect.succeed(false), + }), + ), + ), + Layer.provideMerge(NodeCrypto.layer), + ); + return Effect.gen(function* () { + const checkpoints = yield* CheckpointService.CheckpointServiceV2; + const baseline = checkpoints.captureBaseline({ scope, ordinalWithinScope: 0 }); + const materialize = checkpoints.materializeBaselineCheckpoint({ + scope, + ordinalWithinScope: 0, + }); + const capture = checkpoints.capture({ + scope, + ordinalWithinScope: 1, + runId: scope.runId, + nodeId: scope.nodeId!, + appRunOrdinal: 1, + capturedAt: scope.createdAt, + }); + if (outcome === "not-repo") { + yield* baseline; + assert.equal((yield* materialize).status, "missing"); + assert.equal((yield* capture).status, "missing"); + } else if (outcome === "git-failure") { + const baselineError = yield* Effect.flip(baseline); + assert.equal(baselineError._tag, "CheckpointBaselineCaptureError"); + assert.equal(baselineError.cause, gitFailure); + for (const operation of [materialize, capture]) { + const error = yield* Effect.flip(operation); + assert.equal(error._tag, "CheckpointCaptureError"); + assert.equal(error.cause, gitFailure); + } + } else { + assert.isTrue(Exit.hasInterrupts(yield* Effect.exit(baseline))); + assert.isTrue(Exit.hasInterrupts(yield* Effect.exit(materialize))); + assert.isTrue(Exit.hasInterrupts(yield* Effect.exit(capture))); + } + }).pipe(Effect.provide(testLayer)); + }, +); + it.effect.each([false, true, "interrupt"] as const)( "materializes baseline, lookup fails=%s", (lookupFails) => { diff --git a/apps/server/src/orchestration-v2/CheckpointService.ts b/apps/server/src/orchestration-v2/CheckpointService.ts index cb514529349bb..e4c5af4cbac61 100644 --- a/apps/server/src/orchestration-v2/CheckpointService.ts +++ b/apps/server/src/orchestration-v2/CheckpointService.ts @@ -259,8 +259,7 @@ export const layer: Layer.Layer< const withWorkspaceLock = (cwd: string, effect: Effect.Effect) => workspaceLocks.withLock(cwd, effect); - const isGitCheckpointable = (cwd: string) => - checkpointStore.isGitRepository(cwd).pipe(Effect.orElseSucceed(() => false)); + const isGitCheckpointable = (cwd: string) => checkpointStore.isGitRepository(cwd); const ensureScope: CheckpointServiceV2Shape["ensureScope"] = (scope) => Effect.succeed(scope); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 917ad3a30b8cb..d4944ffdb88bb 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -11,6 +11,8 @@ import * as AgentSessionScanner from "../project/AgentSessionScanner.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts"; +import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts"; +import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; import * as CheckpointCaptureService from "./CheckpointCaptureService.ts"; import * as CheckpointService from "./CheckpointService.ts"; import * as CheckpointRollbackService from "./CheckpointRollbackService.ts"; @@ -173,6 +175,7 @@ const layerRuntimeRequestServiceProvided = RuntimeRequestService.layer.pipe( const layerCheckpointRollbackServiceProvided = CheckpointRollbackService.layer.pipe( Layer.provide( Layer.mergeAll( + WorkspaceEntries.layer.pipe(Layer.provide(WorkspacePaths.layer)), ProjectStore.layer, layerCheckpointServiceProvided, layerEventSinkProvided, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 687e881c46563..81902d54e34fa 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -22,6 +22,7 @@ import * as VcsProcess from "../../vcs/VcsProcess.ts"; import * as CheckpointCaptureService from "../CheckpointCaptureService.ts"; import * as CheckpointService from "../CheckpointService.ts"; import * as CheckpointRollbackService from "../CheckpointRollbackService.ts"; +import * as WorkspaceEntries from "../../workspace/WorkspaceEntries.ts"; import * as CommandPolicy from "../CommandPolicy.ts"; import * as CommandReceiptStore from "../CommandReceiptStore.ts"; import * as ContextHandoffService from "../ContextHandoffService.ts"; @@ -388,6 +389,7 @@ export function layerWithRegistry( const layerCheckpointRollbackServiceProvided = CheckpointRollbackService.layer.pipe( Layer.provide( Layer.mergeAll( + Layer.mock(WorkspaceEntries.WorkspaceEntries)({ refresh: () => Effect.void }), layerCheckpointServiceProvided, layerEventSinkProvided, IdAllocator.layer,