"""Fail-closed unleased worktree GC — historical harnessd cases ported to system-monitor."""

from __future__ import annotations

import base64
import errno
import json
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path

LIB = Path(__file__).resolve().parents[1] / "lib"
sys.path.insert(0, str(LIB))
import worktree_gc  # noqa: E402
from worktree_gc import (  # noqa: E402
    ARCHIVE_REF_PREFIX,
    UNLEASED_DEFAULT_TTL_MS,
    VAULT_WARN_LEAD_MS,
    archive_worktree,
    cleanup_local_branch,
    expire_vault_entries,
    ignored_dir_holds_precious,
    live_session_cwds,
    salvage_push,
    sweep_unleased_worktrees,
    worktree_has_live_session,
)

TEST_SWEEP_BUDGET_MS = 120_000
TEST_GIT_TIMEOUT_S = 120


def git(cwd: Path | str, args: list[str]) -> str:
    result = subprocess.run(
        ["git", "-C", str(cwd), *args],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        raise RuntimeError(f"git {' '.join(args)} failed: {result.stderr}")
    return result.stdout


def read_manifest(entry_dir: Path) -> dict:
    return json.loads((entry_dir / "manifest.json").read_text(encoding="utf-8"))


def archive_names(entry_dir: Path) -> list[str]:
    archive = entry_dir / "archive.tar.zst"
    if not archive.exists():
        return []
    zstd_proc = subprocess.Popen(["zstd", "-dc", str(archive)], stdout=subprocess.PIPE)
    tar_out = subprocess.run(["tar", "-tf", "-"], stdin=zstd_proc.stdout, capture_output=True, text=True)
    zstd_proc.wait()
    return [line for line in tar_out.stdout.splitlines() if line]


def pushed_tree_paths(repo_root: Path, ref: str) -> list[str]:
    origin = repo_root.parent / f"{repo_root.name}-origin.git"
    out = git(origin, ["ls-tree", "-r", "--name-only", f"refs/heads/{ref}"])
    return [line for line in out.splitlines() if line]


def only_vault_entry(vault_root: Path) -> Path:
    entries = [p for p in vault_root.iterdir() if not p.name.startswith(".tmp-")]
    assert len(entries) == 1
    return entries[0]


def make_projects_root(tmp_path: Path) -> Path:
    root = tmp_path / "projects"
    root.mkdir()
    return root


def make_repo(projects_root: Path, name: str = "repo") -> Path:
    repo_root = projects_root / name
    repo_root.mkdir(parents=True)
    git(repo_root, ["init", "-q", "-b", "main"])
    git(repo_root, ["config", "user.email", "test@test"])
    git(repo_root, ["config", "user.name", "test"])
    (repo_root / "README").write_text("x", encoding="utf-8")
    (repo_root / ".gitignore").write_text(".env\nnode_modules/\n", encoding="utf-8")
    git(repo_root, ["add", "-A"])
    git(repo_root, ["commit", "-qm", "init"])
    origin = projects_root / f"{name}-origin.git"
    origin.mkdir(parents=True)
    git(origin, ["init", "-q", "--bare", "-b", "main"])
    git(repo_root, ["remote", "add", "origin", str(origin)])
    git(repo_root, ["push", "-q", "origin", "main"])
    return repo_root


def add_worktree(repo_root: Path, slug: str, container: str = ".worktrees") -> Path:
    wt_path = repo_root / container / slug
    wt_path.parent.mkdir(parents=True, exist_ok=True)
    git(repo_root, ["worktree", "add", "-q", "-b", f"task/{slug}", str(wt_path), "main"])
    return wt_path


def empty_proc_root(tmp_path: Path, name: str = "proc") -> Path:
    proc = tmp_path / name
    proc.mkdir()
    return proc


def proc_root_with_live_cwd(tmp_path: Path, worktree_path: Path, name: str = "proc") -> Path:
    proc = empty_proc_root(tmp_path, name)
    pid_dir = proc / "4242"
    pid_dir.mkdir()
    (pid_dir / "cwd").symlink_to(worktree_path)
    (pid_dir / "cmdline").write_bytes(f"node\0{worktree_path / 'x.js'}\0".encode())
    return proc


def proc_root_with_cmdline_only(tmp_path: Path, worktree_path: Path) -> Path:
    proc = empty_proc_root(tmp_path, "proc-cmdline")
    pid_dir = proc / "4243"
    pid_dir.mkdir()
    (pid_dir / "cwd").symlink_to("/")
    (pid_dir / "cmdline").write_bytes(f"python\0{worktree_path / 'tool.py'}\0".encode())
    return proc


def state_dir(tmp_path: Path) -> Path:
    home = tmp_path / "state"
    home.mkdir()
    return home


def aged_clock() -> float:
    return time.time() * 1000.0 + (365 * 24 * 60 * 60 * 1000)


def _run_git_with_test_timeout(cwd: str, args: list[str], _timeout_s: float | None) -> str | None:
    try:
        result = subprocess.run(
            ["git", "-C", cwd, *args],
            capture_output=True,
            text=True,
            timeout=TEST_GIT_TIMEOUT_S,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        return None
    return result.stdout or "" if result.returncode == 0 else None


def _run_git_env_with_test_timeout(cwd: str, args: list[str], _timeout_s: float | None, env: dict | 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=TEST_GIT_TIMEOUT_S,
            env=full_env,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        return None
    return result.stdout or "" if result.returncode == 0 else None


def sweep(**kwargs):
    kwargs.setdefault("budget_ms", TEST_SWEEP_BUDGET_MS)
    kwargs.setdefault("run_git", _run_git_with_test_timeout)
    kwargs.setdefault("run_git_env", _run_git_env_with_test_timeout)
    kwargs.setdefault("live_session_cwds", lambda: set())
    if "state_dir" in kwargs and "vault_root" not in kwargs:
        kwargs["vault_root"] = kwargs["state_dir"].parent / "vault"
    return sweep_unleased_worktrees(kwargs)


def test_reaps_aged_clean_unoccupied_and_archives(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "abandoned")
    sha = git(wt, ["rev-parse", "HEAD"]).strip()

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
    )

    assert summary["kept"] == []
    assert len(summary["reaped"]) == 1
    assert summary["reaped"][0]["worktree"] == str(wt.resolve())
    assert summary["reaped"][0]["archivedSha"] == sha
    assert not wt.exists()
    archived = git(repo, ["rev-parse", f"{ARCHIVE_REF_PREFIX}{sha}"]).strip()
    assert archived == sha


def test_dry_run_reports_without_touching(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "dry-run")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=False,
    )

    assert len(summary["reaped"]) == 1
    assert summary["reaped"][0]["applied"] is False
    assert wt.exists()


def test_defers_younger_than_ttl(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "fresh")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        apply=True,
    )

    assert summary["reaped"] == []
    assert len(summary["deferred"]) == 1
    assert summary["deferred"][0]["reason"] == "within-ttl"
    assert wt.exists()


def test_reaps_merged_before_full_ttl(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "already-merged")
    git(repo, ["update-ref", "refs/remotes/origin/main", "refs/heads/main"])

    day_old = lambda: time.time() * 1000.0 + (24 * 60 * 60 * 1000)
    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=day_old,
        apply=True,
    )

    assert summary["deferred"] == []
    assert len(summary["reaped"]) == 1
    assert not wt.exists()


def test_unmerged_waits_full_ttl(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "unmerged")
    git(repo, ["update-ref", "refs/remotes/origin/main", "refs/heads/main"])
    (wt / "work.txt").write_text("unique work", encoding="utf-8")
    git(wt, ["add", "-A"])
    git(wt, ["commit", "-qm", "unmerged work"])

    day_old = lambda: time.time() * 1000.0 + (24 * 60 * 60 * 1000)
    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=day_old,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["deferred"][0]["reason"] == "within-ttl"
    assert wt.exists()


def test_merged_waits_short_merged_ttl(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "just-merged")
    git(repo, ["update-ref", "refs/remotes/origin/main", "refs/heads/main"])

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["deferred"][0]["reason"] == "within-ttl"
    assert wt.exists()


def test_salvages_dirty_tracked_and_removes(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "dirty")
    (wt / "README").write_text("edited", encoding="utf-8")
    sdir = state_dir(tmp_path)
    vault = sdir.parent / "vault"

    summary = sweep(
        state_dir=sdir,
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        vault_ttl_ms=10**15,
        apply=True,
    )

    assert summary["kept"] == []
    assert len(summary["salvaged"]) == 1
    assert not wt.exists()
    entry_dir = only_vault_entry(vault)
    manifest = read_manifest(entry_dir)
    assert manifest["branch"] == "task/dirty"
    assert manifest["wip_ref"].startswith("wip/task/dirty-")
    assert sorted(pushed_tree_paths(repo, manifest["wip_ref"])) == [".gitignore", "README"]
    origin = repo.parent / f"{repo.name}-origin.git"
    assert git(origin, [
        "cat-file", "-p", f"refs/heads/{manifest['wip_ref']}:README",
    ]) == "edited"


def test_salvages_dirty_untracked_and_removes(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "untracked")
    (wt / "scratch.txt").write_text("local", encoding="utf-8")
    sdir = state_dir(tmp_path)

    summary = sweep(
        state_dir=sdir,
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        vault_ttl_ms=10**15,
        apply=True,
    )

    assert summary["kept"] == []
    assert len(summary["salvaged"]) == 1
    assert not wt.exists()
    manifest = read_manifest(only_vault_entry(sdir.parent / "vault"))
    assert "scratch.txt" in pushed_tree_paths(repo, manifest["wip_ref"])


def test_salvage_archives_secret_excludes_from_push(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "has-secrets")
    (wt / ".env").write_text("TOKEN=keepme\n", encoding="utf-8")
    sdir = state_dir(tmp_path)

    summary = sweep(
        state_dir=sdir,
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        vault_ttl_ms=10**15,
        apply=True,
    )

    assert summary["kept"] == []
    assert len(summary["salvaged"]) == 1
    assert not wt.exists()
    entry_dir = only_vault_entry(sdir.parent / "vault")
    manifest = read_manifest(entry_dir)
    assert ".env" in archive_names(entry_dir)
    assert ".env" not in pushed_tree_paths(repo, manifest["wip_ref"])


def test_salvage_archives_ordinary_ignored_files(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    (repo / ".gitignore").write_text(".env\nnode_modules/\n*.scratch\nscratch/\n", encoding="utf-8")
    git(repo, ["add", ".gitignore"])
    git(repo, ["commit", "-qm", "expand ignores"])
    git(repo, ["push", "-q", "origin", "main"])
    file_wt = add_worktree(repo, "ignored-file")
    dir_wt = add_worktree(repo, "ignored-dir")
    (file_wt / "notes.scratch").write_text("keep", encoding="utf-8")
    (dir_wt / "scratch").mkdir()
    (dir_wt / "scratch" / "artifact.txt").write_text("keep", encoding="utf-8")
    sdir = state_dir(tmp_path)

    summary = sweep(
        state_dir=sdir,
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        vault_ttl_ms=10**15,
        apply=True,
    )

    assert summary["kept"] == []
    assert len(summary["salvaged"]) == 2
    assert not file_wt.exists()
    assert not dir_wt.exists()
    archived_names = set()
    for entry in sdir.parent.joinpath("vault").iterdir():
        if entry.name.startswith(".tmp-"):
            continue
        archived_names.update(archive_names(entry))
    assert "notes.scratch" in archived_names
    assert any(name.startswith("scratch/") for name in archived_names)


def test_keeps_precious_file_nested_in_reproducible_dir(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "nested-secret")
    nested = wt / "node_modules" / "package" / "cache" / "private.key"
    nested.parent.mkdir(parents=True)
    nested.write_text("keep", encoding="utf-8")
    sdir = state_dir(tmp_path)

    summary = sweep(
        state_dir=sdir,
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        vault_ttl_ms=10**15,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["salvaged"] == []
    assert summary["kept"][0]["reason"] == "precious-ignored"
    assert wt.exists()
    assert nested.exists()


def test_reproducible_scan_budget_exhaustion_keeps(tmp_path: Path):
    cache = tmp_path / "node_modules"
    cache.mkdir()
    (cache / "ordinary.js").write_text("x", encoding="utf-8")

    assert ignored_dir_holds_precious(str(cache), {"files": 0, "depth": 4}) is True


def test_reproducible_scan_unreadable_descendant_keeps(tmp_path: Path, monkeypatch):
    cache = tmp_path / "node_modules"
    nested = cache / "package"
    nested.mkdir(parents=True)
    real_scandir = worktree_gc.os.scandir

    def unreadable(path):
        if str(path) == str(nested):
            raise PermissionError(errno.EACCES, "denied", str(path))
        return real_scandir(path)

    monkeypatch.setattr(worktree_gc.os, "scandir", unreadable)
    assert ignored_dir_holds_precious(str(cache)) is True


def test_reaps_reproducible_ignored_node_modules(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "installed")
    nm = wt / "node_modules" / "left-pad"
    nm.mkdir(parents=True)
    (nm / "index.js").write_text("x", encoding="utf-8")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
    )

    assert summary["kept"] == []
    assert len(summary["reaped"]) == 1
    assert not wt.exists()


def test_proc_scan_skips_protected_same_uid_process(tmp_path: Path, monkeypatch):
    proc = empty_proc_root(tmp_path, "proc-protected")
    protected = proc / "4245"
    protected.mkdir()
    (protected / "cwd").symlink_to("/")
    (protected / "cmdline").write_bytes(b"systemd\0--user\0")
    real_readlink = worktree_gc.os.readlink

    def protected_readlink(path):
        if str(path) == str(protected / "cwd"):
            raise PermissionError(errno.EACCES, "denied", str(path))
        return real_readlink(path)

    monkeypatch.setattr(worktree_gc.os, "readlink", protected_readlink)

    assert worktree_gc.collect_live_dirs(str(proc)) == set()


def test_keeps_occupied_via_cwd(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "busy")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(proc_root_with_live_cwd(tmp_path, wt)),
        clock=aged_clock,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "occupied"
    assert wt.exists()


def test_keeps_occupied_via_cmdline(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "cmdline-busy")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(proc_root_with_cmdline_only(tmp_path, wt)),
        clock=aged_clock,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "occupied"
    assert wt.exists()


def test_keeps_occupied_via_shell_command(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "shell-busy")
    proc = empty_proc_root(tmp_path, "proc-shell")
    pid_dir = proc / "4244"
    pid_dir.mkdir()
    (pid_dir / "cwd").symlink_to("/")
    (pid_dir / "cmdline").write_bytes(f"bash\0-lc\0cd -- {wt}/src && npm test\0".encode())

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(proc),
        clock=aged_clock,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "occupied"
    assert wt.exists()


def test_keeps_worktree_when_proc_is_unreadable(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "proc-unknown")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(tmp_path / "missing-proc"),
        clock=aged_clock,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "proc-unreadable"
    assert wt.exists()


def test_keeps_all_when_same_uid_proc_metadata_is_unreadable(tmp_path: Path, monkeypatch):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "proc-metadata-unknown")
    proc = proc_root_with_live_cwd(tmp_path, wt, "proc-denied")
    denied = proc / "4242" / "cmdline"
    real_open = open

    def guarded_open(path, *args, **kwargs):
        if str(path) == str(denied):
            raise PermissionError(errno.EACCES, "denied", str(path))
        return real_open(path, *args, **kwargs)

    monkeypatch.setattr(worktree_gc, "open", guarded_open, raising=False)
    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(proc),
        clock=aged_clock,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "proc-unreadable"
    assert wt.exists()


def test_ignores_disappeared_proc_entries(tmp_path: Path, monkeypatch):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "gone-process")
    proc = empty_proc_root(tmp_path, "proc-gone")
    pid_dir = proc / "4245"
    pid_dir.mkdir()
    real_readlink = worktree_gc.os.readlink

    def disappear(path):
        if str(path) == str(pid_dir / "cwd"):
            shutil.rmtree(pid_dir)
            raise FileNotFoundError(errno.ENOENT, "gone", str(path))
        return real_readlink(path)

    monkeypatch.setattr(worktree_gc.os, "readlink", disappear)
    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(proc),
        clock=aged_clock,
        apply=True,
    )

    assert len(summary["reaped"]) == 1
    assert not wt.exists()


def test_ignores_other_uid_processes(tmp_path: Path, monkeypatch):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "other-user")
    proc = proc_root_with_live_cwd(tmp_path, wt, "proc-other-uid")
    current_uid = os.getuid()
    monkeypatch.setattr(worktree_gc.os, "getuid", lambda: current_uid + 1)

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(proc),
        clock=aged_clock,
        apply=True,
    )

    assert len(summary["reaped"]) == 1
    assert not wt.exists()


def test_rescans_occupancy_before_archive(tmp_path: Path, monkeypatch):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "busy-before-archive")
    scans = iter([set(), {str(wt.resolve())}])
    monkeypatch.setattr(worktree_gc, "collect_live_dirs", lambda _proc_root: next(scans))

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "occupied"
    assert wt.exists()


def test_rescans_occupancy_before_remove(tmp_path: Path, monkeypatch):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "busy-before-remove")
    sha = git(wt, ["rev-parse", "HEAD"]).strip()
    scans = iter([set(), set(), {str(wt.resolve())}])
    monkeypatch.setattr(worktree_gc, "collect_live_dirs", lambda _proc_root: next(scans))

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "occupied"
    assert git(repo, ["rev-parse", f"{ARCHIVE_REF_PREFIX}{sha}"]).strip() == sha
    assert wt.exists()


def test_keeps_locked(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "locked")
    git(repo, ["worktree", "lock", str(wt)])

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "locked"
    assert wt.exists()


def test_never_removes_primary_under_container_path(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    outer = make_repo(projects, "outer")
    nested = outer / ".worktrees" / "nested-repo"
    nested.mkdir(parents=True)
    git(nested, ["init", "-q", "-b", "main"])
    git(nested, ["config", "user.email", "test@test"])
    git(nested, ["config", "user.name", "test"])
    (nested / "README").write_text("x", encoding="utf-8")
    git(nested, ["add", "-A"])
    git(nested, ["commit", "-qm", "init"])

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["scanned"] == 0
    assert (nested / "README").exists()


def test_never_treats_repo_primary_as_candidate(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    add_worktree(repo, "child")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=False,
    )

    touched = [e.get("worktree") for e in summary["reaped"] + summary["kept"] + summary["deferred"]]
    assert str(repo.resolve()) not in touched
    assert (repo / "README").exists()


def test_scans_claude_worktrees_container(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "skill-made", os.path.join(".claude", "worktrees"))

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
    )

    assert len(summary["reaped"]) == 1
    assert not wt.exists()


def test_budget_cursor_fairness(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo_a = make_repo(projects, "aaa")
    repo_b = make_repo(projects, "bbb")
    wt_a = add_worktree(repo_a, "one")
    wt_b = add_worktree(repo_b, "one")
    home = state_dir(tmp_path)

    base = time.time() * 1000.0 + (365 * 24 * 60 * 60 * 1000)

    def make_clock():
        ticks = {"n": 0}

        def clock():
            ticks["n"] += 1
            return base + ((ticks["n"] - 1) * 1000)

        return clock

    first = sweep(
        state_dir=home,
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path, "proc1")),
        clock=make_clock(),
        budget_ms=2500,
        ttl_ms=0,
        apply=True,
    )
    assert first["budgetHit"] is True
    assert len(first["reaped"]) == 1
    assert first["reaped"][0]["worktree"] == str(wt_a.resolve())
    assert wt_b.exists()

    second = sweep(
        state_dir=home,
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path, "proc2")),
        clock=make_clock(),
        budget_ms=2500,
        ttl_ms=0,
        apply=True,
    )
    assert len(second["reaped"]) == 1
    assert second["reaped"][0]["worktree"] == str(wt_b.resolve())
    assert not wt_b.exists()


def test_symlink_project_root_occupancy(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects, "real")
    wt = add_worktree(repo, "busy")
    (projects / "alias").symlink_to(repo)

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(proc_root_with_live_cwd(tmp_path, projects / "alias" / ".worktrees" / "busy")),
        clock=aged_clock,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["scanned"] == 1
    assert summary["kept"][0]["reason"] == "occupied"
    assert wt.exists()


def test_both_containers_same_repo(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    a = add_worktree(repo, "plain")
    b = add_worktree(repo, "skill", os.path.join(".claude", "worktrees"))

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        budget_ms=TEST_SWEEP_BUDGET_MS,
        apply=True,
    )

    assert len(summary["reaped"]) == 2
    assert not a.exists()
    assert not b.exists()


def test_salvages_dirty_submodule_and_removes(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    sub = make_repo(projects, "sub")
    repo = make_repo(projects, "super")
    git(repo, ["-c", "protocol.file.allow=always", "submodule", "-q", "add", str(sub), "libs/sub"])
    git(repo, ["commit", "-qm", "add submodule"])
    wt = add_worktree(repo, "with-sub")
    git(wt, ["-c", "protocol.file.allow=always", "submodule", "-q", "update", "--init"])
    (wt / "libs" / "sub" / "README").write_text("uncommitted work", encoding="utf-8")
    sdir = state_dir(tmp_path)

    summary = sweep(
        state_dir=sdir,
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        vault_ttl_ms=10**15,
        apply=True,
    )

    assert not any(e.get("worktree") == str(wt.resolve()) for e in summary["kept"])
    assert len(summary["salvaged"]) == 1
    assert not wt.exists()
    entry_dir = only_vault_entry(sdir.parent / "vault")
    names = archive_names(entry_dir)
    assert any(n.endswith("libs/sub/README") or n == "libs/sub/README" for n in names)


def test_git_failure_fail_closed(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "unknowable")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
        run_git=lambda *_args, **_kwargs: None,
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "worktree-list-failed"
    assert wt.exists()


def test_ttl_default_is_three_days():
    assert UNLEASED_DEFAULT_TTL_MS == 3 * 24 * 60 * 60 * 1000


def test_merged_dirty_waits_full_ttl_then_salvages(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "merged-dirty")
    (wt / "README").write_text("edited", encoding="utf-8")
    sdir = state_dir(tmp_path)

    day_old = lambda: time.time() * 1000.0 + (24 * 60 * 60 * 1000)
    summary = sweep(
        state_dir=sdir,
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=day_old,
        vault_ttl_ms=10**15,
        apply=True,
    )
    assert summary["reaped"] == []
    assert summary["salvaged"] == []
    assert summary["deferred"][0]["reason"] == "within-ttl"
    assert wt.exists()

    summary = sweep(
        state_dir=sdir,
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path, "proc2")),
        clock=aged_clock,
        vault_ttl_ms=10**15,
        apply=True,
    )
    assert len(summary["salvaged"]) == 1
    assert not wt.exists()


def test_ttl_override_honored(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "custom-ttl")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        ttl_ms=0,
        merged_ttl_ms=0,
        clock=lambda: time.time() * 1000.0 + 1000,
        apply=True,
    )
    assert len(summary["reaped"]) == 1
    assert not wt.exists()


def test_temp_index_leaves_real_index_and_head_untouched(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "dirty-index")
    (wt / "README").write_text("edited", encoding="utf-8")
    (wt / "scratch.txt").write_text("untracked", encoding="utf-8")
    status_before = git(wt, ["status", "--porcelain", "--ignore-submodules=none"])
    head_before = git(wt, ["rev-parse", "HEAD"]).strip()

    result = salvage_push(str(wt), "task/dirty-index", _run_git_env_with_test_timeout)

    assert result is not None
    assert result["pushed"] is True
    assert git(wt, ["status", "--porcelain", "--ignore-submodules=none"]) == status_before
    assert git(wt, ["rev-parse", "HEAD"]).strip() == head_before
    assert (wt / "scratch.txt").read_text(encoding="utf-8") == "untracked"


def test_push_failure_keeps_worktree(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "no-push")
    (wt / "README").write_text("edited", encoding="utf-8")

    def failing_run_git_env(cwd, args, timeout_s, env):
        if args and args[0] == "push":
            return None
        return _run_git_env_with_test_timeout(cwd, args, timeout_s, env)

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
        run_git_env=failing_run_git_env,
    )
    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "push-failed"
    assert wt.exists()


def test_fetch_failure_keeps_clean_unmerged_worktree(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "no-fetch")
    (wt / "extra.txt").write_text("x", encoding="utf-8")
    git(wt, ["add", "-A"])
    git(wt, ["commit", "-qm", "unpushed"])

    real_run_git = _run_git_with_test_timeout

    def failing_fetch(cwd, args, timeout_s):
        if args and args[0] == "fetch":
            return None
        return real_run_git(cwd, args, timeout_s)

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
        run_git=failing_fetch,
    )
    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "remote-unverified"
    assert wt.exists()


def test_identity_change_before_removal_keeps(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "identity-race")
    (wt / "README").write_text("edited", encoding="utf-8")

    real_run_git = _run_git_with_test_timeout
    calls = {"status": 0}

    def racing_run_git(cwd, args, timeout_s):
        if args and args[0:1] == ["status"]:
            calls["status"] += 1
            if calls["status"] == 2:
                (Path(cwd) / "race.txt").write_text("late edit", encoding="utf-8")
        return real_run_git(cwd, args, timeout_s)

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
        run_git=racing_run_git,
    )
    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "identity-changed"
    assert wt.exists()


def test_session_attached_keeps_past_ttl(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "session-live")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
        live_session_cwds=lambda: {str(wt.resolve())},
    )
    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "session-attached"
    assert wt.exists()


def test_session_oracle_none_keeps_all(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "oracle-down")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
        live_session_cwds=lambda: None,
    )
    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "session-oracle-unreadable"
    assert wt.exists()


def test_dead_session_absent_from_oracle_is_eligible():
    assert worktree_has_live_session("/tmp/wt", {"/tmp/other"}) is False
    assert worktree_has_live_session("/tmp/wt", {"/tmp/wt"}) is True
    assert worktree_has_live_session("/tmp/wt", {"/tmp/wt/sub"}) is True


def test_detached_head_salvage_ref_shape_and_manifest(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "detach-me")
    sha = git(wt, ["rev-parse", "HEAD"]).strip()
    git(wt, ["checkout", "-q", "--detach", sha])
    (wt / "README").write_text("edited", encoding="utf-8")
    sdir = state_dir(tmp_path)

    summary = sweep(
        state_dir=sdir,
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        vault_ttl_ms=10**15,
        apply=True,
    )
    assert summary["kept"] == []
    assert len(summary["salvaged"]) == 1
    manifest = read_manifest(only_vault_entry(sdir.parent / "vault"))
    assert "branch" not in manifest
    assert manifest["detached_sha"] == sha
    assert manifest["wip_ref"].startswith("wip/detached-")
    git(repo, ["check-ref-format", f"refs/heads/{manifest['wip_ref']}"])
    assert "task/detach-me" in git(repo, ["branch", "--list", "task/detach-me"])


def test_archive_transactionality_no_tmp_left_on_tar_failure(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "tar-fail")
    (wt / "scratch.txt").write_text("x", encoding="utf-8")
    vault = tmp_path / "vault"

    with_env = dict(os.environ)
    with_env["PATH"] = str(tmp_path / "empty-bin")
    (tmp_path / "empty-bin").mkdir()

    def broken_run_git(cwd, args, timeout_s):
        return _run_git_with_test_timeout(cwd, args, timeout_s)

    import subprocess as _sp
    real_popen = _sp.Popen

    def failing_popen(cmd, *a, **kw):
        if cmd and cmd[0] == "tar":
            raise FileNotFoundError("simulated missing tar")
        return real_popen(cmd, *a, **kw)

    orig_popen = worktree_gc.subprocess.Popen
    worktree_gc.subprocess.Popen = failing_popen
    try:
        result = archive_worktree(str(wt), vault, broken_run_git, lambda: time.time() * 1000.0)
    finally:
        worktree_gc.subprocess.Popen = orig_popen

    assert result is None
    remaining = list(vault.iterdir()) if vault.exists() else []
    assert remaining == []


def test_symlinks_archived_as_links(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "with-symlink")
    (wt / "target.txt").write_text("real", encoding="utf-8")
    git(wt, ["add", "-A"])
    git(wt, ["commit", "-qm", "add target"])
    (wt / "scratch.txt").write_text("x", encoding="utf-8")
    (wt / "link.txt").symlink_to("target.txt")
    vault = tmp_path / "vault"

    result = archive_worktree(str(wt), vault, _run_git_with_test_timeout, lambda: time.time() * 1000.0)
    assert result is not None
    entry_dir = Path(result["path"])
    zstd_proc = subprocess.Popen(["zstd", "-dc", str(entry_dir / "archive.tar.zst")], stdout=subprocess.PIPE)
    listing = subprocess.run(["tar", "-tvf", "-"], stdin=zstd_proc.stdout, capture_output=True, text=True)
    zstd_proc.wait()
    link_lines = [line for line in listing.stdout.splitlines() if "link.txt" in line]
    assert link_lines
    assert link_lines[0].startswith("l")


def test_vault_expiry_warns_one_sweep_ahead_and_deletes(tmp_path: Path):
    vault = tmp_path / "vault"
    vault.mkdir()
    entry = vault / "repo--task--20200101-000000--abc1234"
    entry.mkdir(mode=0o700)
    (entry / "manifest.json").write_text("{}", encoding="utf-8")
    ttl_ms = 60 * 24 * 60 * 60 * 1000
    warn_age_ms = ttl_ms - VAULT_WARN_LEAD_MS + 1000
    now_ms = time.time() * 1000.0
    mtime = (now_ms - warn_age_ms) / 1000.0
    os.utime(entry, (mtime, mtime))

    warned = expire_vault_entries(vault, lambda: now_ms, ttl_ms)
    assert len(warned) == 1
    assert warned[0]["entry"] == entry.name
    assert entry.exists()

    expired = expire_vault_entries(vault, lambda: now_ms + VAULT_WARN_LEAD_MS + 60_000, ttl_ms)
    assert expired == []
    assert not entry.exists()


def test_vault_expiry_skips_entry_outside_canonical_root(tmp_path: Path):
    vault = tmp_path / "vault"
    vault.mkdir()
    outside = tmp_path / "outside-entry"
    outside.mkdir()
    (outside / "manifest.json").write_text("{}", encoding="utf-8")
    (vault / "escape-link").symlink_to(outside)
    old = time.time() - (365 * 24 * 60 * 60)
    os.utime(outside, (old, old))

    expire_vault_entries(vault, lambda: time.time() * 1000.0, 60 * 24 * 60 * 60 * 1000)
    assert outside.exists()


def test_vault_expiry_keeps_unreadable_manifest(tmp_path: Path):
    vault = tmp_path / "vault"
    vault.mkdir()
    entry = vault / "repo--task--20200101-000000--abc1234"
    entry.mkdir(mode=0o700)
    old = time.time() - (365 * 24 * 60 * 60)
    os.utime(entry, (old, old))

    expire_vault_entries(vault, lambda: time.time() * 1000.0, 60 * 24 * 60 * 60 * 1000)
    assert entry.exists()


def test_local_branch_deleted_only_on_verified_tip(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    git(repo, ["branch", "verified"])
    git(repo, ["push", "-q", "origin", "verified"])
    assert cleanup_local_branch(str(repo), "verified", _run_git_with_test_timeout) is True
    assert git(repo, ["branch", "--list", "verified"]).strip() == ""

    git(repo, ["branch", "unpushed"])
    assert cleanup_local_branch(str(repo), "unpushed", _run_git_with_test_timeout) is False
    assert "unpushed" in git(repo, ["branch", "--list", "unpushed"])


def test_local_branch_deleted_when_ancestor_of_pushed_salvage(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    git(repo, ["branch", "salvaged"])
    (repo / "extra.txt").write_text("x", encoding="utf-8")
    git(repo, ["add", "extra.txt"])
    git(repo, ["commit", "-qm", "salvage child"])
    child = git(repo, ["rev-parse", "HEAD"]).strip()
    git(repo, ["push", "-q", "origin", f"{child}:refs/heads/wip/salvaged-x"])
    git(repo, ["reset", "-q", "--hard", "HEAD~1"])

    assert cleanup_local_branch(str(repo), "salvaged", _run_git_with_test_timeout, pushed_sha=child) is True
    assert git(repo, ["branch", "--list", "salvaged"]).strip() == ""

    git(repo, ["branch", "unrelated"])
    assert cleanup_local_branch(str(repo), "unrelated", _run_git_with_test_timeout, pushed_sha="0" * 40) is False


# ── FIRE: reaper must never touch worktrees belonging to live work ─────────────────────


def _write_landq_ticket(repo_root: Path, wt_path: Path, ticket_id: str = "aa11bb22cc33dd44", verdict: bool = False) -> None:
    landq_dir = repo_root / ".git" / "harness" / "landq"
    landq_dir.mkdir(parents=True, exist_ok=True)
    wt_b64 = base64.b64encode(str(wt_path.resolve()).encode()).decode()
    (landq_dir / f"ticket.{ticket_id}.job").write_text(f"wt {wt_b64}\n", encoding="utf-8")
    if verdict:
        (landq_dir / f"ticket.{ticket_id}.verdict").write_text("0\n{}\n", encoding="utf-8")


def _own_proc_start_ticks() -> int:
    raw = Path("/proc/self/stat").read_text(encoding="utf-8")
    fields = raw[raw.rfind(")") + 2:].split()
    return int(fields[19])


def _write_session_ledger_row(ledger_dir: Path, *, pid: int, pid_start_ticks: int, cwd: str, finished: bool = False) -> None:
    ledger_dir.mkdir(parents=True, exist_ok=True)
    row = {
        "pid": pid,
        "pidStartTicks": pid_start_ticks,
        "cwd": cwd,
        "finishedAt": "2026-08-01T00:00:00Z" if finished else None,
    }
    (ledger_dir / "victim.json").write_text(json.dumps(row), encoding="utf-8")


def test_keeps_worktree_with_live_landq_ticket(tmp_path: Path):
    """Victim 1: a worktree with a live land-queue ticket must never be reaped
    while the conductor still owns its lifecycle."""
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "queued-ticket")
    _write_landq_ticket(repo, wt)

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "land-queue-ticket"
    assert wt.exists()


def test_reaps_worktree_once_landq_ticket_has_terminal_verdict(tmp_path: Path):
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "resolved-ticket")
    _write_landq_ticket(repo, wt, verdict=True)

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
    )

    assert summary["kept"] == []
    assert len(summary["reaped"]) == 1
    assert not wt.exists()


def test_keeps_worktree_with_recent_index_lock(tmp_path: Path):
    """Requirement (d): a checkout in progress (fresh index.lock) is untouchable."""
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "checkout-in-progress")
    gitdir_rel = git(wt, ["rev-parse", "--git-dir"]).strip()
    gitdir = Path(gitdir_rel) if Path(gitdir_rel).is_absolute() else wt / gitdir_rel
    (gitdir / "index.lock").write_text("", encoding="utf-8")

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=lambda: time.time() * 1000.0,
        apply=True,
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "checkout-in-progress"
    assert wt.exists()


def test_keeps_worktree_with_pid_verified_live_session_ledger_entry(tmp_path: Path):
    """Victim 3 (tmux-attached session): a ledger row is only trusted once its pid
    is confirmed alive with matching start-ticks against /proc — never trusted raw."""
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "tmux-attached")
    ledger_dir = tmp_path / "ledger"
    _write_session_ledger_row(
        ledger_dir,
        pid=os.getpid(),
        pid_start_ticks=_own_proc_start_ticks(),
        cwd=str(wt.resolve()),
    )

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
        live_session_cwds=lambda: live_session_cwds(proc_root="/proc", ledger_dir=str(ledger_dir)),
    )

    assert summary["reaped"] == []
    assert summary["kept"][0]["reason"] == "session-attached"
    assert wt.exists()


def test_reaps_worktree_with_stale_ledger_entry_pid_mismatch(tmp_path: Path):
    """A ledger row whose pid/start-ticks no longer match a live process must never
    be trusted — the ledger can carry phantom or reused-pid entries."""
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "stale-ledger-row")
    ledger_dir = tmp_path / "ledger"
    _write_session_ledger_row(
        ledger_dir,
        pid=os.getpid(),
        pid_start_ticks=_own_proc_start_ticks() + 999_999,
        cwd=str(wt.resolve()),
    )

    summary = sweep(
        state_dir=state_dir(tmp_path),
        project_roots=[str(projects)],
        proc_root=str(empty_proc_root(tmp_path)),
        clock=aged_clock,
        apply=True,
        live_session_cwds=lambda: live_session_cwds(proc_root="/proc", ledger_dir=str(ledger_dir)),
    )

    assert summary["kept"] == []
    assert len(summary["reaped"]) == 1
    assert not wt.exists()


def test_idle_ms_detects_nested_file_edit(tmp_path: Path):
    """Requirement (c): idle-age must reflect edits nested under subdirectories, not
    just the worktree's own top-level directory entries (or git-index/HEAD alone)."""
    projects = make_projects_root(tmp_path)
    repo = make_repo(projects)
    wt = add_worktree(repo, "nested-edit")

    gitdir_rel = git(wt, ["rev-parse", "--git-dir"]).strip()
    gitdir = Path(gitdir_rel) if Path(gitdir_rel).is_absolute() else wt / gitdir_rel

    old_s = time.time() - (10 * 24 * 60 * 60)
    for aged_path in (wt, wt / ".git", gitdir, gitdir / "HEAD", gitdir / "index"):
        if aged_path.exists():
            os.utime(aged_path, (old_s, old_s))
    for top_level in wt.iterdir():
        if top_level.name != ".git":
            os.utime(top_level, (old_s, old_s), follow_symlinks=False)

    nested_dir = wt / "modules" / "foo"
    nested_dir.mkdir(parents=True)
    (nested_dir / "bar.py").write_text("x", encoding="utf-8")

    idle_ms = worktree_gc.worktree_idle_ms(str(wt), lambda: time.time() * 1000.0)
    assert idle_ms < 60_000