diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 939458e..390c288 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2274,7 +2274,14 @@ def _enforce_locked_issue_duplicate_recheck( org: str | None = None, repo: str | None = None, ) -> dict | None: - """Re-check duplicate-work gates for the locked issue (#400).""" + """Re-check duplicate-work gates for the locked issue (#400). + + #768 AC2: when the lock being re-checked carries a sanctioned dead-session + recovery, carry that server-derived owning-PR evidence into the gate. The + commit and create-PR phases run in their own calls, long after the recovery + assessment ended, so without this they re-block the very PR the recovery + already proved belongs to this author. + """ lock_data = _load_existing_issue_lock() if not lock_data: return None @@ -2297,6 +2304,9 @@ def _enforce_locked_issue_duplicate_recheck( auth=auth, locked_branch=locked_branch, phase=phase, + recovered_owning_pr=issue_lock_recovery.recovered_owning_pr_from_lock( + lock_data + ), ) if gate.get("block"): return gate @@ -3403,6 +3413,24 @@ def gitea_lock_issue( recovery_pr_number = pull.get("number") break recovery_claimant = _work_lease_claimant(h) + # #768: the recovering author's clean worktree is, by construction, one + # commit ahead of the head recorded at lock time — committing is the + # only sanctioned way to clean it without discarding the work. Observe + # the ancestry here, server-side, so the assessor can tell an honest + # fast-forward from a rewritten head. Probed only when the heads + # actually differ, and never from any caller-supplied value. + recovery_local_head = git_state.get("head_sha") + recovery_ancestry: dict | None = None + if ( + recovery_remote_head + and recovery_local_head + and recovery_remote_head != recovery_local_head + ): + recovery_ancestry = issue_lock_worktree.read_head_ancestry( + resolved_worktree, + ancestor_sha=recovery_remote_head, + descendant_sha=recovery_local_head, + ) recovery_assessment = issue_lock_recovery.assess_dead_session_lock_recovery( existing_issue_lock, issue_number=issue_number, @@ -3422,6 +3450,7 @@ def gitea_lock_issue( competing_live_locks=issue_lock_store.list_live_locks(), candidate_branches=recovery_candidates, current_pid=os.getpid(), + head_ancestry=recovery_ancestry, ) recovery_sanctioned = bool( @@ -3629,6 +3658,15 @@ def gitea_assess_work_issue_duplicate( } h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) + # #768 AC2: recovery evidence comes only from the durable lock this session + # already holds, and only when that lock is for the issue being assessed. + # Nothing here is reachable through this tool's parameters. + recovered_owning_pr = None + lock_data = _load_existing_issue_lock() + if lock_data and int(lock_data.get("issue_number") or 0) == int(issue_number): + recovered_owning_pr = issue_lock_recovery.recovered_owning_pr_from_lock( + lock_data + ) gate = _assess_issue_duplicate_gate( issue_number, h=h, @@ -3637,6 +3675,7 @@ def gitea_assess_work_issue_duplicate( auth=auth, locked_branch=branch_name, phase=phase, + recovered_owning_pr=recovered_owning_pr, ) return {"success": not gate.get("block"), **gate} @@ -16116,6 +16155,16 @@ def _prove_author_ownership_for_pr( "durable lock, or branch lock (fail closed; issue_number need not " "equal pr_number when the tracking issue is linked)" ) + # #768 AC2: when the lock that proved ownership was itself a sanctioned + # dead-session recovery, carry that server-derived evidence into the push + # gate rather than making it re-derive ownership blind. ``recorded_head`` + # and ``accepted_head`` let the caller see that the PR head it is about to + # advance is the one recovery already sanctioned, not a foreign head. + recovered_owning_pr = None + if proven and lock_record: + candidate = issue_lock_recovery.recovered_owning_pr_from_lock(lock_record) + if candidate and int(candidate.get("pr_number") or 0) == int(pr_number): + recovered_owning_pr = candidate return { "proven": proven, "has_author_lock": proven, @@ -16124,6 +16173,7 @@ def _prove_author_ownership_for_pr( "matched_issue": matched_issue, "matched_via": matched_via, "lock_record": lock_record, + "recovered_owning_pr": recovered_owning_pr, "reasons": reasons, } @@ -16586,6 +16636,9 @@ def gitea_update_pr_branch_by_merge( "linked_issues": ownership.get("linked_issues"), "matched_issue": ownership.get("matched_issue"), "matched_via": ownership.get("matched_via"), + # #768 AC2: present only when a sanctioned dead-session recovery already + # proved this exact PR belongs to this author. + "recovered_owning_pr": ownership.get("recovered_owning_pr"), } preflight["host"] = h preflight["org"] = o diff --git a/issue_lock_recovery.py b/issue_lock_recovery.py index 3c2e5e5..bc32f37 100644 --- a/issue_lock_recovery.py +++ b/issue_lock_recovery.py @@ -22,6 +22,18 @@ 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. + +#768 extends the head requirement from strict equality to "equal, or a strict +descendant". Equality alone made remediation after a session death unreachable: +recovery needs a clean worktree, the only sanctioned way to clean one without +discarding work is to commit, and committing advances the head past the value +recorded at lock time. A commit that strictly descends from the recorded head, +on the same branch, in the same worktree, by the same claimant, preserves +everything equality protected — the recorded head is still reachable, still an +ancestor, still unmodified — so it is accepted, and nothing else is. The +descendant fact is observed server-side by +``issue_lock_worktree.read_head_ancestry`` and handed in as ``head_ancestry``; +no caller can assert it. """ from __future__ import annotations @@ -40,6 +52,10 @@ REFUSED = "REFUSED" # Durable fields a lock must carry before it can be considered at all. REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path") +# How the clean local head relates to the head recorded at lock time (#768). +HEAD_RELATION_EQUAL = "equal" +HEAD_RELATION_STRICT_DESCENDANT = "strict_descendant" + def _same_realpath(left: str | None, right: str | None) -> bool: if not left or not right: @@ -87,6 +103,68 @@ def _malformed_reasons(lock: Mapping[str, Any]) -> list[str]: return missing +def _assess_strict_descendant( + head_ancestry: Mapping[str, Any] | None, + *, + recorded_head: str, + local_head: str, +) -> tuple[bool, list[str]]: + """Is ``local_head`` a proven strict descendant of ``recorded_head`` (#768)? + + ``head_ancestry`` is the server-side git observation from + ``issue_lock_worktree.read_head_ancestry``. Its own ``ancestor_sha`` / + ``descendant_sha`` are re-checked against the heads this assessment is + actually reasoning about, so a probe taken for some other pair of commits — + stale, mismatched, or hand-built — can never authorize a waiver. + + Returns ``(proven, notes)``. Notes name the exact missing element so a + refused caller sees why, never a bare "unproven". + """ + if not isinstance(head_ancestry, Mapping): + return False, [ + "no server-derived ancestry observation was available; a local head " + "that differs from the recorded head cannot be accepted" + ] + + notes: list[str] = [] + probe_ancestor = _text(head_ancestry.get("ancestor_sha")) + probe_descendant = _text(head_ancestry.get("descendant_sha")) + if probe_ancestor != recorded_head or probe_descendant != local_head: + return False, [ + f"ancestry observation covers {probe_ancestor or 'unknown'} -> " + f"{probe_descendant or 'unknown'}, not the heads under assessment " + f"({recorded_head} -> {local_head})" + ] + if not head_ancestry.get("probe_ok"): + notes.extend( + list(head_ancestry.get("reasons") or []) + or ["ancestry probe did not complete; ancestry unproven"] + ) + return False, notes + if not head_ancestry.get("ancestor_present"): + return False, [ + f"recorded head {recorded_head} is no longer reachable; a rewritten " + "or force-moved head cannot be recovered" + ] + if not head_ancestry.get("is_strict_descendant"): + notes.extend( + list(head_ancestry.get("reasons") or []) + or [ + f"local head {local_head} is not a strict descendant of the " + f"recorded head {recorded_head}" + ] + ) + return False, notes + + proof = _text(head_ancestry.get("proof")) or ( + f"{recorded_head} is an ancestor of {local_head}" + ) + return True, [ + f"local head {local_head} strictly descends from recorded head " + f"{recorded_head} ({proof})" + ] + + def assess_dead_session_lock_recovery( existing_lock: Mapping[str, Any] | None, *, @@ -107,6 +185,7 @@ def assess_dead_session_lock_recovery( competing_live_locks: Sequence[Mapping[str, Any]] | None = None, candidate_branches: Iterable[str] | None = None, current_pid: int | None = None, + head_ancestry: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Decide whether a dead-session author lock may be natively recovered. @@ -218,31 +297,66 @@ def assess_dead_session_lock_recovery( ) evidence["dirty_files"] = dirty_files - # ── Head agreement: local == remote == PR ─────────────────────────────── + # ── Head agreement: local is the recorded head, or strictly descends it ── + # The recorded head is what the remote branch still carries. A local head + # equal to it is the #753 case. A local head that strictly descends from it + # is the #768 case: the author committed remediation, which is the only way + # to reach the clean worktree recovery itself demands. local_head = _text(head_sha) remote_head = _text(remote_head_sha) + head_relation: str | None = None + ancestry_proof: str | None = None 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}" - ) + if local_head and remote_head: + if local_head == remote_head: + head_relation = HEAD_RELATION_EQUAL + else: + descends, notes = _assess_strict_descendant( + head_ancestry, + recorded_head=remote_head, + local_head=local_head, + ) + if descends: + head_relation = HEAD_RELATION_STRICT_DESCENDANT + ancestry_proof = notes[0] if notes else None + else: + reasons.append( + f"local head {local_head} does not match remote branch head " + f"{remote_head}" + ) + reasons.extend(notes) evidence["local_head"] = local_head or None evidence["remote_head"] = remote_head or None + # ``recorded_head`` is the head recovery is being measured against; + # ``accepted_head`` is the head this recovery actually adopts. They differ + # only in the descendant case, and downstream gates need both (#768 AC2/AC7). + evidence["recorded_head"] = remote_head or None + evidence["accepted_head"] = local_head or None + evidence["head_relation"] = head_relation + evidence["ancestry_proof"] = ancestry_proof 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}" - ) + # A descendant recovery has not been published yet, so the open PR + # legitimately still points at the recorded head. Any other + # disagreement is a real mismatch. + if not ( + head_relation == HEAD_RELATION_STRICT_DESCENDANT + and remote_head + and pr_head == remote_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) @@ -335,16 +449,14 @@ def assess_dead_session_lock_recovery( 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, - ) + proof = [ + 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" + ] + if head_relation == HEAD_RELATION_STRICT_DESCENDANT and ancestry_proof: + proof.append(ancestry_proof) + return _result(RECOVERY_SANCTIONED, True, proof, evidence) def _result( @@ -374,10 +486,16 @@ def owning_pr_recovery_evidence( lock already owns" apart from "a competing duplicate PR". Returns ``None`` unless recovery was actually granted and the assessment's - own evidence names exactly one owning PR whose head agrees with the local - and remote heads. Nothing here is caller-supplied: every field is copied - from evidence the assessor built out of durable lock state plus live + own evidence names exactly one owning PR whose head agrees with the heads + the assessor accepted. Nothing here is caller-supplied: every field is + copied from evidence the assessor built out of durable lock state plus live git/Gitea observation, so a caller cannot manufacture an exemption. + + #768: a descendant recovery carries two heads. ``head_sha`` stays the head + the open PR currently shows (the recorded head, since the remediation is not + published yet) and ``accepted_head`` is the local descendant that + publication will move it to. Downstream gates accept either, so the + exemption survives the very push it exists to permit. """ if not isinstance(assessment, Mapping): return None @@ -391,13 +509,28 @@ def owning_pr_recovery_evidence( pr_head = _text(evidence.get("pr_head")) local_head = _text(evidence.get("local_head")) remote_head = _text(evidence.get("remote_head")) + recorded_head = _text(evidence.get("recorded_head")) or remote_head + accepted_head = _text(evidence.get("accepted_head")) or local_head + relation = _text(evidence.get("head_relation")) or HEAD_RELATION_EQUAL raw_pr_number = evidence.get("pr_number") if raw_pr_number is None or not branch_name or not pr_head: return None # The assessor already required these to agree. Re-check, so a truncated or # hand-built evidence map can never authorize an exemption. - if pr_head != local_head or pr_head != remote_head: + if relation == HEAD_RELATION_EQUAL: + if pr_head != local_head or pr_head != remote_head: + return None + elif relation == HEAD_RELATION_STRICT_DESCENDANT: + # The PR must still be at the recorded head, and the accepted head must + # actually be a different commit — otherwise this is not a descendant. + if not recorded_head or pr_head != recorded_head: + return None + if not accepted_head or accepted_head == recorded_head: + return None + if accepted_head != local_head: + return None + else: return None try: pr_number = int(raw_pr_number) @@ -410,6 +543,77 @@ def owning_pr_recovery_evidence( "pr_number": pr_number, "branch_name": branch_name, "head_sha": pr_head, + "recorded_head": recorded_head or None, + "accepted_head": accepted_head or None, + "head_relation": relation, + } + + +def recovered_owning_pr_from_lock( + lock_record: Mapping[str, Any] | None, +) -> dict[str, Any] | None: + """Rebuild owning-PR recovery evidence from a persisted lock (#768 AC2). + + ``gitea_lock_issue`` holds the live assessment only for the duration of the + lock call. The commit, push, create-PR, and duplicate-assessment gates run + later, in their own calls, and re-derive ownership from scratch — so an open + PR that recovery already proved belongs to this author reappears there as + competing duplicate work. + + This reads the same proof back out of the durable ``dead_session_recovery`` + block that only the server writes, on a lock the caller must already own. + It is a re-read of server-derived state, not a new assertion: a caller that + could forge this could equally forge the lock file itself, which every other + ownership gate already treats as authoritative. + """ + if not isinstance(lock_record, Mapping): + return None + record = lock_record.get("dead_session_recovery") + if not isinstance(record, Mapping) or not record.get("recovered"): + return None + + branch_name = _text(record.get("branch_name")) or _text( + lock_record.get("branch_name") + ) + pr_head = _text(record.get("pr_head")) + recorded_head = _text(record.get("recorded_head")) or _text( + record.get("remote_head") + ) + accepted_head = _text(record.get("accepted_head")) or _text( + record.get("local_head") + ) + relation = _text(record.get("head_relation")) or HEAD_RELATION_EQUAL + raw_pr_number = record.get("pr_number") + raw_issue_number = lock_record.get("issue_number") + + if raw_pr_number is None or raw_issue_number is None: + return None + if not branch_name or not pr_head: + return None + if relation == HEAD_RELATION_EQUAL: + if accepted_head and accepted_head != pr_head: + return None + elif relation == HEAD_RELATION_STRICT_DESCENDANT: + if not recorded_head or pr_head != recorded_head: + return None + if not accepted_head or accepted_head == recorded_head: + return None + else: + return None + try: + pr_number = int(raw_pr_number) + issue_number = int(raw_issue_number) + except (TypeError, ValueError): + return None + + return { + "issue_number": issue_number, + "pr_number": pr_number, + "branch_name": branch_name, + "head_sha": pr_head, + "recorded_head": recorded_head or None, + "accepted_head": accepted_head or None, + "head_relation": relation, } @@ -418,7 +622,13 @@ def build_recovery_record( *, recovered_at: str, ) -> dict[str, Any]: - """Durable, secret-free provenance for a completed recovery (#753 AC2/AC6).""" + """Durable, secret-free provenance for a completed recovery (#753 AC2/AC6). + + #768 AC7: a granted recovery records, atomically with the lock itself, both + session identities, the head it was measured against, the head it adopted, + how those two relate, and the ancestry proof — so a descendant recovery can + be audited after the fact without re-running any probe. + """ evidence = dict(assessment.get("evidence") or {}) return { "recovered": True, @@ -431,6 +641,10 @@ def build_recovery_record( "worktree_path": evidence.get("locked_worktree_path"), "local_head": evidence.get("local_head"), "remote_head": evidence.get("remote_head"), + "recorded_head": evidence.get("recorded_head"), + "accepted_head": evidence.get("accepted_head"), + "head_relation": evidence.get("head_relation"), + "ancestry_proof": evidence.get("ancestry_proof"), "pr_head": evidence.get("pr_head"), "pr_number": evidence.get("pr_number"), "identity": evidence.get("locked_identity"), diff --git a/issue_lock_worktree.py b/issue_lock_worktree.py index d14f5a2..f0f56bd 100644 --- a/issue_lock_worktree.py +++ b/issue_lock_worktree.py @@ -30,6 +30,108 @@ def resolve_author_worktree_path( return os.path.realpath(os.path.abspath(path)) +def read_head_ancestry( + worktree_path: str, + *, + ancestor_sha: str | None, + descendant_sha: str | None, +) -> dict: + """Observe whether ``descendant_sha`` strictly descends from ``ancestor_sha`` (#768). + + Server-side git observation for dead-session lock recovery. The recovering + author's only reachable clean-worktree state is one commit *ahead* of the + head recorded at lock time, so recovery needs to know whether that commit + extends the recorded head or replaces it. + + Reports facts only; the disposition lives in ``issue_lock_recovery``. Every + field is read from git in the declared worktree — nothing here is supplied + by, or reachable from, an MCP caller (#768 AC6). + + ``ancestor_present`` proves the recorded head is still reachable, which is + what separates an honest fast-forward from a rewritten or force-moved + history: a rewritten recorded head leaves the object graph and the probe + fails closed. + """ + path = (worktree_path or "").strip() + ancestor = (ancestor_sha or "").strip() + descendant = (descendant_sha or "").strip() + result: dict = { + "ancestor_sha": ancestor or None, + "descendant_sha": descendant or None, + "probe_ok": False, + "ancestor_present": False, + "descendant_present": False, + "is_ancestor": False, + "is_strict_descendant": False, + "proof": None, + "reasons": [], + } + if not path or not ancestor or not descendant: + result["reasons"].append( + "ancestry probe requires a worktree path and both commit SHAs" + ) + return result + + def _present(sha: str) -> bool: + res = subprocess.run( + ["git", "-C", path, "rev-parse", "--verify", "--quiet", f"{sha}^{{commit}}"], + capture_output=True, + text=True, + check=False, + ) + return res.returncode == 0 + + try: + result["ancestor_present"] = _present(ancestor) + result["descendant_present"] = _present(descendant) + except OSError as exc: # git unavailable — fail closed, never assume + result["reasons"].append(f"ancestry probe could not run: {exc}") + return result + + if not result["ancestor_present"]: + result["reasons"].append( + f"recorded head {ancestor} is not reachable in '{path}'; history may " + "have been rewritten or force-moved" + ) + if not result["descendant_present"]: + result["reasons"].append( + f"local head {descendant} is not reachable in '{path}'" + ) + if not (result["ancestor_present"] and result["descendant_present"]): + return result + + probe = subprocess.run( + ["git", "-C", path, "merge-base", "--is-ancestor", ancestor, descendant], + capture_output=True, + text=True, + check=False, + ) + # 0 = is an ancestor, 1 = is not. Anything else is a failed probe, not a "no". + if probe.returncode not in (0, 1): + result["reasons"].append( + f"ancestry probe failed with exit {probe.returncode}; ancestry unproven" + ) + return result + + result["probe_ok"] = True + result["is_ancestor"] = probe.returncode == 0 + result["is_strict_descendant"] = result["is_ancestor"] and ancestor != descendant + result["proof"] = ( + f"git -C merge-base --is-ancestor {ancestor} {descendant} " + f"-> exit {probe.returncode}" + ) + if not result["is_ancestor"]: + result["reasons"].append( + f"local head {descendant} does not descend from recorded head {ancestor}" + ) + elif not result["is_strict_descendant"]: + result["reasons"].append( + f"local head {descendant} equals the recorded head; no descendant " + "recovery is involved" + ) + return result + + def read_worktree_git_state( worktree_path: str, extra_bases: tuple[str, ...] | list[str] = (), diff --git a/issue_work_duplicate_gate.py b/issue_work_duplicate_gate.py index 81350c7..d327609 100644 --- a/issue_work_duplicate_gate.py +++ b/issue_work_duplicate_gate.py @@ -122,18 +122,30 @@ def _assess_owning_pr_exemption( f"the recovered branch '{token_branch}' (no owning-PR exemption)" ) return False, notes - if not token_head or not only_sha or only_sha != token_head: + # #768: a descendant recovery is measured against the head the PR still + # shows, then publishes the local descendant — so the live PR head is the + # recorded head before that push and the accepted head after it. Both are + # server-derived and name the same owned PR, so both are accepted; anything + # else still fails closed. + token_accepted = str(recovered_owning_pr.get("accepted_head") or "").strip() + acceptable_heads = [head for head in (token_head, token_accepted) if head] + if not acceptable_heads or not only_sha or only_sha not in acceptable_heads: notes.append( f"open PR #{only_number} head {only_sha or 'unknown'} does not " - f"match the recovered head {token_head or 'unknown'} " - "(no owning-PR exemption)" + f"match the recovered head {token_head or 'unknown'}" + + ( + f" or the accepted head {token_accepted}" + if token_accepted and token_accepted != token_head + else "" + ) + + " (no owning-PR exemption)" ) return False, notes return True, [ f"open PR #{only_number} is the exact PR already owned by the " f"recovering lock for issue #{issue_number} (branch '{token_branch}', " - f"head {token_head}); not duplicate work" + f"head {only_sha}); not duplicate work" ] diff --git a/tests/test_issue_755_owning_pr_recovery.py b/tests/test_issue_755_owning_pr_recovery.py index d9fc617..001956f 100644 --- a/tests/test_issue_755_owning_pr_recovery.py +++ b/tests/test_issue_755_owning_pr_recovery.py @@ -73,12 +73,21 @@ def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE): def sanctioned_token( issue_number=ISSUE, pr_number=OWNING_PR, branch=BRANCH, head=HEAD ): - """The evidence shape the server derives from a granted recovery.""" + """The evidence shape the server derives from a granted recovery. + + #768 extends the token with recorded/accepted heads and the head relation + so a strict-descendant recovery can still exempt the owning PR after the + remediation commit lands. Exact-head recovery (#753/#755) reports equal + heads under the same shape. + """ return { "issue_number": issue_number, "pr_number": pr_number, "branch_name": branch, "head_sha": head, + "recorded_head": head, + "accepted_head": head, + "head_relation": issue_lock_recovery.HEAD_RELATION_EQUAL, } diff --git a/tests/test_issue_768_strict_descendant_recovery.py b/tests/test_issue_768_strict_descendant_recovery.py new file mode 100644 index 0000000..7a61852 --- /dev/null +++ b/tests/test_issue_768_strict_descendant_recovery.py @@ -0,0 +1,551 @@ +"""Strict-descendant dead-session recovery (#768). + +After a dead author session, a preserved clean remediation commit that strictly +descends from the head recorded at lock time must be recoverable so the author +can publish. Equality alone is still accepted (#753); every other divergence +must keep failing closed. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import issue_lock_recovery # noqa: E402 +import issue_lock_store # noqa: E402 +import issue_lock_worktree # noqa: E402 +import issue_work_duplicate_gate # noqa: E402 + +ISSUE = 7680 +PR_NUMBER = 7681 +BRANCH = f"fix/issue-{ISSUE}-descendant-recovery" +WORKTREE = "/scratch/wt-768" +RECORDED = "a" * 40 +DESCENDANT = "c" * 40 +DIVERGED = "d" * 40 +BEHIND = "b" * 40 +IDENTITY = "example-user" +PROFILE = "example-author" + + +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") + ) + + +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 ancestry_ok( + *, + ancestor: str = RECORDED, + descendant: str = DESCENDANT, + is_strict: bool = True, + probe_ok: bool = True, + ancestor_present: bool = True, + reasons: list[str] | None = None, +) -> dict: + return { + "ancestor_sha": ancestor, + "descendant_sha": descendant, + "probe_ok": probe_ok, + "ancestor_present": ancestor_present, + "descendant_present": True, + "is_ancestor": is_strict or ancestor == descendant, + "is_strict_descendant": is_strict, + "proof": f"git merge-base --is-ancestor {ancestor} {descendant} -> exit 0", + "reasons": list(reasons or []), + } + + +def assess(**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": RECORDED, + "remote_head_sha": RECORDED, + "pr_head_sha": RECORDED, + "pr_number": PR_NUMBER, + "competing_live_locks": [], + "candidate_branches": [BRANCH], + "current_pid": os.getpid(), + "head_ancestry": None, + } + kwargs.update(overrides) + lock = kwargs.pop("lock", None) + return issue_lock_recovery.assess_dead_session_lock_recovery( + make_lock() if lock is None else lock, **kwargs + ) + + +class TestExactHeadRecoveryStillSucceeds(unittest.TestCase): + def test_equal_heads_still_sanctioned(self): + result = assess() + self.assertTrue(result["recovery_sanctioned"], result["reasons"]) + self.assertEqual( + result["evidence"]["head_relation"], + issue_lock_recovery.HEAD_RELATION_EQUAL, + ) + self.assertEqual(result["evidence"]["recorded_head"], RECORDED) + self.assertEqual(result["evidence"]["accepted_head"], RECORDED) + + def test_exact_match_record_carries_relation(self): + record = issue_lock_recovery.build_recovery_record( + assess(), recovered_at="2026-07-20T00:00:00Z" + ) + self.assertEqual(record["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL) + self.assertEqual(record["recorded_head"], RECORDED) + self.assertEqual(record["accepted_head"], RECORDED) + + +class TestStrictDescendantRecoverySucceeds(unittest.TestCase): + def test_clean_strict_descendant_recovers(self): + result = assess( + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok(), + ) + self.assertTrue(result["recovery_sanctioned"], result["reasons"]) + self.assertEqual( + result["evidence"]["head_relation"], + issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT, + ) + self.assertEqual(result["evidence"]["recorded_head"], RECORDED) + self.assertEqual(result["evidence"]["accepted_head"], DESCENDANT) + self.assertIsNotNone(result["evidence"]["ancestry_proof"]) + self.assertTrue( + any("strictly descends" in r for r in result["reasons"]), + result["reasons"], + ) + + def test_pr_still_at_recorded_head_is_ok_for_descendant(self): + # Remediation is local only; open PR still points at the recorded head. + result = assess( + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok(), + ) + self.assertTrue(result["recovery_sanctioned"], result["reasons"]) + + def test_recovery_record_names_both_heads_and_proof(self): + result = assess( + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok(), + ) + record = issue_lock_recovery.build_recovery_record( + result, recovered_at="2026-07-20T00:00:00Z" + ) + self.assertEqual(record["recorded_head"], RECORDED) + self.assertEqual(record["accepted_head"], DESCENDANT) + self.assertEqual( + record["head_relation"], + issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT, + ) + self.assertIn("strictly descends", record["ancestry_proof"] or "") + self.assertEqual(record["prior_session_pid"], result["evidence"]["prior_session_pid"]) + self.assertEqual(record["replacement_session_pid"], os.getpid()) + + +class TestDescendantEvidenceReachesPublicationGates(unittest.TestCase): + def _descendant_assessment(self): + return assess( + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok(), + ) + + def test_owning_pr_evidence_carries_accepted_head(self): + token = issue_lock_recovery.owning_pr_recovery_evidence( + self._descendant_assessment() + ) + self.assertIsNotNone(token) + assert token is not None + self.assertEqual(token["head_sha"], RECORDED) + self.assertEqual(token["accepted_head"], DESCENDANT) + self.assertEqual(token["recorded_head"], RECORDED) + self.assertEqual( + token["head_relation"], + issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT, + ) + + def test_persisted_lock_rebuilds_owning_pr_evidence(self): + assessment = self._descendant_assessment() + record = issue_lock_recovery.build_recovery_record( + assessment, recovered_at="2026-07-20T00:00:00Z" + ) + lock = make_lock(dead_session_recovery=record) + token = issue_lock_recovery.recovered_owning_pr_from_lock(lock) + self.assertIsNotNone(token) + assert token is not None + self.assertEqual(token["pr_number"], PR_NUMBER) + self.assertEqual(token["head_sha"], RECORDED) + self.assertEqual(token["accepted_head"], DESCENDANT) + + def test_duplicate_gate_accepts_pr_at_recorded_or_accepted_head(self): + token = issue_lock_recovery.owning_pr_recovery_evidence( + self._descendant_assessment() + ) + for live_sha in (RECORDED, DESCENDANT): + with self.subTest(live_sha=live_sha): + gate = issue_work_duplicate_gate.assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[ + { + "number": PR_NUMBER, + "title": f"Closes #{ISSUE}", + "body": f"Closes #{ISSUE}", + "head": {"ref": BRANCH, "sha": live_sha}, + } + ], + branch_names=[BRANCH], + claim_entry={"status": "unclaimed"}, + locked_branch=BRANCH, + phase=issue_work_duplicate_gate.PHASE_COMMIT, + recovered_owning_pr=token, + ) + self.assertFalse(gate["block"], gate) + self.assertTrue(gate["owning_pr_recovery_exempted"]) + + def test_duplicate_gate_still_rejects_foreign_head(self): + token = issue_lock_recovery.owning_pr_recovery_evidence( + self._descendant_assessment() + ) + gate = issue_work_duplicate_gate.assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[ + { + "number": PR_NUMBER, + "title": f"Closes #{ISSUE}", + "body": f"Closes #{ISSUE}", + "head": {"ref": BRANCH, "sha": DIVERGED}, + } + ], + branch_names=[BRANCH], + claim_entry={"status": "unclaimed"}, + locked_branch=BRANCH, + phase=issue_work_duplicate_gate.PHASE_COMMIT, + recovered_owning_pr=token, + ) + self.assertTrue(gate["block"]) + self.assertFalse(gate["owning_pr_recovery_exempted"]) + + +class TestDirtyDescendantRejected(unittest.TestCase): + def test_dirty_descendant_refused(self): + result = assess( + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok(), + porcelain_status=" M issue_lock_recovery.py\n", + ) + self.assertFalse(result["recovery_sanctioned"]) + self.assertTrue(any("dirty" in r.lower() for r in result["reasons"])) + + +class TestDivergedAndBehindRejected(unittest.TestCase): + def test_diverged_head_refused(self): + result = assess( + head_sha=DIVERGED, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok( + ancestor=RECORDED, + descendant=DIVERGED, + is_strict=False, + reasons=[ + f"local head {DIVERGED} does not descend from recorded head " + f"{RECORDED}" + ], + ), + ) + self.assertFalse(result["recovery_sanctioned"]) + self.assertIsNone(result["evidence"].get("head_relation")) + self.assertTrue( + any("does not match remote" in r for r in result["reasons"]), + result["reasons"], + ) + + def test_local_behind_recorded_refused(self): + # merge-base --is-ancestor RECORDED BEHIND is false when BEHIND is ancestor. + result = assess( + head_sha=BEHIND, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok( + ancestor=RECORDED, + descendant=BEHIND, + is_strict=False, + reasons=[ + f"local head {BEHIND} does not descend from recorded head " + f"{RECORDED}" + ], + ), + ) + self.assertFalse(result["recovery_sanctioned"]) + self.assertTrue( + any("does not match remote" in r or "not a strict descendant" in r + for r in result["reasons"]), + result["reasons"], + ) + + +class TestUnrelatedAndMalformedAncestryRejected(unittest.TestCase): + def test_missing_ancestry_observation_fails_closed(self): + result = assess( + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=None, + ) + self.assertFalse(result["recovery_sanctioned"]) + self.assertTrue( + any("ancestry" in r.lower() for r in result["reasons"]), + result["reasons"], + ) + + def test_mismatched_probe_pair_fails_closed(self): + # Observation for a different commit pair must not authorize this pair. + result = assess( + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok(ancestor=DIVERGED, descendant=DESCENDANT), + ) + self.assertFalse(result["recovery_sanctioned"]) + self.assertTrue( + any("not the heads under assessment" in r for r in result["reasons"]), + result["reasons"], + ) + + def test_rewritten_recorded_head_fails_closed(self): + result = assess( + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok(ancestor_present=False, is_strict=False), + ) + self.assertFalse(result["recovery_sanctioned"]) + self.assertTrue( + any("no longer reachable" in r or "rewritten" in r + for r in result["reasons"]), + result["reasons"], + ) + + def test_failed_probe_fails_closed(self): + result = assess( + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok( + probe_ok=False, + is_strict=False, + reasons=["ancestry probe failed with exit 128; ancestry unproven"], + ), + ) + self.assertFalse(result["recovery_sanctioned"]) + + def test_pr_head_not_equal_to_recorded_blocks_descendant(self): + result = assess( + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=DIVERGED, + head_ancestry=ancestry_ok(), + ) + self.assertFalse(result["recovery_sanctioned"]) + self.assertTrue( + any("open PR" in r and "does not match" in r for r in result["reasons"]), + result["reasons"], + ) + + +class TestLiveOwnerStillRejected(unittest.TestCase): + def test_live_prior_pid_refused_even_with_descendant_proof(self): + result = assess( + lock=make_lock(session_pid=os.getpid()), + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok(), + ) + self.assertFalse(result["recovery_sanctioned"]) + self.assertTrue( + any("still alive" in r or "live" in r.lower() for r in result["reasons"]), + result["reasons"], + ) + + +class TestDiagnosticsIdentifyDisposition(unittest.TestCase): + def test_equal_disposition_named(self): + result = assess() + self.assertEqual( + result["evidence"]["head_relation"], + issue_lock_recovery.HEAD_RELATION_EQUAL, + ) + + def test_descendant_disposition_named(self): + result = assess( + head_sha=DESCENDANT, + remote_head_sha=RECORDED, + pr_head_sha=RECORDED, + head_ancestry=ancestry_ok(), + ) + self.assertEqual( + result["evidence"]["head_relation"], + issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT, + ) + + def test_rejected_divergence_has_no_accepted_relation(self): + result = assess( + head_sha=DIVERGED, + remote_head_sha=RECORDED, + head_ancestry=None, + ) + self.assertIsNone(result["evidence"].get("head_relation")) + message = issue_lock_recovery.format_recovery_refusal(result) + self.assertIn("fail closed", message) + self.assertIn("does not match remote", message) + + +class TestReadHeadAncestryRealGit(unittest.TestCase): + def _git(self, repo: str, *args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["git", "-C", repo, *args], + capture_output=True, + text=True, + check=True, + ) + + def _init_repo_with_chain(self) -> tuple[str, str, str, str]: + """Return (repo, parent_sha, child_sha, sibling_sha).""" + repo = tempfile.mkdtemp(prefix="issue-768-ancestry-") + self._git(repo, "init") + self._git(repo, "config", "user.email", "test@example.com") + self._git(repo, "config", "user.name", "Test") + path = Path(repo) / "f.txt" + path.write_text("one\n") + self._git(repo, "add", "f.txt") + self._git(repo, "commit", "-m", "parent") + parent = self._git(repo, "rev-parse", "HEAD").stdout.strip() + path.write_text("two\n") + self._git(repo, "add", "f.txt") + self._git(repo, "commit", "-m", "child") + child = self._git(repo, "rev-parse", "HEAD").stdout.strip() + # Divergent sibling: branch from parent, then unique commit. + self._git(repo, "checkout", "-B", "side", parent) + path.write_text("side\n") + self._git(repo, "add", "f.txt") + self._git(repo, "commit", "-m", "sibling") + sibling = self._git(repo, "rev-parse", "HEAD").stdout.strip() + self._git(repo, "checkout", "-B", "main", child) + return repo, parent, child, sibling + + def test_strict_descendant_observation(self): + repo, parent, child, _sibling = self._init_repo_with_chain() + obs = issue_lock_worktree.read_head_ancestry( + repo, ancestor_sha=parent, descendant_sha=child + ) + self.assertTrue(obs["probe_ok"]) + self.assertTrue(obs["ancestor_present"]) + self.assertTrue(obs["is_ancestor"]) + self.assertTrue(obs["is_strict_descendant"]) + self.assertEqual(obs["ancestor_sha"], parent) + self.assertEqual(obs["descendant_sha"], child) + + def test_equal_heads_not_strict_descendant(self): + repo, parent, _child, _sibling = self._init_repo_with_chain() + obs = issue_lock_worktree.read_head_ancestry( + repo, ancestor_sha=parent, descendant_sha=parent + ) + self.assertTrue(obs["probe_ok"]) + self.assertTrue(obs["is_ancestor"]) + self.assertFalse(obs["is_strict_descendant"]) + + def test_diverged_not_ancestor(self): + repo, _parent, child, sibling = self._init_repo_with_chain() + # child and sibling share a parent but neither descends from the other. + obs = issue_lock_worktree.read_head_ancestry( + repo, ancestor_sha=child, descendant_sha=sibling + ) + self.assertTrue(obs["probe_ok"]) + self.assertFalse(obs["is_ancestor"]) + self.assertFalse(obs["is_strict_descendant"]) + + def test_missing_sha_fails_closed(self): + repo, _parent, child, _ = self._init_repo_with_chain() + obs = issue_lock_worktree.read_head_ancestry( + repo, ancestor_sha="0" * 40, descendant_sha=child + ) + self.assertFalse(obs["probe_ok"]) + self.assertFalse(obs["ancestor_present"]) + + def test_end_to_end_real_git_descendant_recovery(self): + repo, parent, child, _sibling = self._init_repo_with_chain() + obs = issue_lock_worktree.read_head_ancestry( + repo, ancestor_sha=parent, descendant_sha=child + ) + result = assess( + worktree_path=repo, + head_sha=child, + remote_head_sha=parent, + pr_head_sha=parent, + head_ancestry=obs, + lock=make_lock(worktree_path=repo), + ) + self.assertTrue(result["recovery_sanctioned"], result["reasons"]) + self.assertEqual( + result["evidence"]["head_relation"], + issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT, + ) + + +if __name__ == "__main__": + unittest.main()