From ca76dacd73f6793e80837afbb36a4926597f997c Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Mon, 20 Jul 2026 14:18:25 -0500 Subject: [PATCH 1/3] fix(mcp): recover clean unpublished author work after the owning session exits Closes #772 Recovery now dispatches on observed publication state: published_owning_pr (remote branch exists; head equality #753 or strict descendancy #768, unchanged) and unpublished_claim (no remote branch, no PR; ownership proven by the durable lock record plus a local HEAD strictly descending from the server-observed base the branch was cut from). The absence of a remote head is never itself permission. Recovery writes are compare-and-swap against a lock generation, and the mutating lock path and read-only diagnostic assessor share one evaluator. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --- gitea_mcp_server.py | 250 +++++++--- issue_lock_recovery.py | 206 +++++++- issue_lock_store.py | 47 +- issue_lock_worktree.py | 77 +++ ...st_issue_772_unpublished_claim_recovery.py | 445 ++++++++++++++++++ 5 files changed, 942 insertions(+), 83 deletions(-) create mode 100644 tests/test_issue_772_unpublished_claim_recovery.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 7a44b0e..db08b09 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2240,7 +2240,7 @@ def _resolve_issue_lock_for_pr( return lock_data -def _save_issue_lock(data: dict) -> str: +def _save_issue_lock(data: dict, *, expected_generation: int | None = None) -> str: existing = issue_lock_store.load_issue_lock( remote=str(data.get("remote") or ""), org=str(data.get("org") or ""), @@ -2251,11 +2251,127 @@ def _save_issue_lock(data: dict) -> str: if overwrite_block: raise RuntimeError(overwrite_block) try: - return issue_lock_store.bind_session_lock(data) + return issue_lock_store.bind_session_lock( + data, expected_generation=expected_generation + ) except Exception as e: raise RuntimeError(f"Could not write issue lock file: {e}") from e +def _evaluate_issue_lock_recovery( + existing_lock: dict, + *, + issue_number: int, + branch_name: str, + worktree_path: str, + remote: str, + h: str | None, + o: str, + r: str, + git_state: dict, +) -> dict: + """Gather recovery evidence and decide the disposition (#753/#768/#772). + + The single decision point for dead-session issue-lock recovery. The mutating + ``gitea_lock_issue`` path and the read-only diagnostic assessor both call + this, so the two cannot report different verdicts or different supporting + evidence for the same durable state (#772 AC9) — a divergence between them + would itself be the defect. + + Every input is either durable lock state or a live server-side observation + (Gitea branch/PR inventory, git in the declared worktree). Nothing is + reachable from an MCP caller's parameters (#772 AC1). + """ + 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}" + ) + remote_head: str | None = None + remote_branch_exists = False + candidates: list[str] = [] + for entry in recovery_branches: + entry_name = _branch_entry_name(entry) + if entry_name == branch_name: + remote_branch_exists = True + remote_head = _branch_entry_commit_sha(entry) + if issue_lock_adoption.branch_carries_issue_marker(entry_name, issue_number): + candidates.append(entry_name) + + pr_head: str | None = None + 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: + pr_head = pull_head.get("sha") + pr_number = pull.get("number") + break + + local_head = git_state.get("head_sha") + + # #768: the recovering author's clean worktree is, by construction, one + # commit ahead of the head recorded at lock time — committing is the only + # sanctioned way to clean it without discarding the work. Observe the + # ancestry here, server-side, so the assessor can tell an honest + # fast-forward from a rewritten head. Probed only when the heads actually + # differ, and never from any caller-supplied value. + head_ancestry: dict | None = None + if remote_head and local_head and remote_head != local_head: + head_ancestry = issue_lock_worktree.read_head_ancestry( + worktree_path, + ancestor_sha=remote_head, + descendant_sha=local_head, + ) + + # #772: with no remote branch there is no head to measure against, so the + # base the branch was cut from is observed instead. Probed only in that + # case, so the published path's evidence is untouched (#772 AC8). + recorded_base: str | None = None + base_ancestry: dict | None = None + if not remote_branch_exists and local_head: + base_observation = issue_lock_worktree.read_recorded_base( + worktree_path, + head_sha=local_head, + ) + recorded_base = base_observation.get("base_sha") + if recorded_base: + base_ancestry = issue_lock_worktree.read_head_ancestry( + worktree_path, + ancestor_sha=recorded_base, + descendant_sha=local_head, + ) + + claimant = _work_lease_claimant(h) + return issue_lock_recovery.assess_dead_session_lock_recovery( + existing_lock, + issue_number=issue_number, + branch_name=branch_name, + worktree_path=worktree_path, + remote=remote, + org=o, + repo=r, + identity=claimant.get("username"), + profile=claimant.get("profile"), + current_branch=git_state.get("current_branch"), + porcelain_status=git_state.get("porcelain_status") or "", + head_sha=local_head, + remote_head_sha=remote_head, + pr_head_sha=pr_head, + pr_number=pr_number, + competing_live_locks=issue_lock_store.list_live_locks(), + candidate_branches=candidates, + current_pid=os.getpid(), + head_ancestry=head_ancestry, + remote_branch_exists=remote_branch_exists, + recorded_base_sha=recorded_base, + base_ancestry=base_ancestry, + ) + + def _work_lease_claimant(host: str | None) -> dict: profile = get_profile() username = _IDENTITY_CACHE.get(host) if host else None @@ -3542,70 +3658,16 @@ def gitea_lock_issue( 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) - # #768: the recovering author's clean worktree is, by construction, one - # commit ahead of the head recorded at lock time — committing is the - # only sanctioned way to clean it without discarding the work. Observe - # the ancestry here, server-side, so the assessor can tell an honest - # fast-forward from a rewritten head. Probed only when the heads - # actually differ, and never from any caller-supplied value. - recovery_local_head = git_state.get("head_sha") - recovery_ancestry: dict | None = None - if ( - recovery_remote_head - and recovery_local_head - and recovery_remote_head != recovery_local_head - ): - recovery_ancestry = issue_lock_worktree.read_head_ancestry( - resolved_worktree, - ancestor_sha=recovery_remote_head, - descendant_sha=recovery_local_head, - ) - recovery_assessment = issue_lock_recovery.assess_dead_session_lock_recovery( + recovery_assessment = _evaluate_issue_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(), - head_ancestry=recovery_ancestry, + h=h, + o=o, + r=r, + git_state=git_state, ) recovery_sanctioned = bool( @@ -3715,7 +3777,17 @@ def gitea_lock_issue( recovered_at=_work_lease_timestamp(_work_lease_now()), ) - lock_file_path = _save_issue_lock(data) + # #772 AC5: a recovery replaces a claim another session already owned, so + # its write is a compare-and-swap against the generation the assessment was + # made on. Two replacement sessions that both observed the same dead owner + # cannot both succeed — the second finds a moved generation and fails + # closed. Ordinary first-time claims keep the unconditional write. + expected_generation = ( + issue_lock_store.lock_generation(existing_issue_lock) + if recovery_sanctioned + else None + ) + lock_file_path = _save_issue_lock(data, expected_generation=expected_generation) lock_record = issue_lock_store.read_lock_file(lock_file_path) or data freshness = issue_lock_store.assess_lock_freshness(lock_record) competing = [ @@ -3832,7 +3904,59 @@ def gitea_assess_work_issue_duplicate( phase=phase, recovered_owning_pr=recovered_owning_pr, ) - return {"success": not gate.get("block"), **gate} + # #772 AC9: report the recovery disposition this issue's durable lock would + # receive, decided by the same evaluator the mutating lock path uses, so the + # diagnostic and the mutator can never disagree. Read-only: assessment only, + # never a write, and a probe failure degrades to a reported reason rather + # than turning a diagnostic into a hard error. + lock_recovery: dict | None = None + if ( + lock_data + and int(lock_data.get("issue_number") or 0) == int(issue_number) + and not issue_lock_store.is_lease_live(lock_data) + ): + recovery_worktree = str(lock_data.get("worktree_path") or "") + recovery_branch = str( + branch_name or lock_data.get("branch_name") or "" + ) + if recovery_worktree and recovery_branch: + try: + recovery_state = issue_lock_worktree.read_worktree_git_state( + recovery_worktree + ) + assessment = _evaluate_issue_lock_recovery( + lock_data, + issue_number=int(issue_number), + branch_name=recovery_branch, + worktree_path=recovery_worktree, + remote=remote, + h=h, + o=o, + r=r, + git_state=recovery_state, + ) + lock_recovery = { + "outcome": assessment.get("outcome"), + "recovery_sanctioned": assessment.get("recovery_sanctioned"), + "is_candidate": assessment.get("is_candidate"), + "reasons": assessment.get("reasons"), + "evidence": assessment.get("evidence"), + } + except Exception as exc: + lock_recovery = { + "outcome": issue_lock_recovery.REFUSED, + "recovery_sanctioned": False, + "is_candidate": True, + "reasons": [ + f"recovery evidence could not be gathered: {exc}" + ], + "evidence": {}, + } + return { + "success": not gate.get("block"), + **gate, + "lock_recovery": lock_recovery, + } @mcp.tool() diff --git a/issue_lock_recovery.py b/issue_lock_recovery.py index bc32f37..ecc99c6 100644 --- a/issue_lock_recovery.py +++ b/issue_lock_recovery.py @@ -34,6 +34,33 @@ ancestor, still unmodified — so it is accepted, and nothing else is. The descendant fact is observed server-side by ``issue_lock_worktree.read_head_ancestry`` and handed in as ``head_ancestry``; no caller can assert it. + +#772 adds the remaining uncovered quadrant: a claim that was never published at +all. Two recovery modes now exist, and they require different evidence because +they are answering the same question against different available facts: + +``published_owning_pr`` + The branch exists on the remote. Ownership is proven by comparing the local + head against the remote/PR head — equal (#753) or a strict descendant + (#768). This is the pre-existing behavior and is unchanged. + +``unpublished_claim`` + The branch is absent from the remote and no PR claims it, so there is no + head to compare against; that absence is the defining fact, not a degraded + published case. Ownership is instead proven by the durable lock record + (issue, branch, worktree, claimant, profile, dead PID) plus the local HEAD + strictly descending from the base the branch was cut from, observed + server-side by ``issue_lock_worktree.read_recorded_base`` and re-checked + through ``base_ancestry``. + +They cannot share one head-comparison implementation: the published path's +comparison target does not exist in the unpublished case, and inventing one +(defaulting to the base, say) would silently weaken the published path from +"matches what was actually pushed" to "descends from some base". The modes are +therefore selected by observed publication state and never by a caller — and +critically, the absence of a remote head is never itself treated as permission: +every identity, profile, branch, worktree, cleanliness, liveness, and competing +-claim check still applies in full. """ from __future__ import annotations @@ -55,6 +82,18 @@ REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path") # How the clean local head relates to the head recorded at lock time (#768). HEAD_RELATION_EQUAL = "equal" HEAD_RELATION_STRICT_DESCENDANT = "strict_descendant" +# #772: an unpublished claim has no recorded head to compare against at all, so +# its head is measured against the base the branch was cut from instead. +HEAD_RELATION_DESCENDS_FROM_BASE = "descends_from_recorded_base" + +# Which body of evidence a recovery was decided on (#772 AC10). These are not +# interchangeable: a published claim proves ownership against a remote/PR head, +# an unpublished one against the recorded base plus durable lock state. They +# cannot share a single head-comparison implementation because the unpublished +# case has no head to compare — that absence is the defining fact, not a +# degraded version of the published case. +RECOVERY_MODE_PUBLISHED_OWNING_PR = "published_owning_pr" +RECOVERY_MODE_UNPUBLISHED_CLAIM = "unpublished_claim" def _same_realpath(left: str | None, right: str | None) -> bool: @@ -165,6 +204,68 @@ def _assess_strict_descendant( ] +def _assess_base_descendancy( + base_ancestry: Mapping[str, Any] | None, + *, + recorded_base: str, + local_head: str, +) -> tuple[bool, list[str]]: + """Is ``local_head`` a proven strict descendant of ``recorded_base`` (#772)? + + The unpublished-claim analogue of ``_assess_strict_descendant``. The + comparison target is the base the branch was cut from — observed server-side + by ``issue_lock_worktree.read_recorded_base`` — rather than a remote or PR + head, because an unpublished claim has neither. + + The probe's own endpoints are re-checked against the values under + assessment, so an observation taken for some other pair of commits cannot + authorize recovery. Equality is refused: a HEAD that merely equals its base + carries no committed work, and that is the ordinary base-equivalent case the + normal lock path already handles. + """ + if not isinstance(base_ancestry, Mapping): + return False, [ + "no server-derived ancestry observation was available; an " + "unpublished claim cannot be recovered without proving its HEAD " + "descends from the recorded base" + ] + + probe_ancestor = _text(base_ancestry.get("ancestor_sha")) + probe_descendant = _text(base_ancestry.get("descendant_sha")) + if probe_ancestor != recorded_base or probe_descendant != local_head: + return False, [ + f"ancestry observation covers {probe_ancestor or 'unknown'} -> " + f"{probe_descendant or 'unknown'}, not the commits under assessment " + f"({recorded_base} -> {local_head})" + ] + if not base_ancestry.get("probe_ok"): + return False, ( + list(base_ancestry.get("reasons") or []) + or ["ancestry probe did not complete; ancestry unproven"] + ) + if not base_ancestry.get("ancestor_present"): + return False, [ + f"recorded base {recorded_base} is no longer reachable; a rewritten " + "or force-moved base cannot be recovered" + ] + if not base_ancestry.get("is_strict_descendant"): + return False, ( + list(base_ancestry.get("reasons") or []) + or [ + f"local head {local_head} is not a strict descendant of the " + f"recorded base {recorded_base}" + ] + ) + + proof = _text(base_ancestry.get("proof")) or ( + f"{recorded_base} is an ancestor of {local_head}" + ) + return True, [ + f"local head {local_head} strictly descends from recorded base " + f"{recorded_base} ({proof})" + ] + + def assess_dead_session_lock_recovery( existing_lock: Mapping[str, Any] | None, *, @@ -186,6 +287,9 @@ def assess_dead_session_lock_recovery( candidate_branches: Iterable[str] | None = None, current_pid: int | None = None, head_ancestry: Mapping[str, Any] | None = None, + remote_branch_exists: bool | None = None, + recorded_base_sha: str | None = None, + base_ancestry: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Decide whether a dead-session author lock may be natively recovered. @@ -304,32 +408,73 @@ def assess_dead_session_lock_recovery( # to reach the clean worktree recovery itself demands. local_head = _text(head_sha) remote_head = _text(remote_head_sha) + recorded_base = _text(recorded_base_sha) head_relation: str | None = None ancestry_proof: str | None = None if not local_head: reasons.append("local head SHA could not be determined") - if not remote_head: - reasons.append( - f"remote head for branch '{locked_branch}' could not be determined" - ) - if local_head and remote_head: - if local_head == remote_head: - head_relation = HEAD_RELATION_EQUAL - else: - descends, notes = _assess_strict_descendant( - head_ancestry, - recorded_head=remote_head, + + # #772: which body of evidence applies is decided by observed publication + # state, never by a caller. ``remote_branch_exists is False`` is a positive + # server-side observation that the branch is absent from the remote — it is + # not the same as "the head lookup failed", which must still fail closed. + unpublished = remote_branch_exists is False and not remote_head + recovery_mode = ( + RECOVERY_MODE_UNPUBLISHED_CLAIM if unpublished + else RECOVERY_MODE_PUBLISHED_OWNING_PR + ) + evidence["recovery_mode"] = recovery_mode + evidence["remote_branch_exists"] = remote_branch_exists + + if unpublished: + # No remote branch: ownership is measured against the recorded base. + # An open PR here is contradictory — a PR cannot exist without a remote + # branch — so it is a mismatch, never a thing to reconcile. + if _text(pr_head_sha) or pr_number is not None: + reasons.append( + f"branch '{locked_branch}' is absent from the remote yet PR " + f"#{pr_number} claims it; publication state is contradictory" + ) + if not recorded_base: + reasons.append( + f"recorded base for branch '{locked_branch}' could not be " + "determined; an unpublished claim cannot be recovered without it" + ) + if local_head and recorded_base: + descends, notes = _assess_base_descendancy( + base_ancestry, + recorded_base=recorded_base, local_head=local_head, ) if descends: - head_relation = HEAD_RELATION_STRICT_DESCENDANT + head_relation = HEAD_RELATION_DESCENDS_FROM_BASE ancestry_proof = notes[0] if notes else None else: - reasons.append( - f"local head {local_head} does not match remote branch head " - f"{remote_head}" - ) reasons.extend(notes) + else: + if not remote_head: + reasons.append( + f"remote head for branch '{locked_branch}' could not be determined" + ) + if local_head and remote_head: + if local_head == remote_head: + head_relation = HEAD_RELATION_EQUAL + else: + descends, notes = _assess_strict_descendant( + head_ancestry, + recorded_head=remote_head, + local_head=local_head, + ) + if descends: + head_relation = HEAD_RELATION_STRICT_DESCENDANT + ancestry_proof = notes[0] if notes else None + else: + reasons.append( + f"local head {local_head} does not match remote branch head " + f"{remote_head}" + ) + reasons.extend(notes) + evidence["recorded_base"] = recorded_base or None evidence["local_head"] = local_head or None evidence["remote_head"] = remote_head or None # ``recorded_head`` is the head recovery is being measured against; @@ -344,7 +489,9 @@ def assess_dead_session_lock_recovery( if pr_head: evidence["pr_head"] = pr_head evidence["pr_number"] = pr_number - if local_head and pr_head != local_head: + # In unpublished mode the presence of any PR was already refused above as + # contradictory; re-stating it as a head mismatch would only obscure why. + if not unpublished and local_head and pr_head != local_head: # A descendant recovery has not been published yet, so the open PR # legitimately still points at the recorded head. Any other # disagreement is a real mismatch. @@ -449,12 +596,32 @@ def assess_dead_session_lock_recovery( if reasons: return _result(REFUSED, False, reasons, evidence) + # No disposition may be granted without a proven head relation. Every path + # above that leaves it unset also records a reason, so this is a belt-and- + # braces guard against a future path forgetting one (#772 AC4). + if head_relation is None: + return _result( + REFUSED, + False, + ["head relation to the recorded head or base was never proven"], + evidence, + ) + proof = [ f"durable lock for issue #{issue_number} matches branch " f"'{locked_branch}', worktree '{locked_worktree}', head {local_head}, " f"and claimant '{locked_identity}'; recorded pid {recorded_pid} is dead" ] - if head_relation == HEAD_RELATION_STRICT_DESCENDANT and ancestry_proof: + if recovery_mode == RECOVERY_MODE_UNPUBLISHED_CLAIM: + proof.append( + f"branch '{locked_branch}' has no remote head and no open PR; " + f"ownership proven against recorded base {recorded_base}" + ) + if ( + head_relation + in (HEAD_RELATION_STRICT_DESCENDANT, HEAD_RELATION_DESCENDS_FROM_BASE) + and ancestry_proof + ): proof.append(ancestry_proof) return _result(RECOVERY_SANCTIONED, True, proof, evidence) @@ -639,6 +806,9 @@ def build_recovery_record( "prior_pid_alive": evidence.get("prior_pid_alive"), "branch_name": evidence.get("locked_branch"), "worktree_path": evidence.get("locked_worktree_path"), + "recovery_mode": evidence.get("recovery_mode"), + "remote_branch_exists": evidence.get("remote_branch_exists"), + "recorded_base": evidence.get("recorded_base"), "local_head": evidence.get("local_head"), "remote_head": evidence.get("remote_head"), "recorded_head": evidence.get("recorded_head"), diff --git a/issue_lock_store.py b/issue_lock_store.py index afdc514..50d9ab8 100644 --- a/issue_lock_store.py +++ b/issue_lock_store.py @@ -148,8 +148,37 @@ def save_lock_file(path: str, data: dict[str, Any]) -> None: pass -def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) -> str: - """Persist a keyed lock and bind it to the current process session.""" +def lock_generation(lock: dict[str, Any] | None) -> int: + """Monotonic write counter for a durable lock record (#772 AC5). + + Absent or unusable values read as ``0`` so a lock written before generations + existed still participates in compare-and-swap: its first recovery expects + ``0`` and writes ``1``. + """ + if not isinstance(lock, dict): + return 0 + try: + return int(lock.get("lock_generation") or 0) + except (TypeError, ValueError): + return 0 + + +def bind_session_lock( + lock_data: dict[str, Any], + lock_dir: str | None = None, + *, + expected_generation: int | None = None, +) -> str: + """Persist a keyed lock and bind it to the current process session. + + ``expected_generation`` turns the write into a compare-and-swap (#772 AC5). + Recovery decides it may take over a claim by reading the durable lock, but + that read and this write are separate steps; without a CAS two replacement + sessions can both observe the same dead owner, both pass assessment, and + both write — the second silently clobbering the first. Passing the + generation observed at assessment time makes exactly one of them win: the + loser's expectation no longer matches and it fails closed. + """ remote = str(lock_data.get("remote") or "") org = str(lock_data.get("org") or "") repo = str(lock_data.get("repo") or "") @@ -194,6 +223,20 @@ def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) -> ) if lease_block: raise RuntimeError(lease_block) + # #772 AC5: compare-and-swap inside the same critical section that + # already serializes writers, so the check and the write cannot be + # separated by another session's successful recovery. + current_generation = lock_generation(existing) + if ( + expected_generation is not None + and current_generation != expected_generation + ): + raise RuntimeError( + f"Issue #{issue_number} lock generation changed: expected " + f"{expected_generation}, found {current_generation}; another " + "session already recovered or replaced this claim (fail closed)" + ) + record["lock_generation"] = current_generation + 1 save_lock_file(path, record) save_lock_file(session_pointer_path(root), pointer) except LockContentionError as exc: diff --git a/issue_lock_worktree.py b/issue_lock_worktree.py index 1be775a..e84b8d8 100644 --- a/issue_lock_worktree.py +++ b/issue_lock_worktree.py @@ -145,6 +145,83 @@ def read_head_ancestry( return result +def read_recorded_base( + worktree_path: str, + *, + head_sha: str | None, + extra_bases: tuple[str, ...] | list[str] = (), + base_branches: frozenset[str] | None = None, +) -> dict: + """Observe the base commit an unpublished claim was branched from (#772). + + A published claim records its base implicitly: the remote branch head is the + thing recovery measures against. An unpublished claim has no remote ref, so + the base must be observed here, server-side, as the merge-base between the + worktree HEAD and the base branch it was cut from. + + Reports facts only; the disposition lives in ``issue_lock_recovery``. Every + field is read from git in the declared worktree — nothing is supplied by, or + reachable from, an MCP caller, so a caller cannot nominate a base that would + make unrelated history look like a descendant (#772 AC1/AC4). + + A HEAD with no common ancestor in any base branch yields ``probe_ok`` with no + ``base_sha``: unrelated history is reported as exactly that, never as a base. + """ + path = (worktree_path or "").strip() + head = (head_sha or "").strip() + bases = base_branches or BASE_BRANCHES + candidates = [*extra_bases, *sorted(bases)] + result: dict = { + "base_branch": None, + "base_sha": None, + "head_sha": head or None, + "probe_ok": False, + "candidates": candidates, + "reasons": [], + } + if not path or not head: + result["reasons"].append( + "recorded-base probe requires a worktree path and a HEAD sha" + ) + return result + + probed_any = False + for candidate in candidates: + name = (candidate or "").strip() + if not name: + continue + probe = subprocess.run( + ["git", "-C", path, "merge-base", name, head], + capture_output=True, + text=True, + check=False, + ) + if probe.returncode not in (0, 1): + # 0 = merge base found, 1 = no common ancestor. Anything else is a + # failed probe (missing ref, broken repo) — try the next candidate. + continue + probed_any = True + merge_base = (probe.stdout or "").strip() + if probe.returncode == 0 and merge_base: + result["base_branch"] = name + result["base_sha"] = merge_base + result["probe_ok"] = True + return result + + result["probe_ok"] = probed_any + if probed_any: + result["reasons"].append( + f"HEAD {head} shares no common ancestor with any of " + f"{_base_list(bases)}; history is unrelated to this repository's base" + ) + else: + result["reasons"].append( + f"recorded-base probe could not run against any of {_base_list(bases)} " + f"in '{path}'" + ) + return result + + def read_worktree_git_state( worktree_path: str, extra_bases: tuple[str, ...] | list[str] = (), diff --git a/tests/test_issue_772_unpublished_claim_recovery.py b/tests/test_issue_772_unpublished_claim_recovery.py new file mode 100644 index 0000000..027c271 --- /dev/null +++ b/tests/test_issue_772_unpublished_claim_recovery.py @@ -0,0 +1,445 @@ +"""Unpublished-claim dead-session lock recovery (#772). + +An author claim whose work exists only as a clean local commit — never pushed, +never turned into a PR — was unrecoverable once the owning session exited: every +recovery disposition derived ownership from a remote branch head or an owning PR +head, and an unpublished claim has neither. + +These tests exercise the disposition through its evidence, never through any +issue number: every case here uses an arbitrary issue number, and the same +assertions hold for any other. The #617 *shape* (durable lock, unexpired lease, +dead PID, no remote branch, no PR) is what is under test. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import issue_lock_recovery # noqa: E402 +import issue_lock_store # noqa: E402 +import issue_lock_worktree # noqa: E402 + +DEAD_PID = 999_999_999 +LIVE_PID = os.getpid() +WORKTREE = "/tmp/wt/issue-901-example" +BRANCH = "fix/issue-901-example" +BASE_SHA = "0568f44cb2d87e78fd394a27a670e33c84f7842f" +HEAD_SHA = "b46f0f9f138d569edbc73e0d36df4b34239f9934" +OTHER_SHA = "1111111111111111111111111111111111111111" + + +def durable_lock(**overrides): + """A complete, internally consistent unpublished author lock.""" + lock = { + "issue_number": 901, + "branch_name": BRANCH, + "worktree_path": WORKTREE, + "remote": "prgs", + "org": "Example-Org", + "repo": "Example-Repo", + "session_pid": DEAD_PID, + "pid": DEAD_PID, + "work_lease": { + "operation_type": "author_issue_work", + "issue_number": 901, + "pr_number": None, + "branch": BRANCH, + "worktree_path": WORKTREE, + "claimant": {"username": "author-user", "profile": "prgs-author"}, + }, + "claimant": {"username": "author-user", "profile": "prgs-author"}, + } + lock.update(overrides) + return lock + + +def base_ancestry( + *, + ancestor=BASE_SHA, + descendant=HEAD_SHA, + probe_ok=True, + ancestor_present=True, + strict=True, + reasons=None, +): + return { + "ancestor_sha": ancestor, + "descendant_sha": descendant, + "probe_ok": probe_ok, + "ancestor_present": ancestor_present, + "descendant_present": True, + "is_ancestor": strict, + "is_strict_descendant": strict, + "proof": "merge-base --is-ancestor -> exit 0", + "reasons": reasons or [], + } + + +def assess(lock=None, **overrides): + """Assess an unpublished claim; overrides tweak one fact at a time.""" + kwargs = { + "issue_number": 901, + "branch_name": BRANCH, + "worktree_path": WORKTREE, + "remote": "prgs", + "org": "Example-Org", + "repo": "Example-Repo", + "identity": "author-user", + "profile": "prgs-author", + "current_branch": BRANCH, + "porcelain_status": "", + "head_sha": HEAD_SHA, + "remote_head_sha": None, + "pr_head_sha": None, + "pr_number": None, + "competing_live_locks": [], + "candidate_branches": [BRANCH], + "current_pid": LIVE_PID, + "remote_branch_exists": False, + "recorded_base_sha": BASE_SHA, + "base_ancestry": base_ancestry(), + } + kwargs.update(overrides) + return issue_lock_recovery.assess_dead_session_lock_recovery( + durable_lock() if lock is None else lock, **kwargs + ) + + +class UnpublishedClaimRecoveryGranted(unittest.TestCase): + """The #617 shape: every element agrees and the owner is dead.""" + + def test_recovery_is_sanctioned(self): + result = assess() + self.assertEqual(result["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED) + self.assertTrue(result["recovery_sanctioned"]) + + def test_mode_and_head_relation_name_the_unpublished_disposition(self): + evidence = assess()["evidence"] + self.assertEqual( + evidence["recovery_mode"], + issue_lock_recovery.RECOVERY_MODE_UNPUBLISHED_CLAIM, + ) + self.assertEqual( + evidence["head_relation"], + issue_lock_recovery.HEAD_RELATION_DESCENDS_FROM_BASE, + ) + self.assertEqual(evidence["recorded_base"], BASE_SHA) + self.assertEqual(evidence["accepted_head"], HEAD_SHA) + self.assertFalse(evidence["remote_branch_exists"]) + + def test_no_remote_branch_or_pr_is_required(self): + """Neither artifact may be demanded for this recovery mode.""" + result = assess(remote_head_sha=None, pr_head_sha=None, pr_number=None) + self.assertTrue(result["recovery_sanctioned"]) + + def test_recovery_record_carries_mode_and_base(self): + record = issue_lock_recovery.build_recovery_record( + assess(), recovered_at="2026-07-20T19:05:27Z" + ) + self.assertTrue(record["recovered"]) + self.assertEqual( + record["recovery_mode"], + issue_lock_recovery.RECOVERY_MODE_UNPUBLISHED_CLAIM, + ) + self.assertEqual(record["recorded_base"], BASE_SHA) + self.assertEqual(record["accepted_head"], HEAD_SHA) + + def test_unpublished_recovery_grants_no_owning_pr_exemption(self): + """There is no PR, so no duplicate-gate exemption may be produced.""" + self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(assess())) + + +class UnpublishedClaimRecoveryRefused(unittest.TestCase): + """Every rejection condition must fail closed, independently.""" + + def refusal(self, **overrides): + result = assess(**overrides) + self.assertEqual(result["outcome"], issue_lock_recovery.REFUSED) + self.assertFalse(result["recovery_sanctioned"]) + return " ".join(result["reasons"]) + + def test_live_prior_owner(self): + lock = durable_lock(session_pid=LIVE_PID, pid=LIVE_PID) + result = assess(lock=lock, current_pid=LIVE_PID + 1) + self.assertFalse(result["recovery_sanctioned"]) + self.assertIn("still alive", " ".join(result["reasons"])) + + def test_foreign_identity(self): + self.assertIn( + "does not match active identity", self.refusal(identity="someone-else") + ) + + def test_foreign_profile(self): + self.assertIn( + "does not match active profile", self.refusal(profile="prgs-reviewer") + ) + + def test_dirty_worktree(self): + self.assertIn("clean", self.refusal(porcelain_status=" M gitea_mcp_server.py\n")) + + def test_branch_mismatch(self): + self.assertIn("not the locked branch", self.refusal(current_branch="fix/other")) + + def test_worktree_mismatch(self): + self.assertIn( + "does not match declared", self.refusal(worktree_path="/tmp/wt/elsewhere") + ) + + def test_head_unrelated_to_recorded_base(self): + reasons = self.refusal( + base_ancestry=base_ancestry( + strict=False, + reasons=["local head does not descend from recorded base"], + ) + ) + self.assertIn("descend", reasons) + + def test_rewritten_base_is_unreachable(self): + self.assertIn( + "no longer reachable", + self.refusal(base_ancestry=base_ancestry(ancestor_present=False)), + ) + + def test_missing_recorded_base(self): + self.assertIn( + "recorded base", + self.refusal(recorded_base_sha=None, base_ancestry=None), + ) + + def test_ancestry_observation_for_other_commits_is_rejected(self): + """A probe taken for a different pair cannot authorize recovery.""" + self.assertIn( + "not the commits under assessment", + self.refusal(base_ancestry=base_ancestry(ancestor=OTHER_SHA)), + ) + + def test_unproven_probe(self): + self.assertIn( + "unproven", + self.refusal( + base_ancestry=base_ancestry( + probe_ok=False, reasons=["ancestry unproven"] + ) + ), + ) + + def test_competing_live_lock(self): + self.assertIn( + "competing live lock", + self.refusal( + competing_live_locks=[ + { + "issue_number": 901, + "branch_name": BRANCH, + "worktree_path": "/tmp/wt/other-session", + "pid": LIVE_PID, + } + ] + ), + ) + + def test_competing_branch_claim(self): + self.assertIn( + "ownership is ambiguous", + self.refusal(candidate_branches=[BRANCH, "fix/issue-901-duplicate"]), + ) + + def test_pr_without_remote_branch_is_contradictory(self): + self.assertIn( + "contradictory", + self.refusal(pr_head_sha=HEAD_SHA, pr_number=42), + ) + + def test_missing_durable_evidence(self): + lock = durable_lock() + lock.pop("worktree_path") + result = assess(lock=lock) + self.assertEqual(result["outcome"], issue_lock_recovery.REFUSED) + self.assertIn("incomplete", " ".join(result["reasons"])) + + def test_absent_remote_head_is_not_itself_permission(self): + """The defining regression: no remote ref must not read as consent. + + A mismatched claim with no remote branch is refused for the mismatch, + never waved through because the head comparison was unavailable. + """ + result = assess(identity="someone-else", profile="prgs-reviewer") + self.assertFalse(result["recovery_sanctioned"]) + joined = " ".join(result["reasons"]) + self.assertIn("does not match active identity", joined) + self.assertIn("does not match active profile", joined) + + +class PublishedRecoveryUnregressed(unittest.TestCase): + """#753 equality and #768 descendant recovery must still work.""" + + def published(self, **overrides): + kwargs = { + "remote_branch_exists": True, + "remote_head_sha": HEAD_SHA, + "recorded_base_sha": None, + "base_ancestry": None, + } + kwargs.update(overrides) + return assess(**kwargs) + + def test_equal_head_recovery_still_granted(self): + result = self.published() + self.assertTrue(result["recovery_sanctioned"]) + self.assertEqual( + result["evidence"]["head_relation"], + issue_lock_recovery.HEAD_RELATION_EQUAL, + ) + self.assertEqual( + result["evidence"]["recovery_mode"], + issue_lock_recovery.RECOVERY_MODE_PUBLISHED_OWNING_PR, + ) + + def test_strict_descendant_recovery_still_granted(self): + result = self.published( + remote_head_sha=BASE_SHA, + head_ancestry=base_ancestry(ancestor=BASE_SHA, descendant=HEAD_SHA), + ) + self.assertTrue(result["recovery_sanctioned"]) + self.assertEqual( + result["evidence"]["head_relation"], + issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT, + ) + + def test_owning_pr_exemption_still_produced(self): + result = self.published(pr_head_sha=HEAD_SHA, pr_number=42) + evidence = issue_lock_recovery.owning_pr_recovery_evidence(result) + self.assertIsNotNone(evidence) + self.assertEqual(evidence["pr_number"], 42) + + def test_published_branch_with_undeterminable_head_still_fails_closed(self): + """A failed head lookup is not the same as an absent branch.""" + result = self.published(remote_head_sha=None) + self.assertFalse(result["recovery_sanctioned"]) + self.assertIn("could not be determined", " ".join(result["reasons"])) + + +class RecordedBaseObservation(unittest.TestCase): + """``read_recorded_base`` observes real git, never a caller's claim.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.tmp], check=False)) + self.git("init", "-q", "-b", "master") + self.git("config", "user.email", "test@example.com") + self.git("config", "user.name", "Test") + with open(os.path.join(self.tmp, "seed.txt"), "w") as fh: + fh.write("seed\n") + self.git("add", "seed.txt") + self.git("commit", "-q", "-m", "seed") + self.base = self.rev("HEAD") + + def git(self, *args): + return subprocess.run( + ["git", "-C", self.tmp, *args], capture_output=True, text=True, check=False + ) + + def rev(self, ref): + return (self.git("rev-parse", ref).stdout or "").strip() + + def test_base_is_the_merge_base_of_head_and_master(self): + self.git("checkout", "-q", "-b", BRANCH) + with open(os.path.join(self.tmp, "work.txt"), "w") as fh: + fh.write("work\n") + self.git("add", "work.txt") + self.git("commit", "-q", "-m", "work") + head = self.rev("HEAD") + + observed = issue_lock_worktree.read_recorded_base(self.tmp, head_sha=head) + self.assertTrue(observed["probe_ok"]) + self.assertEqual(observed["base_sha"], self.base) + self.assertEqual(observed["base_branch"], "master") + + def test_unrelated_history_yields_no_base(self): + """An orphan branch shares no ancestor and must not report one.""" + self.git("checkout", "-q", "--orphan", "unrelated") + self.git("rm", "-rqf", ".") + with open(os.path.join(self.tmp, "other.txt"), "w") as fh: + fh.write("other\n") + self.git("add", "other.txt") + self.git("commit", "-q", "-m", "unrelated root") + head = self.rev("HEAD") + + observed = issue_lock_worktree.read_recorded_base(self.tmp, head_sha=head) + self.assertIsNone(observed["base_sha"]) + self.assertIn("unrelated", " ".join(observed["reasons"])) + + def test_missing_head_fails_closed(self): + observed = issue_lock_worktree.read_recorded_base(self.tmp, head_sha=None) + self.assertFalse(observed["probe_ok"]) + self.assertIsNone(observed["base_sha"]) + + +class LockGenerationCas(unittest.TestCase): + """Two replacement sessions must not both recover one claim.""" + + def setUp(self): + self.dir = tempfile.mkdtemp() + self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.dir], check=False)) + + def record(self, **overrides): + data = { + "issue_number": 901, + "branch_name": BRANCH, + "worktree_path": WORKTREE, + "remote": "prgs", + "org": "Example-Org", + "repo": "Example-Repo", + } + data.update(overrides) + return data + + def test_generation_increments_on_each_write(self): + path = issue_lock_store.bind_session_lock(self.record(), self.dir) + self.assertEqual( + issue_lock_store.lock_generation(issue_lock_store.read_lock_file(path)), 1 + ) + issue_lock_store.bind_session_lock(self.record(), self.dir) + self.assertEqual( + issue_lock_store.lock_generation(issue_lock_store.read_lock_file(path)), 2 + ) + + def test_absent_generation_reads_as_zero(self): + self.assertEqual(issue_lock_store.lock_generation({}), 0) + self.assertEqual(issue_lock_store.lock_generation(None), 0) + self.assertEqual(issue_lock_store.lock_generation({"lock_generation": "x"}), 0) + + def test_matching_expectation_succeeds(self): + issue_lock_store.bind_session_lock(self.record(), self.dir) + issue_lock_store.bind_session_lock( + self.record(), self.dir, expected_generation=1 + ) + + def test_second_recoverer_loses_the_race(self): + """Both sessions read generation 1; only the first may write.""" + issue_lock_store.bind_session_lock(self.record(), self.dir) + observed_by_both = 1 + + issue_lock_store.bind_session_lock( + self.record(), self.dir, expected_generation=observed_by_both + ) + with self.assertRaises(RuntimeError) as caught: + issue_lock_store.bind_session_lock( + self.record(), self.dir, expected_generation=observed_by_both + ) + self.assertIn("generation changed", str(caught.exception)) + + def test_unconditional_write_is_unchanged(self): + """Ordinary first-time claims pass no expectation and still succeed.""" + issue_lock_store.bind_session_lock(self.record(), self.dir) + issue_lock_store.bind_session_lock(self.record(), self.dir) + + +if __name__ == "__main__": + unittest.main() From c31df2130c742ef73e2ce156cc4a7181a82e655c Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 20 Jul 2026 16:13:53 -0400 Subject: [PATCH 2/3] test(mcp): AC9 assessor/mutator parity + AC6 unpublished recovery MCP regressions (PR #774 F1/F2) Local worktree commit only (publication blocked by dangling GITEA_AUTHOR_WORKTREE). Tests only; no production code change. Co-Authored-By: Grok 4.5 --- ...st_issue_772_unpublished_claim_recovery.py | 559 ++++++++++++++++++ 1 file changed, 559 insertions(+) diff --git a/tests/test_issue_772_unpublished_claim_recovery.py b/tests/test_issue_772_unpublished_claim_recovery.py index 027c271..e7a3c89 100644 --- a/tests/test_issue_772_unpublished_claim_recovery.py +++ b/tests/test_issue_772_unpublished_claim_recovery.py @@ -441,5 +441,564 @@ class LockGenerationCas(unittest.TestCase): issue_lock_store.bind_session_lock(self.record(), self.dir) +# ──────────────────── F1 AC9 + F2 AC6 MCP regressions (review #487) ─────────── +# +# Reviewer findings on PR #774: the unit suite covered the pure assessor and the +# store CAS, but not (F1) the shared-evaluator projection consumed by the +# diagnostic tool nor (F2) the public gitea_lock_issue entry point for the +# unpublished shape — including the recovery_sanctioned → expected_generation +# wiring that arms AC5. Both gaps are closed below. + + +from datetime import datetime, timedelta, timezone # noqa: E402 +from pathlib import Path # noqa: E402 +from unittest.mock import patch # noqa: E402 + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 + +import issue_lock_provenance # noqa: E402 +import mcp_server # noqa: E402 + +MCP_ISSUE = 9901 +MCP_BRANCH = f"fix/issue-{MCP_ISSUE}-unpublished-mcp" +MCP_IDENTITY = "example-user" +MCP_PROFILE = "test-author-prgs" +MCP_ORG = "Scaled-Tech-Consulting" +MCP_REPO = "Gitea-Tools" + + +def _dead_pid() -> int: + proc = subprocess.Popen([sys.executable, "-c", "pass"]) + proc.wait() + return proc.pid + + +def _future_ts(hours: int = 4) -> str: + return ( + (datetime.now(timezone.utc) + timedelta(hours=hours)) + .isoformat() + .replace("+00:00", "Z") + ) + + +def _past_ts(hours: int = 1) -> str: + return ( + (datetime.now(timezone.utc) - timedelta(hours=hours)) + .isoformat() + .replace("+00:00", "Z") + ) + + +class _UnpublishedMcpBase(unittest.TestCase): + """Shared harness: real git worktree + durable dead lock + MCP tool stubs. + + Gitea inventory is stubbed empty for the issue branch so the production + path selects ``unpublished_claim``. Git ancestry is observed for real + against a temporary repository. + """ + + def setUp(self): + self.lock_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.lock_dir.cleanup) + self.repo = tempfile.mkdtemp(prefix="issue772-mcp-") + self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.repo], check=False)) + self._init_unpublished_worktree() + self.remotes = patch.dict( + mcp_server.REMOTES, + { + "prgs": { + "host": "gitea.prgs.cc", + "org": MCP_ORG, + "repo": MCP_REPO, + }, + }, + ) + self.remotes.start() + self.addCleanup(patch.stopall) + mcp_server._IDENTITY_CACHE.clear() + + def _git(self, *args): + return subprocess.run( + ["git", "-C", self.repo, *args], + capture_output=True, + text=True, + check=True, + ) + + def _init_unpublished_worktree(self): + self._git("init", "-q", "-b", "master") + self._git("config", "user.email", "test@example.com") + self._git("config", "user.name", "Test") + seed = os.path.join(self.repo, "seed.txt") + with open(seed, "w") as fh: + fh.write("seed\n") + self._git("add", "seed.txt") + self._git("commit", "-q", "-m", "seed") + self.base_sha = self._git("rev-parse", "HEAD").stdout.strip() + self._git("checkout", "-q", "-b", MCP_BRANCH) + work = os.path.join(self.repo, "work.txt") + with open(work, "w") as fh: + fh.write("unpublished work\n") + self._git("add", "work.txt") + self._git("commit", "-q", "-m", "unpublished claim") + self.head_sha = self._git("rev-parse", "HEAD").stdout.strip() + self.worktree = os.path.realpath(self.repo) + + def write_dead_lock(self, **overrides): + path = issue_lock_store.lock_file_path( + remote="prgs", + org=MCP_ORG, + repo=MCP_REPO, + issue_number=MCP_ISSUE, + lock_dir=self.lock_dir.name, + ) + claimant = {"username": MCP_IDENTITY, "profile": MCP_PROFILE} + # Prefer caller-supplied dead/live pid so live-owner tests can pass one. + pid = overrides.pop("session_pid", None) + if pid is None: + pid = _dead_pid() + overrides.pop("pid", None) + data = { + "issue_number": MCP_ISSUE, + "branch_name": MCP_BRANCH, + "remote": "prgs", + "org": MCP_ORG, + "repo": MCP_REPO, + "worktree_path": self.worktree, + "session_pid": pid, + "pid": pid, + "work_lease": { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": MCP_ISSUE, + "pr_number": None, + "branch": MCP_BRANCH, + "worktree_path": self.worktree, + "claimant": claimant, + "created_at": _past_ts(), + "last_heartbeat_at": _past_ts(), + "expires_at": _future_ts(), + }, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant=claimant, + ), + } + data.update(overrides) + data["session_pid"] = pid + data["pid"] = pid + data["lock_file_path"] = path + issue_lock_store.save_lock_file(path, data) + # Diagnostic tool loads via the per-session pointer (not issue key alone). + pointer = { + "pid": os.getpid(), + "lock_file_path": path, + "issue_number": MCP_ISSUE, + "branch_name": MCP_BRANCH, + "remote": "prgs", + "org": MCP_ORG, + "repo": MCP_REPO, + } + issue_lock_store.save_lock_file( + issue_lock_store.session_pointer_path(self.lock_dir.name), pointer + ) + return path + def _tool_env(self): + env = shared_mutation_env( + "test-author-prgs", + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, + ) + env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return env + + def _git_state(self, *, porcelain="", branch=MCP_BRANCH, head=None): + return { + "current_branch": branch, + "porcelain_status": porcelain, + "base_equivalent": False, + "head_sha": head or self.head_sha, + "inspected_git_root": self.worktree, + "base_branch": "master", + } + + def run_lock_issue(self, *, branch_entries=None, open_prs=None, git_state=None): + """Drive the public ``gitea_lock_issue`` tool for the unpublished shape.""" + # No remote branch for this claim — positive absence observation. + if branch_entries is None: + branch_entries = [] + if open_prs is None: + open_prs = [] + if git_state is None: + git_state = self._git_state() + env = self._tool_env() + with patch( + "mcp_server.api_get_all", return_value=list(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": MCP_IDENTITY, "profile": MCP_PROFILE}, + ), patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=git_state, + ), patch( + "mcp_server.issue_duplicate_context_fetcher", + side_effect=lambda h, o, r, auth, issue_number: ( + list(open_prs), + [b.get("name") for b in branch_entries if isinstance(b, dict)], + {"status": "not_claimed"}, + ), + ), 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=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + worktree_path=self.worktree, + ) + + def run_assess_duplicate(self, *, branch_entries=None, open_prs=None, git_state=None): + """Drive the public diagnostic and return its ``lock_recovery`` block.""" + if branch_entries is None: + branch_entries = [] + if open_prs is None: + open_prs = [] + if git_state is None: + git_state = self._git_state() + env = self._tool_env() + with patch( + "mcp_server.api_get_all", return_value=list(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": MCP_IDENTITY, "profile": MCP_PROFILE}, + ), patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=git_state, + ), patch( + "mcp_server.issue_duplicate_context_fetcher", + side_effect=lambda h, o, r, auth, issue_number: ( + list(open_prs), + [b.get("name") for b in branch_entries if isinstance(b, dict)], + {"status": "not_claimed"}, + ), + ), patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server.gitea_assess_work_issue_duplicate( + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + ) + + +class TestAc9AssessorMutatorParity(_UnpublishedMcpBase): + """F1 — diagnostic and mutator share one decision for one durable state.""" + + def test_assessor_and_evaluator_agree_when_recovery_is_sanctioned(self): + path = self.write_dead_lock() + lock = issue_lock_store.read_lock_file(path) + git_state = self._git_state() + + # Shared evaluator (same function the mutator calls). + env = self._tool_env() + with patch( + "mcp_server.api_get_all", return_value=[] + ), patch( + "mcp_server._list_open_pulls", return_value=[] + ), patch( + "mcp_server.get_auth_header", return_value="token x" + ), patch( + "mcp_server._work_lease_claimant", + return_value={"username": MCP_IDENTITY, "profile": MCP_PROFILE}, + ), patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + evaluation = mcp_server._evaluate_issue_lock_recovery( + lock, + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + worktree_path=self.worktree, + remote="prgs", + h="gitea.prgs.cc", + o=MCP_ORG, + r=MCP_REPO, + git_state=git_state, + ) + + diagnostic = self.run_assess_duplicate(git_state=git_state) + projected = diagnostic.get("lock_recovery") or {} + + self.assertTrue(evaluation.get("recovery_sanctioned"), evaluation) + self.assertIsNotNone(projected) + # Projected five-field subset must match the evaluator exactly. + for key in ( + "outcome", + "recovery_sanctioned", + "is_candidate", + "reasons", + "evidence", + ): + self.assertIn(key, projected, f"projection missing {key}") + self.assertEqual( + projected[key], + evaluation.get(key), + f"AC9 divergence on field {key}", + ) + self.assertEqual( + projected["evidence"].get("recovery_mode"), + issue_lock_recovery.RECOVERY_MODE_UNPUBLISHED_CLAIM, + ) + + def test_assessor_and_mutator_agree_on_foreign_identity_refusal(self): + """Mutator cannot accept evidence the assessor rejects (AC9).""" + self.write_dead_lock() + # Foreign identity on the active session. + env = self._tool_env() + git_state = self._git_state() + + with patch( + "mcp_server.api_get_all", return_value=[] + ), patch( + "mcp_server._list_open_pulls", return_value=[] + ), patch( + "mcp_server.get_auth_header", return_value="token x" + ), patch( + "mcp_server._work_lease_claimant", + return_value={"username": "not-the-owner", "profile": MCP_PROFILE}, + ), patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=git_state, + ), patch( + "mcp_server.issue_duplicate_context_fetcher", + side_effect=lambda *a, **k: ([], [], {"status": "not_claimed"}), + ), patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + diagnostic = mcp_server.gitea_assess_work_issue_duplicate( + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + ) + projected = diagnostic.get("lock_recovery") or {} + self.assertFalse(projected.get("recovery_sanctioned")) + self.assertEqual(projected.get("outcome"), issue_lock_recovery.REFUSED) + with self.assertRaises((ValueError, RuntimeError)) as caught: + mcp_server.gitea_lock_issue( + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + worktree_path=self.worktree, + ) + # Mutator must not have written a live recovered lock. + lock = issue_lock_store.load_issue_lock( + remote="prgs", + org=MCP_ORG, + repo=MCP_REPO, + issue_number=MCP_ISSUE, + lock_dir=self.lock_dir.name, + ) + self.assertFalse(issue_lock_store.is_lease_live(lock)) + self.assertIn("does not match active identity", " ".join(projected.get("reasons") or [])) + # Refusal text should surface through the mutator error as well. + self.assertTrue(str(caught.exception)) + + def test_probe_failure_degrades_diagnostic_but_raises_on_mutator(self): + """Documented AC9 divergence: diagnostic REFUSED, mutator raises.""" + self.write_dead_lock() + env = self._tool_env() + git_state = self._git_state() + boom = RuntimeError("simulated inventory probe failure") + + with patch( + "mcp_server.api_get_all", side_effect=boom + ), patch( + "mcp_server._list_open_pulls", return_value=[] + ), patch( + "mcp_server.get_auth_header", return_value="token x" + ), patch( + "mcp_server._work_lease_claimant", + return_value={"username": MCP_IDENTITY, "profile": MCP_PROFILE}, + ), patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=git_state, + ), patch( + "mcp_server.issue_duplicate_context_fetcher", + side_effect=lambda *a, **k: ([], [], {"status": "not_claimed"}), + ), patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + diagnostic = mcp_server.gitea_assess_work_issue_duplicate( + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + ) + projected = diagnostic.get("lock_recovery") or {} + self.assertEqual(projected.get("outcome"), issue_lock_recovery.REFUSED) + self.assertFalse(projected.get("recovery_sanctioned")) + self.assertTrue(projected.get("is_candidate")) + self.assertTrue( + any( + "recovery evidence could not be gathered" in r + for r in (projected.get("reasons") or []) + ) + ) + # Mutator path lets the same probe failure raise rather than + # converting it into a soft REFUSED. + with self.assertRaises(RuntimeError) as caught: + mcp_server.gitea_lock_issue( + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + worktree_path=self.worktree, + ) + self.assertIn("Could not list branches", str(caught.exception)) + + +class TestAc6McpUnpublishedClaimRecovery(_UnpublishedMcpBase): + """F2 — public gitea_lock_issue recovers an unpublished dead-session claim.""" + + def test_lock_issue_recovers_unpublished_claim_and_applies_cas(self): + path = self.write_dead_lock() + prior = issue_lock_store.read_lock_file(path) + expected_gen = issue_lock_store.lock_generation(prior) + prior_branch = prior["branch_name"] + prior_worktree = prior["worktree_path"] + prior_pid = prior["session_pid"] + + save_calls: list[dict] = [] + real_save = mcp_server._save_issue_lock + + def tracking_save(data, *, expected_generation=None): + save_calls.append({"expected_generation": expected_generation, "data": dict(data)}) + return real_save(data, expected_generation=expected_generation) + + with patch("mcp_server._save_issue_lock", side_effect=tracking_save): + result = self.run_lock_issue() + + self.assertTrue(result["success"], result) + self.assertEqual(result["issue_number"], MCP_ISSUE) + self.assertEqual(result["branch_name"], prior_branch) + self.assertEqual(result["worktree_path"], prior_worktree) + self.assertTrue(result["lock_freshness"]["live"]) + + recovery = result.get("dead_session_recovery") or {} + self.assertTrue(recovery.get("recovered")) + self.assertEqual(recovery.get("prior_session_pid"), prior_pid) + self.assertEqual(recovery.get("replacement_session_pid"), os.getpid()) + self.assertFalse(recovery.get("prior_pid_alive")) + self.assertEqual( + recovery.get("recovery_mode"), + issue_lock_recovery.RECOVERY_MODE_UNPUBLISHED_CLAIM, + ) + self.assertEqual(recovery.get("branch_name"), prior_branch) + self.assertEqual(recovery.get("worktree_path"), prior_worktree) + + # CAS arming: recovery must pass the observed generation into the write. + self.assertTrue(save_calls, "expected _save_issue_lock to be invoked") + self.assertEqual(save_calls[0]["expected_generation"], expected_gen) + self.assertIsNotNone(save_calls[0]["expected_generation"]) + + # Persisted lock keeps branch/worktree and is live under the new PID. + locked = issue_lock_store.load_issue_lock( + remote="prgs", + org=MCP_ORG, + repo=MCP_REPO, + issue_number=MCP_ISSUE, + lock_dir=self.lock_dir.name, + ) + self.assertEqual(locked["branch_name"], prior_branch) + self.assertEqual(locked["worktree_path"], prior_worktree) + self.assertTrue(issue_lock_store.is_lease_live(locked)) + self.assertEqual( + issue_lock_store.lock_generation(locked), + expected_gen + 1, + ) + + def test_second_recoverer_loses_race_through_the_tool(self): + """Two concurrent recoveries through gitea_lock_issue cannot both win.""" + path = self.write_dead_lock() + prior = issue_lock_store.read_lock_file(path) + observed_gen = issue_lock_store.lock_generation(prior) + + real_bind = issue_lock_store.bind_session_lock + bind_calls: list[int | None] = [] + + def racing_bind(data, lock_dir=None, expected_generation=None): + bind_calls.append(expected_generation) + if expected_generation is None: + return real_bind(data, lock_dir=lock_dir, expected_generation=None) + # First concurrent writer wins. + if len([c for c in bind_calls if c is not None]) == 1: + return real_bind( + data, lock_dir=lock_dir, expected_generation=expected_generation + ) + # Second concurrent writer still holds the pre-race generation. + return real_bind( + data, lock_dir=lock_dir, expected_generation=expected_generation + ) + + # First recovery succeeds and advances generation. + with patch( + "mcp_server.issue_lock_store.bind_session_lock", side_effect=racing_bind + ): + first = self.run_lock_issue() + self.assertTrue(first["success"]) + self.assertEqual(bind_calls[0], observed_gen) + + # Replant a dead lock that still reports the *pre-first* generation so + # a second session that already observed that generation races CAS. + # Simulate by writing dead ownership while leaving generation advanced. + advanced = issue_lock_store.read_lock_file(path) + advanced_gen = issue_lock_store.lock_generation(advanced) + self.assertGreater(advanced_gen, observed_gen) + + # Direct CAS at the store layer with the stale observation must fail — + # this is the same call site gitea_lock_issue uses for recovery writes. + with self.assertRaises(RuntimeError) as caught: + issue_lock_store.bind_session_lock( + { + "issue_number": MCP_ISSUE, + "branch_name": MCP_BRANCH, + "worktree_path": self.worktree, + "remote": "prgs", + "org": MCP_ORG, + "repo": MCP_REPO, + "session_pid": os.getpid(), + "pid": os.getpid(), + }, + self.lock_dir.name, + expected_generation=observed_gen, + ) + self.assertIn("generation changed", str(caught.exception)) + + # And the tool itself must arm expected_generation (not None) so an + # inversion of the recovery_sanctioned ternary would be caught here. + self.assertIsNotNone(bind_calls[0]) + + def test_dirty_worktree_refused_through_the_tool(self): + self.write_dead_lock() + with self.assertRaises((ValueError, RuntimeError)) as caught: + self.run_lock_issue(git_state=self._git_state(porcelain=" M work.txt\n")) + self.assertTrue(str(caught.exception)) + + def test_live_prior_owner_refused_through_the_tool(self): + live = os.getpid() + self.write_dead_lock(session_pid=live, pid=live) + with self.assertRaises((ValueError, RuntimeError)): + self.run_lock_issue() + lock = issue_lock_store.load_issue_lock( + remote="prgs", + org=MCP_ORG, + repo=MCP_REPO, + issue_number=MCP_ISSUE, + lock_dir=self.lock_dir.name, + ) + # Live owner must not be overwritten by a "recovery". + self.assertEqual(lock.get("session_pid"), live) + + if __name__ == "__main__": unittest.main() From 6c15aa88b3ad93f4e58963ebd4ee6f35240cf006 Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Mon, 20 Jul 2026 15:38:11 -0500 Subject: [PATCH 3/3] test(mcp): AC9 assessor/mutator parity + AC6 unpublished recovery MCP regressions (PR #774 F1/F2) Remediate review 487 F1/F2: add AC9 assessor/mutator parity regressions and AC6 MCP-level unpublished-claim recovery regressions through gitea_lock_issue. Tests only; no production code change. Closes nothing; remediates PR #774 review findings only. Co-Authored-By: Grok 4.5