clover's git

the-knife

1	import {
2	GitCommandError,
3	OrchestrationV2AppThreadJson,
4	OrchestrationV2ProviderSessionJson,
5	} from "@t3tools/contracts";
6	import * as Schema from "effect/Schema";
7	import * as SqlClient from "effect/sql/SqlClient";
8	import type {
9	OrchestrationV2ThreadShell,
10	ProjectId,
11	ServerSettings,
12	ServerSettingsError,
13	TerminalSummary,
14	WorktreeCleanupRules,
15	StorageCleanupReport,
16	StorageCleanupReportEntry,
17	} from "@t3tools/contracts";
18	import { resolveWorktreeCleanup } from "@t3tools/shared/projectSettings";
19	import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
20	import * as Context from "effect/Context";
21	import * as Config from "effect/Config";
22	import * as Deferred from "effect/Deferred";
23	import * as Layer from "effect/Layer";
24	import * as Clock from "effect/Clock";
25	import * as DateTime from "effect/DateTime";
26	import * as Effect from "effect/Effect";
27	import * as Equal from "effect/Equal";
28	import * as FileSystem from "effect/FileSystem";
29	import * as Option from "effect/Option";
30	import * as Path from "effect/Path";
31	import * as Ref from "effect/Ref";
32	import type { PlatformError } from "effect/PlatformError";
33	import * as Schedule from "effect/Schedule";
34	import * as Stream from "effect/Stream";
35	import * as SubscriptionRef from "effect/SubscriptionRef";
36
37	import * as ServerConfig from "./config.ts";
38	import * as GitManager from "./git/GitManager.ts";
39	import * as ProjectStore from "./orchestration-v2/ProjectStore.ts";
40	import * as Orchestrator from "./orchestration-v2/Orchestrator.ts";
41	import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts";
42	import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts";
43	import { threadHasQueuedTurnStart } from "./orchestration-v2/ThreadSettlementService.ts";
44	import { forkParked } from "./serverActivation.ts";
45	import * as Settings from "./serverSettings.ts";
46	import * as TerminalManager from "./terminal/Manager.ts";
47	import { isFilesystemRoot, managedWorktreesDirectories } from "./worktreesDirectory.ts";
48	import * as GitVcsDriver from "./vcs/GitVcsDriver.ts";
49	import { withWorkspaceLease } from "./workspace/workspaceLease.ts";
50	import * as HostProcess from "@t3tools/shared/HostProcess";
51	import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
52
53	const decodeCleanupThread = Schema.decodeUnknownEffect(
54	Schema.fromJsonString(OrchestrationV2AppThreadJson),
55	);
56	const decodeCleanupSession = Schema.decodeUnknownEffect(
57	Schema.fromJsonString(OrchestrationV2ProviderSessionJson),
58	);
59
60	const DAY_MS = 86_400_000;
61	// After a settle, the project's settle script needs time to open its
62	// terminal, which then protects the worktree.
63	const SETTLE_SCRIPT_GRACE_MS = 60_000;
64
65	const sentence = (text: string) => text.charAt(0).toUpperCase() + text.slice(1);
66	const REPORT_ENTRY_LIMIT = 200;
67	const isGitCommandError = Schema.is(GitCommandError);
68
69	function cleanupFailureReason(error: { readonly message: string }) {
70	return isGitCommandError(error)
71	? `${error.command}: ${error.detail.split(/\r?\n/)[0]}${error.reason ? ` (${error.reason})` : ""}`
72	: (error.message.split(/\r?\n/)[0] ?? error.message);
73	}
74
75	const worktreeCleanupEnabled = (rules: WorktreeCleanupRules) =>
76	rules.worktreeAfterDays !== null ||
77	rules.worktreeSettledAfterDays !== null ||
78	rules.worktreeOnMerge ||
79	rules.worktreeOnDelete ||
80	rules.worktreeUnchanged;
81
82	function anyWorktreePolicy(
83	settings: ServerSettings,
84	predicate: (rules: WorktreeCleanupRules) => boolean,
85	): boolean {
86	return (
87	predicate(resolveWorktreeCleanup(settings, null)) ||
88	Object.keys(settings.projectSettingsOverrides).some((projectId) =>
89	predicate(resolveWorktreeCleanup(settings, projectId as ProjectId)),
90	)
91	);
92	}
93
94	function sameProjectWorktreePolicies(left: ServerSettings, right: ServerSettings): boolean {
95	return [
96	...new Set([
97	...Object.keys(left.projectSettingsOverrides),
98	...Object.keys(right.projectSettingsOverrides),
99	]),
100	].every((projectId) =>
101	Equal.equals(
102	left.projectSettingsOverrides[projectId as ProjectId]?.worktreeCleanup,
103	right.projectSettingsOverrides[projectId as ProjectId]?.worktreeCleanup,
104	),
105	);
106	}
107
108	/** Live sessions keep their cwd even when no turn is currently running. */
109	export function storageCleanupThreadIdle(thread: OrchestrationV2ThreadShell, now: number): boolean {
110	return (
111	thread.branch !== null &&
112	thread.worktreePath !== null &&
113	thread.activeRunId === null &&
114	(thread.status === "idle" ||
115	thread.status === "completed" ||
116	thread.status === "interrupted" ||
117	thread.status === "failed" ||
118	thread.status === "cancelled" ||
119	thread.status === "rolled_back") &&
120	(thread.pendingBackgroundTasks?.length ?? 0) === 0 &&
121	thread.pendingRuntimeRequest === null &&
122	!threadHasQueuedTurnStart(thread, now)
123	);
124	}
125
126	/** PR metadata refreshes must not reset the inactivity clock. */
127	export function storageCleanupActivityAt(thread: OrchestrationV2ThreadShell): number {
128	return Math.max(
129	...[
130	thread.createdAt,
131	thread.latestUserMessageAt,
132	thread.latestRunRequestedAt,
133	thread.latestRunStartedAt,
134	thread.latestRunCompletedAt,
135	].flatMap((value) => (value == null ? [] : [DateTime.toEpochMillis(value)])),
136	);
137	}
138
139	/**
140	* Whether the host's pull request proves this worktree's head was merged. A
141	* squash or rebase merge leaves the head outside the default branch, so the
142	* merged pull request then has to name this exact commit.
143	*/
144	export function storageCleanupPullRequestMerged(
145	pullRequest: Pick<
146	GitManager.GitBranchPullRequest,
147	"state" | "headRef" | "baseRef" | "headSha"
148	> | null,
149	worktree: {
150	readonly branch: string;
151	readonly defaultBranch: string;
152	readonly headSha: string;
153	readonly integrated: boolean;
154	},
155	): boolean {
156	return (
157	pullRequest?.state === "merged" &&
158	(worktree.integrated ||
159	(pullRequest.headRef === worktree.branch &&
160	pullRequest.baseRef === worktree.defaultBranch &&
161	pullRequest.headSha === worktree.headSha))
162	);
163	}
164
165	export class StorageCleanup extends Context.Service<
166	StorageCleanup,
167	{
168	readonly runNow: Effect.Effect<StorageCleanupReport, ServerSettingsError>;
169	readonly latestReport: Effect.Effect<StorageCleanupReport | null>;
170	readonly reports: Stream.Stream<StorageCleanupReport | null>;
171	readonly drain: Effect.Effect<void>;
172	}
173	>()("t3/storageCleanup") {}
174
175	const make = Effect.gen(function* () {
176	const config = yield* ServerConfig.ServerConfig;
177	const settingsService = yield* Settings.ServerSettingsService;
178	const projectStore = yield* ProjectStore.ProjectStoreV2;
179	const engine = yield* Orchestrator.OrchestratorV2;
180	const projections = yield* ProjectionStore.ProjectionStoreV2;
181	const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
182	const sql = yield* SqlClient.SqlClient;
183	const archiveOffload = yield* Config.Boolean("T3CODE_ARCHIVE_OFFLOAD").pipe(
184	Config.withDefault(false),
185	Effect.orDie,
186	);
187	const git = yield* GitVcsDriver.GitVcsDriver;
188	const gitManager = yield* GitManager.GitManager;
189	const terminals = yield* TerminalManager.TerminalManager;
190	const fs = yield* FileSystem.FileSystem;
191	const path = yield* Path.Path;
192	const liveTerminals = new Map<string, Map<string, TerminalSummary>>();
193	const noteTerminal = (terminal: TerminalSummary) => {
194	const threadTerminals =
195	liveTerminals.get(terminal.threadId) ?? new Map<string, TerminalSummary>();
196	threadTerminals.set(terminal.terminalId, terminal);
197	liveTerminals.set(terminal.threadId, threadTerminals);
198	};
199
200	const inside = (root: string, target: string) => {
201	const relative = path.relative(root, target);
202	return (
203	relative !== "" &&
204	relative !== ".." &&
205	!relative.startsWith(`..${path.sep}`) &&
206	!path.isAbsolute(relative)
207	);
208	};
209	const hasTerminal = (worktreePath: string) =>
210	[...liveTerminals.values()]
211	.flatMap((entries) => [...entries.values()])
212	.some((terminal) => {
213	if (terminal.status !== "starting" && terminal.status !== "running") return false;
214	const cwd = path.resolve(terminal.cwd);
215	return (
216	(terminal.worktreePath !== null &&
217	path.resolve(terminal.worktreePath) === worktreePath) ||
218	cwd === worktreePath ||
219	inside(worktreePath, cwd)
220	);
221	});
222
223	const measureWorktree = (worktreePath: string) =>
224	Effect.gen(function* () {
225	const root = yield* fs.realPath(worktreePath);
226	const pending = [root];
227	let entries = 1;
228	let bytes = 0;
229	while (pending.length > 0) {
230	const target = pending.pop()!;
231	// Effect's stat follows links. Probe with readLink first to get lstat
232	// semantics, including skipping dangling links and directory links.
233	const isLink = yield* fs.readLink(target).pipe(
234	Effect.as(true),
235	Effect.catchIf(
236	(error) =>
237	error.cause instanceof Error &&
238	"code" in error.cause &&
239	error.cause.code === "EINVAL",
240	() => Effect.succeed(false),
241	),
242	);
243	if (isLink) continue;
244	if ((yield* fs.realPath(target)) !== target) return null;
245	const stat = yield* fs.stat(target);
246	if (stat.type === "File") bytes += Number(stat.size);
247	else if (stat.type === "Directory") {
248	const names = yield* fs.readDirectory(target);
249	entries += names.length;
250	if (entries > 2_000_000) return null;
251	for (const name of names) {
252	const child = path.join(target, name);
253	if (!inside(root, child)) return null;
254	pending.push(child);
255	}
256	}
257	}
258	return bytes;
259	}).pipe(
260	Effect.timeout("30 seconds"),
261	Effect.orElseSucceed(() => null),
262	);
263
264	// Active threads only: archived threads are not candidates and do not count
265	// as sharing a worktree. The recheck's getThreadWorktreePaths reads the same set.
266	// The snapshot takes seconds on large databases, so it runs on the reader
267	// worker instead of the event loop.
268	const readThreads = Effect.fn("StorageCleanup.readThreads")(function* () {
269	const { threads } = yield* projections
270	.getShellSnapshot({ location: "active" })
271	.pipe(NodeSqliteClient.readOnly);
272	const projects = yield* projectStore.listShells();
273	return { projects, threads };
274	});
275
276	// Local threads under another project need not have a worktreePath of their own.
277	const containsProjectRoot = Effect.fn("StorageCleanup.containsProjectRoot")(function* (
278	worktreePath: string,
279	projects: ReadonlyArray<{ readonly workspaceRoot: string }>,
280	) {
281	for (const project of projects) {
282	const projectPath = path.resolve(project.workspaceRoot);
283	if (projectPath === worktreePath || inside(worktreePath, projectPath)) return true;
284	const realPath = yield* fs
285	.realPath(projectPath)
286	.pipe(Effect.orElseSucceed(() => projectPath));
287	if (realPath === worktreePath || inside(worktreePath, realPath)) return true;
288	}
289	return false;
290	});
291
292	// Whether cwd stopped being a checkout of this branch. Two small Git reads,
293	// since every worktree in a sweep is asked; statusDetailsLocal also diffs and
294	// counts.
295	const branchChanged = Effect.fn("StorageCleanup.branchChanged")(function* (
296	cwd: string,
297	branch: string | null,
298	) {
299	const failure = (detail: string) =>
300	new GitCommandError({
301	operation: "StorageCleanup.branchChanged",
302	command: "git",
303	cwd,
304	detail,
305	});
306	// Null when cwd is not a repository, or is gone. Git's own messages are
307	// matched, so they must not be translated.
308	const read = Effect.fnUntraced(function* (args: string[], allowedExitCode?: number) {
309	const result = yield* git
310	.execute({
311	operation: "StorageCleanup.branchChanged",
312	cwd,
313	args,
314	env: { LC_ALL: "C" },
315	allowNonZeroExit: true,
316	})
317	.pipe(
318	Effect.catch((error) =>
319	Effect.flatMap(fs.exists(cwd), (exists) =>
320	exists ? Effect.fail(error) : Effect.succeed(null),
321	),
322	),
323	);
324	if (result === null) return null;
325	if (result.exitCode === 0 || result.exitCode === allowedExitCode) return result;
326	if (result.stderr.toLowerCase().includes("not a git repository")) return null;
327	return yield* failure(
328	result.stderr.trim().split(/\r?\n/)[0]?.slice(0, 1000) || "command failed",
329	);
330	});
331	const index = yield* read(["rev-parse", "--git-path", "index"]);
332	if (index === null) return true;
333	// Another Git process is writing to this checkout.
334	if (yield* fs.exists(`${path.resolve(cwd, index.stdout.trim())}.lock`))
335	return yield* failure(
336	"Git index is locked. Status will resume when the index lock is removed.",
337	);
338	// Exits 1 for a detached HEAD. Reads the full ref name, which is never
339	// shortened to "heads/<branch>" when a tag shares the branch name.
340	const head = yield* read(["symbolic-ref", "--quiet", "HEAD"], 1);
341	if (head === null) return true;
342	const ref = head.exitCode === 0 ? head.stdout.trim() : "";
343	const name = ref.startsWith("refs/heads/") ? ref.slice("refs/heads/".length).trim() : null;
344	// Status output writes a detached HEAD as "(detached)", so a branch named
345	// with a leading "(" has always been read as one.
346	return (name?.startsWith("(") ? null : name) !== branch;
347	});
348
349	const localChangesReason = Effect.fn("StorageCleanup.localChangesReason")(function* (
350	cwd: string,
351	rules: WorktreeCleanupRules,
352	) {
353	const status = yield* git.execute({
354	operation: "StorageCleanup.localChanges",
355	cwd,
356	args: [
357	"status",
358	"--porcelain=v1",
359	"-z",
360	"--ignore-submodules=none",
361	rules.worktreeKeepWhen === "tracked-changes"
362	? "--untracked-files=no"
363	: "--untracked-files=all",
364	],
365	maxOutputBytes: 64 * 1024,
366	});
367	if (status.stdoutTruncated) return "local changes exceed the inspection limit";
368	const records = status.stdout.split("\0").filter(Boolean);
369	let count = 0;
370	for (let i = 0; i < records.length; i++) {
371	count++;
372	if (/^[RC]|^.[RC]/.test(records[i]!)) i++;
373	}
374	if (count > 0) return `has uncommitted changes (${count} ${count === 1 ? "file" : "files"})`;
375	if (rules.worktreeKeepWhen !== "any-local-files") return null;
376	const ignored = yield* git.execute({
377	operation: "StorageCleanup.ignoredFiles",
378	cwd,
379	args: ["ls-files", "--others", "--ignored", "--exclude-standard", "--directory", "-z"],
380	maxOutputBytes: 64 * 1024,
381	});
382	if (ignored.stdoutTruncated) return "ignored files exceed the inspection limit";
383	const file = ignored.stdout
384	.split("\0")
385	.find((entry) => entry !== "" && !/(^|\/)node_modules\/$/.test(entry));
386	return file === undefined ? null : `has ignored files (${file})`;
387	});
388
389	const cleanupGit = Effect.fn("StorageCleanup.cleanupGit")(function* (
390	cwd: string,
391	args: string[],
392	command: string,
393	) {
394	const result = yield* git.execute({
395	operation: "StorageCleanup.cleanupGit",
396	cwd,
397	args,
398	allowNonZeroExit: true,
399	timeoutMs: 300_000,
400	});
401	if (result.exitCode !== 0)
402	return yield* Effect.fail(
403	new GitCommandError({
404	operation: "StorageCleanup.cleanupGit",
405	command,
406	cwd,
407	detail: result.stderr.trim().split(/\r?\n/)[0]?.slice(0, 1000) || "command failed",
408	}),
409	);
410	});
411
412	const cleanWorktrees = Effect.fn("StorageCleanup.cleanWorktrees")(function* (
413	serverSettings: ServerSettings,
414	now: number,
415	entries: StorageCleanupReportEntry[],
416	) {
417	const roots: Array<string> = [];
418	for (const directory of managedWorktreesDirectories(
419	serverSettings,
420	config.worktreesDir,
421	path,
422	yield* HostProcess.HomeDirectory,
423	)) {
424	// An unmounted drive only skips its own worktrees.
425	const root = yield* fs.exists(directory).pipe(
426	Effect.flatMap((exists) => (exists ? fs.realPath(directory) : Effect.succeed(null))),
427	Effect.orElseSucceed(() => null),
428	);
429	if (root !== null && !isFilesystemRoot(root, path)) roots.push(root);
430	}
431	const deletedRows = yield* sql<{ payload_json: string; workspaceRoot: string }>`
432	SELECT t.payload_json, p.workspace_root AS "workspaceRoot"
433	FROM orchestration_v2_projection_threads t
434	JOIN projection_projects p ON p.project_id = t.project_id
435	WHERE t.deleted_at IS NOT NULL
436	`;
437	const deletedThreads = (yield* Effect.forEach(deletedRows, (row) =>
438	decodeCleanupThread(row.payload_json).pipe(
439	Effect.map((thread) => ({ ...thread, workspaceRoot: row.workspaceRoot })),
440	),
441	)).filter((thread) => thread.worktreePath !== null && thread.branch !== null);
442	const snapshot = yield* readThreads();
443	const refreshedDefaultRefs = new Map<string, Set<string>>();
444	const groups = Map.groupBy(
445	snapshot.threads.filter((thread) => thread.worktreePath !== null),
446	(thread) => path.resolve(thread.worktreePath!),
447	);
448	const candidates = [
449	...[...groups.values()].map((group) => group[0]!),
450	...deletedThreads.filter((thread) => !groups.has(path.resolve(thread.worktreePath!))),
451	];
452	for (const thread of candidates) {
453	const settings = resolveWorktreeCleanup(serverSettings, thread.projectId);
454	if (!worktreeCleanupEnabled(settings)) continue;
455	const worktreePath = path.resolve(thread.worktreePath!);
456	const deleted = "workspaceRoot" in thread;
457	const project = deleted
458	? { workspaceRoot: thread.workspaceRoot }
459	: snapshot.projects.find((entry) => entry.id === thread.projectId);
460	const entry: StorageCleanupReportEntry = {
461	kind: "worktree",
462	outcome: "kept",
463	reason: "",
464	path: worktreePath,
465	threadId: thread.id,
466	threadTitle: thread.title,
467	bytes: null,
468	files: null,
469	};
470	const keep = (reason: string) => entries.push({ ...entry, reason: sentence(reason) });
471	yield* Effect.gen(function* () {
472	if (!(yield* fs.exists(worktreePath))) return;
473	if (deleted && !settings.worktreeOnDelete) return keep("no rules apply");
474	const shared = groups.get(worktreePath)?.length ?? 0;
475	if (shared > 1) return keep(`shared by ${shared} threads`);
476	if (project === undefined) return keep("project is unavailable");
477	if (!deleted && !storageCleanupThreadIdle(thread, now))
478	return keep("thread is running or has pending work");
479	if (hasTerminal(worktreePath)) return keep("open terminal");
480	// Roots are canonical, so compare canonical paths. A symlinked parent
481	// (a linked drive) is fine; a symlinked worktree directory is not.
482	const realPath = yield* fs.realPath(worktreePath);
483	const realParent = yield* fs.realPath(path.dirname(worktreePath));
484	if (realPath !== path.join(realParent, path.basename(worktreePath)))
485	return keep("worktree is a symbolic link");
486	if (!roots.some((root) => inside(root, realPath)))
487	return keep("outside the managed worktree folder");
488	if (yield* containsProjectRoot(worktreePath, [project, ...snapshot.projects]))
489	return keep("contains a project checkout");
490	// A linked worktree has a .git file. Never remove a main checkout.
491	if ((yield* fs.stat(path.join(worktreePath, ".git"))).type !== "File")
492	return keep("not a linked worktree");
493	if (yield* branchChanged(worktreePath, thread.branch))
494	return keep("repository or branch changed");
495	const changes = yield* localChangesReason(worktreePath, settings);
496	if (changes !== null) return keep(changes);
497	// The project's settle script starts in the worktree right after a
498	// settle. Settling stamps updatedAt now, even when automatic
499	// settlement backdates settledAt to the thread's last activity.
500	if (
501	!deleted &&
502	thread.settledAt !== null &&
503	now - DateTime.toEpochMillis(thread.updatedAt) < SETTLE_SCRIPT_GRACE_MS
504	)
505	return keep("settled moments ago");
506	const head = yield* git.resolveCommit({ cwd: worktreePath, revision: "HEAD" });
507	const old =
508	!deleted &&
509	settings.worktreeAfterDays !== null &&
510	storageCleanupActivityAt(thread) < now - settings.worktreeAfterDays * DAY_MS;
511	const settledFor =
512	deleted || thread.settledAt === null
513	? null
514	: now - DateTime.toEpochMillis(thread.settledAt);
515	const settledDays = settledFor === null ? null : Math.floor(settledFor / DAY_MS);
516	const settled =
517	settledFor !== null &&
518	settings.worktreeSettledAfterDays !== null &&
519	settledFor >= settings.worktreeSettledAfterDays * DAY_MS;
520	let eligible = deleted || old || settled;
521	let removalReason = deleted
522	? "thread was deleted"
523	: old
524	? `inactive for ${Math.floor((now - storageCleanupActivityAt(thread)) / DAY_MS)} days`
525	: `settled for ${settledDays} days`;
526	if (!eligible && (settings.worktreeUnchanged || settings.worktreeOnMerge)) {
527	const repositoryCwd = path.resolve(project.workspaceRoot);
528	const remote = yield* git.resolvePrimaryRemoteName(repositoryCwd);
529	const branch = yield* git.resolveDefaultBranchName(repositoryCwd, remote);
530	if (branch === null) return keep("default branch is unavailable");
531	const defaultRef = `refs/remotes/${remote}/${branch}`;
532	const refreshed = refreshedDefaultRefs.get(repositoryCwd) ?? new Set<string>();
533	if (!refreshed.has(defaultRef)) {
534	yield* git.fetchRemoteTrackingBranch({
535	cwd: repositoryCwd,
536	remoteName: remote,
537	remoteBranch: branch,
538	});
539	refreshed.add(defaultRef);
540	refreshedDefaultRefs.set(repositoryCwd, refreshed);
541	}
542	const base = yield* git.resolveCommit({
543	cwd: worktreePath,
544	revision: defaultRef,
545	});
546	const ancestor = yield* git.execute({
547	operation: "StorageCleanup.integratedBranch",
548	cwd: worktreePath,
549	args: ["merge-base", "--is-ancestor", head.commitSha, base.commitSha],
550	allowNonZeroExit: true,
551	});
552	const integrated = ancestor.exitCode === 0;
553	eligible = integrated && settings.worktreeUnchanged;
554	if (eligible) removalReason = "no commits beyond the default branch";
555	if (!eligible && settings.worktreeOnMerge && thread.branch !== null) {
556	const pullRequest = yield* gitManager.branchPullRequest(
557	{ cwd: worktreePath, branch: thread.branch },
558	{ refresh: true },
559	);
560	eligible = storageCleanupPullRequestMerged(pullRequest, {
561	branch: thread.branch,
562	defaultBranch: branch,
563	headSha: head.commitSha,
564	integrated,
565	});
566	if (eligible) removalReason = "pull request was merged";
567	}
568	}
569	if (!eligible)
570	return keep(
571	!deleted && settings.worktreeAfterDays !== null
572	? `inactive for ${Math.floor((now - storageCleanupActivityAt(thread)) / DAY_MS)} of ${settings.worktreeAfterDays} days`
573	: settings.worktreeSettledAfterDays !== null
574	? settledFor === null
575	? "not settled"
576	: `settled for ${settledDays} of ${settings.worktreeSettledAfterDays} days`
577	: settings.worktreeOnMerge
578	? "not merged"
579	: "has commits beyond the default branch",
580	);
581	// Threads, queued deletion work and provider sessions as they are now, so a
582	// queued turn, resumed session or new thread sharing this path cancels
583	// the removal. Read only this path and this thread: a full snapshot per
584	// worktree takes seconds on large databases.
585	const blockedSinceCheck = Effect.fnUntraced(function* () {
586	if (archiveOffload) {
587	const pendingArchive =
588	yield* sql`SELECT 1 FROM orchestration_v2_projection_threads t WHERE t.archived_at IS NOT NULL AND t.deleted_at IS NULL AND json_extract(t.payload_json, '$.worktreePath') = ${worktreePath} AND NOT EXISTS (SELECT 1 FROM transcript_archives a WHERE a.thread_id = t.thread_id) LIMIT 1`;
589	if (pendingArchive.length) return "transcript archive is not verified";
590	}
591	const latestProjects = yield* projectStore.listShells();
592	if (yield* containsProjectRoot(worktreePath, [project, ...latestProjects]))
593	return "contains a project checkout";
594	const latest = (yield* projections.getThreadWorktreePaths()).filter(
595	(entry) => path.resolve(entry.worktreePath) === worktreePath,
596	);
597	if (hasTerminal(worktreePath)) return "open terminal";
598	if (deleted) {
599	if (
600	latest.length > 0 ||
601	!resolveWorktreeCleanup(yield* settingsService.getSettings, thread.projectId)
602	.worktreeOnDelete
603	)
604	return "thread or cleanup settings changed since check";
605	// V2 deletion queues durable cleanup. Do not remove its checkout until
606	// every effect has finished successfully or was explicitly cancelled.
607	const pendingCleanup = yield* sql`
608	SELECT 1 FROM orchestration_v2_effect_outbox
609	WHERE thread_id = ${thread.id} AND status NOT IN ('succeeded', 'cancelled') LIMIT 1
610	`;
611	if (pendingCleanup.length > 0) return "thread deletion is still pending";
612	} else {
613	const latestShell =
614	latest.length === 1 && latest[0]!.threadId === thread.id
615	? yield* projections.getThreadShell(thread.id)
616	: null;
617	if (
618	latestShell === null ||
619	!storageCleanupThreadIdle(latestShell, now) ||
620	storageCleanupActivityAt(latestShell) !== storageCleanupActivityAt(thread) ||
621	!Equal.equals(latestShell.settledAt, thread.settledAt)
622	)
623	return "thread activity or shared worktree changed since check";
624	}
625	// Sessions can outlive their run and can be shared across app threads.
626	const sessionRows = yield* sql<{ payload_json: string }>`
627	SELECT payload_json FROM orchestration_v2_projection_provider_sessions
628	WHERE status != 'stopped'
629	`;
630	const sessions = yield* Effect.forEach(sessionRows, (row) =>
631	decodeCleanupSession(row.payload_json),
632	);
633	// A stored session can stay ready after its runtime was released,
634	// for example once every thread detached; only a live one holds it.
635	for (const session of sessions) {
636	const cwd = path.resolve(session.cwd);
637	if (
638	(cwd === worktreePath || inside(worktreePath, cwd)) &&
639	(yield* providerSessions.isLive(session.id))
640	)
641	return "provider session is still open";
642	}
643	return null;
644	});
645	// Most cancelled removals are caught here, before the worktree is walked.
646	const blocked = yield* blockedSinceCheck();
647	if (blocked !== null) return keep(blocked);
648	const bytes = yield* measureWorktree(worktreePath);
649	// The walk can take a while; ask again for anything that started during it.
650	const blockedWhileMeasuring = yield* blockedSinceCheck();
651	if (blockedWhileMeasuring !== null) return keep(blockedWhileMeasuring);
652	if (yield* branchChanged(worktreePath, thread.branch))
653	return keep("repository or branch changed since check");
654	if (
655	(yield* git.resolveCommit({ cwd: worktreePath, revision: "HEAD" })).commitSha !==
656	head.commitSha
657	)
658	return keep("commit changed since check");
659	const finalChanges = yield* localChangesReason(worktreePath, settings);
660	if (finalChanges !== null) return keep(finalChanges);
661	const current = resolveWorktreeCleanup(
662	yield* settingsService.getSettings,
663	thread.projectId,
664	);
665	if (
666	Object.keys(settings).some(
667	(key) =>
668	current[key as keyof typeof settings] !== settings[key as keyof typeof settings],
669	)
670	)
671	return keep("settings changed since check");
672	// Clean only untracked files; Git must still refuse removal if a tracked
673	// edit arrives after our last status check.
674	if (settings.worktreeKeepWhen === "tracked-changes")
675	yield* cleanupGit(worktreePath, ["clean", "-ffdx"], "git clean");
676	yield* cleanupGit(
677	project.workspaceRoot,
678	["-c", "status.showUntrackedFiles=normal", "worktree", "remove", worktreePath],
679	"git worktree remove",
680	);
681	entries.push({ ...entry, outcome: "removed", reason: sentence(removalReason), bytes });
682	yield* gitManager.invalidateStatus(project.workspaceRoot);
683	// Preserve branch and path: ProviderTurnStartService recreates the checkout
684	// from that branch when the thread is resumed.
685	yield* Effect.logInfo("storage cleanup removed worktree", { threadId: thread.id });
686	}).pipe(
687	(effect) => withWorkspaceLease(worktreePath, effect),
688	Effect.catch((error) =>
689	Effect.gen(function* () {
690	entries.push({
691	...entry,
692	outcome: "failed",
693	reason: cleanupFailureReason(error),
694	});
695	yield* Effect.logWarning("storage cleanup failed for worktree", {
696	threadId: thread.id,
697	error,
698	});
699	}),
700	),
701	);
702	}
703	});
704
705	const cleanFiles = Effect.fn("StorageCleanup.cleanFiles")(function* (
706	root: string,
707	days: number | null,
708	now: number,
709	rotatedLogs: boolean,
710	// Owned by the caller so files removed before a failure still count.
711	removed: { files: number; bytes: number },
712	) {
713	if (days === null || !(yield* fs.exists(root))) return removed;
714	const realRoot = yield* fs.realPath(root);
715	if (realRoot !== path.resolve(root)) return removed;
716	const visit = Effect.fn("StorageCleanup.visitFiles")(function* (
717	directory: string,
718	): Effect.fn.Return<void, PlatformError | ServerSettingsError> {
719	for (const name of yield* fs.readDirectory(directory)) {
720	const target = path.join(directory, name);
721	if ((yield* fs.realPath(target)) !== target || !inside(realRoot, target)) continue;
722	const stat = yield* fs.stat(target);
723	if (stat.type === "Directory" && rotatedLogs) {
724	yield* visit(target);
725	} else if (stat.type === "File" && (!rotatedLogs || /\.(?:log|ndjson)\.\d+$/.test(name))) {
726	const modified = Option.getOrNull(stat.mtime);
727	if (modified !== null && modified.getTime() < now - days * DAY_MS) {
728	const current = (yield* settingsService.getSettings).storageCleanup;
729	if ((rotatedLogs ? current.logsAfterDays : current.browserArtifactsAfterDays) !== days)
730	return;
731	yield* fs.remove(target);
732	removed.files++;
733	removed.bytes += Number(stat.size);
734	}
735	}
736	}
737	});
738	yield* visit(realRoot);
739	return removed;
740	});
741
742	const reportRef = yield* SubscriptionRef.make<StorageCleanupReport | null>(null);
743	const sweep = Effect.fn("StorageCleanup.sweep")(function* (
744	trigger: StorageCleanupReport["trigger"],
745	) {
746	const now = yield* Clock.currentTimeMillis;
747	const serverSettings = yield* settingsService.getSettings;
748	const settings = serverSettings.storageCleanup;
749	const entries: StorageCleanupReportEntry[] = [];
750	yield* cleanWorktrees(serverSettings, now, entries).pipe(
751	Effect.catch((error) =>
752	Effect.gen(function* () {
753	entries.push({
754	kind: "worktree",
755	outcome: "failed",
756	reason: cleanupFailureReason(error),
757	path: null,
758	threadId: null,
759	threadTitle: null,
760	bytes: null,
761	files: null,
762	});
763	yield* Effect.logWarning("worktree cleanup failed", { error });
764	}),
765	),
766	);
767	for (const category of [
768	{
769	kind: "browser-artifacts" as const,
770	root: config.browserArtifactsDir,
771	days: settings.browserArtifactsAfterDays,
772	label: "browser artifacts",
773	},
774	{
775	kind: "logs" as const,
776	root: config.logsDir,
777	days: settings.logsAfterDays,
778	label: "rotated logs",
779	},
780	]) {
781	if (category.days === null) continue;
782	const removed = { files: 0, bytes: 0 };
783	yield* cleanFiles(category.root, category.days, now, category.kind === "logs", removed).pipe(
784	Effect.map(({ files, bytes }) =>
785	entries.push({
786	kind: category.kind,
787	outcome: files > 0 ? "removed" : "kept",
788	reason: files === 0 ? "No expired files" : `Removed ${files} ${category.label}`,
789	path: null,
790	threadId: null,
791	threadTitle: null,
792	bytes: files > 0 ? bytes : null,
793	files: files > 0 ? files : null,
794	}),
795	),
796	Effect.catch((error) =>
797	Effect.gen(function* () {
798	entries.push({
799	kind: category.kind,
800	outcome: "failed",
801	reason: cleanupFailureReason(error),
802	path: null,
803	threadId: null,
804	threadTitle: null,
805	bytes: removed.files > 0 ? removed.bytes : null,
806	files: removed.files > 0 ? removed.files : null,
807	});
808	yield* Effect.logWarning("storage file cleanup failed", { kind: category.kind, error });
809	}),
810	),
811	);
812	}
813	const counts = { removed: 0, kept: 0, failed: 0 };
814	let bytesFreed = 0;
815	for (const entry of entries) {
816	counts[entry.outcome]++;
817	bytesFreed += entry.bytes ?? 0;
818	}
819	const priority = { failed: 0, removed: 1, kept: 2 };
820	const latestReport: StorageCleanupReport = {
821	trigger,
822	startedAt: DateTime.formatIso(DateTime.makeUnsafe(now)),
823	finishedAt: DateTime.formatIso(yield* DateTime.now),
824	entries: entries
825	.sort((a, b) => priority[a.outcome] - priority[b.outcome])
826	.slice(0, REPORT_ENTRY_LIMIT),
827	counts,
828	bytesFreed,
829	omittedCount: Math.max(0, entries.length - REPORT_ENTRY_LIMIT),
830	};
831	yield* SubscriptionRef.set(reportRef, latestReport);
832	return latestReport;
833	});
834	// At most one automatic sweep waits at a time. An automatic request made
835	// while one is waiting joins it, since that sweep has not read any state yet.
836	// Each manual run still gets its own sweep and report.
837	const automaticSweepQueued = yield* Ref.make(false);
838	const worker = yield* makeDrainableWorker(
839	(completion: Deferred.Deferred<StorageCleanupReport, ServerSettingsError> | undefined) =>
840	(completion === undefined ? Ref.set(automaticSweepQueued, false) : Effect.void).pipe(
841	Effect.andThen(sweep(completion === undefined ? "automatic" : "manual")),
842	Effect.exit,
843	Effect.flatMap((exit) =>
844	completion === undefined
845	? exit.pipe(
846	Effect.asVoid,
847	Effect.catchCause((cause) =>
848	Effect.logWarning("storage cleanup failed", { cause }),
849	),
850	)
851	: Deferred.done(completion, exit).pipe(Effect.asVoid),
852	),
853	),
854	);
855	const requestSweep = Ref.getAndSet(automaticSweepQueued, true).pipe(
856	Effect.flatMap((queued) => (queued ? Effect.void : worker.enqueue(undefined))),
857	);
858	const runNow = Effect.gen(function* () {
859	const completion = yield* Deferred.make<StorageCleanupReport, ServerSettingsError>();
860	yield* worker.enqueue(completion);
861	return yield* Deferred.await(completion);
862	});
863
864	const start = Effect.fn("StorageCleanup.start")(function* () {
865	const unsubscribe = yield* terminals.subscribeMetadata((event) =>
866	Effect.sync(() => {
867	if (event.type === "snapshot") {
868	liveTerminals.clear();
869	for (const terminal of event.terminals) noteTerminal(terminal);
870	} else if (event.type === "upsert") {
871	noteTerminal(event.terminal);
872	} else {
873	const threadTerminals = liveTerminals.get(event.threadId);
874	threadTerminals?.delete(event.terminalId);
875	if (threadTerminals?.size === 0) liveTerminals.delete(event.threadId);
876	}
877	}),
878	);
879	yield* Effect.addFinalizer(() => Effect.sync(unsubscribe));
880	const changes = yield* settingsService.subscribeChanges;
881	const events = engine.streamDomainEvents;
882	let lastSettings = yield* settingsService.getSettings.pipe(Effect.orDie);
883	yield* forkParked(
884	requestSweep.pipe(
885	Effect.andThen(worker.drain),
886	Effect.repeat(Schedule.spaced("1 hour")),
887	Effect.asVoid,
888	),
889	);
890	yield* forkParked(
891	Stream.runForEach(changes, (settings) => {
892	if (
893	Equal.equals(settings.storageCleanup, lastSettings.storageCleanup) &&
894	Equal.equals(settings.worktreeCleanup, lastSettings.worktreeCleanup) &&
895	sameProjectWorktreePolicies(settings, lastSettings)
896	)
897	return Effect.void;
898	lastSettings = settings;
899	return requestSweep;
900	}),
901	);
902	yield* forkParked(
903	// A deleted thread's worktree waits for its sessions to end. Moving
904	// between live statuses, such as running to ready, cannot free one.
905	Stream.runForEach(events, (event) =>
906	(event.type === "thread.deleted" ||
907	(event.type === "provider-session.updated" &&
908	(event.payload.status === "stopped" || event.payload.status === "error"))) &&
909	anyWorktreePolicy(lastSettings, (rules) => rules.worktreeOnDelete)
910	? requestSweep
911	: Effect.void,
912	).pipe(
913	Effect.catchCause((cause) =>
914	Effect.logWarning("Storage cleanup event stream failed", { cause }),
915	),
916	),
917	);
918	});
919	yield* start();
920	return StorageCleanup.of({
921	runNow,
922	latestReport: SubscriptionRef.get(reportRef),
923	reports: SubscriptionRef.changes(reportRef),
924	drain: worker.drain,
925	});
926	});
927
928	export const layer = Layer.effect(StorageCleanup, make);