diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 58f78d1..1b16902 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1650,6 +1650,7 @@ import issue_lock_worktree # noqa: E402 import issue_lock_provenance # noqa: E402 import issue_lock_store # noqa: E402 import issue_lock_adoption # noqa: E402 +import issue_lock_recovery # noqa: E402 import stacked_pr_support # noqa: E402 import merge_approval_gate # noqa: E402 import review_quarantine # noqa: E402 # #695 contaminated formal-review quarantine @@ -3155,8 +3156,11 @@ def gitea_lock_issue( worktree_path, _canonical_local_git_root() ) h, o, r = _resolve(remote, host, org, repo) + existing_issue_lock = _load_existing_issue_lock( + remote=remote, org=o, repo=r, issue_number=issue_number + ) active_lease_block = issue_lock_store.assess_same_issue_lease_conflict( - _load_existing_issue_lock(remote=remote, org=o, repo=r, issue_number=issue_number), + existing_issue_lock, issue_number=issue_number, branch_name=branch_name, worktree_path=resolved_worktree, @@ -3198,6 +3202,72 @@ def gitea_lock_issue( org=org, repo=repo, ) + # ── Dead-session lock recovery assessment (#753) ── + # When the MCP session that took a lock exits, the lock goes non-live + # (stale by dead PID) even inside its lease TTL, and the branch it owns is + # ahead of its base by construction — so the base-equivalence gate below + # makes normal re-lock unreachable for every PR that already exists. + # + # This grants a waiver ONLY for that case, proven against the durable lock + # record plus live git/Gitea observation. A refused assessment never raises: + # it simply withholds the waiver, leaving the pre-existing guard to fail + # closed exactly as before. Recovery can only ever add permission. + recovery_assessment: dict | None = None + if ( + existing_issue_lock + and existing_issue_lock.get("issue_number") == issue_number + and not issue_lock_store.is_lease_live(existing_issue_lock) + ): + recovery_auth = _auth(h) + try: + recovery_branches = api_get_all( + f"{repo_api_url(h, o, r)}/branches", recovery_auth + ) + except Exception as exc: + raise RuntimeError( + f"Could not list branches to verify issue-lock recovery: {exc}" + ) + recovery_remote_head: str | None = None + recovery_candidates: list[str] = [] + for entry in recovery_branches: + entry_name = _branch_entry_name(entry) + if entry_name == branch_name: + recovery_remote_head = _branch_entry_commit_sha(entry) + if issue_lock_adoption.branch_carries_issue_marker(entry_name, issue_number): + recovery_candidates.append(entry_name) + recovery_pr_head: str | None = None + recovery_pr_number: int | None = None + for pull in _list_open_pulls(h, o, r, recovery_auth): + pull_head = pull.get("head") or {} + if str(pull_head.get("ref") or "") == branch_name: + recovery_pr_head = pull_head.get("sha") + recovery_pr_number = pull.get("number") + break + recovery_claimant = _work_lease_claimant(h) + recovery_assessment = issue_lock_recovery.assess_dead_session_lock_recovery( + existing_issue_lock, + issue_number=issue_number, + branch_name=branch_name, + worktree_path=resolved_worktree, + remote=remote, + org=o, + repo=r, + identity=recovery_claimant.get("username"), + profile=recovery_claimant.get("profile"), + current_branch=git_state.get("current_branch"), + porcelain_status=git_state.get("porcelain_status") or "", + head_sha=git_state.get("head_sha"), + remote_head_sha=recovery_remote_head, + pr_head_sha=recovery_pr_head, + pr_number=recovery_pr_number, + competing_live_locks=issue_lock_store.list_live_locks(), + candidate_branches=recovery_candidates, + current_pid=os.getpid(), + ) + + recovery_sanctioned = bool( + recovery_assessment and recovery_assessment.get("recovery_sanctioned") + ) lock_assessment = issue_lock_worktree.assess_issue_lock_worktree( worktree_path=resolved_worktree, current_branch=git_state.get("current_branch"), @@ -3205,10 +3275,20 @@ def gitea_lock_issue( base_equivalent=git_state.get("base_equivalent"), inspected_git_root=git_state.get("inspected_git_root"), base_branch=git_state.get("base_branch"), + recovery_sanctioned=recovery_sanctioned, ) if lock_assessment["block"]: + reasons = list(lock_assessment.get("reasons") or []) + # Surface why recovery was unavailable, so a blocked caller sees the + # exact missing evidence instead of only the base-equivalence text. + if recovery_assessment and recovery_assessment.get("is_candidate"): + reasons.append( + issue_lock_recovery.format_recovery_refusal(recovery_assessment) + ) raise RuntimeError( - issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment) + issue_lock_worktree.format_issue_lock_worktree_error( + {**lock_assessment, "reasons": reasons} + ) ) auth = _auth(h) @@ -3271,6 +3351,14 @@ def gitea_lock_issue( } if stacked_approved: data["approved_stacked_base"] = stacked_approved + if recovery_sanctioned and recovery_assessment: + # #753 AC2/AC6: record that this claim was recovered after session + # death, with the prior and replacement session identity, so the + # takeover is auditable and never looks like an original claim. + data["dead_session_recovery"] = issue_lock_recovery.build_recovery_record( + recovery_assessment, + recovered_at=_work_lease_timestamp(_work_lease_now()), + ) lock_file_path = _save_issue_lock(data) lock_record = issue_lock_store.read_lock_file(lock_file_path) or data @@ -3304,6 +3392,14 @@ def gitea_lock_issue( "lock_freshness": freshness, "lock_proof": lock_proof, } + if recovery_sanctioned and recovery_assessment: + result["dead_session_recovery"] = data["dead_session_recovery"] + result["message"] = ( + f"Recovered the durable lock for issue #{issue_number} on branch " + f"'{branch_name}' after the owning MCP session (pid " + f"{recovery_assessment['evidence'].get('prior_session_pid')}) exited; " + "ownership evidence matched exactly (fail-closed check complete)." + ) if stacked_approved: result["approved_stacked_base"] = stacked_approved result["message"] = ( diff --git a/issue_lock_adoption.py b/issue_lock_adoption.py index 81a5a92..b6d52a4 100644 --- a/issue_lock_adoption.py +++ b/issue_lock_adoption.py @@ -85,6 +85,15 @@ def _branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool: return re.search(pattern, name) is not None +def branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool: + """Public accessor for the exact issue-marker match (#753). + + Dead-session lock recovery needs the same word-boundary matcher to detect + ambiguous branch claims, so it is exposed rather than reached into. + """ + return _branch_carries_issue_marker(branch_name, issue_number) + + def assess_own_branch_adoption( *, issue_number: int, diff --git a/issue_lock_recovery.py b/issue_lock_recovery.py new file mode 100644 index 0000000..ab27f38 --- /dev/null +++ b/issue_lock_recovery.py @@ -0,0 +1,400 @@ +"""Dead-session author issue-lock recovery (#753). + +A durable author issue lock records the PID of the MCP session that took it. +When that process exits, ``issue_lock_store.assess_lock_freshness`` classifies +the lock as ``stale`` (``live=False``) even while its lease is still within TTL, +so every ownership check that requires a *live* lock fails closed. + +Re-taking the lock through ``gitea_lock_issue`` is unreachable for real work: +``issue_lock_worktree.assess_issue_lock_worktree`` demands the worktree be +base-equivalent to ``master``/``main``/``dev``, and a branch that already +carries commits is ahead of its base by construction. The existing +``assess_expired_lock_reclaim`` affordance does not apply either, because +``assess_same_issue_lease_conflict`` only consults it once the lease has +*expired* — a dead PID under an unexpired lease never reaches it. + +This module is the pure evidence assessor for that one narrow case. It grants +recovery only when every element of durable ownership still matches exactly and +the recorded process is demonstrably dead. It never trusts caller assertions: +every field is compared against durable lock state or live observation supplied +by the caller. It performs no mutation and no network I/O. + +Recovery deliberately does **not** relax base-equivalence for brand-new issue +claims — only for a lock whose own prior record already proves the branch, +worktree, head, and author. +""" + +from __future__ import annotations + +import os +from typing import Any, Iterable, Mapping, Sequence + +from issue_lock_store import is_process_alive +from reviewer_worktree import parse_dirty_tracked_files + +# Outcome values +RECOVERY_SANCTIONED = "RECOVERY_SANCTIONED" +NO_CANDIDATE = "NO_CANDIDATE" +REFUSED = "REFUSED" + +# Durable fields a lock must carry before it can be considered at all. +REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path") + + +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 _text(value: Any) -> str: + return str(value or "").strip() + + +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 _malformed_reasons(lock: Mapping[str, Any]) -> list[str]: + """Names of durable fields that are missing or unusable.""" + 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 assess_dead_session_lock_recovery( + existing_lock: Mapping[str, Any] | None, + *, + issue_number: int, + branch_name: str, + worktree_path: str, + remote: str, + org: str, + repo: str, + identity: str | None, + profile: str | None, + current_branch: str | None, + porcelain_status: str, + head_sha: str | None, + remote_head_sha: str | None, + pr_head_sha: str | None = None, + pr_number: int | None = None, + competing_live_locks: Sequence[Mapping[str, Any]] | None = None, + candidate_branches: Iterable[str] | None = None, + current_pid: int | None = None, +) -> dict[str, Any]: + """Decide whether a dead-session author lock may be natively recovered. + + Returns a dict with ``recovery_sanctioned`` (bool), ``outcome``, ``reasons`` + (why it was refused, or the positive proof when sanctioned), and + ``evidence`` (a redaction-safe record for auditing). + + ``NO_CANDIDATE`` means no recovery was attempted at all — there is no + existing lock, or the lock does not describe this issue. The caller must + treat that exactly as it treated the pre-#753 world. ``REFUSED`` means a + candidate existed but the evidence did not agree; the caller fails closed. + """ + 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, + } + + if not existing_lock: + return _result( + NO_CANDIDATE, False, ["no existing durable lock for this issue"], evidence + ) + + lock = dict(existing_lock) + + # ── Candidate identification ──────────────────────────────────────────── + # Recovery only ever applies to a lock that already claims THIS issue. + # Anything else is not a recovery candidate and must not be reinterpreted. + if lock.get("issue_number") != issue_number: + return _result( + NO_CANDIDATE, + False, + [ + f"existing lock targets issue #{lock.get('issue_number')}, " + f"not #{issue_number}; not a recovery candidate" + ], + evidence, + ) + + # A malformed/incomplete durable record can never prove ownership. + missing = _malformed_reasons(lock) + if missing: + return _result( + REFUSED, + False, + [ + "durable lock record is incomplete and cannot prove ownership " + f"(missing/unusable: {', '.join(missing)})" + ], + evidence, + ) + + recorded_pid = _recorded_pid(lock) + evidence["prior_session_pid"] = recorded_pid + evidence["replacement_session_pid"] = ( + current_pid if current_pid is not None else os.getpid() + ) + + # ── Repository scope ──────────────────────────────────────────────────── + 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)}'" + ) + + # ── Branch identity ───────────────────────────────────────────────────── + 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)}'" + ) + evidence["locked_branch"] = locked_branch + + # The worktree must actually be sitting on the locked branch. Without this + # a clean worktree parked elsewhere could stand in for the real work. + 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}'" + ) + + # ── Worktree identity ─────────────────────────────────────────────────── + 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 + + # ── Cleanliness (never waived) ────────────────────────────────────────── + dirty_files = parse_dirty_tracked_files(porcelain_status) + if dirty_files: + reasons.append( + "worktree has tracked local edits; recovery requires a clean " + f"worktree (dirty files: {', '.join(dirty_files)})" + ) + evidence["dirty_files"] = dirty_files + + # ── Head agreement: local == remote == PR ─────────────────────────────── + local_head = _text(head_sha) + remote_head = _text(remote_head_sha) + if not local_head: + reasons.append("local head SHA could not be determined") + if not remote_head: + reasons.append( + f"remote head for branch '{locked_branch}' could not be determined" + ) + if local_head and remote_head and local_head != remote_head: + reasons.append( + f"local head {local_head} does not match remote branch head {remote_head}" + ) + evidence["local_head"] = local_head or None + evidence["remote_head"] = remote_head or None + + pr_head = _text(pr_head_sha) + if pr_head: + evidence["pr_head"] = pr_head + evidence["pr_number"] = pr_number + if local_head and pr_head != local_head: + reasons.append( + f"open PR #{pr_number} head {pr_head} does not match local head " + f"{local_head}" + ) + + # ── Author identity ───────────────────────────────────────────────────── + claimant = _lock_claimant(lock) + locked_identity = _text(claimant.get("username")) + locked_profile = _text(claimant.get("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; " + "author ownership could not be proven" + ) + if not _text(identity) or not _text(profile): + reasons.append( + "active session identity/profile is unknown; author ownership " + "could not be proven" + ) + if locked_identity and _text(identity) and locked_identity != _text(identity): + reasons.append( + f"lock claimant '{locked_identity}' does not match active identity " + f"'{_text(identity)}'" + ) + if locked_profile and _text(profile) and locked_profile != _text(profile): + reasons.append( + f"lock profile '{locked_profile}' does not match active profile " + f"'{_text(profile)}'" + ) + + # ── The defining condition: the recorded owner must be dead ───────────── + prior_alive = is_process_alive(recorded_pid) + evidence["prior_pid_alive"] = prior_alive + if prior_alive: + reasons.append( + f"prior owner pid {recorded_pid} is still alive; this is not a " + "dead-session recovery" + ) + if current_pid is not None and recorded_pid is not None: + try: + if int(recorded_pid) == int(current_pid): + reasons.append( + "recorded pid is the current session; nothing to recover" + ) + except (TypeError, ValueError): + pass + + # ── 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 + # The lock we are recovering is not competition with itself. + if ( + same_issue + and same_branch + and _same_realpath(_text(entry.get("worktree_path")), worktree_path) + ): + 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"), + } + ) + if competing: + described = ", ".join( + f"issue #{c['issue_number']} branch '{c['branch_name']}'" for c in competing + ) + reasons.append(f"competing live lock or lease exists ({described})") + evidence["competing_live_locks"] = competing + + # ── Ambiguous branch claims ───────────────────────────────────────────── + others = [ + name + for name in (candidate_branches or ()) + if _text(name) and _text(name) != locked_branch + ] + if others: + reasons.append( + "multiple branches claim this issue " + f"({', '.join(sorted(set(others)))}); ownership is ambiguous" + ) + evidence["other_candidate_branches"] = sorted(set(others)) + + if reasons: + return _result(REFUSED, False, reasons, evidence) + + return _result( + RECOVERY_SANCTIONED, + True, + [ + f"durable lock for issue #{issue_number} matches branch " + f"'{locked_branch}', worktree '{locked_worktree}', head {local_head}, " + f"and claimant '{locked_identity}'; recorded pid {recorded_pid} is dead" + ], + evidence, + ) + + +def _result( + outcome: str, + sanctioned: bool, + reasons: list[str], + evidence: dict[str, Any], +) -> dict[str, Any]: + return { + "outcome": outcome, + "recovery_sanctioned": sanctioned, + "is_candidate": outcome != NO_CANDIDATE, + "reasons": reasons, + "evidence": evidence, + } + + +def build_recovery_record( + assessment: Mapping[str, Any], + *, + recovered_at: str, +) -> dict[str, Any]: + """Durable, secret-free provenance for a completed recovery (#753 AC2/AC6).""" + evidence = dict(assessment.get("evidence") or {}) + return { + "recovered": True, + "reason": "owning MCP session exited; durable ownership evidence matched", + "recovered_at": recovered_at, + "prior_session_pid": evidence.get("prior_session_pid"), + "replacement_session_pid": evidence.get("replacement_session_pid"), + "prior_pid_alive": evidence.get("prior_pid_alive"), + "branch_name": evidence.get("locked_branch"), + "worktree_path": evidence.get("locked_worktree_path"), + "local_head": evidence.get("local_head"), + "remote_head": evidence.get("remote_head"), + "pr_head": evidence.get("pr_head"), + "pr_number": evidence.get("pr_number"), + "identity": evidence.get("locked_identity"), + "profile": evidence.get("locked_profile"), + "proof": list(assessment.get("reasons") or []), + } + + +def format_recovery_refusal(assessment: Mapping[str, Any]) -> str: + """Single fail-closed message for a refused recovery attempt.""" + reasons = list(assessment.get("reasons") or []) or [ + "dead-session lock recovery evidence did not agree" + ] + return ( + "Dead-session issue-lock recovery refused: " + + "; ".join(reasons) + + " (fail closed)" + ) diff --git a/issue_lock_worktree.py b/issue_lock_worktree.py index 3c23134..d14f5a2 100644 --- a/issue_lock_worktree.py +++ b/issue_lock_worktree.py @@ -92,8 +92,19 @@ def assess_issue_lock_worktree( inspected_git_root: str | None = None, base_branch: str | None = None, base_branches: frozenset[str] | None = None, + recovery_sanctioned: bool = False, ) -> dict: - """Fail closed when lock preconditions are not met on the declared worktree.""" + """Fail closed when lock preconditions are not met on the declared worktree. + + ``recovery_sanctioned`` is set only when ``issue_lock_recovery`` has already + proven, from the durable lock itself, that this is a dead-session recovery of + an existing claim (#753): same issue, branch, worktree, author, and head, with + the recording process dead. In that one case the base-equivalence requirement + is waived, because a branch that already carries the work is ahead of its base + by construction and could never satisfy it. Every other precondition — + notably worktree cleanliness — still applies unchanged, and brand-new issue + claims keep the full base-equivalence requirement. + """ bases = base_branches or BASE_BRANCHES reasons: list[str] = [] path = (worktree_path or "").strip() @@ -111,7 +122,11 @@ def assess_issue_lock_worktree( f"(dirty files: {', '.join(dirty_files)})" ) - if base_equivalent is False: + if recovery_sanctioned: + # Base-equivalence intentionally not evaluated: ownership was proven + # against the durable lock record instead (#753). + pass + elif base_equivalent is False: reasons.append( "issue lock worktree must be base-equivalent to one of " f"{_base_list(bases)} before implementation work; inspected " @@ -139,6 +154,7 @@ def assess_issue_lock_worktree( inspected_git_root=inspected_git_root, base_branch=base_branch, base_equivalent=base_equivalent, + recovery_sanctioned=recovery_sanctioned, ) @@ -197,6 +213,7 @@ def _assessment( inspected_git_root: str | None = None, base_branch: str | None = None, base_equivalent: bool | None = None, + recovery_sanctioned: bool = False, ) -> dict: return { "proven": proven, @@ -208,6 +225,8 @@ def _assessment( "dirty_files": dirty_files, "base_branch": base_branch, "base_equivalent": base_equivalent, + "recovery_sanctioned": recovery_sanctioned, + "base_equivalence_waived": bool(recovery_sanctioned), } diff --git a/tests/test_issue_753_dead_pid_lock_recovery.py b/tests/test_issue_753_dead_pid_lock_recovery.py new file mode 100644 index 0000000..93a8a7c --- /dev/null +++ b/tests/test_issue_753_dead_pid_lock_recovery.py @@ -0,0 +1,364 @@ +"""Dead-session author issue-lock recovery (#753). + +Covers the narrow recovery path that lets an author re-acquire a durable lock +after the MCP session that recorded it exits, plus every rejection condition +that must keep failing closed. +""" + +import os +import subprocess +import sys +import unittest +from datetime import datetime, timedelta, timezone + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import issue_lock_recovery # noqa: E402 +import issue_lock_store # noqa: E402 +import issue_lock_worktree # noqa: E402 + +ISSUE = 4242 +BRANCH = f"fix/issue-{ISSUE}-demo" +WORKTREE = "/scratch/wt" +HEAD = "a" * 40 +OTHER_SHA = "b" * 40 +IDENTITY = "example-user" +PROFILE = "example-author" + + +def dead_pid() -> int: + """A PID that has certainly exited (spawned, then reaped).""" + 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") + ) + + +def make_lock(**overrides): + lock = { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": WORKTREE, + "remote": "prgs", + "org": "ExampleOrg", + "repo": "ExampleRepo", + "session_pid": dead_pid(), + "work_lease": { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": ISSUE, + "branch": BRANCH, + "worktree_path": WORKTREE, + "claimant": {"username": IDENTITY, "profile": PROFILE}, + "expires_at": future_ts(), + }, + } + lock.update(overrides) + return lock + + +def assess(lock=None, **overrides): + kwargs = { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": WORKTREE, + "remote": "prgs", + "org": "ExampleOrg", + "repo": "ExampleRepo", + "identity": IDENTITY, + "profile": PROFILE, + "current_branch": BRANCH, + "porcelain_status": "", + "head_sha": HEAD, + "remote_head_sha": HEAD, + "pr_head_sha": HEAD, + "pr_number": 99, + "competing_live_locks": [], + "candidate_branches": [BRANCH], + "current_pid": os.getpid(), + } + kwargs.update(overrides) + return issue_lock_recovery.assess_dead_session_lock_recovery( + make_lock() if lock is None else lock, **kwargs + ) + + +class TestDeadSessionRecoveryGranted(unittest.TestCase): + def test_dead_pid_with_exact_evidence_recovers(self): + result = assess() + self.assertTrue(result["recovery_sanctioned"], result["reasons"]) + self.assertEqual(result["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED) + + def test_recovery_still_granted_when_no_open_pr_exists(self): + # A locked branch need not have a PR yet; absence must not block. + result = assess(pr_head_sha=None, pr_number=None) + self.assertTrue(result["recovery_sanctioned"], result["reasons"]) + + def test_lease_expiry_is_not_required_for_recovery(self): + # The defining condition is PID death, not TTL expiry (the #601 gap). + lock = make_lock() + self.assertFalse(issue_lock_store.is_lease_expired(lock)) + self.assertFalse(issue_lock_store.assess_lock_freshness(lock)["live"]) + self.assertTrue(assess(lock)["recovery_sanctioned"]) + + +class TestDeadSessionRecoveryRefused(unittest.TestCase): + def assert_refused(self, result, needle): + self.assertFalse(result["recovery_sanctioned"]) + self.assertEqual(result["outcome"], issue_lock_recovery.REFUSED) + self.assertTrue( + any(needle in reason for reason in result["reasons"]), + f"expected {needle!r} in {result['reasons']}", + ) + + def test_live_prior_pid_refused(self): + lock = make_lock(session_pid=os.getpid(), pid=os.getpid()) + # Distinct current pid so the refusal is attributable to liveness. + self.assert_refused(assess(lock, current_pid=os.getpid() + 1), "still alive") + + def test_different_author_identity_refused(self): + self.assert_refused( + assess(identity="someone-else"), "does not match active identity" + ) + + def test_different_profile_refused(self): + self.assert_refused( + assess(profile="other-profile"), "does not match active profile" + ) + + def test_different_branch_refused(self): + lock = make_lock(branch_name=f"fix/issue-{ISSUE}-other") + self.assert_refused(assess(lock), "does not match requested") + + def test_worktree_parked_on_another_branch_refused(self): + self.assert_refused(assess(current_branch="master"), "not the locked branch") + + def test_detached_head_worktree_refused(self): + self.assert_refused(assess(current_branch=None), "detached HEAD") + + def test_different_worktree_refused(self): + self.assert_refused( + assess(worktree_path="/scratch/elsewhere"), "does not match declared" + ) + + def test_dirty_worktree_refused(self): + self.assert_refused( + assess(porcelain_status=" M gitea_mcp_server.py\n"), "requires a clean" + ) + + def test_local_head_differing_from_remote_refused(self): + self.assert_refused( + assess(remote_head_sha=OTHER_SHA), "does not match remote branch head" + ) + + def test_pr_head_differing_refused(self): + self.assert_refused(assess(pr_head_sha=OTHER_SHA), "does not match local head") + + def test_missing_remote_head_refused(self): + self.assert_refused(assess(remote_head_sha=None), "remote head") + + def test_competing_live_lock_refused(self): + competing = [ + { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": "/scratch/other-wt", + "pid": os.getpid(), + } + ] + self.assert_refused( + assess(competing_live_locks=competing), "competing live lock" + ) + + def test_unrelated_live_lock_does_not_block(self): + unrelated = [ + { + "issue_number": 999, + "branch_name": "fix/issue-999-unrelated", + "worktree_path": "/scratch/unrelated", + "pid": os.getpid(), + } + ] + self.assertTrue(assess(competing_live_locks=unrelated)["recovery_sanctioned"]) + + def test_multiple_candidate_branches_refused(self): + self.assert_refused( + assess(candidate_branches=[BRANCH, f"feat/issue-{ISSUE}-rival"]), + "multiple branches claim this issue", + ) + + def test_repository_scope_mismatch_refused(self): + self.assert_refused(assess(repo="OtherRepo"), "does not match requested") + + def test_malformed_lock_missing_worktree_refused(self): + lock = make_lock() + lock.pop("worktree_path") + self.assert_refused(assess(lock), "incomplete") + + def test_malformed_lock_missing_pid_refused(self): + lock = make_lock() + lock.pop("session_pid", None) + lock.pop("pid", None) + self.assert_refused(assess(lock), "incomplete") + + def test_lock_without_claimant_refused(self): + lock = make_lock() + lock["work_lease"] = dict(lock["work_lease"]) + lock["work_lease"].pop("claimant") + self.assert_refused(assess(lock), "claimant identity/profile") + + +class TestNotACandidate(unittest.TestCase): + def test_absent_lock_is_not_a_candidate(self): + result = issue_lock_recovery.assess_dead_session_lock_recovery( + None, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=WORKTREE, + remote="prgs", + org="ExampleOrg", + repo="ExampleRepo", + identity=IDENTITY, + profile=PROFILE, + current_branch=BRANCH, + porcelain_status="", + head_sha=HEAD, + remote_head_sha=HEAD, + ) + self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE) + self.assertFalse(result["recovery_sanctioned"]) + self.assertFalse(result["is_candidate"]) + + def test_lock_for_a_different_issue_is_not_a_candidate(self): + result = assess(make_lock(issue_number=7777)) + self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE) + self.assertFalse(result["recovery_sanctioned"]) + + +class TestWorktreeGateWaiver(unittest.TestCase): + def test_new_issue_claim_still_requires_base_equivalence(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path=WORKTREE, + current_branch=BRANCH, + porcelain_status="", + base_equivalent=False, + ) + self.assertTrue(result["block"]) + self.assertFalse(result["base_equivalence_waived"]) + + def test_sanctioned_recovery_waives_base_equivalence(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path=WORKTREE, + current_branch=BRANCH, + porcelain_status="", + base_equivalent=False, + recovery_sanctioned=True, + ) + self.assertTrue(result["proven"], result["reasons"]) + self.assertTrue(result["base_equivalence_waived"]) + + def test_recovery_never_waives_cleanliness(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path=WORKTREE, + current_branch=BRANCH, + porcelain_status=" M gitea_mcp_server.py\n", + base_equivalent=False, + recovery_sanctioned=True, + ) + self.assertTrue(result["block"]) + self.assertTrue( + any("tracked file edits" in reason for reason in result["reasons"]) + ) + + def test_unproven_base_equivalence_still_blocks_without_recovery(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path=WORKTREE, + current_branch=BRANCH, + porcelain_status="", + base_equivalent=None, + ) + self.assertTrue(result["block"]) + + +class TestRecoveryRecordAndDownstream(unittest.TestCase): + def test_recovery_record_preserves_truthful_provenance(self): + assessment = assess() + prior = assessment["evidence"]["prior_session_pid"] + record = issue_lock_recovery.build_recovery_record( + assessment, recovered_at="2026-07-18T23:21:40Z" + ) + self.assertTrue(record["recovered"]) + self.assertEqual(record["prior_session_pid"], prior) + self.assertEqual(record["replacement_session_pid"], os.getpid()) + self.assertNotEqual( + record["prior_session_pid"], record["replacement_session_pid"] + ) + self.assertFalse(record["prior_pid_alive"]) + self.assertEqual(record["recovered_at"], "2026-07-18T23:21:40Z") + self.assertEqual(record["branch_name"], BRANCH) + self.assertEqual(record["local_head"], HEAD) + self.assertEqual(record["identity"], IDENTITY) + self.assertTrue(record["proof"]) + + def test_recovery_record_carries_no_secret_material(self): + record = issue_lock_recovery.build_recovery_record( + assess(), recovered_at="2026-07-18T23:21:40Z" + ) + blob = repr(record).lower() + for banned in ("token", "password", "authorization", "secret", "api_key"): + self.assertNotIn(banned, blob) + + def test_recovered_lock_satisfies_update_by_merge_ownership(self): + # After recovery the lock is rebound to the live session, so the + # ownership re-check used by gitea_update_pr_branch_by_merge passes. + assessment = assess() + recovered_lock = make_lock(session_pid=os.getpid(), pid=os.getpid()) + recovered_lock["dead_session_recovery"] = ( + issue_lock_recovery.build_recovery_record( + assessment, recovered_at="2026-07-18T23:21:40Z" + ) + ) + freshness = issue_lock_store.assess_lock_freshness(recovered_lock) + self.assertTrue(freshness["live"], freshness) + + verdict = issue_lock_store.verify_lock_for_mutation( + recovered_lock, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=WORKTREE, + ) + self.assertTrue(verdict["proven"], verdict["reasons"]) + self.assertFalse(verdict["block"]) + + def test_pre_recovery_lock_fails_ownership_check(self): + # Guards against a false positive above: the dead-PID lock must fail. + verdict = issue_lock_store.verify_lock_for_mutation( + make_lock(), + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=WORKTREE, + ) + self.assertTrue(verdict["block"]) + self.assertTrue(any("not live" in reason for reason in verdict["reasons"])) + + def test_no_manual_file_seeding_required(self): + # The whole decision is reachable from the durable record plus live + # observation; nothing is written to disk to reach a verdict. + self.assertTrue(assess()["recovery_sanctioned"]) + + def test_refusal_message_is_fail_closed(self): + message = issue_lock_recovery.format_recovery_refusal( + assess(porcelain_status=" M x.py\n") + ) + self.assertIn("fail closed", message) + self.assertIn("recovery refused", message.lower()) + + +if __name__ == "__main__": + unittest.main()