Skip to content
Merged
Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
3694acf
feat(storage): clean up settled thread worktrees
Bil0000 Oct 3, 2026
5b28630
fix(server): guard provider startup against worktree cleanup
Bil0000 Oct 4, 2026
fa441e3
fix(server): release startup leases before provider work
Bil0000 Oct 4, 2026
254eca7
fix(server): restore cleaned worktrees before checkpoint rollback
Bil0000 Oct 4, 2026
ae7a9e7
fix(server): repair worktree cleanup test types on main
Bil0000 Oct 4, 2026
77fc976
fix(server): use physical worktree paths for cleanup leases
Bil0000 Oct 4, 2026
5be57b7
fix(server): refresh workspace state after acquiring its lease
Bil0000 Oct 4, 2026
335659c
fix(server): warn when cleanup cannot resolve a session workspace
Bil0000 Oct 4, 2026
19b891e
perf(server): select workspace leases with thread-only reads
Bil0000 Oct 4, 2026
dfd163f
test(server): preserve worktrees when thread paths cannot resolve
Bil0000 Oct 4, 2026
a0c0406
fix(storage): update settled cleanup for current main
Bil0000 Oct 14, 2026
faaaab3
fix(web): allow manual settled-worktree cleanup
Bil0000 Oct 14, 2026
c81c14e
refactor(storage): simplify settled worktree cleanup
maria-rcks Oct 15, 2026
0e75b7e
fix(storage): let rollback reach removed worktrees and leave settle s…
maria-rcks Oct 15, 2026
dfc7587
fix(storage): give settle scripts a head start and stop detached sess…
maria-rcks Oct 15, 2026
9b1a598
fix(storage): time the settle grace from the settle and sweep again a…
maria-rcks Oct 15, 2026
27ef12b
fix(storage): hold every cleanup rule for the settle script and end s…
maria-rcks Oct 15, 2026
7798267
fix(storage): leave settled cleanup to regular sweeps and trust live …
maria-rcks Oct 15, 2026
6dd347f
fix(storage): check session liveness without renewing its idle timer
maria-rcks Oct 15, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions apps/server/src/environment/ServerEnvironment.ts
Original file line number Diff line number Diff line change
Expand Up @@ -229,6 +229,7 @@ export const make = Effect.gen(function* () {
threadAutoSettlement: true,
storageCleanup: true,
projectWorktreeCleanup: true,
settledWorktreeCleanup: true,
threadRestartContinuation: true,
projectSettingsOverrides: true,
threadSnooze: true,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Path from "effect/Path";
import { symlinksSupported } from "@t3tools/shared/testing/symlinks";
import { GitWorkflowService } from "../git/GitWorkflowService.ts";
import {
CheckpointRollbackServiceV2,
layer as rollbackLayer,
Expand Down Expand Up @@ -106,6 +107,7 @@ it.effect.each([
Layer.mergeAll(
NodeServices.layer,
idAllocatorLayer,
Layer.mock(GitWorkflowService)({}),
Layer.mock(ProjectStore.ProjectStoreV2)({
get: () => Effect.succeed(Option.some({ workspaceRoot: parent } as never)),
}),
Expand Down
424 changes: 353 additions & 71 deletions apps/server/src/orchestration-v2/CheckpointRollbackService.test.ts

Large diffs are not rendered by default.

145 changes: 100 additions & 45 deletions apps/server/src/orchestration-v2/CheckpointRollbackService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,17 +10,20 @@ 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 Option from "effect/Option";
import * as Path from "effect/Path";
import * as Schema from "effect/Schema";

import * as GitWorkflowService from "../git/GitWorkflowService.ts";
import { resolveWorkspacePath, withWorkspaceLease } from "../workspace/workspaceLease.ts";
import {
isCheckpointRestoreIsolated,
SHARED_WORKSPACE_RESTORE_MESSAGE,
} from "./CheckpointRestoreSafety.ts";
import { CheckpointServiceV2 } from "./CheckpointService.ts";
import { EventSinkV2 } from "./EventSink.ts";
import { IdAllocatorV2 } from "./IdAllocator.ts";
import { ProjectionStoreV2 } from "./ProjectionStore.ts";
import * as ProjectionStore from "./ProjectionStore.ts";
import * as ProjectStore from "./ProjectStore.ts";
import type { ProviderAdapterV2RollbackTarget } from "./ProviderAdapter.ts";
import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts";
Expand Down Expand Up @@ -82,9 +85,10 @@ export const layer: Layer.Layer<
CheckpointRollbackServiceV2,
never,
| CheckpointServiceV2
| GitWorkflowService.GitWorkflowService
| EventSinkV2
| IdAllocatorV2
| ProjectionStoreV2
| ProjectionStore.ProjectionStoreV2
| ProviderSessionManagerV2
| RuntimePolicyV2
| FileSystem.FileSystem
Expand All @@ -94,23 +98,18 @@ export const layer: Layer.Layer<
CheckpointRollbackServiceV2,
Effect.gen(function* () {
const checkpoints = yield* CheckpointServiceV2;
const gitWorkflow = yield* GitWorkflowService.GitWorkflowService;
const eventSink = yield* EventSinkV2;
const ids = yield* IdAllocatorV2;
const projections = yield* ProjectionStoreV2;
const projections = yield* ProjectionStore.ProjectionStoreV2;
const sessions = yield* ProviderSessionManagerV2;
const runtimePolicy = yield* RuntimePolicyV2;
const fileSystem = yield* FileSystem.FileSystem;
const projects = yield* ProjectStore.ProjectStoreV2;
const path = yield* Path.Path;

const execute = Effect.fn("orchestrationV2.checkpointRollback.execute")(function* (input: {
readonly threadId: ThreadId;
readonly providerThreadId: ProviderThreadId;
readonly checkpointId: CheckpointId;
readonly scopeId: CheckpointScopeId;
readonly restoreFiles?: boolean;
}) {
const projection = yield* projections.getThreadRecords(input.threadId, [
const readProjection = (threadId: ThreadId) =>
projections.getThreadRecords(threadId, [
"providerThreads",
"providerSessions",
"checkpoints",
Expand All @@ -120,6 +119,17 @@ export const layer: Layer.Layer<
"nodes",
"providerTurns",
]);

const execute = Effect.fn("orchestrationV2.checkpointRollback.execute")(function* (
input: {
readonly threadId: ThreadId;
readonly providerThreadId: ProviderThreadId;
readonly checkpointId: CheckpointId;
readonly scopeId: CheckpointScopeId;
readonly restoreFiles?: boolean;
},
projection: Effect.Success<ReturnType<typeof readProjection>>,
) {
const providerThread = projection.providerThreads.find(
(candidate) => candidate.id === input.providerThreadId,
);
Expand Down Expand Up @@ -154,6 +164,53 @@ export const layer: Layer.Layer<
});
}

const targetOrdinal = checkpoint.appRunOrdinal ?? 0;
const rollbackTarget: ProviderAdapterV2RollbackTarget =
targetOrdinal === 0
? {
type: "thread_start",
checkpointId: checkpoint.id,
appRunOrdinal: 0,
}
: yield* Effect.gen(function* () {
const targetRun = projection.runs.find((run) => run.ordinal === targetOrdinal);
const targetAttempt = projection.attempts.find(
(attempt) => attempt.id === targetRun?.activeAttemptId,
);
const targetTurn = projection.providerTurns.find(
(turn) =>
turn.id === targetAttempt?.providerTurnId ||
turn.runAttemptId === targetAttempt?.id,
);
if (targetTurn === undefined || targetTurn.providerThreadId !== providerThread.id) {
return yield* new CheckpointRollbackExecutionError({
reason: "provider-turn-unavailable",
threadId: input.threadId,
providerThreadId: input.providerThreadId,
checkpointId: input.checkpointId,
});
}
return {
type: "provider_turn" as const,
checkpointId: checkpoint.id,
appRunOrdinal: targetOrdinal,
providerTurn: targetTurn,
};
});

const { worktreePath, branch } = projection.thread;
if (worktreePath != null && branch != null && !(yield* fileSystem.exists(worktreePath))) {
const project = yield* projects.get(projection.thread.projectId);
if (Option.isSome(project)) {
yield* gitWorkflow.pruneWorktrees({ cwd: project.value.workspaceRoot });
yield* gitWorkflow.createWorktree({
cwd: project.value.workspaceRoot,
refName: branch,
path: worktreePath,
});
}
}

if (
input.restoreFiles !== false &&
!(yield* isCheckpointRestoreIsolated(projection.thread, scope, {
Expand Down Expand Up @@ -195,7 +252,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(
Expand All @@ -221,38 +277,6 @@ export const layer: Layer.Layer<
turn.providerThreadId === providerThread.id &&
(turn.runAttemptId === null || !rolledBackAttemptIds.has(turn.runAttemptId)),
);
const rollbackTarget: ProviderAdapterV2RollbackTarget =
targetOrdinal === 0
? {
type: "thread_start",
checkpointId: checkpoint.id,
appRunOrdinal: 0,
}
: yield* Effect.gen(function* () {
const targetRun = projection.runs.find((run) => run.ordinal === targetOrdinal);
const targetAttempt = projection.attempts.find(
(attempt) => attempt.id === targetRun?.activeAttemptId,
);
const targetTurn = projection.providerTurns.find(
(turn) =>
turn.id === targetAttempt?.providerTurnId ||
turn.runAttemptId === targetAttempt?.id,
);
if (targetTurn === undefined || targetTurn.providerThreadId !== providerThread.id) {
return yield* new CheckpointRollbackExecutionError({
reason: "provider-turn-unavailable",
threadId: input.threadId,
providerThreadId: input.providerThreadId,
checkpointId: input.checkpointId,
});
}
return {
type: "provider_turn" as const,
checkpointId: checkpoint.id,
appRunOrdinal: targetOrdinal,
providerTurn: targetTurn,
};
});

const snapshot =
runsToRollback.length === 0
Expand Down Expand Up @@ -344,7 +368,38 @@ export const layer: Layer.Layer<

return CheckpointRollbackServiceV2.of({
execute: (input) =>
execute(input).pipe(
Effect.gen(function* () {
const resolvePath = (worktreePath: string | null) =>
worktreePath == null
? Effect.succeed(null)
: resolveWorkspacePath(worktreePath).pipe(
Effect.provideService(FileSystem.FileSystem, fileSystem),
Effect.provideService(Path.Path, path),
);
const initial = yield* projections.getThreadRecords(input.threadId, []);
let worktreePath = yield* resolvePath(initial.thread.worktreePath);
while (true) {
const effect = Effect.gen(function* () {
const storedProjection = yield* readProjection(input.threadId);
const freshPath = yield* resolvePath(storedProjection.thread.worktreePath);
if (freshPath !== worktreePath) return freshPath;
const projection =
freshPath === null
? storedProjection
: {
...storedProjection,
thread: { ...storedProjection.thread, worktreePath: freshPath },
};
yield* execute(input, projection);
return undefined;
});
const freshPath = yield* worktreePath === null
? effect
: withWorkspaceLease(worktreePath, effect);
if (freshPath === undefined) return;
worktreePath = freshPath;
}
}).pipe(
Effect.mapError((cause) =>
isCheckpointRollbackExecutionError(cause)
? cause
Expand Down
Loading
Loading