Skip to content
Open
73 changes: 73 additions & 0 deletions apps/server/src/checkpointing/CheckpointDiffQuery.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ function layerFor(input: {
readonly projection: Effect.Effect<ProjectionCheckpointContext, OrchestratorProjectionError>;
readonly diffCheckpoints?: CheckpointStore.CheckpointStore["Service"]["diffCheckpoints"];
readonly authoredPaths?: ReadonlySet<string>;
readonly isGitRepository?: (cwd: string) => boolean;
}) {
return CheckpointDiffQuery.layer.pipe(
Layer.provide(
Expand All @@ -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),
}),
),
),
Expand Down Expand Up @@ -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 = (
Expand Down
33 changes: 31 additions & 2 deletions apps/server/src/checkpointing/CheckpointDiffQuery.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand Down
58 changes: 58 additions & 0 deletions apps/server/src/orchestration-v2/CheckpointCaptureService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<OrchestrationV2DomainEvent> = [];
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],
);
}),
);
16 changes: 11 additions & 5 deletions apps/server/src/orchestration-v2/CheckpointCaptureService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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({
Expand Down
51 changes: 36 additions & 15 deletions apps/server/src/orchestration-v2/CheckpointRestoreSafety.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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;
Expand All @@ -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,
Expand Down Expand Up @@ -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 }],
Expand Down Expand Up @@ -165,21 +174,31 @@ 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 {
yield* service.execute({ threadId, providerThreadId, checkpointId, scopeId, restoreFiles });
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", () =>
Expand All @@ -188,20 +207,21 @@ 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: [] }),
} as never,
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([
Expand Down Expand Up @@ -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: [] }),
Expand All @@ -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))),
);
Loading
Loading