diff --git a/dirty_same_claimant_session_rebind.py b/dirty_same_claimant_session_rebind.py new file mode 100644 index 0000000..fc29add --- /dev/null +++ b/dirty_same_claimant_session_rebind.py @@ -0,0 +1,1565 @@ +"""Dirty-preserving same-claimant author-session rebind (#864 / #868). + +A registered issue worktree can be dirty while its durable lock owner PID is +provably dead. Ordinary ``gitea_lock_issue`` refuses dirty trees, and dead-session +recovery (#753) also requires cleanliness. This module is the *only* sanctioned +path that rebinds session/lock provenance onto the *same* worktree without +touching tracked or untracked content. + +This is SEPARATE from #860 dirty-orphan recovery (PID-less + remote sync). +This operation: + +* acts only on an already-registered dirty worktree +* updates only stale lock/session provenance (PID, session pointer, generation, + heartbeat) +* preserves every tracked/untracked byte +* does NOT sync remote, create recovery worktrees, clean, reset, or change heads + +#868 hardens: + +* complete dirty-inventory revalidation (full path set + fingerprints) + immediately before and after ``bind_session_lock`` +* durable recovery-journal operation identity (remote, org, repo, claimant + identity, claimant profile) validated on execute / resume / retry / + already_rebound +""" + +from __future__ import annotations + +import hashlib +import json +import os +import subprocess +import tempfile +from datetime import datetime, timezone +from typing import Any, Mapping, Sequence + +from author_mutation_worktree import is_path_under_branches +from issue_lock_provenance import ( + SOURCE_DIRTY_SAME_CLAIMANT_REBIND, + build_sanctioned_lock_provenance, +) +from issue_lock_store import ( + AUTHOR_ISSUE_WORK_LEASE, + bind_session_lock, + is_process_alive, + lock_file_path, + lock_generation, + read_lock_file, +) +from reviewer_worktree import parse_dirty_tracked_files + +# Outcomes +REBIND_SANCTIONED = "REBIND_SANCTIONED" +REFUSED = "REFUSED" +NO_CANDIDATE = "NO_CANDIDATE" + +# Provenance / tool identity +SOURCE_TOOL = SOURCE_DIRTY_SAME_CLAIMANT_REBIND +SOURCE = SOURCE_DIRTY_SAME_CLAIMANT_REBIND + +# Journal phases (crash-safe apply) +JOURNAL_PHASE_ASSESSED = "assessed" +JOURNAL_PHASE_PRE_BIND = "pre_bind" +JOURNAL_PHASE_BOUND = "bound" +JOURNAL_PHASE_COMPLETE = "complete" +JOURNAL_PHASE_ALREADY_REBOUND = "already_rebound" + +REQUIRED_LOCK_FIELDS = ( + "issue_number", + "branch_name", + "worktree_path", + "remote", + "org", + "repo", +) + +# Durable journal operation identity (#868 F2). All five must be persisted on +# JOURNAL_PHASE_ASSESSED and re-validated on resume / retry / already_rebound. +REQUIRED_JOURNAL_IDENTITY_FIELDS = ( + "remote", + "org", + "repo", + "claimant_identity", + "claimant_profile", +) + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _text(value: Any) -> str: + return str(value or "").strip() + + +def _same_realpath(left: str | None, right: str | None) -> bool: + if not left or not right: + return False + try: + return os.path.realpath(left) == os.path.realpath(right) + except OSError: + return left == right + + +def _lock_claimant(lock: Mapping[str, Any]) -> dict[str, Any]: + claimant = lock.get("claimant") + if not isinstance(claimant, Mapping): + lease = lock.get("work_lease") + claimant = lease.get("claimant") if isinstance(lease, Mapping) else None + return dict(claimant) if isinstance(claimant, Mapping) else {} + + +def _recorded_pid(lock: Mapping[str, Any]) -> Any: + pid = lock.get("session_pid") + if pid is None: + pid = lock.get("pid") + return pid + + +def content_fingerprint(path: str) -> str: + """Return sha256 hex digest of file bytes at *path*. + + Missing or unreadable files raise ``OSError`` / ``FileNotFoundError`` so + callers fail closed rather than inventing an empty hash. + """ + digest = hashlib.sha256() + with open(path, "rb") as handle: + while True: + chunk = handle.read(1024 * 1024) + if not chunk: + break + digest.update(chunk) + return digest.hexdigest() + + +def parse_dirty_paths(porcelain: str) -> list[str]: + """Tracked + untracked paths from ``git status --porcelain -uall`` output.""" + paths: list[str] = [] + seen: set[str] = set() + for line in (porcelain or "").splitlines(): + if not line or len(line) < 4: + continue + if line.startswith("??"): + path = line[3:].strip() + else: + path = line[3:].strip() + if " -> " in path: + path = path.split(" -> ", 1)[1].strip() + if not path or path in seen: + continue + seen.add(path) + paths.append(path) + return paths + + +def collect_dirty_inventory(worktree_path: str) -> dict[str, Any]: + """Observe dirty tracked + untracked paths and content fingerprints. + + Uses ``git status --porcelain -uall`` so every untracked file is listed + individually (not collapsed into a directory). + """ + path = (worktree_path or "").strip() + if not path: + return { + "worktree_path": path, + "porcelain_status": "", + "dirty_paths": [], + "fingerprints": {}, + "ok": False, + "reasons": ["worktree path is empty"], + } + + status_res = subprocess.run( + ["git", "-C", path, "status", "--porcelain", "-uall"], + capture_output=True, + text=True, + check=False, + ) + if status_res.returncode != 0: + err = (status_res.stderr or status_res.stdout or "").strip() + return { + "worktree_path": path, + "porcelain_status": "", + "dirty_paths": [], + "fingerprints": {}, + "ok": False, + "reasons": [f"git status failed in '{path}': {err or 'unknown error'}"], + } + + porcelain = status_res.stdout or "" + dirty_paths = parse_dirty_paths(porcelain) + fingerprints: dict[str, str] = {} + reasons: list[str] = [] + for rel in dirty_paths: + abs_path = os.path.join(path, rel) + if os.path.isdir(abs_path) and not os.path.islink(abs_path): + # Directories appear only if git reported them; fingerprinting a + # directory is not defined — fail closed. + reasons.append(f"dirty path '{rel}' is a directory; cannot fingerprint") + continue + try: + fingerprints[rel] = content_fingerprint(abs_path) + except OSError as exc: + reasons.append(f"could not fingerprint '{rel}': {exc}") + + return { + "worktree_path": os.path.realpath(path), + "porcelain_status": porcelain, + "dirty_paths": dirty_paths, + "fingerprints": fingerprints, + "ok": not reasons, + "reasons": reasons, + "tracked_dirty": parse_dirty_tracked_files(porcelain), + } + + +def revalidate_complete_dirty_inventory( + worktree_path: str, + *, + expected_dirty_paths: Sequence[str] | None, + expected_fingerprints: Mapping[str, str] | None, + phase: str = "inventory", +) -> dict[str, Any]: + """Collect the full dirty inventory and require exact pin equality (#868 F1). + + Unlike fingerprint-only checks over the expected path list, this recollects + the authoritative tracked+untracked inventory and refuses added, removed, + or renamed paths as well as fingerprint movement. + """ + reasons: list[str] = [] + inv = collect_dirty_inventory(worktree_path) + if inv.get("ok") is False: + reasons.extend(list(inv.get("reasons") or []) or [f"{phase}: dirty inventory collection failed"]) + + observed_paths = sorted( + {_text(p) for p in (inv.get("dirty_paths") or []) if _text(p)} + ) + pin_paths = sorted( + {_text(p) for p in (expected_dirty_paths or []) if _text(p)} + ) + if not pin_paths: + reasons.append( + f"{phase}: expected_dirty_paths pin is empty; complete inventory " + "revalidation requires a non-empty pin (fail closed)" + ) + if set(observed_paths) != set(pin_paths): + extra = sorted(set(observed_paths) - set(pin_paths)) + missing = sorted(set(pin_paths) - set(observed_paths)) + if extra: + reasons.append( + f"{phase}: complete dirty inventory path-set disagreement: " + f"unexpected paths {extra}" + ) + if missing: + reasons.append( + f"{phase}: complete dirty inventory path-set disagreement: " + f"missing expected paths {missing}" + ) + + obs_fps = { + _text(k): _text(v) + for k, v in dict(inv.get("fingerprints") or {}).items() + if _text(k) + } + pin_fps = { + _text(k): _text(v) + for k, v in dict(expected_fingerprints or {}).items() + if _text(k) + } + if not pin_fps: + reasons.append( + f"{phase}: expected_fingerprints pin is empty; byte-level pins " + "are required (fail closed)" + ) + else: + for rel, expected_hash in pin_fps.items(): + if rel not in set(pin_paths): + reasons.append( + f"{phase}: expected_fingerprints contains '{rel}' which is " + "not in expected_dirty_paths" + ) + continue + actual_hash = obs_fps.get(rel) + if not actual_hash: + reasons.append( + f"{phase}: fingerprint missing for dirty path '{rel}'" + ) + elif actual_hash != expected_hash: + reasons.append( + f"{phase}: fingerprint disagreement for '{rel}': " + f"observed {actual_hash}, expected {expected_hash}" + ) + for rel in observed_paths: + if rel not in pin_fps: + reasons.append( + f"{phase}: observed dirty path '{rel}' has no fingerprint pin" + ) + + return { + "ok": not reasons, + "reasons": reasons, + "inventory": inv, + "observed_dirty_paths": observed_paths, + "expected_dirty_paths": pin_paths, + "observed_fingerprints": obs_fps, + "expected_fingerprints": pin_fps, + "phase": phase, + } + + +def build_journal_operation_identity( + *, + remote: str, + org: str, + repo: str, + claimant_identity: str | None, + claimant_profile: str | None, +) -> dict[str, str]: + """Return the five-field durable operation identity for the recovery journal.""" + return { + "remote": _text(remote), + "org": _text(org), + "repo": _text(repo), + "claimant_identity": _text(claimant_identity), + "claimant_profile": _text(claimant_profile), + } + + +def validate_journal_operation_identity( + journal: Mapping[str, Any] | None, + *, + remote: str, + org: str, + repo: str, + claimant_identity: str | None, + claimant_profile: str | None, + require_present: bool = True, +) -> list[str]: + """Validate durable journal identity fields (#868 F2). + + Rejects missing, mismatched, stale, cross-repository, or cross-claimant + journal state. When *require_present* is True, incomplete legacy journals + (any of the five fields absent/empty) fail closed. + """ + reasons: list[str] = [] + if not isinstance(journal, Mapping): + if require_present: + reasons.append( + "recovery journal is missing or unreadable; complete operation " + "identity cannot be proven (fail closed)" + ) + return reasons + + expected = build_journal_operation_identity( + remote=remote, + org=org, + repo=repo, + claimant_identity=claimant_identity, + claimant_profile=claimant_profile, + ) + for field in REQUIRED_JOURNAL_IDENTITY_FIELDS: + observed = _text(journal.get(field)) + want = expected[field] + if not observed: + reasons.append( + f"recovery journal omits operation identity field '{field}' " + "(incomplete legacy or malformed journal identity; fail closed)" + ) + continue + if not want: + reasons.append( + f"caller pin for journal identity field '{field}' is empty " + "(fail closed)" + ) + continue + if observed != want: + reasons.append( + f"recovery journal identity mismatch for '{field}': " + f"journal={observed!r}, expected={want!r} " + "(cross-repository / cross-claimant / replay refused)" + ) + return reasons + + +def journal_path(lock_dir: str, issue_number: int) -> str: + root = (lock_dir or "").strip() + return os.path.join(root, f".rebind-journal-{int(issue_number)}.json") + + +def _atomic_write_json(path: str, data: dict[str, Any]) -> None: + parent = os.path.dirname(path) or "." + os.makedirs(parent, mode=0o700, exist_ok=True) + payload = json.dumps(data, indent=2, sort_keys=True) + "\n" + fd, temp_path = tempfile.mkstemp(prefix=".rebind-j-", suffix=".json", dir=parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(payload) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp_path, path) + finally: + if os.path.exists(temp_path): + try: + os.remove(temp_path) + except OSError: + pass + + +def _read_json(path: str) -> dict[str, Any] | None: + if not path or not os.path.exists(path): + return None + try: + with open(path, encoding="utf-8") as handle: + data = json.load(handle) + except (OSError, json.JSONDecodeError): + return None + return data if isinstance(data, dict) else None + + +def _malformed_lock_reasons(lock: Mapping[str, Any]) -> list[str]: + missing: list[str] = [] + for field in REQUIRED_LOCK_FIELDS: + if not _text(lock.get(field)): + missing.append(field) + pid = _recorded_pid(lock) + if pid is None or _text(pid) == "": + missing.append("session_pid/pid") + else: + try: + if int(pid) <= 0: + missing.append("session_pid/pid") + except (TypeError, ValueError): + missing.append("session_pid/pid") + return missing + + +def _canonical_under_branches(worktree_path: str, repo_root: str | None) -> tuple[bool, list[str]]: + """Prove worktree is a realpath under ``/branches/`` with no symlink escape.""" + reasons: list[str] = [] + path = (worktree_path or "").strip() + if not path: + return False, ["worktree path is empty"] + try: + real = os.path.realpath(path) + except OSError as exc: + return False, [f"worktree path could not be realpath-resolved: {exc}"] + if not os.path.isdir(real): + reasons.append(f"worktree path '{path}' is not an existing directory") + + root = (repo_root or "").strip() + if root: + try: + root_real = os.path.realpath(root) + except OSError as exc: + return False, [f"repo root could not be realpath-resolved: {exc}"] + if not is_path_under_branches(real, root_real): + reasons.append( + f"worktree '{real}' is not under branches/ of repo root '{root_real}' " + "(unregistered/noncanonical worktree; fail closed)" + ) + # Symlink escape: the declared path must not resolve outside branches/. + declared_abs = os.path.abspath(path) + if os.path.islink(path) or declared_abs != real: + if not is_path_under_branches(real, root_real): + reasons.append( + f"worktree path '{path}' escapes branches/ via symlink/realpath " + f"(resolves to '{real}')" + ) + else: + # Without an explicit repo root, still require a /branches/ segment. + if not is_path_under_branches(real, None): + reasons.append( + f"worktree '{real}' is not under a branches/ directory " + "(unregistered/noncanonical worktree; fail closed)" + ) + return not reasons, reasons + + +def assess_dirty_same_claimant_session_rebind( + *, + remote: str, + org: str, + repo: str, + issue_number: int, + branch_name: str, + worktree_path: str, + claimant_identity: str | None, + claimant_profile: str | None, + old_pid: int | None, + expected_local_head: str | None, + expected_remote_head: str | None, + expected_dirty_paths: Sequence[str] | None, + expected_fingerprints: Mapping[str, str] | None, + existing_lock: Mapping[str, Any] | None, + current_identity: str | None, + current_profile: str | None, + role_kind: str | None, + current_pid: int | None, + current_branch: str | None, + local_head: str | None, + remote_head: str | None, + porcelain_status: str | None = None, + dirty_inventory: Mapping[str, Any] | None = None, + competing_live_locks: Sequence[Mapping[str, Any]] | None = None, + competing_sessions: Sequence[Mapping[str, Any]] | None = None, + workflow_lease_active: bool = False, + authorize_reconciler_execute: bool = False, + permission_allowed: bool = False, + repo_root: str | None = None, +) -> dict[str, Any]: + """Pure assessment: may this dirty same-claimant lock be session-rebound? + + Every pin must agree. ``permission_allowed=True`` alone is never ownership + proof. Fail closed on live old PID, foreign identity/profile, pin mismatch, + unregistered/noncanonical worktree, head movement, dirty path/fingerprint + disagreement, competing ownership, malformed lock, empty PID, wrong role. + """ + reasons: list[str] = [] + evidence: dict[str, Any] = { + "issue_number": issue_number, + "branch_name": branch_name, + "worktree_path": worktree_path, + "remote": remote, + "org": org, + "repo": repo, + "old_pid": old_pid, + "current_pid": current_pid if current_pid is not None else os.getpid(), + "role_kind": _text(role_kind).lower() or None, + "permission_allowed": bool(permission_allowed), + } + + if not existing_lock: + return _assessment_result( + NO_CANDIDATE, + False, + ["no existing durable lock for this issue; not a rebind candidate"], + evidence, + ) + + lock = dict(existing_lock) + if lock.get("issue_number") != issue_number: + return _assessment_result( + NO_CANDIDATE, + False, + [ + f"existing lock targets issue #{lock.get('issue_number')}, " + f"not #{issue_number}; not a rebind candidate" + ], + evidence, + ) + + missing = _malformed_lock_reasons(lock) + if missing: + return _assessment_result( + REFUSED, + False, + [ + "durable lock record is incomplete and cannot prove ownership " + f"(missing/unusable: {', '.join(missing)})" + ], + evidence, + ) + + recorded_pid = _recorded_pid(lock) + evidence["recorded_pid"] = recorded_pid + evidence["lock_generation"] = lock_generation(lock) + + # ── Role gate ─────────────────────────────────────────────────────────── + role = _text(role_kind).lower() + if role in {"reviewer", "merger"}: + reasons.append( + f"role '{role}' cannot rebind dirty same-claimant author sessions " + "(fail closed)" + ) + elif role == "reconciler": + if not authorize_reconciler_execute: + reasons.append( + "reconciler role requires authorize_reconciler_execute=True " + "to execute dirty same-claimant rebind (fail closed)" + ) + elif role == "author": + pass + elif role: + reasons.append(f"role '{role}' is not authorized for dirty same-claimant rebind") + else: + reasons.append("role_kind is unknown; dirty same-claimant rebind refused") + + # permission_allowed is explicitly NOT ownership proof + evidence["note_permission_not_ownership"] = ( + "permission_allowed is not treated as ownership proof" + ) + + # ── Repository / issue / branch / worktree pins ───────────────────────── + for field, expected in (("remote", remote), ("org", org), ("repo", repo)): + actual = _text(lock.get(field)) + if actual != _text(expected): + reasons.append( + f"lock {field} '{actual}' does not match requested '{_text(expected)}'" + ) + + locked_branch = _text(lock.get("branch_name")) + if locked_branch != _text(branch_name): + reasons.append( + f"lock branch '{locked_branch}' does not match requested " + f"'{_text(branch_name)}'" + ) + + checked_out = _text(current_branch) + if not checked_out: + reasons.append( + "worktree is not on a named branch (detached HEAD); locked-branch " + "occupancy could not be proven" + ) + elif checked_out != locked_branch: + reasons.append( + f"worktree is on branch '{checked_out}', not the locked branch " + f"'{locked_branch}'" + ) + + locked_worktree = _text(lock.get("worktree_path")) + if not _same_realpath(locked_worktree, worktree_path): + reasons.append( + f"lock worktree '{locked_worktree}' does not match declared " + f"'{_text(worktree_path)}'" + ) + evidence["locked_worktree_path"] = locked_worktree + + under_ok, under_reasons = _canonical_under_branches(worktree_path, repo_root) + if not under_ok: + reasons.extend(under_reasons) + + # ── old_pid pin + liveness ────────────────────────────────────────────── + if old_pid is None or _text(old_pid) == "": + reasons.append("old_pid pin is empty; rebind refused (fail closed)") + else: + try: + old_pid_i = int(old_pid) + except (TypeError, ValueError): + reasons.append(f"old_pid '{old_pid}' is not a valid PID") + old_pid_i = None + if old_pid_i is not None: + if old_pid_i <= 0: + reasons.append("old_pid must be a positive integer (fail closed)") + try: + recorded_i = int(recorded_pid) + except (TypeError, ValueError): + recorded_i = None + if recorded_i is None or recorded_i != old_pid_i: + reasons.append( + f"old_pid {old_pid_i} does not match lock session_pid/pid " + f"{recorded_pid}" + ) + if is_process_alive(old_pid_i): + reasons.append( + f"old_pid {old_pid_i} is still alive; dirty same-claimant " + "rebind requires a provably dead owner (fail closed)" + ) + evidence["old_pid_alive"] = is_process_alive(old_pid_i) + if current_pid is not None: + try: + if int(current_pid) == old_pid_i: + reasons.append( + "old_pid is the current session PID; nothing to rebind" + ) + except (TypeError, ValueError): + pass + + # ── Claimant identity / profile ───────────────────────────────────────── + lock_claimant = _lock_claimant(lock) + locked_identity = _text(lock_claimant.get("username")) + locked_profile = _text(lock_claimant.get("profile")) + pin_identity = _text(claimant_identity) + pin_profile = _text(claimant_profile) + active_identity = _text(current_identity) + active_profile = _text(current_profile) + evidence["locked_identity"] = locked_identity or None + evidence["locked_profile"] = locked_profile or None + + if not locked_identity or not locked_profile: + reasons.append( + "durable lock does not record a claimant identity/profile; " + "ownership could not be proven" + ) + if not pin_identity or not pin_profile: + reasons.append( + "claimant_identity/claimant_profile pins are required (fail closed)" + ) + if locked_identity and pin_identity and locked_identity != pin_identity: + reasons.append( + f"claimant_identity pin '{pin_identity}' does not match lock " + f"claimant '{locked_identity}'" + ) + if locked_profile and pin_profile and locked_profile != pin_profile: + reasons.append( + f"claimant_profile pin '{pin_profile}' does not match lock profile " + f"'{locked_profile}'" + ) + + # Author path: active session must be the same claimant. Reconciler execute + # may rebind for the recorded claimant when explicitly authorized. + if role == "author": + if not active_identity or not active_profile: + reasons.append( + "active session identity/profile is unknown; author ownership " + "could not be proven" + ) + if locked_identity and active_identity and locked_identity != active_identity: + reasons.append( + f"lock claimant '{locked_identity}' does not match active " + f"identity '{active_identity}' (foreign claimant refused)" + ) + if locked_profile and active_profile and locked_profile != active_profile: + reasons.append( + f"lock profile '{locked_profile}' does not match active profile " + f"'{active_profile}' (profile mismatch refused)" + ) + if pin_identity and active_identity and pin_identity != active_identity: + reasons.append( + f"claimant_identity pin '{pin_identity}' does not match active " + f"identity '{active_identity}'" + ) + if pin_profile and active_profile and pin_profile != active_profile: + reasons.append( + f"claimant_profile pin '{pin_profile}' does not match active " + f"profile '{active_profile}'" + ) + + # ── Heads (must match pins and each other for this rebind class) ──────── + obs_local = _text(local_head) + obs_remote = _text(remote_head) + pin_local = _text(expected_local_head) + pin_remote = _text(expected_remote_head) + evidence["local_head"] = obs_local or None + evidence["remote_head"] = obs_remote or None + evidence["expected_local_head"] = pin_local or None + evidence["expected_remote_head"] = pin_remote or None + + if not pin_local or not pin_remote: + reasons.append( + "expected_local_head and expected_remote_head pins are required " + "(fail closed)" + ) + if not obs_local: + reasons.append("local head SHA could not be determined") + if not obs_remote: + reasons.append("remote head SHA could not be determined") + if pin_local and obs_local and pin_local != obs_local: + reasons.append( + f"local head moved or mismatched pin: observed {obs_local}, " + f"expected {pin_local}" + ) + if pin_remote and obs_remote and pin_remote != obs_remote: + reasons.append( + f"remote head moved or mismatched pin: observed {obs_remote}, " + f"expected {pin_remote}" + ) + if obs_local and obs_remote and obs_local != obs_remote: + # Dirty rebind does not allow unpublished head movement; heads must agree. + reasons.append( + f"local head {obs_local} does not match remote head {obs_remote}; " + "dirty same-claimant rebind requires matching heads (fail closed)" + ) + + # ── Dirty inventory + fingerprint pins ────────────────────────────────── + inv: dict[str, Any] + if isinstance(dirty_inventory, Mapping) and dirty_inventory.get("dirty_paths") is not None: + inv = dict(dirty_inventory) + if not inv.get("fingerprints") and porcelain_status is not None: + # Allow fingerprints-only refresh via recompute if needed. + pass + elif porcelain_status is not None: + # Porcelain alone proves path set, not bytes. Fingerprints must come from + # dirty_inventory (or apply()'s collect_dirty_inventory) — never from the + # caller's expected_fingerprints pin (that would make the pin tautological). + dirty_paths_obs = parse_dirty_paths(porcelain_status) + inv = { + "porcelain_status": porcelain_status, + "dirty_paths": dirty_paths_obs, + "fingerprints": {}, + "ok": True, + "reasons": [], + } + else: + reasons.append( + "neither dirty_inventory nor porcelain_status was provided; " + "dirty state could not be proven" + ) + inv = {"dirty_paths": [], "fingerprints": {}, "ok": False} + + if inv.get("ok") is False and inv.get("reasons"): + reasons.extend(list(inv.get("reasons") or [])) + + observed_paths = sorted({_text(p) for p in (inv.get("dirty_paths") or []) if _text(p)}) + pin_paths = sorted({_text(p) for p in (expected_dirty_paths or []) if _text(p)}) + evidence["observed_dirty_paths"] = observed_paths + evidence["expected_dirty_paths"] = pin_paths + + if not pin_paths: + reasons.append( + "expected_dirty_paths pin is empty; dirty same-claimant rebind " + "requires a non-empty dirty inventory pin (fail closed)" + ) + if set(observed_paths) != set(pin_paths): + extra = sorted(set(observed_paths) - set(pin_paths)) + missing_p = sorted(set(pin_paths) - set(observed_paths)) + if extra: + reasons.append( + f"dirty path set disagreement: unexpected paths {extra}" + ) + if missing_p: + reasons.append( + f"dirty path set disagreement: missing expected paths {missing_p}" + ) + + obs_fps = { + _text(k): _text(v) + for k, v in dict(inv.get("fingerprints") or {}).items() + if _text(k) + } + pin_fps = { + _text(k): _text(v) + for k, v in dict(expected_fingerprints or {}).items() + if _text(k) + } + evidence["observed_fingerprints"] = obs_fps + evidence["expected_fingerprints"] = pin_fps + + if not pin_fps: + reasons.append( + "expected_fingerprints pin is empty; byte-level pins are required " + "(fail closed)" + ) + else: + for rel, expected_hash in pin_fps.items(): + if rel not in set(pin_paths): + reasons.append( + f"expected_fingerprints contains '{rel}' which is not in " + "expected_dirty_paths" + ) + actual_hash = obs_fps.get(rel) + if not actual_hash: + reasons.append( + f"fingerprint missing for dirty path '{rel}'" + ) + elif actual_hash != expected_hash: + reasons.append( + f"fingerprint disagreement for '{rel}': observed " + f"{actual_hash}, expected {expected_hash}" + ) + for rel in obs_fps: + if rel in set(pin_paths) and rel not in pin_fps: + reasons.append( + f"expected_fingerprints missing pin for observed dirty path '{rel}'" + ) + + # ── Competing ownership ───────────────────────────────────────────────── + competing: list[dict[str, Any]] = [] + for entry in competing_live_locks or (): + if not isinstance(entry, Mapping): + continue + same_issue = entry.get("issue_number") == issue_number + same_branch = _text(entry.get("branch_name")) == locked_branch + if not (same_issue or same_branch): + continue + if ( + same_issue + and same_branch + and _same_realpath(_text(entry.get("worktree_path")), worktree_path) + ): + # The lock we are rebinding is not competition with itself, but a + # *live* competing owner on the same worktree is still a problem. + entry_pid = entry.get("pid") or entry.get("session_pid") + try: + entry_pid_i = int(entry_pid) if entry_pid is not None else None + except (TypeError, ValueError): + entry_pid_i = None + if entry_pid_i is not None and is_process_alive(entry_pid_i): + if old_pid is None or entry_pid_i != int(old_pid): + competing.append( + { + "issue_number": entry.get("issue_number"), + "branch_name": entry.get("branch_name"), + "worktree_path": entry.get("worktree_path"), + "pid": entry_pid_i, + } + ) + continue + competing.append( + { + "issue_number": entry.get("issue_number"), + "branch_name": entry.get("branch_name"), + "worktree_path": entry.get("worktree_path"), + "pid": entry.get("pid") or entry.get("session_pid"), + } + ) + if competing: + described = ", ".join( + f"issue #{c['issue_number']} branch '{c['branch_name']}' pid={c.get('pid')}" + for c in competing + ) + reasons.append(f"competing live lock exists ({described})") + evidence["competing_live_locks"] = competing + + competing_sess: list[dict[str, Any]] = [] + for entry in competing_sessions or (): + if not isinstance(entry, Mapping): + continue + sess_pid = entry.get("pid") or entry.get("session_pid") + try: + sess_pid_i = int(sess_pid) if sess_pid is not None else None + except (TypeError, ValueError): + sess_pid_i = None + if sess_pid_i is None: + continue + if current_pid is not None and sess_pid_i == int(current_pid): + continue + if old_pid is not None: + try: + if sess_pid_i == int(old_pid) and not is_process_alive(sess_pid_i): + continue + except (TypeError, ValueError): + pass + if is_process_alive(sess_pid_i) or entry.get("live") is True: + competing_sess.append( + { + "pid": sess_pid_i, + "lock_file_path": entry.get("lock_file_path"), + } + ) + if competing_sess: + reasons.append( + "competing live session pointer(s) claim this lock: " + + ", ".join(str(s["pid"]) for s in competing_sess) + ) + evidence["competing_sessions"] = competing_sess + + if workflow_lease_active: + reasons.append( + "workflow lease is active for this scope; dirty same-claimant " + "rebind refused (fail closed)" + ) + evidence["workflow_lease_active"] = bool(workflow_lease_active) + + if reasons: + return _assessment_result(REFUSED, False, reasons, evidence) + + proof = [ + f"registered dirty worktree for issue #{issue_number} on branch " + f"'{locked_branch}' matches claimant '{locked_identity}' / profile " + f"'{locked_profile}'; old_pid {recorded_pid} is dead; heads " + f"{obs_local} match; {len(pin_paths)} dirty paths fingerprint-pinned; " + "provenance-only rebind sanctioned" + ] + return _assessment_result(REBIND_SANCTIONED, True, proof, evidence) + + +def _assessment_result( + outcome: str, + sanctioned: bool, + reasons: list[str], + evidence: dict[str, Any], +) -> dict[str, Any]: + return { + "outcome": outcome, + "rebind_sanctioned": sanctioned, + "is_candidate": outcome != NO_CANDIDATE, + "reasons": reasons, + "evidence": evidence, + "expected_generation": evidence.get("lock_generation"), + } + + +def _already_rebound( + *, + existing_lock: Mapping[str, Any], + current_pid: int, + worktree_path: str, + expected_dirty_paths: Sequence[str] | None, + expected_fingerprints: Mapping[str, str], + worktree_for_fps: str, + remote: str, + org: str, + repo: str, + claimant_identity: str | None, + claimant_profile: str | None, + journal: Mapping[str, Any] | None = None, +) -> tuple[bool, list[str]]: + """Return (True, notes) when lock is already rebound to this session. + + #868: require complete matching operation identity (remote/org/repo/ + claimant) and complete dirty-inventory revalidation, not fingerprint-only + checks. Incomplete or mismatched journal identity fails closed. + """ + notes: list[str] = [] + pid = _recorded_pid(existing_lock) + try: + pid_i = int(pid) if pid is not None else None + except (TypeError, ValueError): + return False, [] + if pid_i != int(current_pid): + return False, [] + if not _same_realpath(_text(existing_lock.get("worktree_path")), worktree_path): + return False, [] + + # Durable lock repo binding must still match the caller's target. + for field, expected in (("remote", remote), ("org", org), ("repo", repo)): + observed = _text(existing_lock.get(field)) + want = _text(expected) + if observed and want and observed != want: + notes.append( + f"already_rebound refused: lock {field}={observed!r} does not " + f"match expected {want!r} (cross-repository replay)" + ) + return False, notes + + lock_claimant = _lock_claimant(existing_lock) + locked_identity = _text(lock_claimant.get("username")) + locked_profile = _text(lock_claimant.get("profile")) + pin_identity = _text(claimant_identity) + pin_profile = _text(claimant_profile) + if pin_identity and locked_identity and pin_identity != locked_identity: + notes.append( + f"already_rebound refused: lock claimant '{locked_identity}' does " + f"not match pin '{pin_identity}' (cross-claimant replay)" + ) + return False, notes + if pin_profile and locked_profile and pin_profile != locked_profile: + notes.append( + f"already_rebound refused: lock profile '{locked_profile}' does " + f"not match pin '{pin_profile}' (cross-claimant replay)" + ) + return False, notes + + # When a durable journal is present, require complete matching identity. + if isinstance(journal, Mapping) and journal: + id_reasons = validate_journal_operation_identity( + journal, + remote=remote, + org=org, + repo=repo, + claimant_identity=claimant_identity, + claimant_profile=claimant_profile, + require_present=True, + ) + if id_reasons: + notes.extend(id_reasons) + return False, notes + + inv_check = revalidate_complete_dirty_inventory( + worktree_for_fps, + expected_dirty_paths=expected_dirty_paths, + expected_fingerprints=expected_fingerprints, + phase="already_rebound", + ) + if not inv_check["ok"]: + notes.extend(list(inv_check["reasons"] or [])) + return False, notes + + gen = lock_generation(existing_lock) + if gen < 1: + # A never-written generation is suspicious for a completed rebind, but + # a same-pid lock with matching fingerprints is still "ours". + notes.append("lock generation is 0; treating same-pid match as rebound") + return True, notes or ["lock already bound to current session PID"] + + +def apply_dirty_same_claimant_session_rebind( + *, + remote: str, + org: str, + repo: str, + issue_number: int, + branch_name: str, + worktree_path: str, + claimant_identity: str | None, + claimant_profile: str | None, + old_pid: int | None, + expected_local_head: str | None, + expected_remote_head: str | None, + expected_dirty_paths: Sequence[str] | None, + expected_fingerprints: Mapping[str, str] | None, + existing_lock: Mapping[str, Any] | None, + current_identity: str | None, + current_profile: str | None, + role_kind: str | None, + current_pid: int | None = None, + current_branch: str | None, + local_head: str | None, + remote_head: str | None, + porcelain_status: str | None = None, + dirty_inventory: Mapping[str, Any] | None = None, + competing_live_locks: Sequence[Mapping[str, Any]] | None = None, + competing_sessions: Sequence[Mapping[str, Any]] | None = None, + workflow_lease_active: bool = False, + authorize_reconciler_execute: bool = False, + permission_allowed: bool = False, + repo_root: str | None = None, + dry_run: bool = False, + lock_dir: str | None = None, +) -> dict[str, Any]: + """Assess and (unless dry_run) apply a dirty same-claimant session rebind.""" + pid_now = int(current_pid) if current_pid is not None else os.getpid() + wt = os.path.realpath((worktree_path or "").strip()) if worktree_path else "" + + # Prefer a live inventory when applying so fingerprints are re-observed. + inv = dict(dirty_inventory) if isinstance(dirty_inventory, Mapping) else None + if inv is None and wt: + inv = collect_dirty_inventory(wt) + + assessment = assess_dirty_same_claimant_session_rebind( + remote=remote, + org=org, + repo=repo, + issue_number=issue_number, + branch_name=branch_name, + worktree_path=worktree_path, + claimant_identity=claimant_identity, + claimant_profile=claimant_profile, + old_pid=old_pid, + expected_local_head=expected_local_head, + expected_remote_head=expected_remote_head, + expected_dirty_paths=expected_dirty_paths, + expected_fingerprints=expected_fingerprints, + existing_lock=existing_lock, + current_identity=current_identity, + current_profile=current_profile, + role_kind=role_kind, + current_pid=pid_now, + current_branch=current_branch, + local_head=local_head, + remote_head=remote_head, + porcelain_status=porcelain_status + if porcelain_status is not None + else (inv or {}).get("porcelain_status"), + dirty_inventory=inv, + competing_live_locks=competing_live_locks, + competing_sessions=competing_sessions, + workflow_lease_active=workflow_lease_active, + authorize_reconciler_execute=authorize_reconciler_execute, + permission_allowed=permission_allowed, + repo_root=repo_root, + ) + + base_result: dict[str, Any] = { + "success": False, + "dry_run": bool(dry_run), + "outcome": assessment["outcome"], + "rebind_sanctioned": assessment["rebind_sanctioned"], + "reasons": list(assessment.get("reasons") or []), + "evidence": assessment.get("evidence") or {}, + "old_pid": old_pid, + "new_pid": pid_now, + "already_rebound": False, + "dirty_paths": list(expected_dirty_paths or []), + "fingerprints": dict(expected_fingerprints or {}), + "local_head": local_head, + "remote_head": remote_head, + } + + root_for_journal = (lock_dir or "").strip() or None + if root_for_journal is None and isinstance(existing_lock, Mapping): + root_for_journal = os.path.dirname( + _text(existing_lock.get("lock_file_path")) + or lock_file_path( + remote=remote, org=org, repo=repo, issue_number=issue_number + ) + ) + jpath_probe = ( + journal_path(root_for_journal, issue_number) if root_for_journal else "" + ) + existing_journal = _read_json(jpath_probe) if jpath_probe else None + + # Resume / retry: reject incomplete, mismatched, or cross-repo journal + # identity before treating any prior journal as authoritative (#868 F2). + if isinstance(existing_journal, Mapping) and existing_journal: + journal_id_reasons = validate_journal_operation_identity( + existing_journal, + remote=remote, + org=org, + repo=repo, + claimant_identity=claimant_identity, + claimant_profile=claimant_profile, + require_present=True, + ) + # Incomplete legacy journals from pre-#868 apply paths must fail closed + # when any identity field is missing — even if the rest of the payload + # looks familiar. Only a complete matching identity may proceed. + phase = _text(existing_journal.get("phase")) + if journal_id_reasons and phase not in ("", JOURNAL_PHASE_COMPLETE): + # Allow a completed journal with missing legacy identity only when + # already_rebound path will re-validate lock + inventory; for + # mid-flight incomplete journals, refuse. + if phase in ( + JOURNAL_PHASE_ASSESSED, + JOURNAL_PHASE_PRE_BIND, + JOURNAL_PHASE_BOUND, + "bind_failed", + ): + return { + **base_result, + "success": False, + "reasons": journal_id_reasons, + "journal_path": jpath_probe, + "journal_phase": phase or None, + } + + # Retry-safe: if already rebound to this session, succeed even when assess + # refuses because old_pid no longer matches the (updated) lock. + if ( + isinstance(existing_lock, Mapping) + and expected_fingerprints + and wt + ): + done, notes = _already_rebound( + existing_lock=existing_lock, + current_pid=pid_now, + worktree_path=worktree_path, + expected_dirty_paths=expected_dirty_paths, + expected_fingerprints=expected_fingerprints, + worktree_for_fps=wt, + remote=remote, + org=org, + repo=repo, + claimant_identity=claimant_identity, + claimant_profile=claimant_profile, + journal=existing_journal, + ) + if done: + lock_path = _text(existing_lock.get("lock_file_path")) or lock_file_path( + remote=remote, + org=org, + repo=repo, + issue_number=issue_number, + lock_dir=lock_dir, + ) + session_ptr = os.path.join( + (lock_dir or os.path.dirname(lock_path) or "."), + f"session-{pid_now}.json", + ) + return { + **base_result, + "success": True, + "outcome": REBIND_SANCTIONED, + "rebind_sanctioned": True, + "already_rebound": True, + "reasons": notes, + "lock_path": lock_path, + "session_pointer": session_ptr, + "generation_before": lock_generation(existing_lock), + "generation_after": lock_generation(existing_lock), + "journal_phase": JOURNAL_PHASE_ALREADY_REBOUND, + } + # Same-pid candidate that failed complete identity/inventory checks + # must not fall through into a fresh bind that would re-mint authority. + if _recorded_pid(existing_lock) is not None: + try: + if int(_recorded_pid(existing_lock)) == int(pid_now) and notes: + return { + **base_result, + "success": False, + "already_rebound": False, + "reasons": notes, + "journal_path": jpath_probe or None, + } + except (TypeError, ValueError): + pass + + if not assessment["rebind_sanctioned"]: + return base_result + + if dry_run: + return { + **base_result, + "success": True, + "message": "dry_run: rebind sanctioned; no lock/session writes performed", + "generation_before": assessment.get("expected_generation"), + "generation_after": assessment.get("expected_generation"), + } + + lock = dict(existing_lock or {}) + gen_before = lock_generation(lock) + root = (lock_dir or "").strip() or None + jpath = journal_path( + root or os.path.dirname( + _text(lock.get("lock_file_path")) + or lock_file_path( + remote=remote, org=org, repo=repo, issue_number=issue_number + ) + ), + issue_number, + ) + + op_identity = build_journal_operation_identity( + remote=remote, + org=org, + repo=repo, + claimant_identity=claimant_identity, + claimant_profile=claimant_profile, + ) + # Refuse incomplete caller identity before any durable write. + for field, value in op_identity.items(): + if not value: + return { + **base_result, + "success": False, + "reasons": [ + f"cannot write recovery journal: operation identity field " + f"'{field}' is empty (fail closed)" + ], + } + + journal = { + "phase": JOURNAL_PHASE_ASSESSED, + "issue_number": issue_number, + "branch_name": branch_name, + "worktree_path": wt, + "old_pid": old_pid, + "new_pid": pid_now, + "expected_generation": gen_before, + "expected_fingerprints": dict(expected_fingerprints or {}), + "expected_dirty_paths": list(expected_dirty_paths or []), + "local_head": local_head, + "remote_head": remote_head, + "started_at": _utc_now_iso(), + "source": SOURCE, + # #868 F2 — complete durable operation identity + **op_identity, + } + _atomic_write_json(jpath, journal) + + # #868 F1 — complete dirty-inventory revalidation immediately before mutation. + # Fail closed with no bind so failures cannot leave a newly authoritative + # live session. + pre_inv = revalidate_complete_dirty_inventory( + wt, + expected_dirty_paths=expected_dirty_paths, + expected_fingerprints=expected_fingerprints, + phase="pre-bind", + ) + if not pre_inv["ok"]: + journal["phase"] = "pre_bind_inventory_failed" + journal["pre_bind_inventory"] = { + "observed_dirty_paths": pre_inv.get("observed_dirty_paths"), + "reasons": pre_inv.get("reasons"), + } + _atomic_write_json(jpath, journal) + return { + **base_result, + "success": False, + "reasons": list(pre_inv["reasons"] or []), + "journal_path": jpath, + "journal_phase": "pre_bind_inventory_failed", + } + + pre_fps = dict(pre_inv.get("observed_fingerprints") or {}) + + journal["phase"] = JOURNAL_PHASE_PRE_BIND + journal["pre_bind_fingerprints"] = pre_fps + journal["pre_bind_dirty_paths"] = list(pre_inv.get("observed_dirty_paths") or []) + _atomic_write_json(jpath, journal) + + now = _utc_now_iso() + new_lock = dict(lock) + new_lock["session_pid"] = pid_now + new_lock["pid"] = pid_now + new_lock["last_heartbeat_at"] = now + new_lock["remote"] = remote + new_lock["org"] = org + new_lock["repo"] = repo + new_lock["issue_number"] = issue_number + new_lock["branch_name"] = branch_name + new_lock["worktree_path"] = _text(lock.get("worktree_path")) or wt + + # Preserve work_lease (including expires_at); refresh heartbeat only. + lease = new_lock.get("work_lease") + if isinstance(lease, dict): + lease = dict(lease) + lease["last_heartbeat_at"] = now + if not lease.get("operation_type"): + lease["operation_type"] = AUTHOR_ISSUE_WORK_LEASE + new_lock["work_lease"] = lease + + claimant = _lock_claimant(lock) + new_lock["lock_provenance"] = build_sanctioned_lock_provenance( + tool=SOURCE_TOOL, + source=SOURCE, + claimant=claimant or { + "username": claimant_identity, + "profile": claimant_profile, + }, + ) + new_lock["rebind_record"] = { + "source": SOURCE, + "old_pid": old_pid, + "new_pid": pid_now, + "rebound_at": now, + "local_head": local_head, + "remote_head": remote_head, + "dirty_path_count": len(list(expected_dirty_paths or [])), + "generation_before": gen_before, + "evidence": { + "fingerprints": dict(expected_fingerprints or {}), + "dirty_paths": list(expected_dirty_paths or []), + }, + } + + try: + lock_path = bind_session_lock( + new_lock, + lock_dir=root, + expected_generation=gen_before, + ) + except Exception as exc: + journal["phase"] = "bind_failed" + journal["error"] = str(exc) + _atomic_write_json(jpath, journal) + return { + **base_result, + "success": False, + "reasons": [f"bind_session_lock failed: {exc}"], + "journal_path": jpath, + "generation_before": gen_before, + } + + journal["phase"] = JOURNAL_PHASE_BOUND + journal["lock_path"] = lock_path + _atomic_write_json(jpath, journal) + + # #868 F1 — complete dirty-inventory revalidation immediately after mutation. + # Path set must remain exactly equal; fingerprints must be unchanged. + post_inv = revalidate_complete_dirty_inventory( + wt, + expected_dirty_paths=expected_dirty_paths, + expected_fingerprints=expected_fingerprints, + phase="post-bind", + ) + post_fps = dict(post_inv.get("observed_fingerprints") or {}) + if not post_inv["ok"]: + journal["phase"] = "post_bind_inventory_failed" + journal["post_bind_inventory"] = { + "observed_dirty_paths": post_inv.get("observed_dirty_paths"), + "reasons": post_inv.get("reasons"), + } + journal["post_bind_fingerprints"] = post_fps + _atomic_write_json(jpath, journal) + return { + **base_result, + "success": False, + "reasons": list(post_inv["reasons"] or []) + [ + "post-bind complete inventory revalidation failed after " + "bind_session_lock; lock may be rebound but content/path " + "verification failed (fail closed)" + ], + "lock_path": lock_path, + "journal_path": jpath, + "journal_phase": "post_bind_inventory_failed", + "generation_before": gen_before, + "fingerprints_after": post_fps, + } + + # Remove stale session pointer for old_pid when it points at this lock. + removed_old_pointer = False + if old_pid is not None and root: + old_ptr = os.path.join(root, f"session-{int(old_pid)}.json") + if os.path.exists(old_ptr): + ptr = _read_json(old_ptr) or {} + ptr_lock = _text(ptr.get("lock_file_path")) + if not ptr_lock or os.path.realpath(ptr_lock) == os.path.realpath(lock_path): + try: + os.remove(old_ptr) + removed_old_pointer = True + except OSError: + pass + + bound = read_lock_file(lock_path) or new_lock + gen_after = lock_generation(bound) + session_ptr = os.path.join( + root or os.path.dirname(lock_path), + f"session-{pid_now}.json", + ) + + journal["phase"] = JOURNAL_PHASE_COMPLETE + journal["completed_at"] = _utc_now_iso() + journal["generation_after"] = gen_after + journal["removed_old_session_pointer"] = removed_old_pointer + journal["post_bind_fingerprints"] = post_fps + journal["post_bind_dirty_paths"] = list( + post_inv.get("observed_dirty_paths") or [] + ) + _atomic_write_json(jpath, journal) + + return { + **base_result, + "success": True, + "message": ( + f"Rebound dirty same-claimant author session for issue #{issue_number} " + f"from dead pid {old_pid} to pid {pid_now}; dirty bytes preserved" + ), + "lock_path": lock_path, + "session_pointer": session_ptr, + "generation_before": gen_before, + "generation_after": gen_after, + "fingerprints": post_fps, + "fingerprints_before": pre_fps, + "removed_old_session_pointer": removed_old_pointer, + "journal_path": jpath, + "journal_phase": JOURNAL_PHASE_COMPLETE, + "rebind_record": bound.get("rebind_record") or new_lock.get("rebind_record"), + "lock_provenance": bound.get("lock_provenance"), + } + + +def build_issue_860_regression_fixture_spec() -> dict[str, Any]: + """Data-only fixture describing the #860 class scenario (no real mutation). + + Claimant jcwalker3 / prgs-author, dead PID, no live session pointer, seven + dirty paths with fingerprint pins, matching local/remote heads. + """ + dirty_paths = [ + "dirty_same_claimant_session_rebind.py", + "issue_lock_provenance.py", + "task_capability_map.py", + "gitea_mcp_server.py", + "tests/test_dirty_same_claimant_session_rebind.py", + "docs/runbook-dirty-rebind.md", + "scratch/notes-untracked.txt", + ] + # Stable placeholder digests — tests replace with real fingerprints when + # constructing on-disk fixtures. These exist so the spec is self-describing. + fingerprints = { + path: hashlib.sha256(f"issue-860-fixture:{path}".encode()).hexdigest() + for path in dirty_paths + } + head = "a" * 40 + dead = 424860 + return { + "issue_class": "issue-860-dirty-orphan-class-fixture", + "description": ( + "Registered dirty worktree, same claimant, dead owner PID, no live " + "session pointer, seven fingerprint-pinned dirty paths, matching heads. " + "Data only — does not mutate any real worktree." + ), + "remote": "prgs", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + "issue_number": 860, + "branch_name": "fix/issue-860-dirty-orphan-recovery", + "worktree_path": "/scratch/branches/fix-issue-860-dirty-orphan-recovery", + "claimant_identity": "jcwalker3", + "claimant_profile": "prgs-author", + "old_pid": dead, + "old_pid_alive": False, + "live_session_pointer": None, + "expected_local_head": head, + "expected_remote_head": head, + "expected_dirty_paths": dirty_paths, + "expected_fingerprints": fingerprints, + "dirty_path_count": 7, + "role_kind": "author", + "notes": [ + "Distinct from #864 apply path: this fixture documents the #860 class " + "inputs (dead PID + dirty inventory) without remote sync or recovery " + "worktree creation.", + ], + } diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index d059c25..8e1661a 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -440,13 +440,32 @@ def _session_author_lock_worktree() -> str | None: Used to derive the author mutation workspace when no explicit ``worktree_path`` or env binding is provided. Never invents a path. + + #864: a session pointer whose owner PID is dead and is not this process + must not force workspace binding for other issues — rebind is required for + that issue, and a stale dead-owner pointer must not poison unrelated work. """ try: lock = issue_lock_store.read_session_issue_lock() or {} except Exception: return None path = (lock.get("worktree_path") or "").strip() - return path or None + if not path: + return None + pid = lock.get("session_pid") + if pid is None: + pid = lock.get("pid") + try: + pid_i = int(pid) if pid is not None else None + except (TypeError, ValueError): + pid_i = None + if ( + pid_i is not None + and pid_i != os.getpid() + and not issue_lock_store.is_process_alive(pid_i) + ): + return None + return path def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: @@ -2031,6 +2050,7 @@ import issue_lock_store # noqa: E402 import issue_lock_adoption # noqa: E402 import issue_lock_recovery # noqa: E402 import issue_lock_renewal # noqa: E402 +import dirty_same_claimant_session_rebind # noqa: E402 # #864 import stacked_pr_support # noqa: E402 import merge_approval_gate # noqa: E402 import review_quarantine # noqa: E402 # #695 contaminated formal-review quarantine @@ -4342,6 +4362,263 @@ def gitea_lock_issue( return result +@mcp.tool() +def gitea_rebind_dirty_same_claimant_author_session( + issue_number: int, + branch_name: str, + worktree_path: str, + old_pid: int, + expected_local_head: str, + expected_remote_head: str, + expected_dirty_paths: list[str], + expected_fingerprints: dict, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + dry_run: bool = False, + authorize_reconciler_execute: bool = False, +) -> dict: + """Rebind a dirty registered issue worktree to this session (#864). + + Sanctioned only when every pin agrees: same claimant, dead old_pid matching + the durable lock, matching local/remote heads, exact dirty path set, and + per-path sha256 fingerprints. Preserves every tracked/untracked byte. + Does not sync remote, create recovery worktrees, clean, reset, or move heads. + + Role gate: + * author — must match the lock claimant identity/profile + * reconciler — execute only when ``authorize_reconciler_execute=True`` + * reviewer/merger — always refuse + + ``gitea.issue.comment`` (author map entry) is required for mutation; dry_run + still assesses fully but writes nothing. Permission alone is never ownership + proof — every pin is re-checked server-side. + + Args: + issue_number: Tracking issue number on the durable lock. + branch_name: Exact locked branch name. + worktree_path: Registered dirty worktree path (must be under branches/). + old_pid: Dead owner PID recorded on the lock (must match session_pid/pid). + expected_local_head: Full local HEAD sha the caller observed. + expected_remote_head: Full remote-tracking HEAD sha the caller observed. + expected_dirty_paths: Exact set of dirty relative paths (tracked+untracked). + expected_fingerprints: Map of relative path -> sha256 hex of file bytes. + remote: Known instance — 'dadeschools' or 'prgs'. + host/org/repo: Optional target overrides (validated against binding). + dry_run: When true, assess only (no lock/session writes). + authorize_reconciler_execute: Reconciler-only execute gate. + """ + role = _profile_role_kind(get_profile()) + role_norm = (role or "").strip().lower() + + # Permission: authors need comment; dry_run assess is reachable under read + # for diagnosis, but execute always needs comment. Reconciler execute also + # needs comment when authorized. + if dry_run: + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "dry_run": True, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + else: + blocked = _profile_permission_block( + task_capability_map.required_permission( + "rebind_dirty_same_claimant_author_session" + ), + issue_number=issue_number, + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) + if blocked: + return blocked + + if role_norm in {"reviewer", "merger"}: + return { + "success": False, + "dry_run": bool(dry_run), + "outcome": dirty_same_claimant_session_rebind.REFUSED, + "reasons": [ + f"role '{role_norm}' cannot rebind dirty same-claimant author " + "sessions (fail closed)" + ], + } + if role_norm == "reconciler" and not authorize_reconciler_execute and not dry_run: + return { + "success": False, + "dry_run": False, + "outcome": dirty_same_claimant_session_rebind.REFUSED, + "reasons": [ + "reconciler role requires authorize_reconciler_execute=True " + "to execute dirty same-claimant rebind (fail closed)" + ], + } + + h, o, r = _resolve(remote, host, org, repo) + try: + identity = _authenticated_username(h) + except Exception: + identity = None + profile = get_profile() + profile_name = profile.get("profile_name") + + existing = _load_existing_issue_lock( + remote=remote, org=o, repo=r, issue_number=issue_number + ) + resolved_wt = os.path.realpath(os.path.abspath((worktree_path or "").strip())) + inv = dirty_same_claimant_session_rebind.collect_dirty_inventory(resolved_wt) + + branch_res = subprocess.run( + ["git", "-C", resolved_wt, "branch", "--show-current"], + capture_output=True, + text=True, + check=False, + ) + current_branch = (branch_res.stdout or "").strip() or None + head_res = subprocess.run( + ["git", "-C", resolved_wt, "rev-parse", "HEAD"], + capture_output=True, + text=True, + check=False, + ) + local_head = (head_res.stdout or "").strip() if head_res.returncode == 0 else None + + # Observe remote-tracking head without network when possible. + remote_head = None + for ref in ( + f"refs/remotes/origin/{branch_name}", + f"origin/{branch_name}", + f"refs/remotes/{remote}/{branch_name}", + f"{remote}/{branch_name}", + ): + rh = subprocess.run( + ["git", "-C", resolved_wt, "rev-parse", "--verify", "--quiet", ref], + capture_output=True, + text=True, + check=False, + ) + if rh.returncode == 0 and (rh.stdout or "").strip(): + remote_head = (rh.stdout or "").strip() + break + if remote_head is None: + # Fall back to caller's pin only for observation absence — assessment + # still requires pin==observed, so missing observation fails closed. + remote_head = None + + # Competing live locks (other issues / other worktrees). + competing_live = [] + for entry in issue_lock_store.list_live_locks(): + competing_live.append(entry) + + # Session pointers that claim this issue lock. + competing_sessions = [] + lock_dir = issue_lock_store.default_lock_dir() + lock_path = issue_lock_store.lock_file_path( + remote=remote, org=o, repo=r, issue_number=issue_number, lock_dir=lock_dir + ) + try: + for name in os.listdir(lock_dir): + if not name.startswith("session-") or not name.endswith(".json"): + continue + ptr = issue_lock_store.read_lock_file(os.path.join(lock_dir, name)) + if not ptr: + continue + ptr_lock = str(ptr.get("lock_file_path") or "").strip() + if not ptr_lock: + continue + try: + same = os.path.realpath(ptr_lock) == os.path.realpath(lock_path) + except OSError: + same = ptr_lock == lock_path + if not same: + continue + try: + sess_pid = int(str(name)[len("session-") : -len(".json")]) + except ValueError: + sess_pid = ptr.get("pid") + competing_sessions.append( + { + "pid": sess_pid, + "lock_file_path": ptr_lock, + "live": issue_lock_store.is_process_alive(sess_pid), + } + ) + except OSError: + pass + + # Best-effort workflow-lease scan: any live lock file whose work_lease is a + # non-author workflow lease on this issue/branch counts as active. + workflow_lease_active = False + for path in issue_lock_store.iter_lock_files(lock_dir): + rec = issue_lock_store.read_lock_file(path) + if not rec: + continue + lease = rec.get("work_lease") if isinstance(rec.get("work_lease"), dict) else {} + op = str(lease.get("operation_type") or "") + if op and op != issue_lock_store.AUTHOR_ISSUE_WORK_LEASE: + if rec.get("issue_number") == issue_number or str( + rec.get("branch_name") or "" + ) == branch_name: + if issue_lock_store.is_lease_live(rec): + workflow_lease_active = True + break + + repo_root = _canonical_local_git_root() + # permission_allowed reflects profile gate only — never ownership proof. + permission_allowed = True + + result = dirty_same_claimant_session_rebind.apply_dirty_same_claimant_session_rebind( + remote=remote, + org=o, + repo=r, + issue_number=issue_number, + branch_name=branch_name, + worktree_path=resolved_wt, + claimant_identity=identity, + claimant_profile=profile_name, + old_pid=old_pid, + expected_local_head=expected_local_head, + expected_remote_head=expected_remote_head, + expected_dirty_paths=list(expected_dirty_paths or []), + expected_fingerprints=dict(expected_fingerprints or {}), + existing_lock=existing, + current_identity=identity, + current_profile=profile_name, + role_kind=role_norm or role, + current_pid=os.getpid(), + current_branch=current_branch, + local_head=local_head, + remote_head=remote_head, + dirty_inventory=inv, + competing_live_locks=competing_live, + competing_sessions=competing_sessions, + workflow_lease_active=workflow_lease_active, + authorize_reconciler_execute=bool(authorize_reconciler_execute), + permission_allowed=permission_allowed, + repo_root=repo_root, + dry_run=bool(dry_run), + lock_dir=lock_dir, + ) + result["observed"] = { + "local_head": local_head, + "remote_head": remote_head, + "current_branch": current_branch, + "dirty_paths": inv.get("dirty_paths"), + "fingerprints": inv.get("fingerprints"), + "identity": identity, + "profile": profile_name, + "role_kind": role_norm, + } + return result + + @mcp.tool() def gitea_assess_work_issue_duplicate( issue_number: int, @@ -11634,6 +11911,7 @@ def gitea_audit_worktree_cleanup( org: str | None = None, repo: str | None = None, ttl_hours: float = worktree_cleanup_audit.DEFAULT_TTL_HOURS, + merged_pr_limit: int = 200, ) -> dict: """Read-only: classify every session-owned worktree under ``branches/`` (#401). @@ -11644,17 +11922,26 @@ def gitea_audit_worktree_cleanup( the active issue-lock branch is read from the local lock file and treated as active work. Deletes nothing and mutates no Gitea state. - Fails closed if the live open-PR list cannot be fetched: without it, - removability cannot be proven, so no candidates are returned. + Merged PRs are fetched as well, so an issue worktree can be linked to the + PR that owns its branch (#858). Such a worktree only becomes removable + when that owning PR is unambiguous and merged, the worktree head is + already contained in authoritative master, and nothing else protects it — + no open or competing PR, lease, issue lock, live session, dirty file, or + protected/control checkout. Anything unproven keeps it classified as + active issue work. + + Fails closed if the live open-PR list, the merged-PR list, or the + control-plane lease state cannot be read: without them removability + cannot be proven, so no candidates are returned. Args: remote: Known instance — 'dadeschools' or 'prgs'. host: Override the Gitea host. org: Override the owner/organization. repo: Override the repository name. - ttl_hours: Age (hours) after which a clean issue/conflict-fix - worktree becomes stale-removable (default from - GITEA_WORKTREE_TTL_HOURS). + ttl_hours: Age (hours) after which a clean conflict-fix worktree + becomes stale-removable (default from GITEA_WORKTREE_TTL_HOURS). + merged_pr_limit: Max closed PRs scanned for merged-PR ownership. Returns: dict with per-worktree classifications, counts, removable @@ -11690,22 +11977,84 @@ def gitea_audit_worktree_cleanup( if (pr.get("head") or {}).get("ref") } + # #858: merged PRs are the ownership evidence that lets a landed issue + # worktree stop being reported as active work. Without them the audit can + # never agree with the PR-scoped reconciler, so treat a fetch failure the + # same way an open-PR fetch failure is treated: fail closed. + try: + closed_prs = api_get_all( + f"{repo_api_url(h, o, r)}/pulls?state=closed", auth, limit=merged_pr_limit + ) + except Exception as exc: + return { + "success": False, + "performed": False, + "open_pr_state_verified": True, + "merged_pr_state_verified": False, + "reasons": [ + "could not fetch merged PRs; worktree ownership unverified " + f"(fail closed): {_redact(str(exc))}" + ], + } + merged_prs = [pr for pr in closed_prs if (pr.get("merged") or pr.get("merged_at"))] + pr_index = worktree_cleanup_audit.build_pr_index(list(open_prs) + merged_prs) + + # #858: the auditor already accepted lease evidence but nothing ever + # supplied it, so every worktree looked unleased. Removability is now + # reachable for issue worktrees, so authoritative control-plane leases + # must be readable or the audit fails closed. + db, lease_errs = _control_plane_db_or_error() + if db is None: + return { + "success": False, + "performed": False, + "open_pr_state_verified": True, + "merged_pr_state_verified": True, + "lease_state_verified": False, + "reasons": [ + "could not read control-plane leases; worktree protection " + "unverified (fail closed)", + *lease_errs, + ], + } + lease_result = lease_lifecycle.list_active_leases( + db, remote=remote, org=o, repo=r, include_non_active=False, limit=500 + ) + leased_issue_numbers: set[int] = set() + live_session_paths: set[str] = set() + for lease in lease_result.get("leases") or []: + if lease.get("work_kind") == "issue" and lease.get("work_number") is not None: + try: + leased_issue_numbers.add(int(lease["work_number"])) + except (TypeError, ValueError): + pass + if lease.get("worktree_path"): + live_session_paths.add(str(lease["worktree_path"])) + active_issue_branches: set[str] = set() lock = merged_cleanup_reconcile.read_issue_lock(ISSUE_LOCK_FILE) if lock and lock.get("branch_name"): active_issue_branches.add(str(lock["branch_name"]).strip()) + master_ref = f"{remote}/master" if remote in REMOTES else "origin/master" report = worktree_cleanup_audit.audit_branches_directory( _canonical_local_git_root(), open_pr_branches=open_pr_branches, active_issue_branches=active_issue_branches, now=datetime.now(timezone.utc), ttl_hours=ttl_hours, + pr_index=pr_index, + leased_issue_numbers=leased_issue_numbers, + live_session_paths=live_session_paths, + master_ref=master_ref, ) return { "success": True, "performed": False, "open_pr_state_verified": True, + "merged_pr_state_verified": True, + "lease_state_verified": True, + "master_ref": master_ref, "task_mode": "work-issue", **report, } diff --git a/issue_lock_provenance.py b/issue_lock_provenance.py index 87ee38f..544e017 100644 --- a/issue_lock_provenance.py +++ b/issue_lock_provenance.py @@ -16,11 +16,16 @@ ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock SOURCE_LOCK_ISSUE = "gitea_lock_issue" SOURCE_LOCK_ADOPTION = "gitea_lock_issue_adoption" SOURCE_OPERATOR_OVERRIDE = "operator_override" +# #864: dirty-preserving same-claimant author-session rebind (dead owner PID). +SOURCE_DIRTY_SAME_CLAIMANT_REBIND = ( + "gitea_rebind_dirty_same_claimant_author_session" +) SANCTIONED_LOCK_SOURCES = frozenset({ SOURCE_LOCK_ISSUE, SOURCE_LOCK_ADOPTION, SOURCE_OPERATOR_OVERRIDE, + SOURCE_DIRTY_SAME_CLAIMANT_REBIND, }) _OPERATOR_OVERRIDE_ENV = "GITEA_ISSUE_LOCK_OPERATOR_OVERRIDE" diff --git a/task_capability_map.py b/task_capability_map.py index 0b8ac0b..7d7b76b 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -32,6 +32,17 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.issue.comment", "role": "author", }, + # #864: dirty-preserving same-claimant author-session rebind (dead owner PID). + # Author MCP tool path. Reconciler execute is gated inside the tool via + # authorize_reconciler_execute + role_kind checks (not this map entry). + "rebind_dirty_same_claimant_author_session": { + "permission": "gitea.issue.comment", + "role": "author", + }, + "gitea_rebind_dirty_same_claimant_author_session": { + "permission": "gitea.issue.comment", + "role": "author", + }, "set_issue_labels": { "permission": "gitea.issue.comment", "role": "author", diff --git a/tests/test_dirty_same_claimant_session_rebind.py b/tests/test_dirty_same_claimant_session_rebind.py new file mode 100644 index 0000000..c40933e --- /dev/null +++ b/tests/test_dirty_same_claimant_session_rebind.py @@ -0,0 +1,1346 @@ +"""Integration tests for dirty same-claimant author-session rebind (#864 / #868). + +Uses real temp git repos/worktrees and a temp GITEA_ISSUE_LOCK_DIR. Does not +mutate any real #860/#864/#868 worktree on disk. + +#868 adds complete dirty-inventory revalidation around bind_session_lock and +complete recovery-journal operation identity (remote/org/repo/claimant). +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import dirty_same_claimant_session_rebind as rebind # noqa: E402 +import issue_lock_provenance # noqa: E402 +import issue_lock_store as ils # noqa: E402 +import issue_lock_worktree # noqa: E402 + +ISSUE = 864 +BRANCH = f"fix/issue-{ISSUE}-dirty-same-claimant-session-rebind" +REMOTE = "prgs" +ORG = "Scaled-Tech-Consulting" +REPO = "Gitea-Tools" +IDENTITY = "jcwalker3" +PROFILE = "prgs-author" + + +def _git(cwd: str, *args: str, check: bool = True) -> subprocess.CompletedProcess: + return subprocess.run( + ["git", "-C", cwd, *args], + capture_output=True, + text=True, + check=check, + ) + + +def dead_pid() -> int: + proc = subprocess.Popen([sys.executable, "-c", "pass"]) + proc.wait() + return proc.pid + + +def future_ts(hours: int = 4) -> str: + return ( + (datetime.now(timezone.utc) + timedelta(hours=hours)) + .isoformat() + .replace("+00:00", "Z") + ) + + +@pytest.fixture +def lock_dir(tmp_path, monkeypatch): + d = tmp_path / "issue-locks" + d.mkdir() + monkeypatch.setenv("GITEA_ISSUE_LOCK_DIR", str(d)) + return str(d) + + +@pytest.fixture +def dirty_repo(tmp_path): + """Canonical repo root with branches/ worktree and dirty content.""" + root = tmp_path / "repo" + root.mkdir() + main = root / "main" + main.mkdir() + subprocess.run(["git", "init", "-q", str(main)], check=True, capture_output=True) + _git(str(main), "config", "user.email", "t@t") + _git(str(main), "config", "user.name", "t") + (main / "README.md").write_text("base\n", encoding="utf-8") + _git(str(main), "add", "README.md") + _git(str(main), "commit", "-q", "-m", "base") + _git(str(main), "branch", "-M", "master") + + # Bare remote + origin tracking so remote head is observable offline. + bare = tmp_path / "remote.git" + subprocess.run( + ["git", "init", "--bare", "-q", str(bare)], check=True, capture_output=True + ) + _git(str(main), "remote", "add", "origin", str(bare)) + _git(str(main), "push", "-q", "origin", "master:master") + + branches = root / "branches" + branches.mkdir() + wt_name = f"fix-issue-{ISSUE}-dirty-same-claimant-session-rebind" + wt = branches / wt_name + _git(str(main), "worktree", "add", "-q", "-b", BRANCH, str(wt)) + _git(str(wt), "push", "-q", "-u", "origin", BRANCH) + + # Seed committed files we will dirty. + tracked = [ + "dirty_same_claimant_session_rebind.py", + "issue_lock_provenance.py", + "task_capability_map.py", + ] + for rel in tracked: + p = wt / rel + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(f"seed {rel}\n", encoding="utf-8") + _git(str(wt), "add", *tracked) + _git(str(wt), "commit", "-q", "-m", "seed tracked") + _git(str(wt), "push", "-q", "origin", BRANCH) + + # Dirty tracked + untracked. + for rel in tracked: + (wt / rel).write_text(f"dirty {rel}\n", encoding="utf-8") + untracked = [ + "tests/test_dirty_same_claimant_session_rebind.py", + "docs/runbook-dirty-rebind.md", + "scratch/notes-untracked.txt", + "extra_untracked.txt", + ] + for rel in untracked: + p = wt / rel + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(f"untracked {rel}\n", encoding="utf-8") + + inv = rebind.collect_dirty_inventory(str(wt)) + assert inv["ok"], inv.get("reasons") + head = _git(str(wt), "rev-parse", "HEAD").stdout.strip() + remote_head = _git( + str(wt), "rev-parse", f"refs/remotes/origin/{BRANCH}" + ).stdout.strip() + assert head == remote_head + + return { + "root": str(root), + "main": str(main), + "worktree": str(wt), + "branch": BRANCH, + "inventory": inv, + "local_head": head, + "remote_head": remote_head, + "dirty_paths": list(inv["dirty_paths"]), + "fingerprints": dict(inv["fingerprints"]), + } + + +def _make_lock( + *, + worktree: str, + pid: int, + lock_dir: str, + identity: str = IDENTITY, + profile: str = PROFILE, + **overrides, +) -> dict: + lease = { + "operation_type": ils.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": ISSUE, + "branch": BRANCH, + "worktree_path": worktree, + "claimant": {"username": identity, "profile": profile}, + "created_at": "2026-01-01T00:00:00Z", + "expires_at": future_ts(), + "last_heartbeat_at": "2026-01-01T00:00:00Z", + } + lock = { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": worktree, + "remote": REMOTE, + "org": ORG, + "repo": REPO, + "session_pid": pid, + "pid": pid, + "work_lease": lease, + "lock_generation": 1, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant={"username": identity, "profile": profile}, + ), + } + lock.update(overrides) + path = ils.lock_file_path( + remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, lock_dir=lock_dir + ) + lock["lock_file_path"] = path + ils.save_lock_file(path, lock) + # Stale session pointer for the dead owner. + ptr = { + "pid": pid, + "lock_file_path": path, + "issue_number": ISSUE, + "branch_name": BRANCH, + "remote": REMOTE, + "org": ORG, + "repo": REPO, + } + ils.save_lock_file(os.path.join(lock_dir, f"session-{pid}.json"), ptr) + return ils.read_lock_file(path) or lock + + +def _apply_kwargs(repo, lock, lock_dir, **overrides): + kwargs = { + "remote": REMOTE, + "org": ORG, + "repo": REPO, + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": repo["worktree"], + "claimant_identity": IDENTITY, + "claimant_profile": PROFILE, + "old_pid": lock.get("session_pid") or lock.get("pid"), + "expected_local_head": repo["local_head"], + "expected_remote_head": repo["remote_head"], + "expected_dirty_paths": repo["dirty_paths"], + "expected_fingerprints": repo["fingerprints"], + "existing_lock": lock, + "current_identity": IDENTITY, + "current_profile": PROFILE, + "role_kind": "author", + "current_pid": os.getpid(), + "current_branch": BRANCH, + "local_head": repo["local_head"], + "remote_head": repo["remote_head"], + "dirty_inventory": repo["inventory"], + "competing_live_locks": [], + "competing_sessions": [], + "workflow_lease_active": False, + "repo_root": repo["root"], + "dry_run": False, + "lock_dir": lock_dir, + } + kwargs.update(overrides) + return kwargs + + +# ── 1. Successful dead-PID same-claimant dirty rebind ─────────────────────── + + +def test_successful_dead_pid_same_claimant_dirty_rebind(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert result["success"], result + assert result["outcome"] == rebind.REBIND_SANCTIONED + assert result["old_pid"] == old + assert result["new_pid"] == os.getpid() + assert result["generation_after"] == result["generation_before"] + 1 + + rebound = ils.read_lock_file(result["lock_path"]) + assert rebound is not None + assert int(rebound["session_pid"]) == os.getpid() + assert int(rebound["pid"]) == os.getpid() + assert ( + rebound.get("lock_provenance", {}).get("source") + == issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND + ) + assert rebound.get("rebind_record", {}).get("old_pid") == old + + +# ── 2. Byte-for-byte preservation ─────────────────────────────────────────── + + +def test_byte_for_byte_preservation(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + before = { + rel: rebind.content_fingerprint(os.path.join(dirty_repo["worktree"], rel)) + for rel in dirty_repo["dirty_paths"] + } + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert result["success"], result + after = { + rel: rebind.content_fingerprint(os.path.join(dirty_repo["worktree"], rel)) + for rel in dirty_repo["dirty_paths"] + } + assert before == after + assert result["fingerprints"] == before + + +# ── 3. Exact dirty-path and fingerprint enforcement ───────────────────────── + + +def test_extra_dirty_path_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + pins = list(dirty_repo["dirty_paths"])[:-1] # missing one observed path + fps = {p: dirty_repo["fingerprints"][p] for p in pins} + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + expected_dirty_paths=pins, + expected_fingerprints=fps, + ) + ) + assert not result["success"] + assert any("unexpected paths" in r for r in result["reasons"]) + + +def test_missing_expected_dirty_path_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + pins = list(dirty_repo["dirty_paths"]) + ["not_really_dirty.txt"] + fps = dict(dirty_repo["fingerprints"]) + fps["not_really_dirty.txt"] = "0" * 64 + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + expected_dirty_paths=pins, + expected_fingerprints=fps, + ) + ) + assert not result["success"] + assert any("missing expected" in r for r in result["reasons"]) + + +def test_modified_fingerprint_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + fps = dict(dirty_repo["fingerprints"]) + victim = dirty_repo["dirty_paths"][0] + fps[victim] = "f" * 64 + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + expected_fingerprints=fps, + ) + ) + assert not result["success"] + assert any("fingerprint disagreement" in r for r in result["reasons"]) + + +# ── 4. Atomic session-pointer replacement ─────────────────────────────────── + + +def test_session_pointer_points_to_lock(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert result["success"], result + new_ptr_path = os.path.join(lock_dir, f"session-{os.getpid()}.json") + assert os.path.exists(new_ptr_path) + ptr = ils.read_lock_file(new_ptr_path) + assert ptr is not None + assert os.path.realpath(ptr["lock_file_path"]) == os.path.realpath( + result["lock_path"] + ) + # Old pointer removed when it targeted this lock. + old_ptr = os.path.join(lock_dir, f"session-{old}.json") + assert not os.path.exists(old_ptr) + assert result.get("removed_old_session_pointer") is True + + +# ── 5. Retry after interruption (journal mid-state) ───────────────────────── + + +def test_retry_after_journal_mid_state(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + jpath = rebind.journal_path(lock_dir, ISSUE) + # Complete operation identity required for mid-flight resume (#868 F2). + rebind._atomic_write_json( + jpath, + { + "phase": rebind.JOURNAL_PHASE_PRE_BIND, + "issue_number": ISSUE, + "old_pid": old, + "new_pid": os.getpid(), + "expected_generation": 1, + "remote": REMOTE, + "org": ORG, + "repo": REPO, + "claimant_identity": IDENTITY, + "claimant_profile": PROFILE, + "source": rebind.SOURCE, + }, + ) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert result["success"], result + assert result["journal_phase"] == rebind.JOURNAL_PHASE_COMPLETE + + # Second apply is already_rebound (retry-safe). + rebound_lock = ils.read_lock_file(result["lock_path"]) + result2 = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + rebound_lock, + lock_dir, + old_pid=old, + existing_lock=rebound_lock, + ) + ) + assert result2["success"], result2 + assert result2["already_rebound"] is True + + +# ── 6. Active-PID refusal ─────────────────────────────────────────────────── + + +def test_active_pid_refused(dirty_repo, lock_dir): + live = os.getpid() + # Use a different "current" identity of session via fake current_pid... + # Owner is live (this process). Rebind must refuse. + lock = _make_lock(worktree=dirty_repo["worktree"], pid=live, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=live, + current_pid=live + 10_000_000, # distinct "new" session id for pin check + ) + ) + assert not result["success"] + assert any("still alive" in r for r in result["reasons"]) + + +# ── 7. Foreign claimant refusal ───────────────────────────────────────────── + + +def test_foreign_claimant_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + current_identity="someone-else", + claimant_identity="someone-else", + ) + ) + assert not result["success"] + assert any("foreign claimant" in r or "does not match" in r for r in result["reasons"]) + + +# ── 8. Profile mismatch refusal ───────────────────────────────────────────── + + +def test_profile_mismatch_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + current_profile="other-profile", + claimant_profile="other-profile", + ) + ) + assert not result["success"] + assert any("profile" in r for r in result["reasons"]) + + +# ── 9. Competing session/lock/lease refusal ───────────────────────────────── + + +def test_competing_live_lock_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + competing = [ + { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": dirty_repo["worktree"] + "-other", + "pid": os.getpid(), + } + ] + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + competing_live_locks=competing, + ) + ) + assert not result["success"] + assert any("competing live lock" in r for r in result["reasons"]) + + +def test_competing_session_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + competing_sessions=[ + {"pid": os.getpid(), "lock_file_path": lock["lock_file_path"], "live": True} + ], + ) + ) + # current_pid is os.getpid(), so same session is skipped — use another live pid. + # Spawn a long-lived process to act as competing live session. + rival = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) + try: + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + competing_sessions=[ + { + "pid": rival.pid, + "lock_file_path": lock["lock_file_path"], + "live": True, + } + ], + ) + ) + assert not result["success"] + assert any("competing live session" in r for r in result["reasons"]) + finally: + rival.kill() + rival.wait() + + +def test_workflow_lease_active_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + workflow_lease_active=True, + ) + ) + assert not result["success"] + assert any("workflow lease" in r for r in result["reasons"]) + + +# ── 10. Local- and remote-head movement refusal ───────────────────────────── + + +def test_local_head_movement_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + expected_local_head="b" * 40, + ) + ) + assert not result["success"] + assert any("local head" in r for r in result["reasons"]) + + +def test_remote_head_movement_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + expected_remote_head="c" * 40, + ) + ) + assert not result["success"] + assert any("remote head" in r for r in result["reasons"]) + + +# ── 11. Path/symlink/registration mismatches ──────────────────────────────── + + +def test_worktree_not_under_branches_refused(dirty_repo, lock_dir, tmp_path): + old = dead_pid() + # Use a path outside branches/ as the declared worktree (still real dir). + outside = tmp_path / "outside-wt" + outside.mkdir() + lock = _make_lock(worktree=str(outside), pid=old, lock_dir=lock_dir) + # Inventory empty for outside path; use empty pins to hit path gate first + # by providing matching empty-ish inventory after we force path checks. + inv = { + "dirty_paths": dirty_repo["dirty_paths"], + "fingerprints": dirty_repo["fingerprints"], + "ok": True, + "reasons": [], + } + result = rebind.assess_dirty_same_claimant_session_rebind( + remote=REMOTE, + org=ORG, + repo=REPO, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=str(outside), + claimant_identity=IDENTITY, + claimant_profile=PROFILE, + old_pid=old, + expected_local_head=dirty_repo["local_head"], + expected_remote_head=dirty_repo["remote_head"], + expected_dirty_paths=dirty_repo["dirty_paths"], + expected_fingerprints=dirty_repo["fingerprints"], + existing_lock=lock, + current_identity=IDENTITY, + current_profile=PROFILE, + role_kind="author", + current_pid=os.getpid(), + current_branch=BRANCH, + local_head=dirty_repo["local_head"], + remote_head=dirty_repo["remote_head"], + dirty_inventory=inv, + repo_root=dirty_repo["root"], + ) + assert not result["rebind_sanctioned"] + assert any("branches/" in r for r in result["reasons"]) + + +def test_lock_worktree_mismatch_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock( + worktree=dirty_repo["worktree"] + "-elsewhere", + pid=old, + lock_dir=lock_dir, + ) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert not result["success"] + assert any("does not match declared" in r for r in result["reasons"]) + + +# ── 12. Malformed lock/session records ────────────────────────────────────── + + +def test_malformed_lock_missing_pid_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + lock.pop("session_pid", None) + lock.pop("pid", None) + ils.save_lock_file(lock["lock_file_path"], lock) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert not result["success"] + assert any("incomplete" in r or "session_pid" in r for r in result["reasons"]) + + +def test_empty_old_pid_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=None) + ) + assert not result["success"] + assert any("old_pid" in r for r in result["reasons"]) + + +def test_reviewer_role_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, role_kind="reviewer") + ) + assert not result["success"] + assert any("reviewer" in r for r in result["reasons"]) + + +def test_reconciler_without_authorize_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + role_kind="reconciler", + authorize_reconciler_execute=False, + ) + ) + assert not result["success"] + assert any("authorize_reconciler_execute" in r for r in result["reasons"]) + + +# ── 13. No duplicate ownership after success or retry ─────────────────────── + + +def test_no_duplicate_ownership_after_success_or_retry(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + r1 = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert r1["success"], r1 + rebound = ils.read_lock_file(r1["lock_path"]) + r2 = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, rebound, lock_dir, old_pid=old, existing_lock=rebound) + ) + assert r2["success"], r2 + assert r2["already_rebound"] is True + + # Only one durable lock file for this issue; session pointer is current pid. + # Skip session pointers and rebind journals (dotfiles / non-lock records). + matching = [] + for p in ils.iter_lock_files(lock_dir): + name = os.path.basename(p) + if name.startswith(".") or name.startswith("session-"): + continue + rec = ils.read_lock_file(p) + if not rec: + continue + if ( + rec.get("issue_number") == ISSUE + and rec.get("remote") == REMOTE + and rec.get("branch_name") == BRANCH + and rec.get("session_pid") is not None + ): + matching.append(rec) + assert len(matching) == 1 + assert int(matching[0]["session_pid"]) == os.getpid() + # No live session pointer for the dead old pid. + assert not os.path.exists(os.path.join(lock_dir, f"session-{old}.json")) + + +# ── 14. Ordinary dirty-worktree locking remains fail-closed ───────────────── + + +def test_ordinary_dirty_lock_worktree_assessment_blocks(dirty_repo): + porcelain = dirty_repo["inventory"]["porcelain_status"] + assessment = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path=dirty_repo["worktree"], + current_branch=BRANCH, + porcelain_status=porcelain, + base_equivalent=False, + ) + assert assessment["block"] is True + assert any( + "tracked file edits exist before issue lock" in r + for r in assessment["reasons"] + ) + + +# ── 15. Fixture matching #860 class with 7 fingerprint-pinned dirty paths ─── + + +def test_issue_860_regression_fixture_spec(): + spec = rebind.build_issue_860_regression_fixture_spec() + assert spec["claimant_identity"] == "jcwalker3" + assert spec["claimant_profile"] == "prgs-author" + assert spec["old_pid_alive"] is False + assert spec["live_session_pointer"] is None + assert spec["dirty_path_count"] == 7 + assert len(spec["expected_dirty_paths"]) == 7 + assert len(spec["expected_fingerprints"]) == 7 + assert spec["expected_local_head"] == spec["expected_remote_head"] + for path in spec["expected_dirty_paths"]: + assert path in spec["expected_fingerprints"] + assert len(spec["expected_fingerprints"][path]) == 64 + + +def test_dry_run_does_not_write(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + before = ils.read_lock_file(lock["lock_file_path"]) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, dry_run=True) + ) + assert result["success"], result + assert result["dry_run"] is True + after = ils.read_lock_file(lock["lock_file_path"]) + assert after["session_pid"] == before["session_pid"] + assert not os.path.exists(os.path.join(lock_dir, f"session-{os.getpid()}.json")) + + +def test_provenance_source_is_sanctioned(): + assert ( + issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND + in issue_lock_provenance.SANCTIONED_LOCK_SOURCES + ) + assessment = issue_lock_provenance.assess_lock_file_for_create_pr( + { + "work_lease": {"operation_type": "author_issue_work"}, + "lock_provenance": { + "source": issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND, + "written_by_tool": issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND, + "written_at": "2026-01-01T00:00:00Z", + }, + } + ) + assert assessment["proven"] is True + + +def test_permission_allowed_is_not_ownership_proof(dirty_repo, lock_dir): + """permission_allowed=True must not bypass foreign claimant refusal.""" + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.assess_dirty_same_claimant_session_rebind( + remote=REMOTE, + org=ORG, + repo=REPO, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=dirty_repo["worktree"], + claimant_identity=IDENTITY, + claimant_profile=PROFILE, + old_pid=old, + expected_local_head=dirty_repo["local_head"], + expected_remote_head=dirty_repo["remote_head"], + expected_dirty_paths=dirty_repo["dirty_paths"], + expected_fingerprints=dirty_repo["fingerprints"], + existing_lock=lock, + current_identity="intruder", + current_profile=PROFILE, + role_kind="author", + current_pid=os.getpid(), + current_branch=BRANCH, + local_head=dirty_repo["local_head"], + remote_head=dirty_repo["remote_head"], + dirty_inventory=dirty_repo["inventory"], + permission_allowed=True, + repo_root=dirty_repo["root"], + ) + assert not result["rebind_sanctioned"] + assert any("does not match active identity" in r for r in result["reasons"]) + + +def test_content_fingerprint_stable(tmp_path): + p = tmp_path / "f.txt" + p.write_bytes(b"abc123") + a = rebind.content_fingerprint(str(p)) + b = rebind.content_fingerprint(str(p)) + assert a == b + assert len(a) == 64 + + +# ── #868 F1 — Complete dirty-inventory revalidation ───────────────────────── + + +def test_extra_tracked_dirty_path_before_bind_refused(dirty_repo, lock_dir): + """Extra tracked dirty path appearing immediately before binding fails closed.""" + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + wt = dirty_repo["worktree"] + # Seed a tracked file, then dirty it without including it in the pin set. + tracked_extra = "tracked_extra_before_bind.txt" + path = Path(wt) / tracked_extra + path.write_text("seed tracked extra\n", encoding="utf-8") + _git(wt, "add", tracked_extra) + _git(wt, "commit", "-q", "-m", "seed extra tracked") + # Heads moved — re-pin heads so only inventory disagreement is tested. + head = _git(wt, "rev-parse", "HEAD").stdout.strip() + _git(wt, "push", "-q", "origin", BRANCH) + remote_head = _git(wt, "rev-parse", f"refs/remotes/origin/{BRANCH}").stdout.strip() + path.write_text("dirty tracked extra\n", encoding="utf-8") + # Pins still describe the original inventory (without tracked_extra). + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + expected_local_head=head, + expected_remote_head=remote_head, + local_head=head, + remote_head=remote_head, + dirty_inventory=None, # force live recollect in apply + ) + ) + assert not result["success"] + joined = " ".join(result["reasons"]) + assert "unexpected paths" in joined or "path-set disagreement" in joined + # Must not leave a newly authoritative live session for this pid. + rebound = ils.read_lock_file(lock["lock_file_path"]) + assert rebound is not None + assert int(rebound.get("session_pid") or 0) == old + + +def test_extra_untracked_path_before_bind_refused(dirty_repo, lock_dir): + """Extra untracked path appearing immediately before binding fails closed.""" + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + wt = dirty_repo["worktree"] + extra = Path(wt) / "surprise_untracked_before_bind.txt" + extra.write_text("sneaky\n", encoding="utf-8") + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + dirty_inventory=None, + ) + ) + assert not result["success"] + joined = " ".join(result["reasons"]) + assert "unexpected paths" in joined or "path-set disagreement" in joined + rebound = ils.read_lock_file(lock["lock_file_path"]) + assert int(rebound.get("session_pid") or 0) == old + + +def test_path_added_during_mutation_window_refused(dirty_repo, lock_dir, monkeypatch): + """Path added during the mutation window is detected by post-bind inventory.""" + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + wt = dirty_repo["worktree"] + real_bind = ils.bind_session_lock + + def _bind_then_add_path(lock_payload, **kwargs): + path = real_bind(lock_payload, **kwargs) + surprise = Path(wt) / "added_during_bind.txt" + surprise.write_text("during bind\n", encoding="utf-8") + return path + + monkeypatch.setattr(rebind, "bind_session_lock", _bind_then_add_path) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, dirty_inventory=None) + ) + assert not result["success"] + joined = " ".join(result["reasons"]) + assert "post-bind" in joined + assert "unexpected paths" in joined or "path-set disagreement" in joined + assert result.get("journal_phase") == "post_bind_inventory_failed" + + +def test_path_removed_during_mutation_window_refused(dirty_repo, lock_dir, monkeypatch): + """Path removed during the mutation window is detected by post-bind inventory.""" + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + wt = dirty_repo["worktree"] + victim = dirty_repo["dirty_paths"][-1] # prefer untracked for easy remove + real_bind = ils.bind_session_lock + + def _bind_then_remove_path(lock_payload, **kwargs): + path = real_bind(lock_payload, **kwargs) + target = Path(wt) / victim + if target.exists(): + target.unlink() + return path + + monkeypatch.setattr(rebind, "bind_session_lock", _bind_then_remove_path) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, dirty_inventory=None) + ) + assert not result["success"] + joined = " ".join(result["reasons"]) + assert "post-bind" in joined + assert "missing expected" in joined or "path-set disagreement" in joined + + +def test_fingerprint_movement_unchanged_path_set_refused(dirty_repo, lock_dir, monkeypatch): + """Fingerprint movement with unchanged path set fails pre- or post-bind check.""" + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + wt = dirty_repo["worktree"] + victim = dirty_repo["dirty_paths"][0] + real_bind = ils.bind_session_lock + + def _bind_then_mutate_bytes(lock_payload, **kwargs): + path = real_bind(lock_payload, **kwargs) + target = Path(wt) / victim + target.write_text(target.read_text(encoding="utf-8") + "mutated\n", encoding="utf-8") + return path + + monkeypatch.setattr(rebind, "bind_session_lock", _bind_then_mutate_bytes) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, dirty_inventory=None) + ) + assert not result["success"] + joined = " ".join(result["reasons"]) + assert "fingerprint" in joined + assert "post-bind" in joined + + +# ── #868 F2 — Complete recovery-journal identity ──────────────────────────── + + +def _complete_journal(**overrides): + base = { + "phase": rebind.JOURNAL_PHASE_ASSESSED, + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": "/tmp/wt", + "old_pid": 1, + "new_pid": os.getpid(), + "remote": REMOTE, + "org": ORG, + "repo": REPO, + "claimant_identity": IDENTITY, + "claimant_profile": PROFILE, + "source": rebind.SOURCE, + } + base.update(overrides) + return base + + +def test_journal_remote_mismatch_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + jpath = rebind.journal_path(lock_dir, ISSUE) + rebind._atomic_write_json( + jpath, + _complete_journal( + phase=rebind.JOURNAL_PHASE_PRE_BIND, + old_pid=old, + remote="dadeschools", + ), + ) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert not result["success"] + assert any("remote" in r and "mismatch" in r for r in result["reasons"]) + + +def test_journal_org_mismatch_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + jpath = rebind.journal_path(lock_dir, ISSUE) + rebind._atomic_write_json( + jpath, + _complete_journal( + phase=rebind.JOURNAL_PHASE_PRE_BIND, + old_pid=old, + org="Other-Org", + ), + ) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert not result["success"] + assert any("org" in r and "mismatch" in r for r in result["reasons"]) + + +def test_journal_repo_mismatch_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + jpath = rebind.journal_path(lock_dir, ISSUE) + rebind._atomic_write_json( + jpath, + _complete_journal( + phase=rebind.JOURNAL_PHASE_PRE_BIND, + old_pid=old, + repo="Other-Repo", + ), + ) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert not result["success"] + assert any("repo" in r and "mismatch" in r for r in result["reasons"]) + + +def test_journal_claimant_identity_mismatch_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + jpath = rebind.journal_path(lock_dir, ISSUE) + rebind._atomic_write_json( + jpath, + _complete_journal( + phase=rebind.JOURNAL_PHASE_PRE_BIND, + old_pid=old, + claimant_identity="intruder", + ), + ) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert not result["success"] + assert any("claimant_identity" in r for r in result["reasons"]) + + +def test_journal_claimant_profile_mismatch_refused(dirty_repo, lock_dir): + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + jpath = rebind.journal_path(lock_dir, ISSUE) + rebind._atomic_write_json( + jpath, + _complete_journal( + phase=rebind.JOURNAL_PHASE_PRE_BIND, + old_pid=old, + claimant_profile="prgs-reviewer", + ), + ) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert not result["success"] + assert any("claimant_profile" in r for r in result["reasons"]) + + +def test_incomplete_legacy_journal_identity_refused(dirty_repo, lock_dir): + """Pre-#868 journals missing the five identity fields fail closed mid-flight.""" + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + jpath = rebind.journal_path(lock_dir, ISSUE) + rebind._atomic_write_json( + jpath, + { + "phase": rebind.JOURNAL_PHASE_PRE_BIND, + "issue_number": ISSUE, + "old_pid": old, + "new_pid": os.getpid(), + # deliberately omit remote/org/repo/claimant_* + }, + ) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert not result["success"] + assert any("omits operation identity" in r for r in result["reasons"]) + + +def test_journal_replay_cross_repository_refused(dirty_repo, lock_dir): + """Replaying a journal from another repository is refused.""" + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + jpath = rebind.journal_path(lock_dir, ISSUE) + rebind._atomic_write_json( + jpath, + _complete_journal( + phase=rebind.JOURNAL_PHASE_ASSESSED, + old_pid=old, + remote="dadeschools", + org="Other-Org", + repo="Other-Repo", + ), + ) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert not result["success"] + assert any("replay" in r or "mismatch" in r for r in result["reasons"]) + + +def test_journal_replay_cross_claimant_refused(dirty_repo, lock_dir): + """Replaying a journal from another claimant is refused.""" + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + jpath = rebind.journal_path(lock_dir, ISSUE) + rebind._atomic_write_json( + jpath, + _complete_journal( + phase=rebind.JOURNAL_PHASE_ASSESSED, + old_pid=old, + claimant_identity="other-user", + claimant_profile="other-profile", + ), + ) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert not result["success"] + assert any("claimant" in r for r in result["reasons"]) + + +def test_successful_exact_retry_with_complete_identity(dirty_repo, lock_dir): + """Exact retry after success is already_rebound with complete matching identity.""" + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + r1 = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert r1["success"], r1 + # Journal must persist the five identity fields. + jpath = rebind.journal_path(lock_dir, ISSUE) + journal = rebind._read_json(jpath) + assert journal is not None + assert journal["phase"] == rebind.JOURNAL_PHASE_COMPLETE + for field in rebind.REQUIRED_JOURNAL_IDENTITY_FIELDS: + assert journal.get(field), field + assert journal["remote"] == REMOTE + assert journal["org"] == ORG + assert journal["repo"] == REPO + assert journal["claimant_identity"] == IDENTITY + assert journal["claimant_profile"] == PROFILE + + rebound = ils.read_lock_file(r1["lock_path"]) + r2 = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + rebound, + lock_dir, + old_pid=old, + existing_lock=rebound, + ) + ) + assert r2["success"], r2 + assert r2["already_rebound"] is True + + +def test_already_rebound_requires_complete_matching_identity(dirty_repo, lock_dir): + """already_rebound with mismatched journal identity fails closed.""" + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + r1 = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) + ) + assert r1["success"], r1 + # Corrupt journal identity after success. + jpath = rebind.journal_path(lock_dir, ISSUE) + journal = rebind._read_json(jpath) + assert journal is not None + journal["claimant_identity"] = "not-the-owner" + rebind._atomic_write_json(jpath, journal) + + rebound = ils.read_lock_file(r1["lock_path"]) + r2 = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + rebound, + lock_dir, + old_pid=old, + existing_lock=rebound, + ) + ) + # Same pid on lock would look like already_rebound, but identity must match. + assert not r2["success"] + assert any("claimant_identity" in r or "mismatch" in r for r in r2["reasons"]) + + +def test_ordinary_dirty_worktree_refusal_preserved(dirty_repo): + """#868 must not weaken ordinary dirty-worktree refusal on lock_issue path.""" + porcelain = dirty_repo["inventory"]["porcelain_status"] + assessment = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path=dirty_repo["worktree"], + current_branch=BRANCH, + porcelain_status=porcelain, + base_equivalent=False, + ) + assert assessment["block"] is True + + +# ── #868 — Reconciler success path ────────────────────────────────────────── + + +def test_reconciler_success_path_tightly_pinned(dirty_repo, lock_dir): + """Reconciler with authorize_reconciler_execute=True may execute rebind. + + Reconciler execution grants no commit/push/publication/review/merge + capability — only the tightly pinned session rebind. + """ + old = dead_pid() + lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) + result = rebind.apply_dirty_same_claimant_session_rebind( + **_apply_kwargs( + dirty_repo, + lock, + lock_dir, + old_pid=old, + role_kind="reconciler", + authorize_reconciler_execute=True, + # Reconciler may act for the recorded claimant without being that + # identity in the active session (still pin-checked against lock). + current_identity="sysadmin", + current_profile="prgs-reconciler", + ) + ) + assert result["success"], result + assert result["outcome"] == rebind.REBIND_SANCTIONED + rebound = ils.read_lock_file(result["lock_path"]) + assert int(rebound["session_pid"]) == os.getpid() + # Provenance records the rebind tool; no publication authority is granted. + assert ( + rebound.get("lock_provenance", {}).get("source") + == issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND + ) + # Reconciler rebind does not stamp commit/push/review/merge capabilities. + prov = rebound.get("lock_provenance") or {} + blob = json.dumps(prov) + for forbidden in ( + "gitea.repo.commit", + "gitea.branch.push", + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.create", + ): + assert forbidden not in blob + + +def test_revalidate_complete_dirty_inventory_helper(dirty_repo): + ok = rebind.revalidate_complete_dirty_inventory( + dirty_repo["worktree"], + expected_dirty_paths=dirty_repo["dirty_paths"], + expected_fingerprints=dirty_repo["fingerprints"], + phase="unit", + ) + assert ok["ok"] is True + + bad = rebind.revalidate_complete_dirty_inventory( + dirty_repo["worktree"], + expected_dirty_paths=dirty_repo["dirty_paths"][:-1], + expected_fingerprints={ + p: dirty_repo["fingerprints"][p] for p in dirty_repo["dirty_paths"][:-1] + }, + phase="unit", + ) + assert bad["ok"] is False + assert any("unexpected paths" in r for r in bad["reasons"]) + + +def test_validate_journal_operation_identity_helper(): + complete = _complete_journal() + assert ( + rebind.validate_journal_operation_identity( + complete, + remote=REMOTE, + org=ORG, + repo=REPO, + claimant_identity=IDENTITY, + claimant_profile=PROFILE, + ) + == [] + ) + incomplete = {"phase": "assessed", "remote": REMOTE} + reasons = rebind.validate_journal_operation_identity( + incomplete, + remote=REMOTE, + org=ORG, + repo=REPO, + claimant_identity=IDENTITY, + claimant_profile=PROFILE, + ) + assert any("omits operation identity" in r for r in reasons) + assert any("org" in r for r in reasons) diff --git a/tests/test_issue_858_audit_merged_pr_aware.py b/tests/test_issue_858_audit_merged_pr_aware.py new file mode 100644 index 0000000..3232318 --- /dev/null +++ b/tests/test_issue_858_audit_merged_pr_aware.py @@ -0,0 +1,551 @@ +"""Merged-PR awareness for the worktree cleanup audit (#858). + +Before #858 an ``issue_work`` worktree could never leave ``active_issue_work``: +the audit had no PR linkage at all (``pr_number`` was structurally ``None``) +and its only route to ``clean_stale_removable`` was a TTL derived from a +``last_used_at`` that nothing ever populated. A merged, clean, unprotected +worktree was therefore reported as active work forever, disagreeing with the +PR-scoped reconciler. + +These tests use fabricated temporary repositories and synthetic PR records +only. Nothing here removes a worktree or deletes a branch. +""" + +import os +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import merged_cleanup_reconcile as mcr # noqa: E402 +import worktree_cleanup_audit as wca # noqa: E402 + + +MERGED_BRANCH = "feat/issue-777-timeline" +MERGED_PATH = "/repo/branches/issue-777-timeline" +HEAD_SHA = "a" * 40 + + +def _pr(number, branch, *, merged=True, sha=HEAD_SHA, state=None): + """Synthetic Gitea PR payload.""" + return { + "number": number, + "head": {"ref": branch, "sha": sha}, + "merged_at": "2026-07-24T01:00:00Z" if merged else None, + "state": state or ("closed" if merged else "open"), + } + + +def _porcelain(*entries): + out = [] + for path, branch, sha in entries: + out.append(f"worktree {path}") + out.append(f"HEAD {sha}") + if branch is None: + out.append("detached") + else: + out.append(f"branch refs/heads/{branch}") + out.append("") + return "\n".join(out) + + +class _AuditHarness(unittest.TestCase): + """Runs audit_branches_directory over a fabricated worktree listing.""" + + PORCELAIN = _porcelain( + ("/repo", "master", "f" * 40), + (MERGED_PATH, MERGED_BRANCH, HEAD_SHA), + ) + + def run_audit(self, *, dirty_paths=(), contained=True, **kwargs): + def fake_dirty(path): + if path in dirty_paths: + return {"exists": True, "dirty": True, "dirty_files": [" M x.py"]} + return {"exists": True, "dirty": False, "dirty_files": []} + + with patch.object( + wca, "list_worktrees", + return_value=wca.parse_worktree_porcelain(self.PORCELAIN), + ), patch.object( + wca, "read_worktree_dirty", side_effect=fake_dirty + ), patch.object( + wca, "git_worktree_list", return_value="(mocked)" + ), patch.object( + wca, "is_head_ancestor_of_ref", return_value=contained + ): + report = wca.audit_branches_directory("/repo", **kwargs) + return {wt["path"]: wt for wt in report["worktrees"]}, report + + def merged_audit(self, **kwargs): + kwargs.setdefault("pr_index", wca.build_pr_index([_pr(849, MERGED_BRANCH)])) + kwargs.setdefault("master_ref", "prgs/master") + return self.run_audit(**kwargs) + + +class TestMergedWorktreeBecomesRemovable(_AuditHarness): + def test_clean_merged_issue_worktree_is_linked_and_removable(self): + by_path, report = self.merged_audit() + entry = by_path[MERGED_PATH] + + self.assertEqual(entry["classification"], wca.CLASS_CLEAN_STALE_REMOVABLE) + self.assertTrue(entry["removable"]) + self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_MERGED) + self.assertEqual(entry["merged_pr_cleanup"]["block_reasons"], []) + self.assertIn(MERGED_PATH, [c["path"] for c in report["removable_candidates"]]) + + def test_pr_number_populated_from_authoritative_linkage(self): + by_path, _ = self.merged_audit() + self.assertEqual(by_path[MERGED_PATH]["pr_number"], 849) + + def test_regression_without_pr_evidence_stays_active_issue_work(self): + """The pre-#858 behaviour, still correct when no PR state is supplied.""" + by_path, _ = self.run_audit() + entry = by_path[MERGED_PATH] + self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) + self.assertFalse(entry["removable"]) + self.assertIsNone(entry["pr_number"]) + + +class TestProtectiveSignalsSurvive(_AuditHarness): + def test_open_pr_worktree_is_not_removable(self): + index = wca.build_pr_index([_pr(900, MERGED_BRANCH, merged=False)]) + by_path, _ = self.run_audit( + pr_index=index, + master_ref="prgs/master", + open_pr_branches={MERGED_BRANCH}, + ) + entry = by_path[MERGED_PATH] + self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_OPEN_PR) + self.assertFalse(entry["removable"]) + # linkage still reports the owning PR, it just is not merge proof + self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_OPEN) + self.assertEqual(entry["pr_number"], 900) + + def test_dirty_tracked_worktree_is_not_removable(self): + by_path, _ = self.merged_audit(dirty_paths=(MERGED_PATH,)) + entry = by_path[MERGED_PATH] + self.assertEqual(entry["classification"], wca.CLASS_DIRTY_LOCAL) + self.assertFalse(entry["removable"]) + self.assertIn( + "worktree has uncommitted changes", + entry["merged_pr_cleanup"]["block_reasons"], + ) + + def test_untracked_only_worktree_is_not_removable(self): + """``git status --porcelain`` reports untracked files as dirty too.""" + def untracked(path): + if path == MERGED_PATH: + return {"exists": True, "dirty": True, "dirty_files": ["?? scratch.txt"]} + return {"exists": True, "dirty": False, "dirty_files": []} + + with patch.object( + wca, "list_worktrees", + return_value=wca.parse_worktree_porcelain(self.PORCELAIN), + ), patch.object( + wca, "read_worktree_dirty", side_effect=untracked + ), patch.object( + wca, "git_worktree_list", return_value="(mocked)" + ), patch.object( + wca, "is_head_ancestor_of_ref", return_value=True + ): + report = wca.audit_branches_directory( + "/repo", + pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]), + master_ref="prgs/master", + ) + entry = {wt["path"]: wt for wt in report["worktrees"]}[MERGED_PATH] + self.assertEqual(entry["classification"], wca.CLASS_DIRTY_LOCAL) + self.assertFalse(entry["removable"]) + + def test_active_lease_by_issue_number_is_protective(self): + by_path, _ = self.merged_audit(leased_issue_numbers={777}) + entry = by_path[MERGED_PATH] + self.assertTrue(entry["has_active_lease"]) + self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) + self.assertFalse(entry["removable"]) + + def test_active_lease_by_branch_is_protective(self): + by_path, _ = self.merged_audit(leased_branches={MERGED_BRANCH}) + entry = by_path[MERGED_PATH] + self.assertTrue(entry["has_active_lease"]) + self.assertFalse(entry["removable"]) + + def test_active_issue_lock_is_protective(self): + by_path, _ = self.merged_audit(active_issue_branches={MERGED_BRANCH}) + entry = by_path[MERGED_PATH] + self.assertTrue(entry["has_active_issue_lock"]) + self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) + self.assertFalse(entry["removable"]) + + def test_live_session_worktree_is_protective(self): + by_path, _ = self.merged_audit(live_session_paths={MERGED_PATH}) + entry = by_path[MERGED_PATH] + self.assertTrue(entry["has_live_session"]) + self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) + self.assertFalse(entry["removable"]) + + def test_head_not_contained_in_master_is_not_removable(self): + by_path, _ = self.merged_audit(contained=False) + entry = by_path[MERGED_PATH] + self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) + self.assertFalse(entry["removable"]) + self.assertIn( + "worktree head is not contained in authoritative master " + "(unmerged commits remain)", + entry["merged_pr_cleanup"]["block_reasons"], + ) + + def test_unknown_containment_fails_closed(self): + by_path, _ = self.merged_audit(contained=None) + entry = by_path[MERGED_PATH] + self.assertFalse(entry["removable"]) + self.assertIn( + "containment of the worktree head in master is unknown", + entry["merged_pr_cleanup"]["block_reasons"], + ) + + def test_missing_master_ref_fails_closed(self): + by_path, _ = self.run_audit( + pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]) + ) + self.assertFalse(by_path[MERGED_PATH]["removable"]) + + def test_unmerged_owning_pr_is_not_removable(self): + index = wca.build_pr_index([_pr(901, MERGED_BRANCH, merged=False)]) + by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master") + entry = by_path[MERGED_PATH] + self.assertFalse(entry["removable"]) + self.assertIn( + "owning PR #901 is not merged", + entry["merged_pr_cleanup"]["block_reasons"], + ) + + def test_control_checkout_is_never_removable(self): + by_path, _ = self.merged_audit() + control = by_path["/repo"] + self.assertTrue(control["is_protected"]) + self.assertEqual(control["classification"], wca.CLASS_UNSAFE_UNKNOWN) + self.assertFalse(control["removable"]) + + def test_control_checkout_not_removable_even_if_linked_and_merged(self): + """A merged PR on the control checkout must not unlock removal.""" + porcelain = _porcelain(("/repo", MERGED_BRANCH, HEAD_SHA)) + with patch.object( + wca, "list_worktrees", return_value=wca.parse_worktree_porcelain(porcelain) + ), patch.object( + wca, "read_worktree_dirty", + return_value={"exists": True, "dirty": False, "dirty_files": []}, + ), patch.object( + wca, "git_worktree_list", return_value="(mocked)" + ), patch.object( + wca, "is_head_ancestor_of_ref", return_value=True + ): + report = wca.audit_branches_directory( + "/repo", + pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]), + master_ref="prgs/master", + ) + entry = report["worktrees"][0] + self.assertEqual(entry["classification"], wca.CLASS_UNSAFE_UNKNOWN) + self.assertFalse(entry["removable"]) + + +class TestAmbiguousLinkageFailsClosed(_AuditHarness): + def test_competing_prs_on_one_branch_fail_closed(self): + index = wca.build_pr_index( + [_pr(849, MERGED_BRANCH), _pr(860, MERGED_BRANCH)] + ) + by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master") + entry = by_path[MERGED_PATH] + self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_AMBIGUOUS) + self.assertIsNone(entry["pr_number"]) + self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) + self.assertFalse(entry["removable"]) + + def test_merged_plus_open_pr_on_one_branch_fails_closed(self): + index = wca.build_pr_index( + [_pr(849, MERGED_BRANCH), _pr(861, MERGED_BRANCH, merged=False)] + ) + by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master") + entry = by_path[MERGED_PATH] + self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_AMBIGUOUS) + self.assertFalse(entry["removable"]) + + def test_no_owning_pr_fails_closed(self): + by_path, _ = self.run_audit( + pr_index=wca.build_pr_index([_pr(849, "feat/other-branch")]), + master_ref="prgs/master", + ) + entry = by_path[MERGED_PATH] + self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_NONE) + self.assertFalse(entry["removable"]) + + def test_malformed_pr_records_are_dropped_not_guessed(self): + index = wca.build_pr_index( + [ + {"number": None, "head": {"ref": MERGED_BRANCH}}, + {"number": 5, "head": {}}, + {"number": "not-an-int", "head": {"ref": MERGED_BRANCH}}, + ] + ) + self.assertEqual(index, {}) + self.assertEqual( + wca.resolve_owning_pr(branch=MERGED_BRANCH, pr_index=index)["status"], + wca.LINKAGE_NONE, + ) + + def test_detached_worktree_has_no_branch_linkage(self): + self.assertEqual( + wca.resolve_owning_pr(branch=None, pr_index={})["status"], + wca.LINKAGE_UNKNOWN, + ) + + +class TestUnrelatedClassificationsUnchanged(unittest.TestCase): + """Non-issue_work worktrees keep their pre-#858 classifications.""" + + PORCELAIN = _porcelain( + ("/repo", "master", "f" * 40), + ("/repo/branches/review-pr42", "review-pr42", "2" * 40), + ("/repo/branches/baseline-master-x", "baseline-master-x", "3" * 40), + ("/repo/branches/conflict-fix-pr50", "conflict-fix-pr50", "4" * 40), + ("/repo/branches/review-pr99", None, "5" * 40), + ) + + def _audit(self, **kwargs): + with patch.object( + wca, "list_worktrees", + return_value=wca.parse_worktree_porcelain(self.PORCELAIN), + ), patch.object( + wca, "read_worktree_dirty", + return_value={"exists": True, "dirty": False, "dirty_files": []}, + ), patch.object( + wca, "git_worktree_list", return_value="(mocked)" + ), patch.object( + wca, "is_head_ancestor_of_ref", return_value=True + ): + report = wca.audit_branches_directory("/repo", **kwargs) + return {wt["path"]: wt for wt in report["worktrees"]} + + def test_classifications_identical_with_and_without_pr_evidence(self): + without = self._audit() + with_evidence = self._audit( + pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]), + master_ref="prgs/master", + ) + self.assertEqual( + {p: e["classification"] for p, e in without.items()}, + {p: e["classification"] for p, e in with_evidence.items()}, + ) + + def test_lease_on_issue_does_not_capture_similarly_named_scratch_trees(self): + """A lease on issue 777 protects issue work, not baseline/review trees.""" + porcelain = _porcelain( + ("/repo/branches/baseline-master-issue-777", "baseline-issue-777", "7" * 40), + ("/repo/branches/issue-777-timeline", MERGED_BRANCH, HEAD_SHA), + ) + with patch.object( + wca, "list_worktrees", return_value=wca.parse_worktree_porcelain(porcelain) + ), patch.object( + wca, "read_worktree_dirty", + return_value={"exists": True, "dirty": False, "dirty_files": []}, + ), patch.object( + wca, "git_worktree_list", return_value="(mocked)" + ), patch.object( + wca, "is_head_ancestor_of_ref", return_value=True + ): + report = wca.audit_branches_directory( + "/repo", + pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]), + master_ref="prgs/master", + leased_issue_numbers={777}, + ) + by_path = {wt["path"]: wt for wt in report["worktrees"]} + + baseline = by_path["/repo/branches/baseline-master-issue-777"] + self.assertFalse(baseline["has_active_lease"]) + self.assertEqual(baseline["classification"], wca.CLASS_CLEAN_STALE_REMOVABLE) + + issue_work = by_path["/repo/branches/issue-777-timeline"] + self.assertTrue(issue_work["has_active_lease"]) + self.assertFalse(issue_work["removable"]) + + def test_review_and_baseline_still_removable(self): + by_path = self._audit( + pr_index=wca.build_pr_index([]), master_ref="prgs/master" + ) + self.assertEqual( + by_path["/repo/branches/review-pr42"]["classification"], + wca.CLASS_CLEAN_STALE_REMOVABLE, + ) + self.assertEqual( + by_path["/repo/branches/baseline-master-x"]["classification"], + wca.CLASS_CLEAN_STALE_REMOVABLE, + ) + self.assertEqual( + by_path["/repo/branches/review-pr99"]["classification"], + wca.CLASS_DETACHED_REVIEW_LEFTOVER, + ) + + def test_conflict_fix_ttl_behaviour_unchanged(self): + """conflict_fix still needs only TTL expiry; #858 did not touch it.""" + self.assertEqual( + wca.classify_worktree( + workflow_type=wca.WORKFLOW_CONFLICT_FIX, + is_dirty=False, + ttl_expired=True, + ), + wca.CLASS_CLEAN_STALE_REMOVABLE, + ) + self.assertEqual( + wca.classify_worktree( + workflow_type=wca.WORKFLOW_CONFLICT_FIX, + is_dirty=False, + ttl_expired=False, + ), + wca.CLASS_ACTIVE_ISSUE_WORK, + ) + + def test_issue_work_ttl_alone_no_longer_grants_removal(self): + """Age is not landing proof: TTL alone must not reclaim issue work.""" + self.assertEqual( + wca.classify_worktree( + workflow_type=wca.WORKFLOW_ISSUE_WORK, + is_dirty=False, + ttl_expired=True, + ), + wca.CLASS_ACTIVE_ISSUE_WORK, + ) + + +class TestAssessorPerformsNoDeletion(_AuditHarness): + def test_audit_never_removes_a_worktree(self): + with patch.object(wca, "remove_worktree") as removal: + self.merged_audit() + removal.assert_not_called() + + def test_audit_shells_out_to_no_destructive_git_command(self): + seen = [] + real_run = subprocess.run + + def recording_run(cmd, *args, **kwargs): + seen.append(cmd) + return real_run(["true"], *args, **kwargs) + + with patch.object(subprocess, "run", side_effect=recording_run): + wca.audit_branches_directory("/nonexistent-repo-for-audit") + + joined = [" ".join(c) if isinstance(c, list) else str(c) for c in seen] + for cmd in joined: + self.assertNotIn("worktree remove", cmd) + self.assertNotIn("branch -D", cmd) + self.assertNotIn("push", cmd) + + +class TestAgreementWithPrScopedReconciler(unittest.TestCase): + """The audit and merged_cleanup_reconcile must agree on identical input. + + Uses a real throwaway git repository so containment is computed by git + rather than asserted. Nothing outside the temporary directory is touched. + """ + + def _git(self, *args): + subprocess.run( + ["git", "-C", self.root, *args], + check=True, + capture_output=True, + text=True, + ) + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.root = os.path.realpath(self._tmp.name) + self._git("init", "-b", "master", ".") + self._git("config", "user.email", "test@example.invalid") + self._git("config", "user.name", "Test") + with open(os.path.join(self.root, "seed.txt"), "w") as fh: + fh.write("seed\n") + self._git("add", "seed.txt") + self._git("commit", "-m", "seed") + + self.branch = "feat/issue-777-timeline" + self._git("checkout", "-b", self.branch) + with open(os.path.join(self.root, "feature.txt"), "w") as fh: + fh.write("feature\n") + self._git("add", "feature.txt") + self._git("commit", "-m", "feature") + self.head_sha = subprocess.run( + ["git", "-C", self.root, "rev-parse", "HEAD"], + capture_output=True, text=True, check=True, + ).stdout.strip() + self._git("checkout", "master") + self._git("merge", "--no-ff", "-m", "merge feature", self.branch) + + self.worktree = os.path.join(self.root, "branches", "issue-777-timeline") + self._git("worktree", "add", self.worktree, self.branch) + + def tearDown(self): + self._tmp.cleanup() + + def _pr_index(self): + return wca.build_pr_index( + [ + { + "number": 849, + "head": {"ref": self.branch, "sha": self.head_sha}, + "merged_at": "2026-07-24T01:00:00Z", + } + ] + ) + + def _audit_entry(self): + report = wca.audit_branches_directory( + self.root, pr_index=self._pr_index(), master_ref="master" + ) + return next(wt for wt in report["worktrees"] if wt["path"] == self.worktree) + + def _reconciler_entry(self): + return mcr.assess_local_worktree_cleanup( + pr_number=849, + head_branch=self.branch, + merged=True, + worktree_state=mcr.resolve_cleanup_worktree_state( + project_root=self.root, + head_branch=self.branch, + issue_number=777, + pr_head_sha=self.head_sha, + target_ref="master", + ), + active_lock=False, + ) + + def test_both_assessors_agree_the_worktree_is_safe(self): + audit_entry = self._audit_entry() + reconciler = self._reconciler_entry() + + self.assertTrue(reconciler["safe_to_remove_worktree"], reconciler) + self.assertTrue(audit_entry["removable"], audit_entry) + self.assertEqual(audit_entry["pr_number"], reconciler["pr_number"]) + self.assertEqual(audit_entry["merged_pr_cleanup"]["block_reasons"], []) + self.assertEqual(reconciler["block_reasons"], []) + + def test_both_assessors_agree_a_dirty_worktree_is_unsafe(self): + with open(os.path.join(self.worktree, "feature.txt"), "a") as fh: + fh.write("local edit\n") + + audit_entry = self._audit_entry() + reconciler = self._reconciler_entry() + + self.assertFalse(audit_entry["removable"]) + self.assertFalse(reconciler["safe_to_remove_worktree"]) + + def test_worktree_still_present_after_audit(self): + self._audit_entry() + self.assertTrue(os.path.isdir(self.worktree)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_worktree_cleanup_audit.py b/tests/test_worktree_cleanup_audit.py index f55b652..f3fc510 100644 --- a/tests/test_worktree_cleanup_audit.py +++ b/tests/test_worktree_cleanup_audit.py @@ -134,13 +134,35 @@ class TestClassification(unittest.TestCase): self.assertEqual(cls, wca.CLASS_ACTIVE_OPEN_PR) self.assertFalse(wca.is_removable(cls)) - def test_stale_clean_issue_worktree_removable(self): - # Scenario 5: clean issue worktree, TTL expired, no lock -> removable. + def test_stale_clean_issue_worktree_needs_merged_pr_proof(self): + # Scenario 5 (#858): age is not proof that the branch landed, so a + # TTL-expired issue worktree stays active work. Only authoritative + # merged-PR evidence makes it removable, which is what keeps a + # worktree holding unmerged commits from being reclaimed by age. cls = wca.classify_worktree( workflow_type=wca.WORKFLOW_ISSUE_WORK, is_dirty=False, ttl_expired=True, ) + self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK) + self.assertFalse(wca.is_removable(cls)) + + cls = wca.classify_worktree( + workflow_type=wca.WORKFLOW_ISSUE_WORK, + is_dirty=False, + ttl_expired=True, + merged_pr_cleanup={"proven": True}, + ) + self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE) + self.assertTrue(wca.is_removable(cls)) + + def test_stale_clean_conflict_fix_worktree_removable(self): + # conflict_fix keeps the original TTL rule; #858 changed issue work only. + cls = wca.classify_worktree( + workflow_type=wca.WORKFLOW_CONFLICT_FIX, + is_dirty=False, + ttl_expired=True, + ) self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE) self.assertTrue(wca.is_removable(cls)) diff --git a/worktree_cleanup_audit.py b/worktree_cleanup_audit.py index 261164a..9eb4b35 100644 --- a/worktree_cleanup_audit.py +++ b/worktree_cleanup_audit.py @@ -34,7 +34,11 @@ import subprocess from datetime import datetime, timezone from typing import Any -from merged_cleanup_reconcile import branch_worktree_folder, read_local_worktree_state +from merged_cleanup_reconcile import ( + branch_worktree_folder, + is_head_ancestor_of_ref, + read_local_worktree_state, +) from reviewer_worktree import parse_dirty_tracked_files, REVIEW_WORKTREE_RE PROTECTED_BRANCHES = frozenset({"master", "main", "dev"}) @@ -67,6 +71,14 @@ REMOVABLE_CLASSES = frozenset( {CLASS_CLEAN_STALE_REMOVABLE, CLASS_DETACHED_REVIEW_LEFTOVER} ) +# Merged-PR linkage outcomes for issue worktrees (#858). Only ``LINKAGE_MERGED`` +# is ownership proof; every other outcome leaves the worktree protected. +LINKAGE_MERGED = "merged_pr" +LINKAGE_OPEN = "open_pr" +LINKAGE_NONE = "no_owning_pr" +LINKAGE_AMBIGUOUS = "ambiguous" +LINKAGE_UNKNOWN = "unknown" + _ISSUE_REF_RE = re.compile(r"issue-(\d+)", re.IGNORECASE) _ISSUE_BRANCH_PREFIXES = ("feat/", "fix/", "docs/", "chore/") @@ -169,6 +181,186 @@ def is_ttl_expired( return (now_dt - last).total_seconds() > ttl_hours * 3600.0 +def build_pr_index(prs: list[dict[str, Any]] | None) -> dict[str, list[dict[str, Any]]]: + """Index PR records by head branch for deterministic worktree linkage (#858). + + Accepts Gitea PR payloads (``head`` as a dict) and pre-flattened records + (``head_branch``/``head_sha``). Records without a usable head branch or + number are dropped rather than guessed at, so a branch is only ever linked + to a PR the caller actually proved. + """ + index: dict[str, list[dict[str, Any]]] = {} + for pr in prs or []: + head = pr.get("head") + if isinstance(head, dict): + head_branch = head.get("ref") + head_sha = head.get("sha") + else: + head_branch = pr.get("head_branch") or (head if isinstance(head, str) else None) + head_sha = pr.get("head_sha") + number = pr.get("number") + if not head_branch or number is None: + continue + try: + pr_number = int(number) + except (TypeError, ValueError): + continue + index.setdefault(str(head_branch).strip(), []).append( + { + "pr_number": pr_number, + "head_branch": str(head_branch).strip(), + "head_sha": head_sha, + "merged": bool(pr.get("merged") or pr.get("merged_at")), + "state": pr.get("state"), + } + ) + return index + + +def resolve_owning_pr( + *, + branch: str | None, + pr_index: dict[str, list[dict[str, Any]]] | None, +) -> dict[str, Any]: + """Resolve the single PR that owns ``branch``, failing closed when unclear. + + Ownership is only ``LINKAGE_MERGED`` when exactly one PR claims the branch + and that PR is merged. Several distinct PRs on one branch is a competing + claim (``LINKAGE_AMBIGUOUS``), and a still-open owner is reported as + ``LINKAGE_OPEN`` — both keep the worktree protected while still exposing + the PR number the audit resolved. + """ + if pr_index is None: + return { + "status": LINKAGE_UNKNOWN, + "pr_number": None, + "candidate_pr_numbers": [], + "reasons": ["live PR state was not supplied; ownership unproven"], + } + branch_name = (branch or "").strip() + if not branch_name: + return { + "status": LINKAGE_UNKNOWN, + "pr_number": None, + "candidate_pr_numbers": [], + "reasons": ["worktree has no attached branch; ownership unproven"], + } + + candidates = list(pr_index.get(branch_name) or []) + numbers = sorted({c["pr_number"] for c in candidates}) + if not candidates: + return { + "status": LINKAGE_NONE, + "pr_number": None, + "candidate_pr_numbers": [], + "reasons": [f"no PR claims branch '{branch_name}'"], + } + if len(numbers) > 1: + return { + "status": LINKAGE_AMBIGUOUS, + "pr_number": None, + "candidate_pr_numbers": numbers, + "reasons": [ + f"branch '{branch_name}' is claimed by competing PRs {numbers}; " + "ownership is ambiguous" + ], + } + + owner = candidates[0] + pr_number = owner["pr_number"] + if owner.get("head_branch") != branch_name: + return { + "status": LINKAGE_UNKNOWN, + "pr_number": pr_number, + "candidate_pr_numbers": numbers, + "reasons": [ + f"PR #{pr_number} head branch '{owner.get('head_branch')}' does not " + f"match worktree branch '{branch_name}'" + ], + } + if not owner.get("merged"): + return { + "status": LINKAGE_OPEN, + "pr_number": pr_number, + "candidate_pr_numbers": numbers, + "pr_head_sha": owner.get("head_sha"), + "reasons": [f"owning PR #{pr_number} is not merged"], + } + return { + "status": LINKAGE_MERGED, + "pr_number": pr_number, + "candidate_pr_numbers": numbers, + "pr_head_sha": owner.get("head_sha"), + "reasons": [], + } + + +def assess_merged_pr_worktree_cleanup( + *, + linkage: dict[str, Any] | None, + head_sha: str | None, + head_in_master: bool | None, + is_dirty: bool, + has_open_pr: bool, + has_active_lease: bool, + has_active_issue_lock: bool, + is_protected: bool, + has_live_session: bool = False, +) -> dict[str, Any]: + """Decide whether a merged issue worktree satisfies the full cleanup policy. + + Every condition must be independently proven: conclusive merged-PR + ownership, agreement between the worktree branch and the PR head branch, + containment of the worktree head in authoritative master (which is what + proves no unmerged commits remain), absence of any open/competing PR, + lease, issue lock, or live session, a clean tree, and a worktree that is + not the protected control checkout. Anything unknown blocks. + """ + link = linkage or { + "status": LINKAGE_UNKNOWN, + "pr_number": None, + "reasons": ["no linkage assessment supplied"], + } + status = link.get("status") + reasons: list[str] = [] + + if status != LINKAGE_MERGED: + reasons.extend( + link.get("reasons") or ["owning PR could not be conclusively identified"] + ) + if is_protected: + reasons.append("worktree is protected or the stable control checkout") + if is_dirty: + reasons.append("worktree has uncommitted changes") + if has_open_pr: + reasons.append("worktree branch has an open PR") + if has_active_lease: + reasons.append("worktree has an active lease") + if has_active_issue_lock: + reasons.append("an active issue lock references this branch") + if has_live_session: + reasons.append("a live process or session is using this worktree") + if not head_sha: + reasons.append("worktree head sha is unknown") + if head_in_master is None: + reasons.append("containment of the worktree head in master is unknown") + elif not head_in_master: + reasons.append( + "worktree head is not contained in authoritative master " + "(unmerged commits remain)" + ) + + proven = not reasons + return { + "linkage_status": status, + "pr_number": link.get("pr_number"), + "pr_head_sha": link.get("pr_head_sha"), + "head_in_master": head_in_master, + "proven": proven, + "block_reasons": reasons, + } + + def classify_worktree( *, workflow_type: str, @@ -181,6 +373,8 @@ def classify_worktree( ttl_expired: bool = False, is_protected: bool = False, metadata_known: bool = True, + merged_pr_cleanup: dict[str, Any] | None = None, + has_live_session: bool = False, ) -> str: """Classify a worktree, safety-first: any preservation signal wins. @@ -199,6 +393,8 @@ def classify_worktree( return CLASS_ACTIVE_ISSUE_WORK # never auto-deleted (criterion 8) if has_active_issue_lock: return CLASS_ACTIVE_ISSUE_WORK + if has_live_session: + return CLASS_ACTIVE_ISSUE_WORK # a live session still owns this tree if not metadata_known or workflow_type == WORKFLOW_UNKNOWN: return CLASS_UNSAFE_UNKNOWN # never auto-deleted without proof @@ -207,7 +403,15 @@ def classify_worktree( if is_detached or branch_gone: return CLASS_DETACHED_REVIEW_LEFTOVER return CLASS_CLEAN_STALE_REMOVABLE - # issue_work / conflict_fix: only removable once the TTL has expired. + if workflow_type == WORKFLOW_ISSUE_WORK: + # #858: an issue worktree becomes removable only on authoritative + # merged-PR evidence satisfying the whole cleanup policy. Age alone + # never proves the branch landed, so TTL cannot qualify one by itself + # — otherwise a worktree holding unmerged commits would be reclaimed. + if (merged_pr_cleanup or {}).get("proven"): + return CLASS_CLEAN_STALE_REMOVABLE + return CLASS_ACTIVE_ISSUE_WORK + # conflict_fix: only removable once the TTL has expired. if ttl_expired: return CLASS_CLEAN_STALE_REMOVABLE return CLASS_ACTIVE_ISSUE_WORK @@ -400,6 +604,20 @@ def remove_worktree(project_root: str, path: str) -> dict[str, Any]: } +def head_contained_in_ref( + project_root: str, head_sha: str | None, ref: str | None +) -> bool | None: + """Return True when ``head_sha`` is already contained in ``ref``. + + Shares :mod:`merged_cleanup_reconcile`'s ancestry check so the audit and + the PR-scoped reconciler agree on what "already landed" means (#858). + Returns None when containment cannot be determined, which fails closed. + """ + if not head_sha or not ref: + return None + return is_head_ancestor_of_ref(project_root, head_sha, ref) + + def _is_under_branches(project_root: str, path: str) -> bool: branches_root = os.path.join(os.path.abspath(project_root), "branches") return os.path.abspath(path or "").startswith(branches_root + os.sep) @@ -413,16 +631,30 @@ def audit_branches_directory( active_issue_branches: set[str] | None = None, now: datetime | str | None = None, ttl_hours: float = DEFAULT_TTL_HOURS, + pr_index: dict[str, list[dict[str, Any]]] | None = None, + leased_issue_numbers: set[int] | None = None, + live_session_paths: set[str] | None = None, + master_ref: str | None = None, ) -> dict[str, Any]: """Classify every session-owned worktree under ``branches/``. Read-only: shells out to git for discovery and dirty state, then applies the pure classifier. Returns per-worktree classifications, counts, the list of removable candidates, and the ``git worktree list`` proof. + + ``pr_index`` (see :func:`build_pr_index`) supplies the authoritative PR + ownership used to link issue worktrees to their merged PR (#858). + ``master_ref`` is the ref a worktree head must be contained in before it + can be considered landed. Both are optional and their absence only ever + fails closed: without them no issue worktree becomes removable. """ open_pr_branches = open_pr_branches or set() leased_branches = leased_branches or set() active_issue_branches = active_issue_branches or set() + leased_issue_numbers = leased_issue_numbers or set() + live_session_paths = { + os.path.abspath(p) for p in (live_session_paths or set()) if p + } worktrees: list[dict[str, Any]] = [] for entry in list_worktrees(project_root): @@ -433,12 +665,42 @@ def audit_branches_directory( ) dirty_state = read_worktree_dirty(path) is_dirty = bool(dirty_state.get("dirty")) + head_sha = entry.get("head") + linkage = resolve_owning_pr(branch=branch, pr_index=pr_index) metadata = build_worktree_metadata( - path=path, branch=branch, head_sha=entry.get("head") + path=path, + branch=branch, + head_sha=head_sha, + pr_number=linkage.get("pr_number"), ) has_open_pr = bool(branch) and branch in open_pr_branches - has_active_lease = bool(branch) and branch in leased_branches + # A lease on issue N protects that issue's own work worktree. It must + # not incidentally protect a baseline/review scratch tree that merely + # carries the same issue marker in its name, which would change the + # classification of worktrees this policy does not own. + has_active_lease = (bool(branch) and branch in leased_branches) or ( + metadata["workflow_type"] == WORKFLOW_ISSUE_WORK + and metadata.get("issue_number") is not None + and metadata["issue_number"] in leased_issue_numbers + ) has_active_lock = bool(branch) and branch in active_issue_branches + has_live_session = bool(path) and os.path.abspath(path) in live_session_paths + head_in_master = ( + head_contained_in_ref(project_root, head_sha, master_ref) + if master_ref + else None + ) + merged_pr_cleanup = assess_merged_pr_worktree_cleanup( + linkage=linkage, + head_sha=head_sha, + head_in_master=head_in_master, + is_dirty=is_dirty, + has_open_pr=has_open_pr, + has_active_lease=has_active_lease, + has_active_issue_lock=has_active_lock, + is_protected=is_protected, + has_live_session=has_live_session, + ) ttl_expired = is_ttl_expired( last_used_at=metadata.get("last_used_at"), now=now, ttl_hours=ttl_hours ) @@ -452,6 +714,8 @@ def audit_branches_directory( branch_gone=branch is None and not entry.get("detached"), ttl_expired=ttl_expired, is_protected=is_protected, + merged_pr_cleanup=merged_pr_cleanup, + has_live_session=has_live_session, ) metadata["cleanup_eligibility"] = classification worktrees.append( @@ -463,7 +727,10 @@ def audit_branches_directory( "has_open_pr": has_open_pr, "has_active_lease": has_active_lease, "has_active_issue_lock": has_active_lock, + "has_live_session": has_live_session, "is_protected": is_protected, + "merged_pr_linkage": linkage, + "merged_pr_cleanup": merged_pr_cleanup, "classification": classification, "removable": is_removable(classification), }