diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts index aae4a4a38b662..f1fc2808ba469 100644 --- a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts +++ b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts @@ -51,6 +51,7 @@ function layerFor(input: { readonly projection: Effect.Effect; readonly diffCheckpoints?: CheckpointStore.CheckpointStore["Service"]["diffCheckpoints"]; readonly authoredPaths?: ReadonlySet; + readonly isGitRepository?: (cwd: string) => boolean; }) { return CheckpointDiffQuery.layer.pipe( Layer.provide( @@ -61,6 +62,7 @@ function layerFor(input: { Layer.mock(CheckpointStore.CheckpointStore)({ diffCheckpoints: input.diffCheckpoints ?? (() => Effect.succeed("diff")), listAuthoredPaths: () => Effect.succeed(input.authoredPaths ?? null), + isGitRepository: (cwd) => Effect.succeed(input.isGitRepository?.(cwd) ?? true), }), ), ), @@ -106,6 +108,77 @@ it.effect("excludes ready checkpoints from rolled-back runs", () => { }).pipe(Effect.provide(layer)); }); +it.effect("diffs from the thread start after the thread moved to another worktree", () => { + const projection = makeProjection(); + const diffCheckpoints = vi.fn((_input: CheckpointStore.DiffCheckpointsInput) => + Effect.succeed(""), + ); + const layer = layerFor({ + projection: Effect.succeed({ + ...projection, + // The worktree the thread moved to sorts first and has no thread start. + checkpointScopes: projection.checkpointScopes.toReversed(), + checkpoints: [ + { + scopeId: firstScopeId, + runId: null, + ordinalWithinScope: 0, + appRunOrdinal: null, + status: "ready", + ref: CheckpointRef.make("refs/t3/test/thread-start"), + }, + ...projection.checkpoints, + ], + }), + diffCheckpoints, + }); + return Effect.gen(function* () { + const query = yield* CheckpointDiffQuery.CheckpointDiffQuery; + yield* query.getFullThreadDiff({ threadId, toTurnCount: 2 }); + assert.strictEqual( + diffCheckpoints.mock.calls[0]?.[0].fromCheckpointRef, + yield* checkpointRefForScopeOrdinal({ scopeId: firstScopeId, ordinalWithinScope: 0 }), + ); + }).pipe(Effect.provide(layer)); +}); + +it.effect("reads a removed worktree's turn through the thread's latest worktree", () => { + const projection = makeProjection(); + const diffCheckpoints = vi.fn((_input: CheckpointStore.DiffCheckpointsInput) => + Effect.succeed(""), + ); + const layer = layerFor({ + projection: Effect.succeed({ + ...projection, + checkpointScopes: [ + { id: firstScopeId, runId: firstRunId, kind: "root_run", cwd: "/worktrees/a" }, + { id: secondScopeId, runId: secondRunId, kind: "root_run", cwd: "/worktrees/b" }, + ], + checkpoints: [ + { + scopeId: firstScopeId, + runId: firstRunId, + ordinalWithinScope: 1, + appRunOrdinal: 1, + status: "ready", + ref: CheckpointRef.make("refs/t3/test/first"), + }, + ...projection.checkpoints, + ], + }), + diffCheckpoints, + isGitRepository: (cwd) => cwd !== "/worktrees/a", + }); + return Effect.gen(function* () { + const query = yield* CheckpointDiffQuery.CheckpointDiffQuery; + yield* query.getTurnDiff({ threadId, fromTurnCount: 0, toTurnCount: 1 }); + assert.deepEqual( + new Set(diffCheckpoints.mock.calls.map(([call]) => call.cwd)), + new Set(["/worktrees/b"]), + ); + }).pipe(Effect.provide(layer)); +}); + function makeProjectionAfterStoppedRun(stoppedRunCheckpoint: boolean) { const thirdRunId = RunId.make("run:checkpoint-diff-v2:3"); const checkpoint = ( diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.ts index 062405d533216..9205e121d1eea 100644 --- a/apps/server/src/checkpointing/CheckpointDiffQuery.ts +++ b/apps/server/src/checkpointing/CheckpointDiffQuery.ts @@ -175,9 +175,21 @@ export const make = Effect.gen(function* () { ) : undefined; + // A thread that moved worktrees has a root scope per worktree; only the first holds the thread start. + const rootScopes = projection.checkpointScopes.filter((scope) => scope.kind === "root_run"); + const threadStartScopeIds = new Set( + projection.checkpoints + .filter( + (checkpoint) => + checkpoint.status === "ready" && + checkpoint.ordinalWithinScope === 0 && + checkpoint.appRunOrdinal === null, + ) + .map((checkpoint) => checkpoint.scopeId), + ); const firstScope = input.fromTurnCount === 0 - ? projection.checkpointScopes.find((scope) => scope.kind === "root_run") + ? (rootScopes.find((scope) => threadStartScopeIds.has(scope.id)) ?? rootScopes[0]) : undefined; const fromCheckpointRef = input.fromTurnCount === 0 @@ -202,8 +214,25 @@ export const make = Effect.gen(function* () { }); } + // Worktrees of a repository share refs, so a removed worktree's turns read through the latest one. + const runOrdinals = new Map(projection.runs.map((run) => [run.id, run.ordinal])); + const runOrdinal = (scope: (typeof rootScopes)[number]) => + (scope.runId === null ? undefined : runOrdinals.get(scope.runId)) ?? 0; + const latestRootScope = rootScopes.reduce<(typeof rootScopes)[number] | undefined>( + (latest, scope) => + latest === undefined || runOrdinal(scope) > runOrdinal(latest) ? scope : latest, + undefined, + ); + const cwd = + latestRootScope === undefined || + latestRootScope.id === toScope.id || + (yield* checkpointStore + .isGitRepository(toScope.cwd) + .pipe(Effect.orElseSucceed(() => false))) + ? toScope.cwd + : latestRootScope.cwd; const comparison = { - cwd: toScope.cwd, + cwd, fromCheckpointRef, toCheckpointRef: toCheckpoint.ref, fallbackFromToHead: false, diff --git a/apps/server/src/orchestration-v2/CheckpointCaptureService.test.ts b/apps/server/src/orchestration-v2/CheckpointCaptureService.test.ts index 17fe7224773d5..7dd39a0983ecb 100644 --- a/apps/server/src/orchestration-v2/CheckpointCaptureService.test.ts +++ b/apps/server/src/orchestration-v2/CheckpointCaptureService.test.ts @@ -906,3 +906,61 @@ it.layer(layerProjectionStoreTest)("CheckpointCaptureServiceV2", (it) => { }), ); }); + +it.effect("records no thread start for a root scope opened mid-thread in another worktree", () => + Effect.gen(function* () { + const committed: Array = []; + const checkpoint = (ordinalWithinScope: number, status: "ready" | "missing") => + ({ + id: `checkpoint:${ordinalWithinScope}`, + scopeId, + ordinalWithinScope, + status, + files: [], + }) as never; + const layerCapture = CheckpointCaptureService.layer.pipe( + Layer.provide( + Layer.mergeAll( + IdAllocator.layer, + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getCheckpointCaptureContext: () => + Effect.succeed({ + run: { id: runId, threadId, ordinal: 2, status: "waiting", checkpointId: null }, + rootNode: { id: rootNodeId, checkpointScopeId: scopeId }, + scope: { id: scopeId }, + providerThread: { id: providerThreadId, driver }, + readyCheckpointOrdinals: [], + sideChat: false, + } as never), + }), + // Only the previous turn's baseline exists in this worktree's scope. + Layer.mock(CheckpointService.CheckpointServiceV2)({ + materializeBaselineCheckpoint: ({ ordinalWithinScope }) => + Effect.succeed( + checkpoint(ordinalWithinScope, ordinalWithinScope === 0 ? "missing" : "ready"), + ), + capture: ({ ordinalWithinScope }) => + Effect.succeed(checkpoint(ordinalWithinScope, "ready")), + }), + Layer.mock(EventSink.EventSinkV2)({ + commitCommand: ({ events }) => + Effect.sync(() => committed.push(...events)).pipe( + Effect.as({ committed: true } as never), + ), + }), + ), + ), + ); + + yield* CheckpointCaptureService.CheckpointCaptureServiceV2.use((service) => + service.execute({ threadId, runId, scopeId }), + ).pipe(Effect.provide(layerCapture)); + + assert.deepEqual( + committed.flatMap((event) => + event.type === "checkpoint.captured" ? [event.payload.ordinalWithinScope] : [], + ), + [1, 2], + ); + }), +); diff --git a/apps/server/src/orchestration-v2/CheckpointCaptureService.ts b/apps/server/src/orchestration-v2/CheckpointCaptureService.ts index 9d7d3cfd4ae2a..b40dd876d29ee 100644 --- a/apps/server/src/orchestration-v2/CheckpointCaptureService.ts +++ b/apps/server/src/orchestration-v2/CheckpointCaptureService.ts @@ -103,14 +103,20 @@ export const layer: Layer.Layer< const baselineOrdinalWithinScope = Math.max(0, run.ordinal - 1); const hasReadyCheckpoint = (ordinalWithinScope: number) => readyCheckpointOrdinals.includes(ordinalWithinScope); + // A root scope opened mid-thread in another worktree has no thread start to record, and a + // missing record there would shadow the first scope's ready one in revert lookups. const threadStartCheckpoint = baselineOrdinalWithinScope === 0 || hasReadyCheckpoint(0) ? null - : yield* checkpoints.materializeBaselineCheckpoint({ - scope, - ordinalWithinScope: 0, - forceMissing: sideChat, - }); + : yield* checkpoints + .materializeBaselineCheckpoint({ + scope, + ordinalWithinScope: 0, + forceMissing: sideChat, + }) + .pipe( + Effect.map((checkpoint) => (checkpoint.status === "ready" ? checkpoint : null)), + ); const baselineCheckpoint = hasReadyCheckpoint(baselineOrdinalWithinScope) ? null : yield* checkpoints.materializeBaselineCheckpoint({ diff --git a/apps/server/src/orchestration-v2/CheckpointRestoreSafety.test.ts b/apps/server/src/orchestration-v2/CheckpointRestoreSafety.test.ts index 145de122f387a..6941cdb32fd7f 100644 --- a/apps/server/src/orchestration-v2/CheckpointRestoreSafety.test.ts +++ b/apps/server/src/orchestration-v2/CheckpointRestoreSafety.test.ts @@ -17,9 +17,12 @@ import * as ServerConfig from "../config.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as GitWorkflowService from "../git/GitWorkflowService.ts"; import * as CheckpointRollbackService from "./CheckpointRollbackService.ts"; -import { isCheckpointRestoreIsolated } from "./CheckpointRestoreSafety.ts"; +import { + checkpointRestoreRefusal, + LEGACY_CHECKPOINT_RESTORE_MESSAGE, +} from "./CheckpointRestoreSafety.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; -import { CheckpointServiceV2 } from "./CheckpointService.ts"; +import { CheckpointServiceV2, legacyRootScopeId } from "./CheckpointService.ts"; import { EventSinkV2 } from "./EventSink.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import { ProjectionStoreV2, type ProjectionCheckpointRollbackContext } from "./ProjectionStore.ts"; @@ -34,6 +37,8 @@ it.effect.each([ "sibling", "shared-provider", "conversation", + "legacy", + "legacy-conversation", ] as const)("preserves overlapping workspace files, owner=%s", (owner) => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; @@ -53,7 +58,10 @@ it.effect.each([ const providerSessionId = ProviderSessionId.make("restore-session"); const instanceId = ProviderInstanceId.make("restore-instance"); const checkpointId = CheckpointId.make("restore-checkpoint"); - const scopeId = CheckpointScopeId.make("restore-scope"); + // A thread that never moved still keeps one legacy root scope from before per-workspace scopes. + const scopeId = owner.startsWith("legacy") + ? yield* legacyRootScopeId(yield* IdAllocator.IdAllocatorV2, threadId) + : CheckpointScopeId.make("restore-scope"); const calls: string[] = []; const providerThread = { id: providerThreadId, @@ -87,6 +95,7 @@ it.effect.each([ targetAttempt: undefined, rolledBackAttemptIds: [], staleCheckpoints: [], + staleCheckpointScopes: [], checkpoint: { id: checkpointId, scopeId, status: "ready", appRunOrdinal: null }, scope: { id: scopeId, cwd }, runsToRollback: [{ id: "later-run", ordinal: 1, status: "completed", rootNodeId: null }], @@ -165,13 +174,23 @@ it.effect.each([ const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2.pipe( Effect.provide(layerTest), ); - const restoreFiles = owner !== "conversation"; - const rejected = !["sibling", "shared-provider", "conversation"].includes(owner); + const restoreFiles = !owner.endsWith("conversation"); + const rejected = ![ + "sibling", + "shared-provider", + "conversation", + "legacy-conversation", + ].includes(owner); if (rejected) { const error = yield* service .execute({ threadId, providerThreadId, checkpointId, scopeId, restoreFiles }) .pipe(Effect.flip); - assert.equal(error.reason, "shared-workspace"); + if (owner === "legacy") { + assert.equal(error.reason, "legacy-checkpoint"); + assert.equal(error.message, LEGACY_CHECKPOINT_RESTORE_MESSAGE); + } else { + assert.equal(error.reason, "shared-workspace"); + } assert.deepEqual(calls, []); assert.equal(yield* fs.readFileString(otherFile), "other thread's uncommitted work"); } else { @@ -179,7 +198,7 @@ it.effect.each([ assert.deepEqual(calls, restoreFiles ? ["provider", "files"] : ["provider"]); assert.equal(yield* fs.exists(otherFile), !restoreFiles); } - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))), ); it.effect("lets rollback recreate a worktree storage cleanup removed", () => @@ -188,11 +207,12 @@ it.effect("lets rollback recreate a worktree storage cleanup removed", () => const path = yield* Path.Path; const parent = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-v2-restore-removed-" }); const worktreePath = path.join(parent, "worktree"); - const isolated = yield* isCheckpointRestoreIsolated( + const refusal = yield* checkpointRestoreRefusal( { id: ThreadId.make("restore-removed"), worktreePath }, - { cwd: worktreePath }, + { id: CheckpointScopeId.make("restore-removed-scope"), cwd: worktreePath }, { fileSystem, + idAllocator: yield* IdAllocator.IdAllocatorV2, path, projections: { getShellSnapshot: () => Effect.succeed({ threads: [], archivedThreads: [] }), @@ -200,8 +220,8 @@ it.effect("lets rollback recreate a worktree storage cleanup removed", () => projects: {} as never, }, ); - assert.isTrue(isolated); - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + assert.isNull(refusal); + }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))), ); it.effect.each([ @@ -231,11 +251,12 @@ it.effect.each([ sideChat: true, }, }; - const result = yield* isCheckpointRestoreIsolated( + const refusal = yield* checkpointRestoreRefusal( { id: parentId, worktreePath }, - { cwd: worktreePath }, + { id: CheckpointScopeId.make("restore-side-chat-scope"), cwd: worktreePath }, { fileSystem, + idAllocator: yield* IdAllocator.IdAllocatorV2, path, projections: { getShellSnapshot: () => Effect.succeed({ threads: [sideChat], archivedThreads: [] }), @@ -249,6 +270,6 @@ it.effect.each([ projects: {} as never, }, ); - assert.equal(result, isolated); - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + assert.equal(refusal === null, isolated); + }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))), ); diff --git a/apps/server/src/orchestration-v2/CheckpointRestoreSafety.ts b/apps/server/src/orchestration-v2/CheckpointRestoreSafety.ts index b7dc08018a896..200c7a2f4780c 100644 --- a/apps/server/src/orchestration-v2/CheckpointRestoreSafety.ts +++ b/apps/server/src/orchestration-v2/CheckpointRestoreSafety.ts @@ -3,25 +3,49 @@ import * as Effect from "effect/Effect"; import type * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; import type * as Path from "effect/Path"; +import type * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import { legacyRootScopeId } from "./CheckpointService.ts"; import type { ProjectionStoreV2 } from "./ProjectionStore.ts"; import type * as ProjectStore from "./ProjectStore.ts"; -export const SHARED_WORKSPACE_RESTORE_MESSAGE = +const SHARED_WORKSPACE_RESTORE_MESSAGE = "File restore requires an isolated worktree. This workspace may contain changes from another thread or a side chat. Rewind the conversation without restoring files instead, or delete side chats you no longer need."; +export const LEGACY_CHECKPOINT_RESTORE_MESSAGE = + "This turn was recorded before T3 Code tracked which worktree each turn ran in, so its files can't be restored. Rewind the conversation without restoring files instead."; + +const OTHER_WORKSPACE_RESTORE_MESSAGE = + "This turn ran in a different worktree, so its files can't be restored into this one. Rewind the conversation without restoring files instead."; + +export type CheckpointRestoreRefusal = "shared-workspace" | "legacy-checkpoint" | "other-workspace"; + +const REFUSAL_MESSAGES: Record = { + "shared-workspace": SHARED_WORKSPACE_RESTORE_MESSAGE, + "legacy-checkpoint": LEGACY_CHECKPOINT_RESTORE_MESSAGE, + "other-workspace": OTHER_WORKSPACE_RESTORE_MESSAGE, +}; + +export const checkpointRestoreRefusalMessage = (refusal: CheckpointRestoreRefusal) => + REFUSAL_MESSAGES[refusal]; + // Check isolation at admission and rollback; checkpoints restore whole checkouts, which may become shared. -export const isCheckpointRestoreIsolated = Effect.fn("orchestrationV2.isCheckpointRestoreIsolated")( +export const checkpointRestoreRefusal = Effect.fn("orchestrationV2.checkpointRestoreRefusal")( function* ( thread: Pick, - scope: Pick, + scope: Pick, dependencies: { readonly fileSystem: FileSystem.FileSystem; + readonly idAllocator: IdAllocator.IdAllocatorV2["Service"]; readonly projections: ProjectionStoreV2["Service"]; readonly projects: ProjectStore.ProjectStoreV2["Service"]; readonly path: Path.Path; }, ) { - const { fileSystem, projections, projects, path } = dependencies; + const { fileSystem, idAllocator, projections, projects, path } = dependencies; + // The legacy scope's projected cwd is only its latest; older checkpoints may hold another workspace. + if (scope.id === (yield* legacyRootScopeId(idAllocator, thread.id))) { + return "legacy-checkpoint" as const; + } const contains = (parent: string, child: string) => { const relative = path.relative(parent, child); return ( @@ -41,8 +65,7 @@ export const isCheckpointRestoreIsolated = Effect.fn("orchestrationV2.isCheckpoi ), ); const cwd = yield* resolve(scope.cwd); - const worktreeCwd = yield* resolve(worktreePath); - shared = cwd !== worktreeCwd; + if (cwd !== (yield* resolve(worktreePath))) return "other-workspace" as const; if (!shared) { const shell = yield* projections.getShellSnapshot(); const checkedPaths = new Set(); @@ -80,7 +103,7 @@ export const isCheckpointRestoreIsolated = Effect.fn("orchestrationV2.isCheckpoi ].filter((value): value is string => value !== null); if (otherThread.worktreePath === null) { const project = yield* projects.get(otherThread.projectId, { includeDeleted: true }); - if (Option.isNone(project)) return false; + if (Option.isNone(project)) return "shared-workspace" as const; paths.push(project.value.workspaceRoot); } for (const candidate of paths) { @@ -101,6 +124,6 @@ export const isCheckpointRestoreIsolated = Effect.fn("orchestrationV2.isCheckpoi } } } - return !shared; + return shared ? ("shared-workspace" as const) : null; }, ); diff --git a/apps/server/src/orchestration-v2/CheckpointRollbackService.test.ts b/apps/server/src/orchestration-v2/CheckpointRollbackService.test.ts index c61f9bc175edc..c0ade1dd4b43c 100644 --- a/apps/server/src/orchestration-v2/CheckpointRollbackService.test.ts +++ b/apps/server/src/orchestration-v2/CheckpointRollbackService.test.ts @@ -216,6 +216,7 @@ it.effect.each([ targetAttempt: undefined, rolledBackAttemptIds: [], staleCheckpoints: [], + staleCheckpointScopes: [], rootNodes: [], providerTurns: [], } as unknown as ProjectionStore.ProjectionCheckpointRollbackContext; @@ -571,10 +572,12 @@ const seedRollbackHistory = Effect.gen(function* () { Effect.sync(() => { restored.push(checkpoint.id); }), - deleteStaleRefs: ({ checkpoints }) => - Effect.sync(() => { - deletedRefs.push(...checkpoints.map((checkpoint) => checkpoint.id)); - }), + deleteStaleRefs: ({ scope, checkpoints }) => + checkpoints.every((checkpoint) => checkpoint.scopeId === scope.id) + ? Effect.sync(() => { + deletedRefs.push(...checkpoints.map((checkpoint) => checkpoint.id)); + }) + : Effect.die("stale refs deleted through another scope's workspace"), }), Layer.mock(EventSink.EventSinkV2)({ write: ({ events }) => @@ -731,3 +734,65 @@ it.layer(ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMe ); }, ); + +it.layer(ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)))( + "rollback across workspaces", + (it) => { + it.effect("discards later checkpoints taken in the thread's other workspace", () => + Effect.gen(function* () { + const history = yield* seedRollbackHistory; + const otherScopeId = CheckpointScopeId.make("rollback-history-other-workspace"); + const otherCheckpointId = CheckpointId.make("rollback-history-other-workspace-6"); + const event = { threadId: history.threadId, occurredAt: history.now }; + yield* history.store.apply({ + ...event, + id: EventId.make("rollback-history-other-workspace-scope"), + type: "checkpoint-scope.created", + payload: { + id: otherScopeId, + threadId: history.threadId, + runId: history.runId(6), + nodeId: history.nodeId(6), + parentScopeId: null, + providerThreadId: null, + kind: "root_run", + ordinalWithinParent: 0, + advancesAppRunCount: true, + cwd: import.meta.dirname, + createdAt: history.now, + }, + }); + yield* history.store.apply({ + ...event, + id: EventId.make("rollback-history-other-workspace-checkpoint"), + type: "checkpoint.captured", + payload: { + id: otherCheckpointId, + threadId: history.threadId, + scopeId: otherScopeId, + runId: history.runId(6), + nodeId: history.nodeId(6), + parentCheckpointId: null, + ordinalWithinScope: 6, + appRunOrdinal: 6, + ref: CheckpointRef.make("refs/t3/rollback-history/other-workspace/6"), + status: "ready", + files: [], + capturedAt: history.now, + }, + }); + + yield* history.rollback(2); + + assert.include(history.deletedRefs, otherCheckpointId); + const { checkpoints } = yield* history.store.getThreadRecords(history.threadId, [ + "checkpoints", + ]); + assert.equal( + checkpoints.find((checkpoint) => checkpoint.id === otherCheckpointId)?.status, + "stale", + ); + }), + ); + }, +); diff --git a/apps/server/src/orchestration-v2/CheckpointRollbackService.ts b/apps/server/src/orchestration-v2/CheckpointRollbackService.ts index 2c9a4070db9aa..dec20af06fb59 100644 --- a/apps/server/src/orchestration-v2/CheckpointRollbackService.ts +++ b/apps/server/src/orchestration-v2/CheckpointRollbackService.ts @@ -22,8 +22,8 @@ import type * as ServerSettings from "../serverSettings.ts"; import { makeIsManagedWorktreePath } from "../worktreesDirectory.ts"; import { withWorkspaceLease } from "../workspace/workspaceLease.ts"; import { - isCheckpointRestoreIsolated, - SHARED_WORKSPACE_RESTORE_MESSAGE, + checkpointRestoreRefusal, + checkpointRestoreRefusalMessage, } from "./CheckpointRestoreSafety.ts"; import { CheckpointRestoreError, CheckpointServiceV2 } from "./CheckpointService.ts"; import { EventSinkV2 } from "./EventSink.ts"; @@ -48,6 +48,8 @@ export class CheckpointRollbackExecutionError extends Schema.TaggedError 0) { - yield* checkpoints.deleteStaleRefs({ scope, checkpoints: staleCheckpoints }); + const staleByScope = Map.groupBy(staleCheckpoints, (stale) => stale.scopeId); + const currentWorkspace = + context.thread.worktreePath ?? + Option.match(yield* projects.get(context.thread.projectId), { + onNone: () => scope.cwd, + onSome: (project) => project.workspaceRoot, + }); + for (const staleScope of context.staleCheckpointScopes) { + yield* checkpoints.deleteStaleRefs({ + scope: staleScope, + checkpoints: staleByScope.get(staleScope.id) ?? [], + fallbackCwd: currentWorkspace, + }); } // Lock against full-row thread writes that could restore the old native session after rollback. diff --git a/apps/server/src/orchestration-v2/CheckpointService.test.ts b/apps/server/src/orchestration-v2/CheckpointService.test.ts index 39643ff38cf48..a91b2cb208541 100644 --- a/apps/server/src/orchestration-v2/CheckpointService.test.ts +++ b/apps/server/src/orchestration-v2/CheckpointService.test.ts @@ -1,4 +1,5 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { CheckpointScopeId, @@ -12,12 +13,61 @@ import { } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; +import * as ServerConfig from "../config.ts"; +import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; +import * as VcsProcess from "../vcs/VcsProcess.ts"; +import { checkpointRestoreRefusal } from "./CheckpointRestoreSafety.ts"; import * as CheckpointService from "./CheckpointService.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +it.effect("fails stale ref cleanup when the repository probe fails", () => { + const probeError = new VcsProcessTimeoutError({ + operation: "test.detect", + command: "git", + cwd: "/repo/a", + timeoutMs: 30000, + }); + const layerTest = CheckpointService.layer.pipe( + Layer.provide( + Layer.mergeAll( + IdAllocator.layer, + Layer.mock(CheckpointStore.CheckpointStore)({ + isGitRepository: () => Effect.fail(probeError), + }), + ), + ), + Layer.provideMerge(NodeCrypto.layer), + ); + return Effect.gen(function* () { + const checkpoints = yield* CheckpointService.CheckpointServiceV2; + const result = yield* Effect.result( + checkpoints.deleteStaleRefs({ + scope: { + id: CheckpointScopeId.make("checkpoint-scope:probe-failure"), + threadId: ThreadId.make("thread:probe-failure"), + runId: RunId.make("run:probe-failure:1"), + nodeId: NodeId.make("node:probe-failure:1"), + parentScopeId: null, + providerThreadId: null, + kind: "root_run", + ordinalWithinParent: 0, + advancesAppRunCount: true, + cwd: "/repo/a", + createdAt: DateTime.makeUnsafe("2026-07-28T00:00:00.000Z"), + }, + checkpoints: [], + fallbackCwd: "/repo/b", + }), + ); + assert.strictEqual(result._tag, "Failure"); + }).pipe(Effect.provide(layerTest)); +}); + it.effect( "skips unsupported baseline and turn checkpoints without swallowing capture failures", () => @@ -94,3 +144,157 @@ it.effect( } }), ); + +const layerVcsProcess = VcsProcess.layer.pipe( + Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), +); +const layerGitCheckpoints = CheckpointService.layer.pipe( + Layer.provideMerge(IdAllocator.layer), + Layer.provide(CheckpointStore.layer), + Layer.provide(VcsDriverRegistry.layer.pipe(Layer.provide(layerVcsProcess))), + Layer.provideMerge(layerVcsProcess), + Layer.provideMerge( + ServerConfig.ServerConfig.layerTest(process.cwd(), { prefix: "t3-checkpoint-service-" }), + ), + Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), +); + +// Turn 1 runs in worktree A, then the thread moves to worktree B (a branch with its own +// committed file) of the same repository, and turn 2 adds one file there. A legacy thread +// kept its one root scope across both, as threads did before root scopes were per workspace. +const runTurnsAcrossWorktrees = Effect.fn("runTurnsAcrossWorktrees")(function* ( + scopes: "per-workspace" | "legacy", +) { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const vcsProcess = yield* VcsProcess.VcsProcess; + const checkpoints = yield* CheckpointService.CheckpointServiceV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const git = (cwd: string, ...args: ReadonlyArray) => + vcsProcess.run({ operation: "CheckpointService.test.git", command: "git", cwd, args }); + const root = yield* fileSystem.realPath( + yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-checkpoint-worktrees-" }), + ); + const repo = path.join(root, "repo"); + const worktreeA = path.join(root, "worktree-a"); + const worktreeB = path.join(root, "worktree-b"); + yield* fileSystem.makeDirectory(repo); + yield* git(repo, "init", "--initial-branch=main"); + yield* git(repo, "config", "user.email", "test@test.com"); + yield* git(repo, "config", "user.name", "Test"); + yield* fileSystem.writeFileString(path.join(repo, "shared.txt"), "shared\n"); + yield* git(repo, "add", "."); + yield* git(repo, "commit", "-m", "initial"); + yield* git(repo, "worktree", "add", "-b", "branch-a", worktreeA); + yield* git(repo, "worktree", "add", "-b", "branch-b", worktreeB); + yield* fileSystem.writeFileString(path.join(worktreeB, "b-only.txt"), "only on branch b\n"); + yield* git(worktreeB, "add", "."); + yield* git(worktreeB, "commit", "-m", "branch b work"); + + const threadId = ThreadId.make("thread:worktree-move"); + const providerThreadId = ProviderThreadId.make("provider-thread:worktree-move"); + const createdAt = DateTime.makeUnsafe("2026-11-04T00:00:00.000Z"); + // Scopes are projected by id; a later run's scope record replaces an earlier one with that id. + const projectedScopes = new Map(); + const runTurn = Effect.fn("runTurn")(function* (ordinal: number, cwd: string, file: string) { + const runId = RunId.make(`run:worktree-move:${ordinal}`); + const nodeId = NodeId.make(`node:worktree-move:${ordinal}`); + const prepared = yield* checkpoints.prepareRootRunScope({ + threadId, + runId, + rootNodeId: nodeId, + providerThreadId, + cwd, + createdAt, + }); + const scope = + scopes === "legacy" + ? { ...prepared, id: yield* CheckpointService.legacyRootScopeId(idAllocator, threadId) } + : prepared; + projectedScopes.set(scope.id, scope); + yield* checkpoints.captureBaseline({ scope, ordinalWithinScope: ordinal - 1 }); + yield* fileSystem.writeFileString(path.join(cwd, file), `${file}\n`); + return yield* checkpoints.capture({ + scope, + runId, + nodeId, + ordinalWithinScope: ordinal, + appRunOrdinal: ordinal, + capturedAt: createdAt, + }); + }); + + const turn1 = yield* runTurn(1, worktreeA, "turn-1.txt"); + const turn2 = yield* runTurn(2, worktreeB, "turn-2.txt"); + return { git, repo, threadId, worktreeA, worktreeB, turn1, turn2, projectedScopes, runTurn }; +}); + +it.live("lists only turn 2's edit after the thread moves to another worktree of the repo", () => + Effect.gen(function* () { + const { turn2 } = yield* runTurnsAcrossWorktrees("per-workspace"); + + assert.deepEqual( + turn2.files.map((file) => file.path), + ["turn-2.txt"], + "turn 2's checkpoint lists differences between worktree A's and worktree B's trees", + ); + }).pipe(Effect.provide(layerGitCheckpoints)), +); + +it.live("a worktree recreated after a rollback lists only its next turn's edit", () => + Effect.gen(function* () { + const checkpoints = yield* CheckpointService.CheckpointServiceV2; + const { git, repo, worktreeA, worktreeB, turn2, projectedScopes, runTurn } = + yield* runTurnsAcrossWorktrees("per-workspace"); + + // Back in worktree A with B removed, a rollback to the thread start discards turn 2. + yield* git(repo, "worktree", "remove", "--force", worktreeB); + yield* checkpoints.deleteStaleRefs({ + scope: projectedScopes.get(turn2.scopeId)!, + checkpoints: [turn2], + fallbackCwd: worktreeA, + }); + yield* git(repo, "worktree", "add", worktreeB, "branch-b"); + const turn3 = yield* runTurn(3, worktreeB, "turn-3.txt"); + + assert.deepEqual( + turn3.files.map((file) => file.path), + ["turn-3.txt"], + ); + }).pipe(Effect.provide(layerGitCheckpoints)), +); + +it.live.each(["per-workspace", "legacy"] as const)( + "does not restore a checkpoint taken in another worktree into the thread's worktree, %s scopes", + (scopes) => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const checkpoints = yield* CheckpointService.CheckpointServiceV2; + const { threadId, worktreeB, turn1, projectedScopes } = + yield* runTurnsAcrossWorktrees(scopes); + const scope = projectedScopes.get(turn1.scopeId)!; + + // "Edit from here?" on turn 1 with "Revert files too", as CheckpointRollbackService runs it. + const refusal = yield* checkpointRestoreRefusal( + { id: threadId, worktreePath: worktreeB }, + scope, + { + fileSystem, + path, + idAllocator: yield* IdAllocator.IdAllocatorV2, + projections: { + getShellSnapshot: () => Effect.succeed({ threads: [], archivedThreads: [] }), + } as never, + projects: {} as never, + }, + ); + if (refusal === null) yield* checkpoints.restore({ scope, checkpoint: turn1 }); + + assert.deepEqual( + (yield* fileSystem.readDirectory(worktreeB)).filter((entry) => entry !== ".git").toSorted(), + ["b-only.txt", "shared.txt", "turn-2.txt"], + ); + assert.strictEqual(refusal, scopes === "legacy" ? "legacy-checkpoint" : "other-workspace"); + }).pipe(Effect.provide(layerGitCheckpoints)), +); diff --git a/apps/server/src/orchestration-v2/CheckpointService.ts b/apps/server/src/orchestration-v2/CheckpointService.ts index fc9d84cddfbee..1583fe0b13a2f 100644 --- a/apps/server/src/orchestration-v2/CheckpointService.ts +++ b/apps/server/src/orchestration-v2/CheckpointService.ts @@ -137,6 +137,8 @@ export interface CheckpointServiceV2Shape { readonly deleteStaleRefs: (input: { readonly scope: OrchestrationV2CheckpointScope; readonly checkpoints: ReadonlyArray; + /** Deletes through this checkout when the scope's own is gone; worktrees of a repository share refs. */ + readonly fallbackCwd: string; }) => Effect.Effect; } @@ -171,6 +173,20 @@ function checkpointIdForScopeOrdinal( }); } +// Worktrees of one repository share checkpoint refs, so a thread gets a root scope per workspace. +export const rootScopeId = ( + idAllocator: IdAllocator.IdAllocatorV2["Service"], + threadId: ThreadId, + cwd: string, +) => + idAllocator.allocate.checkpointScope({ threadId, name: `${ROOT_CHECKPOINT_SCOPE_NAME}:${cwd}` }); + +/** Before root scopes were per workspace, every workspace a thread used shared this one scope. */ +export const legacyRootScopeId = ( + idAllocator: IdAllocator.IdAllocatorV2["Service"], + threadId: ThreadId, +) => idAllocator.allocate.checkpointScope({ threadId, name: ROOT_CHECKPOINT_SCOPE_NAME }); + const makeRootRunScope = Effect.fn("CheckpointService.makeRootRunScope")(function* (input: { readonly idAllocator: IdAllocator.IdAllocatorV2["Service"]; readonly threadId: ThreadId; @@ -180,10 +196,7 @@ const makeRootRunScope = Effect.fn("CheckpointService.makeRootRunScope")(functio readonly cwd: string; readonly createdAt: DateTime.Utc; }) { - const scopeId = yield* input.idAllocator.allocate.checkpointScope({ - threadId: input.threadId, - name: ROOT_CHECKPOINT_SCOPE_NAME, - }); + const scopeId = yield* rootScopeId(input.idAllocator, input.threadId, input.cwd); return { id: scopeId, threadId: input.threadId, @@ -538,23 +551,43 @@ export const layer: Layer.Layer< ), ); - const deleteStaleRefs: CheckpointServiceV2Shape["deleteStaleRefs"] = (input) => - withWorkspaceLock( - input.scope.cwd, - checkpointStore.deleteCheckpointRefs({ - cwd: input.scope.cwd, - checkpointRefs: input.checkpoints.map((checkpoint) => checkpoint.ref), - }), - ).pipe( - Effect.mapError( - (cause) => - new CheckpointDeleteStaleRefsError({ - scopeId: input.scope.id, - checkpointIds: input.checkpoints.map((checkpoint) => checkpoint.id), - cause, - }), + const deleteStaleRefs: CheckpointServiceV2Shape["deleteStaleRefs"] = Effect.fn( + "CheckpointService.deleteStaleRefs", + )( + function* (input) { + // A recreated workspace gets the same scope id, so a surviving stale ref would become its baseline. + // A probe failure fails cleanup instead of passing for a removed workspace. + const cwd = (yield* checkpointStore.isGitRepository(input.scope.cwd)) + ? input.scope.cwd + : (yield* checkpointStore.isGitRepository(input.fallbackCwd)) + ? input.fallbackCwd + : null; + if (cwd === null) { + return yield* Effect.logWarning( + "orchestration V2 rollback could not reach stale checkpoint refs of a removed workspace", + { scopeId: input.scope.id, cwd: input.scope.cwd, fallbackCwd: input.fallbackCwd }, + ); + } + yield* withWorkspaceLock( + cwd, + checkpointStore.deleteCheckpointRefs({ + cwd, + checkpointRefs: input.checkpoints.map((checkpoint) => checkpoint.ref), + }), + ); + }, + (effect, input) => + effect.pipe( + Effect.mapError( + (cause) => + new CheckpointDeleteStaleRefsError({ + scopeId: input.scope.id, + checkpointIds: input.checkpoints.map((checkpoint) => checkpoint.id), + cause, + }), + ), ), - ); + ); return CheckpointServiceV2.of({ prepareRootRunScope: (input) => diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b4bdbd705c290..23d8463ad662a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -98,8 +98,8 @@ import * as Stream from "effect/Stream"; import * as ProjectStore from "./ProjectStore.ts"; import { - isCheckpointRestoreIsolated, - SHARED_WORKSPACE_RESTORE_MESSAGE, + checkpointRestoreRefusal, + checkpointRestoreRefusalMessage, } from "./CheckpointRestoreSafety.ts"; import { CheckpointServiceV2 } from "./CheckpointService.ts"; import { @@ -12079,7 +12079,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } if (command.restoreFiles !== false) { - const isolated = yield* isCheckpointRestoreIsolated(projection.thread, targetScope, { + const refusal = yield* checkpointRestoreRefusal(projection.thread, targetScope, { + idAllocator, projects, path, fileSystem, @@ -12094,11 +12095,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), ), ); - if (!isolated) + if (refusal !== null) return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, - cause: SHARED_WORKSPACE_RESTORE_MESSAGE, + cause: checkpointRestoreRefusalMessage(refusal), }); } diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index b5a0f2cfb5556..03486f86ab352 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -283,6 +283,7 @@ export interface ProjectionCheckpointRollbackContext { readonly runsToRollback: ReadonlyArray; readonly rootNodes: ReadonlyArray; readonly staleCheckpoints: ReadonlyArray; + readonly staleCheckpointScopes: ReadonlyArray; } export interface ProjectionProviderControlContext { @@ -5350,6 +5351,7 @@ export const layer: Layer.Layer = runsToRollback: [], rootNodes: [], staleCheckpoints: [], + staleCheckpointScopes: [], }; if ( providerThread === undefined || @@ -5387,14 +5389,22 @@ export const layer: Layer.Layer = JOIN orchestration_v2_projection_runs AS run ON run.run_id = attempt.run_id WHERE run.thread_id = ${threadId} AND run.status = 'rolled_back' AND attempt.thread_id = ${threadId}`, + // A thread that changed workspaces has later runs in other root scopes too. sql`SELECT payload_json FROM orchestration_v2_projection_checkpoints - WHERE thread_id = ${threadId} AND scope_id = ${scope.id} AND status = 'ready' + WHERE thread_id = ${threadId} AND status = 'ready' AND app_run_ordinal > ${targetOrdinal} + AND (scope_id = ${scope.id} OR scope_id IN ( + SELECT scope_id FROM orchestration_v2_projection_checkpoint_scopes + WHERE thread_id = ${threadId} AND kind = 'root_run')) ORDER BY ordinal_within_scope ASC`, ]); const targetAttempt = (yield* decodeRollbackTargetAttempts(attemptRows))[0]; const runsToRollback = yield* decodeRows(decodeRunPayload, threadId)(runRows); - const [nodeRows, turnRows] = yield* Effect.all([ + const staleCheckpoints = yield* decodeRows( + decodeCheckpointPayload, + threadId, + )(staleRows); + const [nodeRows, turnRows, staleScopeRows] = yield* Effect.all([ sql`SELECT payload_json FROM orchestration_v2_projection_nodes WHERE thread_id = ${threadId} AND ${sql.in( "node_id", @@ -5407,6 +5417,12 @@ export const layer: Layer.Layer = OR provider_turn_id = ${targetAttempt?.providerTurnId ?? null} OR run_attempt_id = ${targetAttempt?.id ?? null}) ORDER BY provider_thread_id ASC, ordinal ASC`, + sql`SELECT payload_json FROM orchestration_v2_projection_checkpoint_scopes + WHERE thread_id = ${threadId} AND ${sql.in( + "scope_id", + Array.from(new Set(staleCheckpoints.map((stale) => stale.scopeId))), + )} + ORDER BY scope_id ASC`, ]); return { ...context, @@ -5419,7 +5435,11 @@ export const layer: Layer.Layer = rolledBackAttemptIds: rolledBackAttempts.map(({ id }) => id), rootNodes: yield* decodeRows(decodeNodePayload, threadId)(nodeRows), providerTurns: yield* decodeRows(decodeProviderTurnPayload, threadId)(turnRows), - staleCheckpoints: yield* decodeRows(decodeCheckpointPayload, threadId)(staleRows), + staleCheckpoints, + staleCheckpointScopes: yield* decodeRows( + decodeCheckpointScopePayload, + threadId, + )(staleScopeRows), }; }), ) diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayFixtures.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayFixtures.ts index 1683439c96411..0d4d2d9e6db19 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayFixtures.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayFixtures.ts @@ -130,6 +130,7 @@ const runFixtureProvider = Effect.fn("runOrchestratorReplayFixture")(function* < const materialized = yield* materializeFixtureInput({ scenario: input.fixtureName, fixtureInput, + workspace, driver: input.driver.driver, modelSelection: input.driver.modelSelection, }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime); diff --git a/apps/server/src/orchestration-v2/testkit/ThreadFork.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ThreadFork.integration.test.ts index d90886835a234..a1a4bbc347090 100644 --- a/apps/server/src/orchestration-v2/testkit/ThreadFork.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ThreadFork.integration.test.ts @@ -19,6 +19,7 @@ import { ChildProcess, ChildProcessSpawner } from "effect/process"; import { ClaudeOrchestratorReplayHarness } from "../Adapters/ClaudeAdapterV2.testkit.ts"; import { CodexOrchestratorReplayHarness } from "../Adapters/CodexAdapterV2.testkit.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import { rootScopeId } from "../CheckpointService.ts"; import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts"; import { THREAD_FORK_NATIVE_PRIOR_TURN_ALPHA_PROMPT, @@ -490,10 +491,7 @@ describe("orchestration V2 thread fork", () => { ); const firstRunId = ids.derive.run({ threadId: sourceThreadId, ordinal: 1 }); const secondRunId = ids.derive.run({ threadId: sourceThreadId, ordinal: 2 }); - const checkpointScopeId = yield* ids.allocate.checkpointScope({ - threadId: sourceThreadId, - name: "root", - }); + const checkpointScopeId = yield* rootScopeId(ids, sourceThreadId, cwd); const firstCheckpointId = yield* ids.allocate.checkpoint({ checkpointScopeId, name: "1", @@ -696,10 +694,7 @@ describe("orchestration V2 thread fork", () => { }); const targetThreadId = ThreadId.make("thread-fork-native-fork-local-rollback-target"); const targetSecondRunId = ids.derive.run({ threadId: targetThreadId, ordinal: 2 }); - const targetCheckpointScopeId = yield* ids.allocate.checkpointScope({ - threadId: targetThreadId, - name: "root", - }); + const targetCheckpointScopeId = yield* rootScopeId(ids, targetThreadId, cwd); const targetFirstCheckpointId = yield* ids.allocate.checkpoint({ checkpointScopeId: targetCheckpointScopeId, name: "1", diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_revert/input.ts b/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_revert/input.ts index 14404d84b54ed..a319c63aee886 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_revert/input.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_revert/input.ts @@ -9,7 +9,7 @@ export function openCode2RevertInput(): OrchestratorFixtureInput { steps: [ { type: "message", text: OPENCODE2_REVERT_FIRST_PROMPT }, { type: "message", text: OPENCODE2_REVERT_SECOND_PROMPT }, - { type: "rollback", checkpointScopeSuffix: "root", checkpointSuffix: "1" }, + { type: "rollback", checkpointSuffix: "1" }, { type: "await_run_status", targetRunIndex: 2, status: "rolled_back" }, ], }; diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/shared.ts b/apps/server/src/orchestration-v2/testkit/fixtures/shared.ts index 1a550cd6786dc..27d8d6248e36d 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/shared.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/shared.ts @@ -29,6 +29,7 @@ import type { OrchestratorV2ScenarioStep, } from "../OrchestratorScenario.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import { rootScopeId } from "../../CheckpointService.ts"; import type { RuntimePolicyV2Override } from "../../RuntimePolicy.ts"; export const SIMPLE_PROMPT = "Respond with the following text: fixture simple ok"; @@ -262,7 +263,6 @@ export type OrchestratorFixtureInputStep = } | { readonly type: "rollback"; - readonly checkpointScopeSuffix: string; readonly checkpointSuffix: string; } | { @@ -484,6 +484,8 @@ function dispatchMessageCommand(input: { export function materializeFixtureInput(input: { readonly scenario: string; readonly fixtureInput: OrchestratorFixtureInput; + /** The workspace runs check out, which names their root checkpoint scope. */ + readonly workspace?: string; readonly driver: ProviderDriverKind; readonly modelSelection: ModelSelection; }): Effect.Effect< @@ -937,10 +939,11 @@ export function materializeFixtureInput(input: { break; case "rollback": { - const scopeId = yield* idAllocator.allocate.checkpointScope({ - threadId: ids.threadId, - name: step.checkpointScopeSuffix, - }); + const scopeId = yield* rootScopeId( + idAllocator, + ids.threadId, + input.workspace ?? process.cwd(), + ); pushDispatch({ type: "checkpoint.rollback", restoreFiles: false, diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/input.ts b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/input.ts index 47371f6be7669..ad400dc6bd325 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/input.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/input.ts @@ -12,7 +12,6 @@ export function threadRollbackInput(): OrchestratorFixtureInput { { type: "message", text: THREAD_ROLLBACK_SECOND_PROMPT }, { type: "rollback", - checkpointScopeSuffix: "root", checkpointSuffix: "1", }, { type: "message", text: THREAD_ROLLBACK_AFTER_PROMPT }, diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_restart/input.ts b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_restart/input.ts index 7452f72bfbf5b..c51045d8d6a91 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_restart/input.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_restart/input.ts @@ -14,7 +14,6 @@ export function threadRollbackAfterRestartInput(): OrchestratorFixtureInput { { type: "advance_clock", duration: "31 minutes" }, { type: "rollback", - checkpointScopeSuffix: "root", checkpointSuffix: "1", }, { type: "message", text: THREAD_ROLLBACK_AFTER_PROMPT }, diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/input.ts b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/input.ts index d31d186c56f11..c3779c0c22012 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/input.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/input.ts @@ -16,7 +16,6 @@ export function threadRollbackAfterStopInput(): OrchestratorFixtureInput { { type: "message", text: THREAD_ROLLBACK_SECOND_PROMPT }, { type: "rollback", - checkpointScopeSuffix: "root", checkpointSuffix: "1", }, { type: "message", text: THREAD_ROLLBACK_AFTER_PROMPT }, diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_to_stopped_turn/input.ts b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_to_stopped_turn/input.ts index 2c304197ec0d6..b5fa63c6dc8b5 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_to_stopped_turn/input.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_to_stopped_turn/input.ts @@ -15,7 +15,6 @@ export function threadRollbackToStoppedTurnInput(): OrchestratorFixtureInput { { type: "message", text: THREAD_ROLLBACK_SECOND_PROMPT }, { type: "rollback", - checkpointScopeSuffix: "root", checkpointSuffix: "2", }, { type: "message", text: THREAD_ROLLBACK_AFTER_PROMPT },