diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 81771f5..83bfd05 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1700,6 +1700,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 @@ -2130,6 +2131,7 @@ def _assess_issue_duplicate_gate( auth: str, locked_branch: str | None = None, phase: str, + recovered_owning_pr: dict | None = None, ) -> dict: open_prs, branch_names, claim_entry = _collect_issue_duplicate_context( h, o, r, auth, issue_number @@ -2141,6 +2143,7 @@ def _assess_issue_duplicate_gate( claim_entry=claim_entry, locked_branch=locked_branch, phase=phase, + recovered_owning_pr=recovered_owning_pr, ) @@ -3205,8 +3208,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, @@ -3248,6 +3254,83 @@ 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") + ) + # #755: a sanctioned dead-session recovery always has an owning open PR — + # that is what makes it a recovery rather than a fresh claim. Carry the + # server-derived owning-PR evidence into the duplicate-work gate below so + # the PR this lock already owns is not mistaken for competing duplicate + # work. Withheld (None) unless recovery was granted, so the ordinary + # duplicate blocker is untouched on every other path. + recovered_owning_pr = ( + issue_lock_recovery.owning_pr_recovery_evidence(recovery_assessment) + if recovery_sanctioned + else None + ) lock_assessment = issue_lock_worktree.assess_issue_lock_worktree( worktree_path=resolved_worktree, current_branch=git_state.get("current_branch"), @@ -3255,10 +3338,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) @@ -3270,6 +3363,7 @@ def gitea_lock_issue( auth=auth, locked_branch=branch_name, phase=issue_work_duplicate_gate.PHASE_LOCK, + recovered_owning_pr=recovered_owning_pr, ) if duplicate_gate.get("block"): raise ValueError("; ".join(duplicate_gate.get("reasons") or [ @@ -3321,6 +3415,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 @@ -3354,6 +3456,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"] = ( @@ -15417,46 +15527,151 @@ def _count_commits_behind( return None +def _matching_protection_rule(protections: Any, branch: str) -> dict | None: + """Select the branch-protection rule governing ``branch`` (#751).""" + if not isinstance(protections, list): + return None + for rule in protections: + if not isinstance(rule, dict): + continue + name = (rule.get("branch_name") or rule.get("rule_name") or "").strip() + # Exact match or glob-ish contains for common patterns. + if name == branch or name in ("*", f"{branch}"): + return rule + # Fallback: any protection that mentions the base branch. + for rule in protections: + if not isinstance(rule, dict): + continue + name = (rule.get("branch_name") or rule.get("rule_name") or "").strip() + if name and (branch in name or name.endswith(branch)): + return rule + return None + + +def _branch_protection_policy( + base_url: str, + auth: dict, + *, + base_branch: str, +) -> dict: + """Read the live branch-protection policy for ``base_branch`` (#751). + + Exposes both the current-base rule (``block_on_outdated_branch``) and the + status-check requirement (``enable_status_check`` / ``status_check_contexts``) + from the same payload, so the checks assessor can tell "no checks required" + apart from "required checks pending". + + ``determinable`` is False only when the policy genuinely could not be read + (missing branch or API failure) — never merely because no rule exists. A + successful read that finds no protection for the branch is authoritative + evidence that status checks are not required. + """ + policy: dict[str, Any] = { + "determinable": False, + "protection_found": False, + "requires_current_base": None, + "checks_enabled": None, + "required_contexts": [], + "base_branch": (base_branch or "").strip() or None, + } + branch = (base_branch or "").strip() + if not branch: + return policy + + def _apply(rule: dict) -> None: + policy["protection_found"] = True + if "block_on_outdated_branch" in rule: + policy["requires_current_base"] = bool( + rule.get("block_on_outdated_branch") + ) + if "enable_status_check" in rule: + policy["checks_enabled"] = bool(rule.get("enable_status_check")) + contexts = rule.get("status_check_contexts") + if isinstance(contexts, list): + policy["required_contexts"] = [ + str(ctx).strip() for ctx in contexts if str(ctx or "").strip() + ] + + try: + protections = api_request( + "GET", f"{base_url}/branch_protections", auth + ) + rule = _matching_protection_rule(protections, branch) + if rule is not None: + _apply(rule) + if not policy["protection_found"]: + # Branch payload may embed effective protection. + br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {} + prot = br.get("protection") or br.get("effective_branch_protection") or {} + if isinstance(prot, dict) and prot: + _apply(prot) + policy["determinable"] = True + except Exception: + # Genuine read failure — leave determinable False so callers fail closed. + return policy + + if not policy["protection_found"]: + # Authoritative absence: no protection governs the branch, so no status + # check is required by policy. + policy["checks_enabled"] = False + elif policy["checks_enabled"] is None: + # Protection exists but omits the status-check field entirely: Gitea + # only enforces contexts when the flag is set, so absence means off. + policy["checks_enabled"] = False + + return policy + + def _branch_protection_requires_current_base( base_url: str, auth: dict, *, base_branch: str, ) -> bool | None: - """Read Gitea branch protection ``block_on_outdated_branch`` when present.""" - branch = (base_branch or "").strip() - if not branch: - return None + """Read Gitea branch protection ``block_on_outdated_branch`` when present. + + Thin accessor over :func:`_branch_protection_policy`; return semantics are + unchanged (``None`` when the rule is absent or unreadable). + """ + policy = _branch_protection_policy(base_url, auth, base_branch=base_branch) + return policy.get("requires_current_base") + + +def _commit_checks_snapshot( + base_url: str, + auth: dict, + *, + sha: str, +) -> dict: + """Read the combined commit status *and its context collection* (#751). + + The combined ``state`` alone is not decisive: Gitea reports ``pending`` for + a commit with an empty status-context collection, which is indistinguishable + from executing CI unless the collection itself is inspected. + """ + snapshot: dict[str, Any] = { + "determinable": False, + "combined_state": None, + "statuses": [], + } + head = (sha or "").strip() + if not head: + return snapshot try: - # Prefer the named protection rule matching the base branch. - protections = api_request( - "GET", f"{base_url}/branch_protections", auth - ) or [] - if isinstance(protections, list): - for rule in protections: - if not isinstance(rule, dict): - continue - name = (rule.get("branch_name") or rule.get("rule_name") or "").strip() - # Exact match or glob-ish contains for common patterns. - if name == branch or name in ("*", f"{branch}"): - if "block_on_outdated_branch" in rule: - return bool(rule.get("block_on_outdated_branch")) - # Fallback: any protection that mentions the base branch. - for rule in protections: - if not isinstance(rule, dict): - continue - name = (rule.get("branch_name") or rule.get("rule_name") or "").strip() - if branch in name or name.endswith(branch): - if "block_on_outdated_branch" in rule: - return bool(rule.get("block_on_outdated_branch")) - # Branch payload may embed effective protection. - br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {} - prot = br.get("protection") or br.get("effective_branch_protection") or {} - if isinstance(prot, dict) and "block_on_outdated_branch" in prot: - return bool(prot.get("block_on_outdated_branch")) + payload = api_request( + "GET", f"{base_url}/commits/{head}/status", auth + ) except Exception: - return None - return None + return snapshot + if payload is None: + return snapshot + snapshot["determinable"] = True + if not isinstance(payload, dict): + return snapshot + snapshot["combined_state"] = (payload.get("state") or "").strip().lower() or None + statuses = payload.get("statuses") + snapshot["statuses"] = statuses if isinstance(statuses, list) else [] + return snapshot def _prove_author_ownership_for_pr( @@ -15701,11 +15916,14 @@ def gitea_assess_pr_sync_status( # Fail closed on unknown behind-count only when SHAs differ. commits_behind = 0 if pr_head_sha == base_head_sha else None + # Single live read of the base-branch protection policy; it carries both + # the current-base rule and the status-check requirement (#751). + protection_policy = _branch_protection_policy( + base, auth, base_branch=base_branch + ) if branch_protection_requires_current_base is None: - branch_protection_requires_current_base = ( - _branch_protection_requires_current_base( - base, auth, base_branch=base_branch - ) + branch_protection_requires_current_base = protection_policy.get( + "requires_current_base" ) # When protection cannot be read, fail closed by requiring current base # whenever the PR is behind (safer for protected repos). Callers may pass @@ -15772,23 +15990,30 @@ def gitea_assess_pr_sync_status( except Exception: pass + # ── Live checks derivation (#751) ──────────────────────────────────── + # Classify from the actual status-context collection plus the live + # protection policy. The combined ``state`` is never treated as proof that + # CI is executing, because Gitea reports ``pending`` for an empty + # collection. ``checks_required`` is always derived from live evidence and + # is deliberately not a caller-supplied input, so no session can declare + # checks optional without proof. + checks_snapshot = _commit_checks_snapshot(base, auth, sha=pr_head_sha or "") + checks_classification = pr_sync_status.classify_commit_checks( + combined_state=checks_snapshot.get("combined_state"), + statuses=checks_snapshot.get("statuses"), + checks_enabled=protection_policy.get("checks_enabled"), + required_contexts=protection_policy.get("required_contexts"), + policy_determinable=bool(protection_policy.get("determinable")), + status_determinable=bool(checks_snapshot.get("determinable")), + ) + checks_required = bool(checks_classification.get("checks_required", True)) + caller_supplied_checks_status = checks_status if checks_status is None: - checks_status = "unknown" - # Combined status on PR head when Actions/status API is available. - if pr_head_sha: - try: - st = api_request( - "GET", - f"{base}/commits/{pr_head_sha}/status", - auth, - ) or {} - state = (st.get("state") or "").strip().lower() - if state: - checks_status = state - elif not st: - checks_status = "none" - except Exception: - checks_status = "unknown" + checks_status = checks_classification.get("checks_status") + elif checks_classification.get("checks_status") != pr_sync_status.CHECKS_UNKNOWN: + # Live evidence outranks a caller-supplied value; the override only + # applies when live status could not be classified at all. + checks_status = checks_classification.get("checks_status") assessment = pr_sync_status.assess_pr_sync_status( host=h, @@ -15809,9 +16034,26 @@ def gitea_assess_pr_sync_status( active_reviewer_lease=active_reviewer_lease, active_merger_lease=active_merger_lease, prepared_verdict_head_sha=prepared_verdict_head_sha, + checks_required=checks_required, ) assessment["remote"] = remote if remote in REMOTES else None assessment["base_branch"] = base_branch + # Evidence for the checks decision (#751) — no secrets, read-only. + assessment["checks_evidence"] = { + "combined_state": checks_classification.get("combined_state"), + "context_count": checks_classification.get("context_count"), + "observed_contexts": checks_classification.get("observed_contexts"), + "required_contexts": checks_classification.get("required_contexts"), + "missing_required_contexts": checks_classification.get( + "missing_required_contexts" + ), + "policy_determinable": checks_classification.get("policy_determinable"), + "status_determinable": checks_classification.get("status_determinable"), + "protection_found": protection_policy.get("protection_found"), + "checks_enabled": protection_policy.get("checks_enabled"), + "caller_supplied_checks_status": caller_supplied_checks_status, + "reasons": checks_classification.get("reasons"), + } assessment["success"] = True assessment["performed"] = False return assessment 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..3c2e5e5 --- /dev/null +++ b/issue_lock_recovery.py @@ -0,0 +1,451 @@ +"""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 owning_pr_recovery_evidence( + assessment: Mapping[str, Any] | None, +) -> dict[str, Any] | None: + """Server-derived proof of the open PR a sanctioned recovery already owns (#755). + + A dead-session recovery is, by construction, recovery of work that already + has an open PR — so the duplicate-work gate's linked-open-PR blocker would + otherwise discard every sanctioned recovery. This distils the completed + assessment into the minimum evidence that gate needs to tell "the PR this + 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 + git/Gitea observation, so a caller cannot manufacture an exemption. + """ + if not isinstance(assessment, Mapping): + return None + if assessment.get("outcome") != RECOVERY_SANCTIONED: + return None + if not assessment.get("recovery_sanctioned"): + return None + + evidence = assessment.get("evidence") or {} + branch_name = _text(evidence.get("locked_branch")) + pr_head = _text(evidence.get("pr_head")) + local_head = _text(evidence.get("local_head")) + remote_head = _text(evidence.get("remote_head")) + 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: + return None + try: + pr_number = int(raw_pr_number) + issue_number = int(evidence.get("issue_number")) + except (TypeError, ValueError): + return None + + return { + "issue_number": issue_number, + "pr_number": pr_number, + "branch_name": branch_name, + "head_sha": pr_head, + } + + +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/issue_work_duplicate_gate.py b/issue_work_duplicate_gate.py index fc70437..81350c7 100644 --- a/issue_work_duplicate_gate.py +++ b/issue_work_duplicate_gate.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Any +from typing import Any, Mapping import issue_claim_heartbeat as claim_hb @@ -27,6 +27,116 @@ def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None: return claim_hb._linked_open_pr(issue_number, open_prs) +def _pr_links_issue(issue_number: int, pr: Mapping[str, Any]) -> bool: + """Same linkage rule ``claim_hb._linked_open_pr`` applies, per PR. + + ``_linked_open_pr`` only yields the *first* match, which cannot answer + "is there exactly one linked PR?" — a question the owning-PR exemption + below must answer before it can trust any of them. + """ + pattern = _issue_pattern(issue_number) + head = (pr.get("head") or {}).get("ref") or "" + text = f"{pr.get('title', '')} {pr.get('body', '')}".lower() + if pattern in head.lower(): + return True + return ( + f"closes #{int(issue_number)}" in text + or f"fixes #{int(issue_number)}" in text + ) + + +def _all_linked_open_prs( + issue_number: int, open_prs: list[dict] +) -> list[Mapping[str, Any]]: + return [pr for pr in (open_prs or []) if _pr_links_issue(issue_number, pr)] + + +def _assess_owning_pr_exemption( + issue_number: int, + *, + linked_open_prs: list[Mapping[str, Any]], + locked_branch: str | None, + recovered_owning_pr: Mapping[str, Any] | None, +) -> tuple[bool, list[str]]: + """Is the linked open PR provably the one a sanctioned recovery owns (#755)? + + ``recovered_owning_pr`` is produced by + ``issue_lock_recovery.owning_pr_recovery_evidence`` from a completed + server-side recovery assessment — it is never a caller-supplied field. + Every element is re-checked here against the live PR list this gate was + given, so a stale or partial token cannot widen the exemption. + + Returns ``(exempt, diagnostic_reasons)``. Diagnostics are only emitted when + a token was offered and rejected, so a blocked caller can see which element + of ownership disagreed. + """ + if not recovered_owning_pr: + return False, [] + + notes: list[str] = [] + token_issue = recovered_owning_pr.get("issue_number") + token_pr = recovered_owning_pr.get("pr_number") + token_branch = str(recovered_owning_pr.get("branch_name") or "").strip() + token_head = str(recovered_owning_pr.get("head_sha") or "").strip() + locked = (locked_branch or "").strip() + + if token_issue is not None and int(token_issue) != int(issue_number): + notes.append( + f"recovery evidence is for issue #{token_issue}, not " + f"#{issue_number} (no owning-PR exemption)" + ) + return False, notes + if not locked or not token_branch or locked != token_branch: + notes.append( + f"recovery evidence branch '{token_branch or 'unknown'}' does not " + f"match the branch being locked '{locked or 'unknown'}' " + "(no owning-PR exemption)" + ) + return False, notes + if len(linked_open_prs) != 1: + numbers = ", ".join( + f"#{pr.get('number')}" for pr in linked_open_prs + ) or "none" + notes.append( + f"{len(linked_open_prs)} open PRs link issue #{issue_number} " + f"({numbers}); recovery may only own exactly one " + "(no owning-PR exemption)" + ) + return False, notes + + only = linked_open_prs[0] + head_obj = only.get("head") or {} + only_number = only.get("number") + only_ref = str(head_obj.get("ref") or "").strip() + only_sha = str(head_obj.get("sha") or "").strip() + + if token_pr is None or only_number is None or int(only_number) != int(token_pr): + notes.append( + f"linked open PR #{only_number} is not the recovered owning PR " + f"#{token_pr} (no owning-PR exemption)" + ) + return False, notes + if only_ref != token_branch: + notes.append( + f"open PR #{only_number} head branch '{only_ref}' does not match " + 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: + 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)" + ) + 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" + ] + + def _matching_branches( issue_number: int, branch_names: list[str], @@ -52,8 +162,15 @@ def assess_work_issue_duplicate_gate( claim_entry: dict | None = None, locked_branch: str | None = None, phase: str = PHASE_LOCK, + recovered_owning_pr: Mapping[str, Any] | None = None, ) -> dict[str, Any]: - """Fail closed when duplicate work is already in flight for an issue.""" + """Fail closed when duplicate work is already in flight for an issue. + + ``recovered_owning_pr`` (#755) is server-derived evidence that a sanctioned + dead-session lock recovery already owns one specific open PR. It exempts + *only* that exact PR from the linked-open-PR blocker; every other duplicate + signal, and every mismatch, keeps failing closed. + """ reasons: list[str] = [] outcome = OUTCOME_DUPLICATE_WORK_NOT_PREVENTED prs = list(open_prs or []) @@ -61,12 +178,23 @@ def assess_work_issue_duplicate_gate( pattern = _issue_pattern(issue_number) linked = _linked_open_pr(issue_number, prs) + linked_open_prs = _all_linked_open_prs(issue_number, prs) + owning_pr_exempted = False + exemption_notes: list[str] = [] if linked: - reasons.append( - f"open PR #{linked.get('number')} already covers issue " - f"#{issue_number} (fail closed)" + owning_pr_exempted, exemption_notes = _assess_owning_pr_exemption( + issue_number, + linked_open_prs=linked_open_prs, + locked_branch=locked_branch, + recovered_owning_pr=recovered_owning_pr, ) - outcome = OUTCOME_DUPLICATE_PR_PREVENTED + if not owning_pr_exempted: + reasons.append( + f"open PR #{linked.get('number')} already covers issue " + f"#{issue_number} (fail closed)" + ) + reasons.extend(exemption_notes) + outcome = OUTCOME_DUPLICATE_PR_PREVENTED conflicting_branches = _matching_branches( issue_number, branches, locked_branch=locked_branch @@ -122,6 +250,9 @@ def assess_work_issue_duplicate_gate( "phase": phase, "outcome": outcome, "linked_open_pr": linked.get("number") if linked else entry.get("linked_open_pr"), + "linked_open_pr_count": len(linked_open_prs), + "owning_pr_recovery_exempted": owning_pr_exempted, + "owning_pr_recovery_notes": list(exemption_notes), "conflicting_branches": conflicting_branches, "claim_status": status or None, "reasons": reasons, diff --git a/pr_sync_status.py b/pr_sync_status.py index 546c42c..4d0e691 100644 --- a/pr_sync_status.py +++ b/pr_sync_status.py @@ -41,6 +41,186 @@ _DENIED_UPDATE_ROLES = frozenset({"reviewer", "merger", "reconciler", "mixed", " UPDATE_STYLE_MERGE = "merge" _FORBIDDEN_UPDATE_STYLES = frozenset({"rebase", "rebase-merge", "squash", "force"}) +# ── Commit check classifications (#751) ────────────────────────────────── +# Gitea's *combined* commit status reports ``state: pending`` both when a real +# check is executing and when the status-context collection is empty. Reading +# ``state`` alone therefore cannot distinguish "CI is running" from "no CI +# exists", which permanently blocks a merge-ready PR that no check will ever +# report on. These classifications are derived from the actual context +# collection plus the live branch-protection policy. +CHECKS_SUCCESS = "success" +CHECKS_FAILURE = "failure" +CHECKS_PENDING = "pending" +CHECKS_NONE = "none" # configured/produced nothing +CHECKS_NOT_REQUIRED = "not_required" # protection does not require checks +CHECKS_MISSING_REQUIRED = "missing_required" # required contexts have no result +CHECKS_UNKNOWN = "unknown" # indeterminable — fail closed + +# Values that permit merge_now when checks are required. +_CHECKS_OK = frozenset({"success", "passed", "ok", "skipped", "not_required"}) + +# Raw per-context state vocabularies reported by Gitea. +_CTX_SUCCESS = frozenset({"success", "passed", "ok"}) +_CTX_FAILURE = frozenset({"failure", "failed", "error", "cancelled", "canceled"}) +_CTX_PENDING = frozenset({"pending", "running", "queued", "expected"}) +_CTX_SKIPPED = frozenset({"skipped", "neutral"}) + + +def _normalize_context_rows(statuses: Any) -> list[dict[str, str]]: + """Reduce a raw status collection to newest-wins ``{context, state}`` rows. + + Gitea returns the status collection newest-first, so the first row seen for + a context wins. Rows without a usable state are discarded rather than being + silently treated as passing. + """ + rows: list[dict[str, str]] = [] + seen: set[str] = set() + if not isinstance(statuses, list): + return rows + for raw in statuses: + if not isinstance(raw, dict): + continue + context = (raw.get("context") or raw.get("name") or "").strip() + state = (raw.get("status") or raw.get("state") or "").strip().lower() + if not state: + continue + key = context or f"__unnamed__{len(rows)}" + if key in seen: + continue + seen.add(key) + rows.append({"context": context, "state": state}) + return rows + + +def _aggregate_context_states(rows: list[dict[str, str]]) -> str: + """Fail-closed aggregate: failure > pending > unknown-state > success.""" + states = {row["state"] for row in rows} + if states & _CTX_FAILURE: + return CHECKS_FAILURE + if states & _CTX_PENDING: + return CHECKS_PENDING + unresolved = states - _CTX_SUCCESS - _CTX_SKIPPED + if unresolved: + # An unrecognized context state must never read as success. + return CHECKS_UNKNOWN + return CHECKS_SUCCESS + + +def classify_commit_checks( + *, + combined_state: str | None = None, + statuses: Any = None, + checks_enabled: bool | None = None, + required_contexts: Any = None, + policy_determinable: bool = True, + status_determinable: bool = True, +) -> dict[str, Any]: + """Classify head checks from live evidence (#751). + + ``combined_state`` is deliberately **not** authoritative: it is recorded for + observability but never used to infer that CI is executing. The context + collection and the live protection policy decide. + + Returns ``checks_status`` (one of the ``CHECKS_*`` values), the derived + ``checks_required`` flag, and structured ``reasons``. + """ + reasons: list[str] = [] + rows = _normalize_context_rows(statuses) + required = [ + str(ctx).strip() + for ctx in (required_contexts or []) + if str(ctx or "").strip() + ] + observed_combined = (combined_state or "").strip().lower() or None + + result: dict[str, Any] = { + "checks_status": CHECKS_UNKNOWN, + "checks_required": True, + "combined_state": observed_combined, + "context_count": len(rows), + "observed_contexts": [row["context"] for row in rows], + "required_contexts": required, + "missing_required_contexts": [], + "policy_determinable": bool(policy_determinable), + "status_determinable": bool(status_determinable), + "reasons": reasons, + } + + # Policy unreadable → never assume checks are optional. + if not policy_determinable: + reasons.append( + "branch-protection check policy could not be read; cannot prove " + "whether status checks are required (fail closed)" + ) + return result + + if checks_enabled is False: + result["checks_required"] = False + result["checks_status"] = CHECKS_NOT_REQUIRED + reasons.append( + "live branch protection does not require status checks for the base " + "branch; head status contexts do not gate merge" + ) + return result + + if checks_enabled is None: + reasons.append( + "branch-protection status-check requirement is indeterminate " + "(fail closed)" + ) + return result + + # Checks are required from here on. + if not status_determinable: + reasons.append( + "head commit status collection could not be read while branch " + "protection requires status checks (fail closed)" + ) + return result + + if required: + by_context = {row["context"]: row["state"] for row in rows if row["context"]} + missing = [ctx for ctx in required if ctx not in by_context] + if missing: + result["missing_required_contexts"] = missing + result["checks_status"] = CHECKS_MISSING_REQUIRED + reasons.append( + "branch protection requires status context(s) " + f"{', '.join(missing)} but no matching status result exists at " + "the head commit (fail closed)" + ) + return result + matched = [ + {"context": ctx, "state": by_context[ctx]} for ctx in required + ] + result["checks_status"] = _aggregate_context_states(matched) + reasons.append( + f"evaluated {len(matched)} required status context(s) from live " + "branch protection; unrelated contexts were ignored" + ) + return result + + # Status checks enabled with no specific required contexts configured. + if not rows: + result["checks_status"] = CHECKS_NONE + reasons.append( + "branch protection enables status checks but no status context was " + "produced for the head commit" + ) + if observed_combined in _CTX_PENDING: + reasons.append( + f"combined commit state '{observed_combined}' does not indicate " + "executing CI because the status-context collection is empty" + ) + return result + + result["checks_status"] = _aggregate_context_states(rows) + reasons.append( + f"aggregated {len(rows)} reported status context(s); branch protection " + "configures no explicit required-context list" + ) + return result + def _normalize_sha(value: str | None) -> str | None: text = (value or "").strip().lower() @@ -120,6 +300,7 @@ def assess_pr_sync_status( "branch_protection_requires_current_base": requires_current, "approval_at_current_head": approval_ok if approval_at_current_head is not None else None, "checks_status": checks, + "checks_required": bool(checks_required), "active_locks_and_leases": { "author_lock": bool(active_author_lock) if active_author_lock is not None else None, "reviewer_lease": bool(active_reviewer_lease) if active_reviewer_lease is not None else None, @@ -258,21 +439,41 @@ def assess_pr_sync_status( result["recommended_next_action"] = ACTION_BLOCKED return result - # ── Checks gate for merge_now ──────────────────────────────────────── - if checks_required and checks not in ("success", "passed", "ok", "none", "skipped", "not_required"): - if checks in ("pending", "running", "queued"): + # ── Checks gate for merge_now (#751) ───────────────────────────────── + # ``checks_required`` is derived from the live branch-protection policy by + # the production caller. When protection does not require status checks, + # head contexts cannot gate the merge and this whole gate is skipped. + if not checks_required: + reasons.append( + "live branch protection does not require status checks; head check " + f"state ({checks}) does not gate merge" + ) + elif checks not in _CHECKS_OK: + if checks in _CTX_PENDING: reasons.append(f"required checks are not finished (status={checks})") - result["recommended_next_action"] = ACTION_BLOCKED - return result - if checks in ("failure", "failed", "error", "cancelled"): + elif checks in _CTX_FAILURE: reasons.append(f"required checks failed (status={checks})") - result["recommended_next_action"] = ACTION_BLOCKED - return result - # unknown — fail closed when checks_required - if checks == "unknown": + elif checks == CHECKS_MISSING_REQUIRED: + reasons.append( + "branch protection configures required status context(s) but no " + "matching status result exists at the current head (fail closed)" + ) + elif checks == CHECKS_NONE: + reasons.append( + "branch protection requires status checks but no status context " + "was produced for the current head (fail closed); an empty " + "status collection is not executing CI" + ) + elif checks == CHECKS_UNKNOWN: reasons.append("checks status unknown (fail closed)") - result["recommended_next_action"] = ACTION_BLOCKED - return result + else: + # Unrecognized vocabulary must never fall through to merge_now. + reasons.append( + f"unrecognized checks status '{checks}' cannot prove required " + "checks passed (fail closed)" + ) + result["recommended_next_action"] = ACTION_BLOCKED + return result # ── Ready to merge without update ──────────────────────────────────── # Includes: current with approval; outdated when update is NOT required. diff --git a/tests/test_issue_751_checks_assessor.py b/tests/test_issue_751_checks_assessor.py new file mode 100644 index 0000000..36ce374 --- /dev/null +++ b/tests/test_issue_751_checks_assessor.py @@ -0,0 +1,602 @@ +"""Regression coverage for the PR checks assessor defect (#751). + +Gitea's *combined* commit status reports ``state: pending`` both when a check is +executing and when the status-context collection is empty. The assessor read +``state`` alone and defaulted ``checks_required`` to ``True``, so a PR whose head +had no status contexts — and never would — was routed to ``blocked`` forever. + +These tests pin the corrected semantics end to end: live branch protection +decides whether checks are required, and the actual context collection decides +what the checks say. +""" + +from __future__ import annotations + +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import pr_sync_status # noqa: E402 +from pr_sync_status import ( # noqa: E402 + ACTION_BLOCKED, + ACTION_MERGE_NOW, + ACTION_UPDATE_BRANCH_BY_MERGE, + CHECKS_FAILURE, + CHECKS_MISSING_REQUIRED, + CHECKS_NONE, + CHECKS_NOT_REQUIRED, + CHECKS_PENDING, + CHECKS_SUCCESS, + CHECKS_UNKNOWN, + assess_pr_sync_status, + classify_commit_checks, +) + + +def _sha(prefix: str) -> str: + return (prefix + "0" * 40)[:40] + + +PR_HEAD = _sha("aaaaaaaa") +BASE_HEAD = _sha("bbbbbbbb") + + +def _base_kwargs(**overrides): + data = { + "host": "gitea.prgs.cc", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + "pr_number": 751, + "pr_state": "open", + "source_branch": "fix/issue-751-checks-assessor", + "pr_head_sha": PR_HEAD, + "base_head_sha": BASE_HEAD, + "commits_behind": 0, + "mergeable": True, + "has_conflicts": False, + "branch_protection_requires_current_base": False, + "approval_at_current_head": True, + "checks_status": "success", + "active_author_lock": True, + "active_reviewer_lease": False, + "active_merger_lease": False, + } + data.update(overrides) + return data + + +def _reasons(result) -> str: + return " | ".join(result["reasons"]).lower() + + +class TestClassifyCommitChecks(unittest.TestCase): + """Pure classification from live evidence.""" + + def test_empty_collection_with_combined_pending_is_not_executing_ci(self): + # The exact PR #750 failure mode at the classification layer. + result = classify_commit_checks( + combined_state="pending", + statuses=[], + checks_enabled=True, + required_contexts=[], + ) + self.assertNotEqual(result["checks_status"], CHECKS_PENDING) + self.assertEqual(result["checks_status"], CHECKS_NONE) + self.assertEqual(result["context_count"], 0) + joined = " ".join(result["reasons"]).lower() + self.assertIn("empty", joined) + self.assertIn("does not indicate", joined) + + def test_protection_disables_status_checks(self): + result = classify_commit_checks( + combined_state="pending", + statuses=[], + checks_enabled=False, + required_contexts=[], + ) + self.assertFalse(result["checks_required"]) + self.assertEqual(result["checks_status"], CHECKS_NOT_REQUIRED) + + def test_no_required_contexts_aggregates_reported_contexts(self): + result = classify_commit_checks( + combined_state="success", + statuses=[{"context": "build", "status": "success"}], + checks_enabled=True, + required_contexts=[], + ) + self.assertTrue(result["checks_required"]) + self.assertEqual(result["checks_status"], CHECKS_SUCCESS) + + def test_required_checks_pending(self): + result = classify_commit_checks( + combined_state="pending", + statuses=[{"context": "build", "status": "pending"}], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_PENDING) + + def test_required_checks_failed(self): + result = classify_commit_checks( + combined_state="failure", + statuses=[{"context": "build", "status": "failure"}], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_FAILURE) + + def test_required_checks_successful(self): + result = classify_commit_checks( + combined_state="success", + statuses=[{"context": "build", "status": "success"}], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_SUCCESS) + self.assertTrue(result["checks_required"]) + + def test_required_context_configured_with_no_matching_result(self): + result = classify_commit_checks( + combined_state="success", + statuses=[{"context": "lint", "status": "success"}], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED) + self.assertEqual(result["missing_required_contexts"], ["build"]) + + def test_mixed_required_and_unrelated_contexts_ignores_unrelated(self): + # An unrelated failing context must not fail a satisfied required set. + result = classify_commit_checks( + combined_state="failure", + statuses=[ + {"context": "build", "status": "success"}, + {"context": "optional-scan", "status": "failure"}, + ], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_SUCCESS) + # ...and a failing *required* context still fails despite passing extras. + failing = classify_commit_checks( + combined_state="success", + statuses=[ + {"context": "build", "status": "failure"}, + {"context": "optional-scan", "status": "success"}, + ], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(failing["checks_status"], CHECKS_FAILURE) + + def test_policy_unreadable_fails_closed(self): + result = classify_commit_checks( + combined_state=None, + statuses=[], + checks_enabled=None, + required_contexts=[], + policy_determinable=False, + ) + self.assertTrue(result["checks_required"]) + self.assertEqual(result["checks_status"], CHECKS_UNKNOWN) + self.assertIn("fail closed", " ".join(result["reasons"]).lower()) + + def test_malformed_policy_indeterminate_flag_fails_closed(self): + result = classify_commit_checks( + combined_state="success", + statuses=[{"context": "build", "status": "success"}], + checks_enabled=None, + required_contexts=[], + policy_determinable=True, + ) + self.assertTrue(result["checks_required"]) + self.assertEqual(result["checks_status"], CHECKS_UNKNOWN) + + def test_status_collection_unreadable_fails_closed_when_required(self): + result = classify_commit_checks( + checks_enabled=True, + required_contexts=["build"], + status_determinable=False, + ) + self.assertTrue(result["checks_required"]) + self.assertEqual(result["checks_status"], CHECKS_UNKNOWN) + + def test_unrecognized_context_state_never_reads_as_success(self): + result = classify_commit_checks( + combined_state="success", + statuses=[{"context": "build", "status": "banana"}], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_UNKNOWN) + + def test_newest_wins_per_context(self): + # Gitea returns newest-first; the stale failure must not win. + result = classify_commit_checks( + combined_state="success", + statuses=[ + {"context": "build", "status": "success"}, + {"context": "build", "status": "failure"}, + ], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_SUCCESS) + + def test_malformed_status_rows_are_discarded_not_treated_as_passing(self): + result = classify_commit_checks( + combined_state="pending", + statuses=[{"context": "build"}, "not-a-dict", None], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED) + + +class TestChecksGateSemantics(unittest.TestCase): + """``assess_pr_sync_status`` routing and blocker reasons.""" + + def test_not_required_allows_merge_now_with_empty_checks(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_NOT_REQUIRED), + checks_required=False, + ) + self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + self.assertTrue(result["approval_valid_for_merge"]) + self.assertFalse(result["checks_required"]) + self.assertIn("does not require status checks", _reasons(result)) + + def test_none_blocks_when_checks_required(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_NONE), + checks_required=True, + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("no status context", _reasons(result)) + self.assertIn("not executing ci", _reasons(result)) + + def test_missing_required_blocks(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_MISSING_REQUIRED), + checks_required=True, + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("no matching status result", _reasons(result)) + + def test_required_pending_blocks(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_PENDING), checks_required=True + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("not finished", _reasons(result)) + + def test_required_failure_blocks(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_FAILURE), checks_required=True + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("failed", _reasons(result)) + + def test_required_success_merges(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_SUCCESS), checks_required=True + ) + self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + + def test_unknown_blocks_when_required(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_UNKNOWN), checks_required=True + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("unknown", _reasons(result)) + + def test_unrecognized_status_does_not_fall_through_to_merge(self): + # Regression: the previous gate only handled a fixed vocabulary and let + # anything else reach merge_now. + result = assess_pr_sync_status( + **_base_kwargs(checks_status="totally-unexpected"), + checks_required=True, + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("unrecognized checks status", _reasons(result)) + + def test_checks_gate_does_not_bypass_approval_requirement(self): + result = assess_pr_sync_status( + **_base_kwargs( + checks_status=CHECKS_NOT_REQUIRED, approval_at_current_head=False + ), + checks_required=False, + ) + self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + self.assertFalse(result["approval_valid_for_merge"]) + + def test_checks_gate_does_not_bypass_current_base_protection(self): + # Existing current-base behavior stays intact (#727). + result = assess_pr_sync_status( + **_base_kwargs( + checks_status=CHECKS_NOT_REQUIRED, + commits_behind=3, + branch_protection_requires_current_base=True, + ), + checks_required=False, + ) + self.assertEqual( + result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE + ) + + def test_checks_gate_does_not_bypass_conflicts(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_NOT_REQUIRED, mergeable=False), + checks_required=False, + ) + self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + + +class TestBranchProtectionPolicy(unittest.TestCase): + """Live protection reader derives the status-check requirement.""" + + def setUp(self): + import gitea_mcp_server as gms + + self.gms = gms + + def _policy(self, responses): + def fake_api_request(method, url, auth, *args, **kwargs): + for fragment, payload in responses.items(): + if fragment in url: + if isinstance(payload, Exception): + raise payload + return payload + return None + + with patch.object(self.gms, "api_request", side_effect=fake_api_request): + return self.gms._branch_protection_policy( + "https://example/api/v1/repos/o/r", {"h": "1"}, base_branch="master" + ) + + def test_status_checks_enabled_with_contexts(self): + policy = self._policy({ + "branch_protections": [{ + "branch_name": "master", + "block_on_outdated_branch": True, + "enable_status_check": True, + "status_check_contexts": ["ci/build", " "], + }], + }) + self.assertTrue(policy["determinable"]) + self.assertTrue(policy["checks_enabled"]) + self.assertTrue(policy["requires_current_base"]) + self.assertEqual(policy["required_contexts"], ["ci/build"]) + + def test_status_checks_disabled(self): + policy = self._policy({ + "branch_protections": [{ + "branch_name": "master", + "enable_status_check": False, + }], + }) + self.assertTrue(policy["determinable"]) + self.assertFalse(policy["checks_enabled"]) + + def test_no_protection_rule_means_checks_not_required(self): + # PR #750's repository shape: protection list readable but empty. + policy = self._policy({"branch_protections": [], "branches/master": {}}) + self.assertTrue(policy["determinable"]) + self.assertFalse(policy["protection_found"]) + self.assertFalse(policy["checks_enabled"]) + self.assertIsNone(policy["requires_current_base"]) + + def test_protection_without_status_field_means_not_enabled(self): + policy = self._policy({ + "branch_protections": [{ + "branch_name": "master", + "block_on_outdated_branch": True, + }], + }) + self.assertTrue(policy["determinable"]) + self.assertFalse(policy["checks_enabled"]) + self.assertTrue(policy["requires_current_base"]) + + def test_api_failure_is_not_determinable(self): + policy = self._policy({ + "branch_protections": RuntimeError("boom"), + }) + self.assertFalse(policy["determinable"]) + self.assertIsNone(policy["checks_enabled"]) + + def test_missing_branch_is_not_determinable(self): + with patch.object(self.gms, "api_request", return_value=[]): + policy = self.gms._branch_protection_policy( + "https://example/api/v1/repos/o/r", {}, base_branch=" " + ) + self.assertFalse(policy["determinable"]) + + def test_requires_current_base_accessor_preserved(self): + def fake(method, url, auth, *a, **k): + if "branch_protections" in url: + return [{"branch_name": "master", + "block_on_outdated_branch": True}] + return None + + with patch.object(self.gms, "api_request", side_effect=fake): + value = self.gms._branch_protection_requires_current_base( + "https://example/api/v1/repos/o/r", {}, base_branch="master" + ) + self.assertTrue(value) + + +class TestCommitChecksSnapshot(unittest.TestCase): + def setUp(self): + import gitea_mcp_server as gms + + self.gms = gms + + def test_reads_state_and_context_collection(self): + payload = {"state": "pending", "statuses": [{"context": "b", + "status": "pending"}]} + with patch.object(self.gms, "api_request", return_value=payload): + snap = self.gms._commit_checks_snapshot( + "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD + ) + self.assertTrue(snap["determinable"]) + self.assertEqual(snap["combined_state"], "pending") + self.assertEqual(len(snap["statuses"]), 1) + + def test_empty_collection_recorded_as_determinable(self): + with patch.object( + self.gms, "api_request", return_value={"state": "pending", "statuses": []} + ): + snap = self.gms._commit_checks_snapshot( + "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD + ) + self.assertTrue(snap["determinable"]) + self.assertEqual(snap["statuses"], []) + + def test_api_failure_is_not_determinable(self): + with patch.object(self.gms, "api_request", side_effect=RuntimeError("x")): + snap = self.gms._commit_checks_snapshot( + "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD + ) + self.assertFalse(snap["determinable"]) + + +class TestMcpWrapperForwarding(unittest.TestCase): + """The production MCP path must reach the derived ``checks_required``.""" + + def setUp(self): + import gitea_mcp_server as gms + + self.gms = gms + self.base = ( + "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools" + ) + + def _patches(self, fake_api_request, spy): + return [ + patch.object(self.gms, "_profile_operation_gate", return_value=None), + patch.object(self.gms, "_resolve", return_value=( + "gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")), + patch.object(self.gms, "_auth", return_value={"Authorization": "x"}), + patch.object(self.gms, "repo_api_url", return_value=self.base), + patch.object(self.gms, "api_request", side_effect=fake_api_request), + patch.object(self.gms, "gitea_get_pr_review_feedback", return_value={ + "success": True, "approval_at_current_head": True}), + patch.object(self.gms, "_prove_author_ownership_for_pr", return_value={ + "has_author_lock": True}), + patch.object(pr_sync_status, "assess_pr_sync_status", side_effect=spy), + ] + + def _run(self, *, protections, status_payload, pr_number=750, + caller_checks_status=None): + captured = {} + real_assess = pr_sync_status.assess_pr_sync_status + + def spy(**kwargs): + captured.update(kwargs) + return real_assess(**kwargs) + + def fake_api_request(method, url, auth, *args, **kwargs): + if f"/pulls/{pr_number}" in url: + return { + "number": pr_number, + "state": "open", + "title": "t", + "body": "b", + "mergeable": True, + "head": {"sha": PR_HEAD, "ref": "fix/x"}, + "base": {"sha": BASE_HEAD, "ref": "master"}, + } + if "/branch_protections" in url: + if isinstance(protections, Exception): + raise protections + return protections + if "/branches/master" in url: + return {"commit": {"id": BASE_HEAD}} + if "/status" in url: + if isinstance(status_payload, Exception): + raise status_payload + return status_payload + if "/compare/" in url: + return {"total_commits": 0} + if "/comments" in url: + return [] + return None + + stack = self._patches(fake_api_request, spy) + for p in stack: + p.start() + try: + result = self.gms.gitea_assess_pr_sync_status( + pr_number=pr_number, remote="prgs", + checks_status=caller_checks_status, + ) + finally: + for p in reversed(stack): + p.stop() + return result, captured + + def test_derived_checks_required_is_forwarded(self): + result, captured = self._run( + protections=[{"branch_name": "master", "enable_status_check": True, + "status_check_contexts": ["ci/build"]}], + status_payload={"state": "pending", "statuses": [ + {"context": "ci/build", "status": "pending"}]}, + ) + self.assertIn("checks_required", captured) + self.assertTrue(captured["checks_required"]) + self.assertEqual(captured["checks_status"], CHECKS_PENDING) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + + def test_pr750_empty_status_with_no_protection_is_merge_ready(self): + # The exact reported failure: combined pending, zero contexts, and no + # branch protection requiring checks. + result, captured = self._run( + protections=[], + status_payload={"state": "pending", "statuses": []}, + ) + self.assertFalse(captured["checks_required"]) + self.assertEqual(captured["checks_status"], CHECKS_NOT_REQUIRED) + self.assertNotEqual(result["checks_status"], CHECKS_PENDING) + self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + + def test_protection_read_failure_blocks(self): + result, captured = self._run( + protections=RuntimeError("protection unavailable"), + status_payload={"state": "success", "statuses": []}, + ) + self.assertTrue(captured["checks_required"]) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + + def test_caller_supplied_status_cannot_mask_live_required_failure(self): + # A caller-declared "success" must not override live evidence. + result, captured = self._run( + protections=[{"branch_name": "master", "enable_status_check": True, + "status_check_contexts": ["ci/build"]}], + status_payload={"state": "failure", "statuses": [ + {"context": "ci/build", "status": "failure"}]}, + caller_checks_status="success", + ) + self.assertEqual(captured["checks_status"], CHECKS_FAILURE) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertEqual( + result["checks_evidence"]["caller_supplied_checks_status"], "success" + ) + + def test_checks_evidence_is_reported_without_secrets(self): + result, _ = self._run( + protections=[], + status_payload={"state": "pending", "statuses": []}, + ) + evidence = result["checks_evidence"] + self.assertEqual(evidence["context_count"], 0) + self.assertEqual(evidence["combined_state"], "pending") + self.assertFalse(evidence["protection_found"]) + blob = repr(result).lower() + self.assertNotIn("authorization", blob) + self.assertNotIn("token", blob) + + +if __name__ == "__main__": + unittest.main() 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() diff --git a/tests/test_issue_755_owning_pr_recovery.py b/tests/test_issue_755_owning_pr_recovery.py new file mode 100644 index 0000000..d9fc617 --- /dev/null +++ b/tests/test_issue_755_owning_pr_recovery.py @@ -0,0 +1,609 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 +"""Dead-session lock recovery when the issue already owns an open PR (#755). + +#753 added the recovery *assessor*, but the production ``gitea_lock_issue`` +path still rejected every sanctioned recovery: a dead-session lock is by +construction a lock for work that already has an open PR, and the #400 +duplicate-work gate blocked unconditionally on any linked open PR. These tests +drive the real MCP handler, not just the pure assessor, so that gap cannot +reopen. +""" +import os +import subprocess +import sys +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import issue_lock_provenance # noqa: E402 +import issue_lock_recovery # noqa: E402 +import issue_lock_store # noqa: E402 +import mcp_server # noqa: E402 +from issue_work_duplicate_gate import ( # noqa: E402 + OUTCOME_DUPLICATE_PR_PREVENTED, + OUTCOME_DUPLICATE_WORK_NOT_PREVENTED, + PHASE_LOCK, + assess_work_issue_duplicate_gate, +) + +ISSUE = 4755 +BRANCH = f"fix/issue-{ISSUE}-owning-pr" +OTHER_BRANCH = f"fix/issue-{ISSUE}-competing" +HEAD = "c" * 40 +OTHER_HEAD = "d" * 40 +OWNING_PR = 4756 +OTHER_PR = 4757 +IDENTITY = "example-user" +PROFILE = "test-author-prgs" +ORG = "Scaled-Tech-Consulting" +REPO = "Gitea-Tools" + + +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 shifted_ts(hours: int = 4) -> str: + return ( + (datetime.now(timezone.utc) + timedelta(hours=hours)) + .isoformat() + .replace("+00:00", "Z") + ) + + +def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE): + return { + "number": number, + "title": f"fix: something (Closes #{issue})", + "body": f"Closes #{issue}.", + "head": {"ref": ref, "sha": sha}, + } + + +def sanctioned_token( + issue_number=ISSUE, pr_number=OWNING_PR, branch=BRANCH, head=HEAD +): + """The evidence shape the server derives from a granted recovery.""" + return { + "issue_number": issue_number, + "pr_number": pr_number, + "branch_name": branch, + "head_sha": head, + } + + +# ───────────────────────── duplicate gate: exemption ───────────────────────── + + +class TestOwningPrExemptionGranted(unittest.TestCase): + def test_exact_owning_pr_is_not_duplicate_work(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assertFalse(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED) + self.assertEqual(result["linked_open_pr"], OWNING_PR) + self.assertEqual(result["linked_open_pr_count"], 1) + + def test_unrelated_open_pr_alongside_owning_pr_is_ignored(self): + unrelated = { + "number": 999, + "title": "chore: unrelated", + "body": "no linkage", + "head": {"ref": "chore/unrelated", "sha": OTHER_HEAD}, + } + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[unrelated, owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assertFalse(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["linked_open_pr_count"], 1) + + +class TestOwningPrExemptionRefused(unittest.TestCase): + def assert_blocked(self, result): + self.assertTrue(result["block"]) + self.assertFalse(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED) + + def test_no_recovery_evidence_keeps_ordinary_blocker(self): + self.assert_blocked( + assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + ) + ) + + def test_competing_pr_number_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(number=OTHER_PR)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_multiple_linked_open_prs_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + self.assertEqual(result["linked_open_pr_count"], 2) + + def test_different_branch_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(ref=OTHER_BRANCH)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_locked_branch_differing_from_evidence_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=OTHER_BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_different_head_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(sha=OTHER_HEAD)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_evidence_for_another_issue_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(issue_number=ISSUE + 1), + ) + self.assert_blocked(result) + + def test_missing_head_in_live_pr_refused(self): + pr = owning_pr() + pr["head"] = {"ref": BRANCH} + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[pr], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + +class TestOrdinaryDuplicateBehaviorUnchanged(unittest.TestCase): + def test_clean_issue_still_passes(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[], + branch_names=["feat/other-issue-99"], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + ) + self.assertFalse(result["block"]) + self.assertFalse(result["owning_pr_recovery_exempted"]) + + def test_competing_branch_still_blocks_even_with_owning_pr_evidence(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH, OTHER_BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + # The owning PR is exempt, but the competing branch is not. + self.assertTrue(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + self.assertIn(OTHER_BRANCH, result["conflicting_branches"]) + + +# ─────────────────── server-derived evidence cannot be forged ─────────────────── + + +class TestOwningPrEvidenceDerivation(unittest.TestCase): + def granted(self, **evidence_overrides): + evidence = { + "issue_number": ISSUE, + "locked_branch": BRANCH, + "local_head": HEAD, + "remote_head": HEAD, + "pr_head": HEAD, + "pr_number": OWNING_PR, + } + evidence.update(evidence_overrides) + return { + "outcome": issue_lock_recovery.RECOVERY_SANCTIONED, + "recovery_sanctioned": True, + "is_candidate": True, + "reasons": [], + "evidence": evidence, + } + + def test_granted_recovery_yields_evidence(self): + token = issue_lock_recovery.owning_pr_recovery_evidence(self.granted()) + self.assertEqual(token, sanctioned_token()) + + def test_none_assessment_yields_nothing(self): + self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(None)) + + def test_refused_assessment_yields_nothing(self): + refused = self.granted() + refused["outcome"] = issue_lock_recovery.REFUSED + refused["recovery_sanctioned"] = False + self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(refused)) + + def test_sanctioned_flag_without_outcome_yields_nothing(self): + forged = self.granted() + forged["outcome"] = "SOMETHING_ELSE" + self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(forged)) + + def test_missing_pr_number_yields_nothing(self): + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence( + self.granted(pr_number=None) + ) + ) + + def test_head_disagreement_in_evidence_yields_nothing(self): + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence( + self.granted(remote_head=OTHER_HEAD) + ) + ) + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence( + self.granted(local_head=OTHER_HEAD) + ) + ) + + def test_real_refused_assessment_yields_nothing(self): + """End-to-end against the real assessor, not a hand-built dict.""" + lock = { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": "/scratch/wt", + "remote": "prgs", + "org": "Example-Org", + "repo": "Example-Repo", + "session_pid": os.getpid(), # alive → must refuse + "work_lease": { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": ISSUE, + "branch": BRANCH, + "worktree_path": "/scratch/wt", + "claimant": {"username": IDENTITY, "profile": PROFILE}, + "expires_at": shifted_ts(), + }, + } + assessment = issue_lock_recovery.assess_dead_session_lock_recovery( + lock, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt", + remote="prgs", + org="Example-Org", + repo="Example-Repo", + identity=IDENTITY, + profile=PROFILE, + current_branch=BRANCH, + porcelain_status="", + head_sha=HEAD, + remote_head_sha=HEAD, + pr_head_sha=HEAD, + pr_number=OWNING_PR, + competing_live_locks=[], + candidate_branches=[BRANCH], + current_pid=os.getpid(), + ) + self.assertFalse(assessment["recovery_sanctioned"]) + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence(assessment) + ) + + +# ──────────────────── end-to-end: the real gitea_lock_issue ──────────────────── + + +class LockIssueEndToEndBase(unittest.TestCase): + """Drives ``mcp_server.gitea_lock_issue`` with live git/Gitea observation + stubbed at the module boundary — the production gate chain itself runs.""" + + def setUp(self): + self.lock_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.lock_dir.cleanup) + self.worktree = os.path.realpath(os.getcwd()) + # Bind host/org/repo to what the ``test-author-prgs`` fixture profile is + # pinned to, so the session-context gate under test is the real one and + # not a cross-host denial. The issue number and lock dir stay synthetic. + self.remotes = patch.dict(mcp_server.REMOTES, { + "prgs": { + "host": "gitea.prgs.cc", + "org": ORG, + "repo": REPO, + }, + }) + self.remotes.start() + self.addCleanup(patch.stopall) + mcp_server._IDENTITY_CACHE.clear() + + def write_durable_lock(self, *, pid, branch=BRANCH, worktree=None): + path = issue_lock_store.lock_file_path( + remote="prgs", + org=ORG, + repo=REPO, + issue_number=ISSUE, + lock_dir=self.lock_dir.name, + ) + claimant = {"username": IDENTITY, "profile": PROFILE} + data = { + "issue_number": ISSUE, + "branch_name": branch, + "remote": "prgs", + "org": ORG, + "repo": REPO, + "worktree_path": worktree or self.worktree, + "session_pid": pid, + "pid": pid, + "work_lease": { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": ISSUE, + "branch": branch, + "worktree_path": worktree or self.worktree, + "claimant": claimant, + "created_at": shifted_ts(-1), + "last_heartbeat_at": shifted_ts(-1), + "expires_at": shifted_ts(), + }, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant=claimant, + ), + } + issue_lock_store.save_lock_file(path, data) + return path + + def run_lock( + self, + *, + open_prs, + porcelain="", + current_branch=BRANCH, + base_equivalent=False, + branch_names=None, + head_sha=HEAD, + remote_head=HEAD, + ): + branch_names = branch_names if branch_names is not None else [BRANCH] + branch_entries = [ + {"name": name, "commit": {"id": remote_head}} for name in branch_names + ] + env = shared_mutation_env( + "test-author-prgs", + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, + ) + with patch( + "mcp_server.api_get_all", return_value=branch_entries + ), patch( + "mcp_server._list_open_pulls", return_value=list(open_prs) + ), patch( + "mcp_server.get_auth_header", return_value="token x" + ), patch( + "mcp_server._work_lease_claimant", + return_value={"username": IDENTITY, "profile": PROFILE}, + ), patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={ + "current_branch": current_branch, + "porcelain_status": porcelain, + "base_equivalent": base_equivalent, + "head_sha": head_sha, + "inspected_git_root": self.worktree, + "base_branch": "master", + }, + ), patch( + "mcp_server.issue_duplicate_context_fetcher", + side_effect=lambda h, o, r, auth, issue_number: ( + list(open_prs), list(branch_names), {"status": "not_claimed"} + ), + ): + with patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server.gitea_lock_issue( + issue_number=ISSUE, + branch_name=BRANCH, + remote="prgs", + worktree_path=self.worktree, + ) + + +class TestRecoveryWithOwningPrSucceeds(LockIssueEndToEndBase): + def test_dead_session_recovery_with_owning_pr_relocks(self): + self.write_durable_lock(pid=dead_pid()) + result = self.run_lock(open_prs=[owning_pr()]) + + self.assertTrue(result["success"]) + self.assertEqual(result["issue_number"], ISSUE) + self.assertEqual(result["branch_name"], BRANCH) + self.assertTrue(result["lock_freshness"]["live"]) + self.assertTrue(result["lock_freshness"]["pid_alive"]) + + def test_recovered_lock_records_truthful_provenance(self): + prior = dead_pid() + self.write_durable_lock(pid=prior) + self.run_lock(open_prs=[owning_pr()]) + + lock = issue_lock_store.load_issue_lock( + remote="prgs", + org=ORG, + repo=REPO, + issue_number=ISSUE, + lock_dir=self.lock_dir.name, + ) + record = lock.get("dead_session_recovery") or {} + self.assertTrue(record.get("recovered")) + self.assertEqual(record.get("prior_session_pid"), prior) + self.assertEqual(record.get("replacement_session_pid"), os.getpid()) + self.assertFalse(record.get("prior_pid_alive")) + self.assertEqual(record.get("pr_number"), OWNING_PR) + self.assertEqual(record.get("branch_name"), BRANCH) + self.assertEqual(record.get("identity"), IDENTITY) + + def test_recovered_lock_is_live_and_proves_pr_ownership(self): + """AC6: the persisted lock satisfies update-by-merge's ownership prover.""" + self.write_durable_lock(pid=dead_pid()) + self.run_lock(open_prs=[owning_pr()]) + + lock = issue_lock_store.load_issue_lock( + remote="prgs", + org=ORG, + repo=REPO, + issue_number=ISSUE, + lock_dir=self.lock_dir.name, + ) + self.assertTrue(issue_lock_store.is_lease_live(lock)) + + env = shared_mutation_env( + "test-author-prgs", + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, + ) + with patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + ownership = mcp_server._prove_author_ownership_for_pr( + pr_number=OWNING_PR, + pr_title=f"fix: something (Closes #{ISSUE})", + pr_body=f"Closes #{ISSUE}.", + source_branch=BRANCH, + remote="prgs", + host="gitea.prgs.cc", + org=ORG, + repo=REPO, + worktree_path=self.worktree, + ) + self.assertTrue(ownership["proven"], ownership["reasons"]) + self.assertTrue(ownership["has_author_lock"]) + self.assertEqual(ownership["matched_issue"], ISSUE) + + +class TestRecoveryRejectionsEndToEnd(LockIssueEndToEndBase): + def assert_lock_refused(self, **kwargs): + with self.assertRaises((ValueError, RuntimeError)) as ctx: + self.run_lock(**kwargs) + return str(ctx.exception) + + def test_competing_pr_still_blocked(self): + self.write_durable_lock(pid=dead_pid()) + message = self.assert_lock_refused( + open_prs=[owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)], + branch_names=[BRANCH], + ) + self.assertIn("already covers issue", message) + + def test_multiple_linked_open_prs_blocked(self): + self.write_durable_lock(pid=dead_pid()) + message = self.assert_lock_refused( + open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)], + ) + self.assertIn("already covers issue", message) + + def test_owning_pr_on_a_different_head_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused(open_prs=[owning_pr(sha=OTHER_HEAD)]) + + def test_lock_registered_to_a_different_worktree_blocked(self): + self.write_durable_lock( + pid=dead_pid(), worktree=os.path.join(self.worktree, "elsewhere") + ) + self.assert_lock_refused(open_prs=[owning_pr()]) + + def test_dirty_worktree_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused( + open_prs=[owning_pr()], porcelain=" M gitea_mcp_server.py" + ) + + def test_worktree_parked_on_another_branch_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused( + open_prs=[owning_pr()], current_branch="master" + ) + + def test_local_head_differing_from_remote_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused( + open_prs=[owning_pr()], head_sha=OTHER_HEAD + ) + + def test_live_prior_pid_blocked(self): + self.write_durable_lock(pid=os.getpid()) + self.assert_lock_refused(open_prs=[owning_pr()]) + + def test_new_claim_without_prior_lock_still_requires_base_equivalence(self): + """AC10: no durable lock → no recovery → base-equivalence still rules.""" + self.assert_lock_refused(open_prs=[], branch_names=[]) + + +if __name__ == "__main__": + unittest.main()