#!/usr/bin/env python3 """Fail-closed GC for abandoned linked worktrees under .worktrees / .claude/worktrees. Sole periodic owner is system-monitor-disk-maintain (via disk-check.timer). Age alone never authorizes deletion; unknown/unreadable/Git error keeps the tree. """ from __future__ import annotations import argparse import base64 import errno import hashlib import json import os import re import shutil import stat import subprocess import sys import time from pathlib import Path from typing import Any, Callable GC_GIT_READ_TIMEOUT_S = 5.0 GC_GIT_REMOVE_TIMEOUT_S = 20.0 GC_GIT_PUSH_TIMEOUT_S = 60.0 INDEX_LOCK_KEEP_MS = 15 * 60 * 1000 IDLE_SCAN_FILE_BUDGET = 4_000 LANDQ_TICKET_RE = re.compile(r"^ticket\.([0-9a-fA-F]{16,64})\.job$") UNLEASED_WORKTREE_DIRS = (".worktrees", os.path.join(".claude", "worktrees")) UNLEASED_DEFAULT_TTL_MS = 3 * 24 * 60 * 60 * 1000 UNLEASED_MERGED_TTL_MS = 24 * 60 * 60 * 1000 UNLEASED_TRUNK_REFS = ("origin/main", "origin/master") UNLEASED_TICK_BUDGET_MS = 4_000 UNLEASED_CURSOR_FILE = "worktree-gc-cursor.json" ARCHIVE_REF_PREFIX = "refs/system-monitor/worktree-archive/" SALVAGE_REF_PREFIX = "wip/" JOURNAL_FILE = "worktree-gc.jsonl" EMBEDDED_ABSOLUTE_PATH = re.compile(r'''(?:^|[\s=:'"(,])(/[^\s"'`;|&()<>,\]}]+)''') VAULT_DEFAULT_TTL_MS = 60 * 24 * 60 * 60 * 1000 VAULT_WARN_LEAD_MS = 7 * 24 * 60 * 60 * 1000 DEFAULT_VAULT_ROOT = os.path.join(os.path.expanduser("~"), ".local", "state", "overdeck", "worktree-vault") PRECIOUS_IGNORED_GLOBS = ( re.compile(r"^\.env(\..+)?$", re.I), re.compile(r"\.(pem|key|p12|pfx|jks|keystore|sqlite|sqlite3|db)$", re.I), re.compile(r"^id_(rsa|ed25519|ecdsa|dsa)(\..+)?$", re.I), re.compile(r"^(\.npmrc|\.netrc|\.pgpass|\.htpasswd)$", re.I), re.compile(r"^(credentials|secrets?|service-account).*$", re.I), ) REPRODUCIBLE_IGNORED_DIRS = frozenset({ "node_modules", "dist", "build", "out", "target", "coverage", "tmp", "vendor", ".next", ".nuxt", ".turbo", ".cache", ".parcel-cache", ".svelte-kit", ".astro", ".venv", "venv", "__pycache__", ".pytest_cache", ".mypy_cache", ".ruff_cache", }) RunGit = Callable[[str, list[str], float | None], str | None] RunGitEnv = Callable[[str, list[str], float | None, dict[str, str] | None], str | None] Clock = Callable[[], float] def default_state_dir() -> Path: xdg = os.environ.get("XDG_STATE_HOME") or os.path.join(os.path.expanduser("~"), ".local", "state") return Path(xdg) / "system-monitor" def git_lines(cwd: str, args: list[str], timeout_s: float | None = GC_GIT_READ_TIMEOUT_S) -> str | None: try: result = subprocess.run( ["git", "-C", cwd, *args], capture_output=True, text=True, timeout=timeout_s, check=False, ) except (OSError, subprocess.TimeoutExpired): return None if result.returncode != 0: return None return result.stdout or "" def git_lines_env(cwd: str, args: list[str], timeout_s: float | None, env: dict[str, str] | None) -> str | None: full_env = os.environ.copy() if env: full_env.update(env) try: result = subprocess.run( ["git", "-C", cwd, *args], capture_output=True, text=True, timeout=timeout_s, env=full_env, check=False, ) except (OSError, subprocess.TimeoutExpired): return None if result.returncode != 0: return None return result.stdout or "" def default_vault_root() -> Path: return Path(os.environ.get("SM_WORKTREE_GC_VAULT_ROOT") or DEFAULT_VAULT_ROOT) def default_session_ledger_dir() -> Path: override = os.environ.get("SM_WORKTREE_GC_SESSION_LEDGER_DIR") if override: return Path(override) return Path(os.path.expanduser("~")) / ".local" / "state" / "agent-sessions" / "sessions" def _read_proc_start_ticks(pid: int, proc_root: str) -> int | None: try: with open(os.path.join(proc_root, str(pid), "stat"), encoding="utf-8") as handle: raw = handle.read() except OSError: return None close_paren = raw.rfind(")") if close_paren == -1: return None fields = raw[close_paren + 2:].split() # fields[0] is stat field 3 (state); field 22 (starttime) is fields[19] from there. if len(fields) < 20: return None try: return int(fields[19]) except ValueError: return None def _pid_is_verified_live(pid: Any, start_ticks: Any, proc_root: str) -> bool: if not isinstance(pid, int) or pid <= 0: return False actual_start_ticks = _read_proc_start_ticks(pid, proc_root) if actual_start_ticks is None: return False if not isinstance(start_ticks, int): return False return actual_start_ticks == start_ticks def live_session_cwds(proc_root: str = "/proc", ledger_dir: Path | str | None = None) -> set[str] | None: """Pid-verified session liveness: read the session ledger directly and confirm each row's pid is still the same process (via /proc start-ticks) before trusting its cwd — a ledger entry alone (stale, phantom, or reused pid) is never enough.""" directory = Path(ledger_dir) if ledger_dir else default_session_ledger_dir() try: names = os.listdir(directory) except FileNotFoundError: names = [] # ledger never created (no session tracking here) == no live sessions, not an error except OSError: return None cwds: set[str] = set() for name in names: if not name.endswith(".json"): continue try: with open(os.path.join(directory, name), encoding="utf-8") as handle: row = json.load(handle) except (OSError, json.JSONDecodeError): return None if not isinstance(row, dict): return None if row.get("finishedAt"): continue if not _pid_is_verified_live(row.get("pid"), row.get("pidStartTicks"), proc_root): continue cwd = row.get("cwd") if isinstance(cwd, str) and cwd: cwds.add(cwd) return cwds def worktree_has_live_session(worktree_path: str, session_cwds: set[str]) -> bool: prefix = worktree_path + os.sep for cwd in session_cwds: if cwd == worktree_path or cwd.startswith(prefix): return True return False def landq_dir_for_repo(repo_root: str, run_git: RunGit) -> str | None: common = run_git(repo_root, ["rev-parse", "--git-common-dir"], GC_GIT_READ_TIMEOUT_S) if common is None: return None common = common.strip() if not common: return None if not os.path.isabs(common): common = os.path.join(repo_root, common) return os.path.join(common, "harness", "landq") def _landq_job_wt_path(job_path: str) -> str | None: try: with open(job_path, encoding="utf-8") as handle: lines = handle.readlines() except OSError: return None for line in lines: parts = line.rstrip("\n").split(" ", 1) if len(parts) != 2 or parts[0] != "wt": continue try: return base64.b64decode(parts[1]).decode("utf-8") except (ValueError, UnicodeDecodeError): return None return None def landq_live_worktree_paths(repo_root: str, run_git: RunGit) -> set[str] | None: """A queued or conducting land-queue ticket owns its worktree's lifecycle — the GC must never race it. A ticket is live until its terminal `.verdict` file lands.""" directory = landq_dir_for_repo(repo_root, run_git) if directory is None: return None if not os.path.isdir(directory): return set() try: names = os.listdir(directory) except OSError: return None paths: set[str] = set() for name in names: match = LANDQ_TICKET_RE.match(name) if not match: continue ticket_id = match.group(1) if os.path.exists(os.path.join(directory, f"ticket.{ticket_id}.verdict")): continue # terminal disposition already recorded — ticket no longer owns the worktree wt_path = _landq_job_wt_path(os.path.join(directory, name)) if wt_path is None: return None # can't verify this ticket's target — fail closed for the whole repo paths.add(wt_path) return paths def worktree_has_landq_ticket(worktree_path: str, landq_paths: set[str]) -> bool: return worktree_path in landq_paths def is_precious_basename(name: str) -> bool: return any(pattern.search(name) for pattern in PRECIOUS_IGNORED_GLOBS) def ignored_dir_holds_precious(dir_path: str, budget: dict[str, int] | None = None) -> bool: limits = budget or {"files": 2000, "depth": 4} stack = [(dir_path, 0)] seen = 0 while stack: directory, depth = stack.pop() if depth > limits["depth"]: return True try: entries = list(os.scandir(directory)) except OSError: return True for entry in entries: seen += 1 if seen > limits["files"]: return True try: is_dir = entry.is_dir(follow_symlinks=False) except OSError: return True if is_dir: stack.append((entry.path, depth + 1)) elif is_precious_basename(entry.name): return True return False def precious_ignored_path(worktree_path: str, run_git: RunGit) -> str | None: listing = run_git(worktree_path, [ "ls-files", "--others", "--ignored", "--exclude-standard", "--directory", "--no-empty-directory", ], GC_GIT_READ_TIMEOUT_S) if listing is None: return None for raw in listing.splitlines(): rel = raw.strip() if not rel: continue base = os.path.basename(rel.rstrip("/")) if rel.endswith("/"): if base not in REPRODUCIBLE_IGNORED_DIRS: return rel if ignored_dir_holds_precious(os.path.join(worktree_path, rel)): return rel else: return rel return "" def resolve_path(target: str) -> str | None: try: return os.path.realpath(target) except OSError: return None def discover_repos(project_roots: list[str]) -> list[str]: repos: list[str] = [] for root in project_roots: try: entries = list(os.scandir(root)) except OSError: continue for entry in entries: if not (entry.is_dir(follow_symlinks=True) or entry.is_symlink()): continue repo_root = resolve_path(entry.path) if repo_root is None or repo_root in repos: continue if os.path.exists(os.path.join(repo_root, ".git")): repos.append(repo_root) return repos def is_unleased_candidate_path(repo_root: str, worktree_path: str) -> bool: return any( worktree_path.startswith(os.path.join(repo_root, container) + os.sep) for container in UNLEASED_WORKTREE_DIRS ) def _process_disappeared(pid_dir: str, exc: OSError, *, pid_stat_failed: bool = False) -> bool: if exc.errno == errno.ESRCH: return True if exc.errno != errno.ENOENT: return False if pid_stat_failed: return True try: os.stat(pid_dir) except OSError as probe_error: return probe_error.errno in {errno.ENOENT, errno.ESRCH} return False def collect_live_dirs(proc_root: str) -> set[str] | None: dirs: set[str] = set() try: pids = [name for name in os.listdir(proc_root) if name.isdigit()] except OSError: return None for pid in pids: pid_dir = os.path.join(proc_root, pid) try: if os.stat(pid_dir).st_uid != os.getuid(): continue except OSError as exc: if _process_disappeared(pid_dir, exc, pid_stat_failed=True): continue return None cwd_link = os.path.join(pid_dir, "cwd") try: cwd = os.readlink(cwd_link) dirs.add(cwd) resolved = resolve_path(cwd) if resolved is not None: dirs.add(resolved) except OSError as exc: if _process_disappeared(pid_dir, exc) or exc.errno in {errno.ENOENT, errno.EACCES, errno.EPERM}: continue return None try: with open(os.path.join(pid_dir, "cmdline"), "rb") as handle: cmdline = handle.read().decode("utf-8", errors="replace") for token in cmdline.split("\0"): for match in EMBEDDED_ABSOLUTE_PATH.finditer(token): path = match.group(1) dirs.add(path) resolved = resolve_path(path) if resolved is not None: dirs.add(resolved) except OSError as exc: if _process_disappeared(pid_dir, exc): continue return None return dirs def worktree_is_occupied(worktree_path: str, live_dirs: set[str]) -> bool: prefix = worktree_path + "/" for directory in live_dirs: if directory == worktree_path or directory.startswith(prefix): return True return False def _resolve_gitdir(worktree_path: str) -> str | None: marker = os.path.join(worktree_path, ".git") try: st = os.lstat(marker) except OSError: return None if stat.S_ISDIR(st.st_mode): return marker try: with open(marker, encoding="utf-8") as handle: raw = handle.read() except OSError: return None gitdir = re.sub(r"^gitdir:\s*", "", raw).strip() if not gitdir: return None if not os.path.isabs(gitdir): gitdir = os.path.join(worktree_path, gitdir) return gitdir def worktree_idle_ms(worktree_path: str, clock: Clock, file_budget: int = IDLE_SCAN_FILE_BUDGET) -> float: """Newest of: any working-tree file mtime (recursive, budget-capped — a top-level-only scan misses edits nested under subdirectories), git index mtime, HEAD mtime, and the per-worktree HEAD reflog mtime. Directory mtime alone is never sufficient.""" newest = 0.0 def bump(path: str) -> None: nonlocal newest try: newest = max(newest, os.stat(path).st_mtime * 1000.0) except OSError: pass bump(worktree_path) bump(os.path.join(worktree_path, ".git")) gitdir = _resolve_gitdir(worktree_path) if gitdir: bump(os.path.join(gitdir, "index")) bump(os.path.join(gitdir, "HEAD")) bump(os.path.join(gitdir, "logs", "HEAD")) scanned = 0 try: for root, dirs, files in os.walk(worktree_path, topdown=True): dirs[:] = [d for d in dirs if d != ".git" and d not in REPRODUCIBLE_IGNORED_DIRS] for name in files: if scanned >= file_budget: dirs[:] = [] break scanned += 1 bump(os.path.join(root, name)) if scanned >= file_budget: break except OSError: pass return -1.0 if newest == 0.0 else clock() - newest def index_lock_recent(worktree_path: str, clock: Clock, threshold_ms: float = INDEX_LOCK_KEEP_MS) -> bool: gitdir = _resolve_gitdir(worktree_path) if gitdir is None: return False try: mtime_ms = os.stat(os.path.join(gitdir, "index.lock")).st_mtime * 1000.0 except OSError: return False return (clock() - mtime_ms) < threshold_ms def is_merged_into_trunk(worktree_path: str, run_git: RunGit) -> bool: for trunk in UNLEASED_TRUNK_REFS: if run_git(worktree_path, ["merge-base", "--is-ancestor", "HEAD", trunk], GC_GIT_READ_TIMEOUT_S) is not None: return True return False def _delta_paths(worktree_path: str, run_git: RunGit) -> list[str] | None: listing = run_git( worktree_path, ["status", "--porcelain=v1", "-z", "--ignored=matching"], GC_GIT_READ_TIMEOUT_S, ) if listing is None: return None paths: list[str] = [] records = listing.split("\0") i = 0 while i < len(records): raw = records[i] i += 1 if len(raw) < 4: continue status_xy = raw[:2] rel = raw[3:].rstrip("/") if status_xy[0] in "RC": i += 1 # -z renames append the source path as its own record if not rel: continue if any(part in REPRODUCIBLE_IGNORED_DIRS for part in rel.split("/")): continue paths.append(rel) return sorted(set(paths)) def _branch_label(worktree_path: str, run_git: RunGit) -> str: branch_raw = run_git(worktree_path, ["symbolic-ref", "-q", "--short", "HEAD"], GC_GIT_READ_TIMEOUT_S) label = branch_raw.strip() if branch_raw else "" return label if label else "detached" def archive_worktree(worktree_path: str, vault_dir: Path, run_git: RunGit, clock: Clock) -> dict | None: paths = _delta_paths(worktree_path, run_git) if paths is None: return None head = run_git(worktree_path, ["rev-parse", "--verify", "HEAD^{commit}"], GC_GIT_READ_TIMEOUT_S) sha = (head or "").strip() branch_label = _branch_label(worktree_path, run_git) common_dir = run_git(worktree_path, ["rev-parse", "--path-format=absolute", "--git-common-dir"], GC_GIT_READ_TIMEOUT_S) if common_dir and common_dir.strip(): repo_name = os.path.basename(os.path.dirname(common_dir.strip().rstrip("/"))) else: repo_name = os.path.basename(worktree_path) sha7 = sha[:7] if sha else "0000000" ts_struct = time.gmtime(clock() / 1000.0) ts = time.strftime("%Y%m%d-%H%M%S", ts_struct) branch_fs = re.sub(r"[^A-Za-z0-9._-]", "-", branch_label) entry_name = f"{repo_name}--{branch_fs}--{ts}--{sha7}" try: vault_dir.mkdir(parents=True, exist_ok=True) os.chmod(vault_dir, 0o700) except OSError: return None tmp_entry = vault_dir / f".tmp-{entry_name}-{os.getpid()}" try: tmp_entry.mkdir(mode=0o700, exist_ok=False) os.chmod(tmp_entry, 0o700) except OSError: return None archive_sha256 = None archive_bytes = 0 archive_name = None try: if paths: archive_name = "archive.tar.zst" archive_path = tmp_entry / archive_name filelist = "\0".join(paths) + "\0" tar_proc = subprocess.Popen( ["tar", "-C", worktree_path, "-cf", "-", "--null", "--verbatim-files-from", "-T", "-"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, ) zstd_proc = subprocess.Popen( ["zstd", "-q", "-f", "-o", str(archive_path)], stdin=tar_proc.stdout, stderr=subprocess.DEVNULL, ) assert tar_proc.stdout is not None tar_proc.stdout.close() assert tar_proc.stdin is not None try: tar_proc.stdin.write(filelist.encode("utf-8", errors="surrogateescape")) finally: tar_proc.stdin.close() tar_rc = tar_proc.wait() zstd_rc = zstd_proc.wait() if tar_rc != 0 or zstd_rc != 0 or not archive_path.exists(): shutil.rmtree(tmp_entry, ignore_errors=True) return None os.chmod(archive_path, 0o600) hasher = hashlib.sha256() with open(archive_path, "rb") as handle: for chunk in iter(lambda: handle.read(1 << 20), b""): hasher.update(chunk) archive_sha256 = hasher.hexdigest() archive_bytes = archive_path.stat().st_size fd = os.open(archive_path, os.O_RDONLY) try: os.fsync(fd) finally: os.close(fd) manifest = { "repo": os.path.dirname(common_dir.strip().rstrip("/")) if common_dir and common_dir.strip() else None, "worktree": worktree_path, "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", ts_struct), "archive_file": archive_name, "archive_sha256": archive_sha256, "archive_bytes": archive_bytes, "head_sha": sha or None, "wip_ref": None, "restore_command": f"od-wip restore {entry_name}", } if branch_label == "detached": manifest["detached_sha"] = sha or None else: manifest["branch"] = branch_label manifest_path = tmp_entry / "manifest.json" manifest_path.write_text(json.dumps(manifest, sort_keys=True, indent=1), encoding="utf-8") os.chmod(manifest_path, 0o600) fd = os.open(manifest_path, os.O_RDONLY) try: os.fsync(fd) finally: os.close(fd) except OSError: shutil.rmtree(tmp_entry, ignore_errors=True) return None target = vault_dir / entry_name for attempt in range(1, 51): try: os.rename(tmp_entry, target) break except OSError: target = vault_dir / f"{entry_name}-{attempt}" else: shutil.rmtree(tmp_entry, ignore_errors=True) return None try: fd = os.open(vault_dir, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(fd) finally: os.close(fd) except OSError: return None return { "entry": target.name, "path": str(target), "sha256": archive_sha256, "bytes": archive_bytes, "branchLabel": branch_label, "headSha": sha, } def _update_manifest_wip_ref(entry_path: Path, ref: str | None) -> None: manifest_path = entry_path / "manifest.json" try: data = json.loads(manifest_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return data["wip_ref"] = ref tmp = entry_path / "manifest.json.tmp" try: tmp.write_text(json.dumps(data, sort_keys=True, indent=1), encoding="utf-8") os.chmod(tmp, 0o600) fd = os.open(tmp, os.O_RDONLY) try: os.fsync(fd) finally: os.close(fd) os.replace(tmp, manifest_path) except OSError: try: tmp.unlink() except OSError: pass def salvage_push(worktree_path: str, branch_label: str, run_git_env: RunGitEnv) -> dict | None: git_dir = run_git_env(worktree_path, ["rev-parse", "--path-format=absolute", "--git-dir"], GC_GIT_READ_TIMEOUT_S, None) if git_dir is None or not git_dir.strip(): return None tmp_index = os.path.join(git_dir.strip(), f"salvage-index-{os.getpid()}-{int(time.time() * 1000)}") env = {"GIT_INDEX_FILE": tmp_index} try: head_tree = run_git_env(worktree_path, ["rev-parse", "--verify", "HEAD^{tree}"], GC_GIT_READ_TIMEOUT_S, env) if head_tree is None: return None head_tree = head_tree.strip() if run_git_env(worktree_path, ["read-tree", head_tree], GC_GIT_READ_TIMEOUT_S, env) is None: return None if run_git_env(worktree_path, ["add", "-A"], GC_GIT_READ_TIMEOUT_S, env) is None: return None ls = run_git_env(worktree_path, ["ls-files"], GC_GIT_READ_TIMEOUT_S, env) if ls is None: return None for rel in ls.splitlines(): rel = rel.strip() if rel and is_precious_basename(os.path.basename(rel)): if run_git_env(worktree_path, ["rm", "--cached", "-q", "--", rel], GC_GIT_READ_TIMEOUT_S, env) is None: return None tree = run_git_env(worktree_path, ["write-tree"], GC_GIT_READ_TIMEOUT_S, env) if tree is None: return None tree = tree.strip() if tree == head_tree: commit_sha = run_git_env(worktree_path, ["rev-parse", "HEAD"], GC_GIT_READ_TIMEOUT_S, env) if commit_sha is None: return None commit_sha = commit_sha.strip() else: commit_env = dict(env) commit_env.update({ "GIT_AUTHOR_NAME": "worktree-gc", "GIT_AUTHOR_EMAIL": "worktree-gc@overdeck.local", "GIT_COMMITTER_NAME": "worktree-gc", "GIT_COMMITTER_EMAIL": "worktree-gc@overdeck.local", }) commit_out = run_git_env( worktree_path, ["commit-tree", tree, "-p", "HEAD", "-m", f"salvage: auto-commit idle worktree {branch_label}"], GC_GIT_READ_TIMEOUT_S, commit_env, ) if commit_out is None: return None commit_sha = commit_out.strip() sha7 = commit_sha[:7] date = time.strftime("%Y%m%d", time.gmtime()) ref = f"{SALVAGE_REF_PREFIX}{branch_label}-{date}-{sha7}" full_ref = f"refs/heads/{ref}" if run_git_env(worktree_path, ["check-ref-format", full_ref], GC_GIT_READ_TIMEOUT_S, None) is None: return None pushed = run_git_env(worktree_path, ["push", "origin", f"{commit_sha}:{full_ref}"], GC_GIT_PUSH_TIMEOUT_S, None) if pushed is None: return None remote = run_git_env(worktree_path, ["ls-remote", "origin", full_ref], GC_GIT_PUSH_TIMEOUT_S, None) if remote is None or not remote.strip(): return None remote_sha = remote.split()[0] if remote_sha != commit_sha: return None return {"pushed": True, "ref": ref, "sha": commit_sha} finally: try: os.remove(tmp_index) except OSError: pass def cleanup_local_branch(repo_root: str, branch: str, run_git: RunGit, pushed_sha: str | None = None) -> bool: if not branch or branch == "detached": return False local_sha = run_git(repo_root, ["rev-parse", "--verify", f"refs/heads/{branch}"], GC_GIT_READ_TIMEOUT_S) if local_sha is None: return False local_sha = local_sha.strip() remote = run_git(repo_root, ["ls-remote", "origin", f"refs/heads/{branch}"], GC_GIT_PUSH_TIMEOUT_S) remote_sha = remote.split()[0] if remote and remote.strip() else None contained = remote_sha == local_sha if not contained and pushed_sha: contained = run_git( repo_root, ["merge-base", "--is-ancestor", local_sha, pushed_sha], GC_GIT_READ_TIMEOUT_S, ) is not None if not contained: return False deleted = run_git(repo_root, ["branch", "-D", branch], GC_GIT_READ_TIMEOUT_S) return deleted is not None def _approx_size_kib(path: str, limit: int = 20000) -> float: total = 0 count = 0 for root, _dirs, files in os.walk(path, onerror=lambda _e: None): for name in files: count += 1 if count > limit: return total / 1024.0 try: total += os.lstat(os.path.join(root, name)).st_size except OSError: pass return total / 1024.0 def expire_vault_entries(vault_root: Path, clock: Clock, ttl_ms: float) -> list[dict]: expiring: list[dict] = [] try: canonical_root = Path(os.path.realpath(vault_root)) except OSError: return expiring try: entries = list(os.scandir(vault_root)) except OSError: return expiring now = clock() warn_from_ms = ttl_ms - VAULT_WARN_LEAD_MS for entry in entries: try: st = entry.stat(follow_symlinks=False) except OSError: continue if entry.name.startswith(".tmp-"): age_ms = now - st.st_mtime * 1000.0 if age_ms > 24 * 60 * 60 * 1000: shutil.rmtree(entry.path, ignore_errors=True) continue if not stat.S_ISDIR(st.st_mode): continue try: resolved = Path(os.path.realpath(entry.path)) except OSError: continue if resolved.parent != canonical_root: continue try: (Path(entry.path) / "manifest.json").read_text(encoding="utf-8") except OSError: continue age_ms = now - st.st_mtime * 1000.0 if age_ms >= ttl_ms: try: shutil.rmtree(entry.path) except OSError: continue elif age_ms >= warn_from_ms: expiring.append({"entry": entry.name, "ageMs": age_ms}) return expiring def read_unleased_cursor(state_dir: Path) -> str: try: parsed = json.loads((state_dir / UNLEASED_CURSOR_FILE).read_text(encoding="utf-8")) last = parsed.get("lastRepo") return last if isinstance(last, str) else "" except (OSError, json.JSONDecodeError, TypeError): return "" def write_unleased_cursor(state_dir: Path, last_repo: str) -> None: try: state_dir.mkdir(parents=True, exist_ok=True) (state_dir / UNLEASED_CURSOR_FILE).write_text( json.dumps({"lastRepo": last_repo, "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}), encoding="utf-8", ) except OSError: pass def append_journal(state_dir: Path, event: str, payload: dict[str, Any]) -> None: try: state_dir.mkdir(parents=True, exist_ok=True) record = {"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "event": event, **payload} with open(state_dir / JOURNAL_FILE, "a", encoding="utf-8") as handle: handle.write(json.dumps(record, sort_keys=True) + "\n") except OSError: pass def _parse_worktree_list(listing: str) -> tuple[str, dict[str, dict[str, Any]]]: registered: dict[str, dict[str, Any]] = {} primary_path = "" current: dict[str, Any] | None = None for line in listing.splitlines(): if line.startswith("worktree "): listed = line[len("worktree "):] current = { "path": resolve_path(listed) or listed, "locked": False, "bare": False, "detached": False, "branch": None, } if primary_path == "": primary_path = current["path"] registered[current["path"]] = current elif current is None: continue elif line.startswith("branch "): current["branch"] = line[len("branch "):] elif line == "bare": current["bare"] = True elif line == "detached": current["detached"] = True elif line == "locked" or line.startswith("locked "): current["locked"] = True return primary_path, registered def sweep_unleased_worktrees(options: dict[str, Any] | None = None) -> dict[str, Any]: opts = options or {} state_dir = Path(opts["state_dir"]) if opts.get("state_dir") else default_state_dir() run_git: RunGit = opts["run_git"] if callable(opts.get("run_git")) else git_lines run_git_env: RunGitEnv = opts["run_git_env"] if callable(opts.get("run_git_env")) else git_lines_env proc_root = opts["proc_root"] if isinstance(opts.get("proc_root"), str) else "/proc" clock: Clock = opts["clock"] if callable(opts.get("clock")) else (lambda: time.time() * 1000.0) apply = opts.get("apply") is True ttl_ms = opts["ttl_ms"] if isinstance(opts.get("ttl_ms"), (int, float)) else UNLEASED_DEFAULT_TTL_MS merged_ttl_ms = ( opts["merged_ttl_ms"] if isinstance(opts.get("merged_ttl_ms"), (int, float)) else min(UNLEASED_MERGED_TTL_MS, ttl_ms) ) budget_ms = opts["budget_ms"] if isinstance(opts.get("budget_ms"), (int, float)) else UNLEASED_TICK_BUDGET_MS project_roots = ( opts["project_roots"] if isinstance(opts.get("project_roots"), list) and opts["project_roots"] else [os.path.join(os.path.expanduser("~"), "Projects")] ) session_oracle: Callable[[], set[str] | None] = ( opts["live_session_cwds"] if callable(opts.get("live_session_cwds")) else live_session_cwds ) landq_oracle: Callable[[str], set[str] | None] = ( opts["landq_live_worktree_paths"] if callable(opts.get("landq_live_worktree_paths")) else (lambda repo_root: landq_live_worktree_paths(repo_root, run_git)) ) vault_root = Path(opts["vault_root"]) if opts.get("vault_root") else default_vault_root() vault_ttl_ms = opts["vault_ttl_ms"] if isinstance(opts.get("vault_ttl_ms"), (int, float)) else VAULT_DEFAULT_TTL_MS started_at = clock() def over_budget() -> bool: return clock() - started_at > budget_ms summary: dict[str, Any] = { "scanned": 0, "reaped": [], "kept": [], "deferred": [], "wouldSalvage": [], "salvaged": [], "freedKib": 0.0, "vaultKib": 0.0, "expiringSoon": [], "budgetHit": False, } discovered = discover_repos(project_roots) if not discovered: return summary discovered.sort() cursor = read_unleased_cursor(state_dir) resume_at = next((i for i, repo in enumerate(discovered) if repo > cursor), -1) start_index = 0 if resume_at == -1 else resume_at ordered = discovered[start_index:] + discovered[:start_index] live_dirs = collect_live_dirs(proc_root) session_cwds = session_oracle() fetch_cache: dict[str, bool] = {} landq_cache: dict[str, set[str] | None] = {} repos_to_prune: set[str] = set() last_repo = cursor def ensure_fetched(repo_root: str) -> bool: if repo_root not in fetch_cache: fetch_cache[repo_root] = run_git(repo_root, ["fetch", "--prune", "origin"], GC_GIT_PUSH_TIMEOUT_S) is not None return fetch_cache[repo_root] def landq_paths_for(repo_root: str, *, fresh: bool = False) -> set[str] | None: if fresh or repo_root not in landq_cache: landq_cache[repo_root] = landq_oracle(repo_root) return landq_cache[repo_root] for repo_root in ordered: if over_budget(): summary["budgetHit"] = True break last_repo = repo_root listing = run_git(repo_root, ["worktree", "list", "--porcelain"], GC_GIT_READ_TIMEOUT_S) if listing is None: entry = {"repoRoot": repo_root, "reason": "worktree-list-failed"} summary["kept"].append(entry) append_journal(state_dir, "worktree-gc.disposition", entry) continue primary_path, registered = _parse_worktree_list(listing) for worktree_path, meta in registered.items(): if not is_unleased_candidate_path(repo_root, worktree_path): continue if over_budget(): summary["budgetHit"] = True break summary["scanned"] += 1 def keep(reason: str, **extra: Any) -> None: entry = {"repoRoot": repo_root, "worktree": worktree_path, "reason": reason, **extra} summary["kept"].append(entry) append_journal(state_dir, "worktree-gc.disposition", entry) def defer() -> None: entry = {"repoRoot": repo_root, "worktree": worktree_path, "reason": "within-ttl"} summary["deferred"].append(entry) append_journal(state_dir, "worktree-gc.disposition", entry) def recheck_occupancy_and_session() -> str | None: nonlocal live_dirs live_dirs = collect_live_dirs(proc_root) if live_dirs is None: return "proc-unreadable" if worktree_is_occupied(worktree_path, live_dirs): return "occupied" fresh_sessions = session_oracle() if fresh_sessions is None: return "session-oracle-unreadable" if worktree_has_live_session(worktree_path, fresh_sessions): return "session-attached" fresh_landq = landq_paths_for(repo_root, fresh=True) if fresh_landq is None: return "landq-oracle-unreadable" if worktree_has_landq_ticket(worktree_path, fresh_landq): return "land-queue-ticket" if index_lock_recent(worktree_path, clock): return "checkout-in-progress" return None if worktree_path == primary_path or worktree_path == repo_root: keep("primary-worktree") continue if meta["bare"] or meta["locked"]: keep("bare" if meta["bare"] else "locked") continue if live_dirs is None: keep("proc-unreadable") continue if worktree_is_occupied(worktree_path, live_dirs): keep("occupied") continue if session_cwds is None: keep("session-oracle-unreadable") continue if worktree_has_live_session(worktree_path, session_cwds): keep("session-attached") continue landq_paths = landq_paths_for(repo_root) if landq_paths is None: keep("landq-oracle-unreadable") continue if worktree_has_landq_ticket(worktree_path, landq_paths): keep("land-queue-ticket") continue if index_lock_recent(worktree_path, clock): keep("checkout-in-progress") continue idle_ms = worktree_idle_ms(worktree_path, clock) if idle_ms < 0: keep("mtime-unreadable") continue if idle_ms < merged_ttl_ms: defer() continue status = run_git( worktree_path, ["status", "--porcelain", "--ignore-submodules=none"], GC_GIT_READ_TIMEOUT_S, ) if status is None: keep("status-failed") continue is_dirty = status.strip() != "" if is_dirty: if idle_ms < ttl_ms: defer() continue else: if idle_ms < ttl_ms and not is_merged_into_trunk(worktree_path, run_git): defer() continue precious = precious_ignored_path(worktree_path, run_git) if precious is None: keep("ignored-scan-failed") continue has_precious = precious != "" if has_precious and any(part in REPRODUCIBLE_IGNORED_DIRS for part in precious.split("/")): keep("precious-ignored", path=precious) continue head = run_git(worktree_path, ["rev-parse", "--verify", "HEAD^{commit}"], GC_GIT_READ_TIMEOUT_S) if head is None or head.strip() == "": keep("unborn-head") continue sha = head.strip() branch_label = _branch_label(worktree_path, run_git) needs_salvage = is_dirty or has_precious if not apply: if needs_salvage: entry = { "repoRoot": repo_root, "worktree": worktree_path, "dirty": is_dirty, "precious": has_precious, } summary["wouldSalvage"].append(entry) append_journal(state_dir, "worktree-gc.disposition", {**entry, "reason": "would-salvage"}) else: entry = {"repoRoot": repo_root, "worktree": worktree_path, "applied": False} summary["reaped"].append(entry) append_journal(state_dir, "worktree-gc.disposition", {**entry, "reason": "would-reap"}) continue blocked = recheck_occupancy_and_session() if blocked: keep(blocked) continue archive_result = archive_worktree(worktree_path, vault_root, run_git, clock) if archive_result is None: keep("archive-vault-failed") continue push_result = None if needs_salvage: push_result = salvage_push(worktree_path, branch_label, run_git_env) if push_result is None: keep("push-failed") continue else: if not ensure_fetched(repo_root): keep("remote-unverified") continue if not is_merged_into_trunk(worktree_path, run_git): push_result = salvage_push(worktree_path, branch_label, run_git_env) if push_result is None: keep("push-failed") continue if push_result and push_result.get("ref"): _update_manifest_wip_ref(Path(archive_result["path"]), push_result["ref"]) status_now = run_git( worktree_path, ["status", "--porcelain", "--ignore-submodules=none"], GC_GIT_READ_TIMEOUT_S, ) head_now = run_git(worktree_path, ["rev-parse", "--verify", "HEAD^{commit}"], GC_GIT_READ_TIMEOUT_S) if status_now is None or head_now is None or status_now != status or head_now.strip() != sha: keep("identity-changed") continue archived = run_git( repo_root, ["update-ref", "-m", "system-monitor pre-remove archive", f"{ARCHIVE_REF_PREFIX}{sha}", sha], GC_GIT_READ_TIMEOUT_S, ) if archived is None: keep("archive-ref-failed") continue blocked = recheck_occupancy_and_session() if blocked: keep(blocked) continue freed_kib = _approx_size_kib(worktree_path) remove_args = ["worktree", "remove", worktree_path] if needs_salvage: remove_args = ["worktree", "remove", "--force", worktree_path] removed = run_git( repo_root, remove_args, GC_GIT_REMOVE_TIMEOUT_S, ) if removed is None: keep("remove-refused") continue repos_to_prune.add(repo_root) summary["freedKib"] += freed_kib summary["vaultKib"] += (archive_result.get("bytes") or 0) / 1024.0 if branch_label != "detached": branch_deleted = cleanup_local_branch( repo_root, branch_label, run_git, (push_result or {}).get("sha") ) if branch_deleted: append_journal(state_dir, "worktree-gc.branch-cleanup", { "repoRoot": repo_root, "branch": branch_label, "sha": sha, "recover": f"git -C {repo_root} branch {branch_label} {sha}", }) entry = { "repoRoot": repo_root, "worktree": worktree_path, "applied": True, "archivedSha": sha, "vaultEntry": archive_result.get("entry"), "wipRef": (push_result or {}).get("ref"), } summary["reaped"].append(entry) if needs_salvage or (push_result and push_result.get("pushed")): summary["salvaged"].append(entry) append_journal(state_dir, "worktree-gc.disposition", {**entry, "reason": "reaped"}) for repo_root in repos_to_prune: pruned = run_git(repo_root, ["worktree", "prune", "--expire=7.days"], GC_GIT_REMOVE_TIMEOUT_S) if pruned is None: append_journal(state_dir, "worktree-gc.prune-failed", {"repoRoot": repo_root}) if apply: summary["expiringSoon"] = expire_vault_entries(vault_root, clock, vault_ttl_ms) write_unleased_cursor(state_dir, last_repo) kept_by_reason: dict[str, int] = {} for entry in summary["kept"]: reason = entry.get("reason", "unknown") kept_by_reason[reason] = kept_by_reason.get(reason, 0) + 1 append_journal(state_dir, "worktree-gc.summary", { "scanned": summary["scanned"], "reaped": len(summary["reaped"]), "salvaged": len(summary["salvaged"]), "kept": len(summary["kept"]), "keptByReason": kept_by_reason, "deferred": len(summary["deferred"]), "wouldSalvage": len(summary["wouldSalvage"]), "freedKib": summary["freedKib"], "vaultKib": summary["vaultKib"], "expiringSoon": len(summary["expiringSoon"]), "budgetHit": summary["budgetHit"], "apply": apply, }) return summary def _parse_project_roots(raw: str | None) -> list[str]: if not raw: return [os.path.join(os.path.expanduser("~"), "Projects")] return [part for part in raw.split(os.pathsep) if part] def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Fail-closed unleased worktree GC") parser.add_argument("--apply", action="store_true", help="Actually remove eligible worktrees") parser.add_argument("--state-dir", default=None) parser.add_argument("--proc-root", default="/proc") parser.add_argument("--project-roots", default=os.environ.get("SM_WORKTREE_GC_PROJECT_ROOTS")) parser.add_argument("--budget-ms", type=float, default=float(os.environ.get("SM_WORKTREE_GC_BUDGET_MS", UNLEASED_TICK_BUDGET_MS))) parser.add_argument("--ttl-ms", type=float, default=float(os.environ.get("SM_WORKTREE_GC_TTL_MS", UNLEASED_DEFAULT_TTL_MS))) parser.add_argument("--merged-ttl-ms", type=float, default=None) parser.add_argument("--vault-root", default=None) args = parser.parse_args(argv) summary = sweep_unleased_worktrees({ "apply": args.apply, "state_dir": args.state_dir or default_state_dir(), "proc_root": args.proc_root, "project_roots": _parse_project_roots(args.project_roots), "budget_ms": args.budget_ms, "ttl_ms": args.ttl_ms, "merged_ttl_ms": args.merged_ttl_ms, "vault_root": Path(args.vault_root) if args.vault_root else default_vault_root(), }) print(json.dumps({ "scanned": summary["scanned"], "reaped": len(summary["reaped"]), "salvaged": len(summary["salvaged"]), "kept": len(summary["kept"]), "deferred": len(summary["deferred"]), "wouldSalvage": [e["worktree"] for e in summary["wouldSalvage"]], "freedKib": summary["freedKib"], "vaultKib": summary["vaultKib"], "expiringSoon": [e["entry"] for e in summary["expiringSoon"]], "budgetHit": summary["budgetHit"], }, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())