feat: require exact per-mutation capability proof in reviewer reports (Closes #405 )

Adds mutation-capability table validation so review_pr cannot authorize merge or delete mutations without their own exact capability rows. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 11:52:41 -04:00
10 changed files with 358 additions and 501 deletions
@@ -1,198 +0,0 @@
 """Early duplicate-work detection for author work-issue flows (#400)."""
 from __future__ import annotations
 import re
 from typing import Any
 from issue_claim_heartbeat import (
    _linked_open_pr,
    _matching_branch_names,
    classify_issue_claim,
 )
 STAGES = (
    "claim",
    "lock",
    "worktree",
    "edit",
    "commit",
    "push",
    "create_pr",
 )
 ELIGIBILITY_OPEN_PR_EXISTS = "OPEN_PR_EXISTS"
 ELIGIBILITY_DUPLICATE_BRANCH_EXISTS = "DUPLICATE_BRANCH_EXISTS"
 ELIGIBILITY_ACTIVE_CLAIM = "ACTIVE_CLAIM_BY_OTHER"
 ELIGIBILITY_CLEAR = "CLEAR"
 def assess_author_duplicate_work(
    issue_number: int,
    *,
    stage: str,
    open_prs: list[dict] | None = None,
    branch_names: list[str] | None = None,
    claim_entry: dict | None = None,
    matching_branches: list[str] | None = None,
    allow_stale_takeover: bool = False,
 ) -> dict[str, Any]:
    """Fail closed when duplicate work is detected before author mutations (#400)."""
    stage_norm = (stage or "").strip().lower()
    if stage_norm not in STAGES:
        return {
            "allowed": False,
            "block": True,
            "eligibility_class": "INVALID_STAGE",
            "stage": stage_norm or None,
            "reasons": [f"unknown duplicate-work stage {stage!r}"],
            "safe_next_action": f"use one of: {', '.join(STAGES)}",
        }
    prs = list(open_prs or [])
    linked_pr = _linked_open_pr(int(issue_number), prs)
    branches = list(
        matching_branches
        if matching_branches is not None
        else _matching_branch_names(int(issue_number), list(branch_names or []))
    )
    reasons: list[str] = []
    eligibility = ELIGIBILITY_CLEAR
    if linked_pr:
        eligibility = ELIGIBILITY_OPEN_PR_EXISTS
        reasons.append(
            f"open PR #{linked_pr.get('number')} already covers issue "
            f"#{issue_number}"
        )
    if branches and stage_norm in {"claim", "lock", "worktree", "edit"}:
        if eligibility == ELIGIBILITY_CLEAR:
            eligibility = ELIGIBILITY_DUPLICATE_BRANCH_EXISTS
        reasons.append(
            f"remote branch(es) already exist for issue #{issue_number}: "
            f"{', '.join(branches)}"
        )
    entry = claim_entry or {}
    claim_status = (entry.get("status") or "").strip()
    if claim_status in {"active", "awaiting_review"} and stage_norm == "claim":
        if entry.get("reclaimable") and allow_stale_takeover:
            pass
        elif claim_status == "active" and not entry.get("reclaimable"):
            if eligibility == ELIGIBILITY_CLEAR:
                eligibility = ELIGIBILITY_ACTIVE_CLAIM
            reasons.append(
                f"issue #{issue_number} has active claim "
                f"(status={claim_status})"
            )
    elif claim_status == "awaiting_review" and stage_norm == "claim":
        if not linked_pr:
            reasons.append(
                f"issue #{issue_number} is awaiting_review but no linked open PR "
                "was supplied for duplicate-work proof"
            )
    allowed = not reasons
    outcome = "duplicate_work_prevented" if not allowed else "clear"
    if stage_norm == "create_pr" and not allowed:
        outcome = "duplicate_pr_prevented"
    elif stage_norm in {"commit", "push"} and not allowed:
        outcome = "duplicate_push_prevented" if stage_norm == "push" else "duplicate_commit_prevented"
    return {
        "allowed": allowed,
        "block": not allowed,
        "eligibility_class": eligibility if not allowed else ELIGIBILITY_CLEAR,
        "stage": stage_norm,
        "linked_open_pr": linked_pr.get("number") if linked_pr else None,
        "matching_branches": branches,
        "claim_status": claim_status or None,
        "outcome": outcome,
        "reasons": reasons,
        "safe_next_action": (
            "stop without edits/commit/push/PR; produce reconciliation handoff "
            "preserving local work only"
            if not allowed
            else "proceed"
        ),
    }
 def build_claim_entry_from_classification(classification: dict) -> dict:
    """Map ``classify_issue_claim`` output to gate claim metadata."""
    return {
        "status": classification.get("status"),
        "reclaimable": classification.get("reclaimable"),
        "linked_open_pr": classification.get("linked_open_pr"),
    }
 _DUPLICATE_OUTCOME_RE = re.compile(
    r"(duplicate\s+(?:pr|branch|commit|push)\s+prevented|"
    r"duplicate\s+work\s+not\s+prevented|reconciliation\s+handoff)",
    re.IGNORECASE,
 )
 def assess_work_issue_duplicate_prevention_report(report_text: str) -> dict:
    """#400: work-issue reports must state duplicate-work prevention outcome."""
    text = report_text or ""
    if "duplicate work" not in text.lower() and "duplicate pr" not in text.lower():
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "safe_next_action": "proceed",
        }
    if _DUPLICATE_OUTCOME_RE.search(text):
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "safe_next_action": "proceed",
        }
    return {
        "proven": False,
        "block": True,
        "reasons": [
            "duplicate-work discussion must name a prevention outcome "
            "(duplicate PR/branch/commit/push prevented, or duplicate work "
            "not prevented, or reconciliation handoff)"
        ],
        "safe_next_action": "state exact duplicate-work prevention class in final report",
    }
 def classify_and_assess(
    issue: dict,
    *,
    stage: str,
    comments: list[dict] | None = None,
    open_prs: list[dict] | None = None,
    branch_names: list[str] | None = None,
    allow_stale_takeover: bool = False,
 ) -> dict[str, Any]:
    """Combine claim classification with duplicate-work gate assessment."""
    issue_number = int(issue.get("number") or 0)
    claim = classify_issue_claim(
        issue=issue,
        comments=comments or [],
        open_prs=open_prs or [],
        branch_names=branch_names or [],
    )
    assessment = assess_author_duplicate_work(
        issue_number,
        stage=stage,
        open_prs=open_prs,
        branch_names=branch_names,
        claim_entry=build_claim_entry_from_classification(claim),
        matching_branches=claim.get("matching_branches"),
        allow_stale_takeover=allow_stale_takeover,
    )
    return {
        "issue_number": issue_number,
        "claim": claim,
        "duplicate_work": assessment,
    }
@@ -889,6 +889,22 @@ def _rule_reviewer_review_mutation(
    )
 def _rule_reviewer_mutation_capability_proof(report_text: str) -> list[dict[str, str]]:
    from reviewer_mutation_capability_proof import assess_mutation_capability_proof
    result = assess_mutation_capability_proof(report_text)
    if not result.get("block"):
        return []
    return _findings_from_reasons(
        "reviewer.mutation_capability_proof",
        result.get("reasons") or [],
        field="Capabilities proven",
        severity="block",
        safe_next_action=result.get("safe_next_action")
        or "document exact per-mutation capability proof before each mutation",
    )
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
    "review_pr": [
        _rule_shared_controller_handoff,
@@ -909,6 +925,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
        _rule_reviewer_target_branch_freshness,
        _rule_reviewer_mutation_ledger,
        _rule_reviewer_review_mutation,
        _rule_reviewer_mutation_capability_proof,
    ],
    "reconcile_already_landed": [
        _rule_reconcile_controller_handoff,
@@ -503,7 +503,6 @@ import issue_lock_worktree  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import author_duplicate_work_gate  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
@@ -1050,76 +1049,6 @@ def gitea_create_issue(
    return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 def _list_repo_branch_names(h: str, o: str, r: str, auth: str, *, limit: int = 200) -> list[str]:
    branches = api_get_all(f"{repo_api_url(h, o, r)}/branches", auth, limit=limit)
    return [_branch_entry_name(branch) for branch in branches]
 def _gather_author_duplicate_work_context(
    issue_number: int,
    *,
    h: str,
    o: str,
    r: str,
    auth: str,
    exclude_branch_name: str | None = None,
 ) -> dict:
    base = repo_api_url(h, o, r)
    issue = api_request("GET", f"{base}/issues/{issue_number}", auth)
    comments = api_request("GET", f"{base}/issues/{issue_number}/comments", auth) or []
    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
    branch_names = _list_repo_branch_names(h, o, r, auth)
    if exclude_branch_name:
        branch_names = [
            name for name in branch_names
            if name != exclude_branch_name
        ]
    return {
        "issue": issue,
        "comments": comments,
        "open_prs": open_prs,
        "branch_names": branch_names,
    }
 def _enforce_author_duplicate_work_gate(
    issue_number: int,
    stage: str,
    *,
    h: str,
    o: str,
    r: str,
    auth: str,
    allow_stale_takeover: bool = False,
    exclude_branch_name: str | None = None,
 ) -> dict:
    """Fail closed when duplicate work is detected (#400)."""
    ctx = _gather_author_duplicate_work_context(
        issue_number,
        h=h,
        o=o,
        r=r,
        auth=auth,
        exclude_branch_name=exclude_branch_name,
    )
    result = author_duplicate_work_gate.classify_and_assess(
        ctx["issue"],
        stage=stage,
        comments=ctx["comments"],
        open_prs=ctx["open_prs"],
        branch_names=ctx["branch_names"],
        allow_stale_takeover=allow_stale_takeover,
    )
    assessment = result.get("duplicate_work") or {}
    if assessment.get("block"):
        reasons = "; ".join(assessment.get("reasons") or ["duplicate work detected"])
        raise RuntimeError(
            f"Author duplicate-work gate (#400) blocked at stage '{stage}': "
            f"{reasons} (fail closed)"
        )
    return result
@mcp.tool()
 def gitea_lock_issue(
    issue_number: int,
@@ -1182,17 +1111,48 @@ def gitea_lock_issue(
            issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
        )
    # 2. Check if the issue already has an open PR (reuse protection)
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
-    _enforce_author_duplicate_work_gate(
+    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
-        issue_number,
+
-        "lock",
+    try:
-        h=h,
+        prs = api_get_all(url, auth)
-        o=o,
+    except Exception as e:
-        r=r,
+        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
-        auth=auth,
+
-        exclude_branch_name=branch_name,
+    for pr in prs:
-    )
+        pr_head = pr.get("head", {}).get("ref", "")
        pr_title = pr.get("title", "")
        pr_body = pr.get("body", "")
        if expected_pattern in pr_head:
            raise ValueError(
                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
            )
        patterns = [
            f"closes #{issue_number}",
            f"fixes #{issue_number}",
        ]
        text_to_check = f"{pr_title} {pr_body}".lower()
        if any(p in text_to_check for p in patterns):
            raise ValueError(
                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
            )
    branch_url = f"{repo_api_url(h, o, r)}/branches"
    try:
        branches = api_get_all(branch_url, auth)
    except Exception as e:
        raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
    for branch in branches:
        name = _branch_entry_name(branch)
        if expected_pattern in name:
            raise ValueError(
                f"Issue #{issue_number} already has matching branch '{name}' "
                "(fail closed)"
            )
    work_lease = _build_author_issue_work_lease(
        issue_number=issue_number,
@@ -1313,17 +1273,6 @@ def gitea_create_pr(
            f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
        )
    auth = _auth(h)
    _enforce_author_duplicate_work_gate(
        int(locked_issue),
        "create_pr",
        h=h,
        o=o,
        r=r,
        auth=auth,
        exclude_branch_name=locked_branch,
    )
    # Check for forbidden terms anywhere in title/body
    forbidden_terms = ["equivalent", "related", "same as"]
    text_to_check = f"{title} {body}".lower()
@@ -1340,6 +1289,7 @@ def gitea_create_pr(
            f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
        )
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/pulls"
    payload = {"title": title, "body": body, "head": head, "base": base}
    meta = {"title": title, "head": head, "base": base}
@@ -6053,14 +6003,6 @@ def gitea_mark_issue(
        )
    if action == "start":
        _enforce_author_duplicate_work_gate(
            issue_number,
            "claim",
            h=h,
            o=o,
            r=r,
            auth=auth,
        )
        with _audited("label_issue", host=h, remote=remote, org=o, repo=r,
                      issue_number=issue_number,
                      request_metadata={"op": "add", "label": "status:in-progress"}):
@@ -6142,65 +6084,6 @@ def gitea_post_heartbeat(
    )
@mcp.tool()
 def gitea_assess_author_duplicate_work(
    issue_number: int,
    stage: str,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    allow_stale_takeover: bool = False,
    exclude_branch_name: str | None = None,
 ) -> dict:
    """Read-only: assess duplicate-work risk before author mutations (#400).
    Call at claim, lock, worktree, edit, commit, push, and create_pr stages.
    """
    read_block = _profile_operation_gate("gitea.read")
    if read_block:
        return {
            "success": False,
            "reasons": read_block,
            "permission_report": _permission_block_report("gitea.read"),
        }
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    ctx = _gather_author_duplicate_work_context(
        issue_number,
        h=h,
        o=o,
        r=r,
        auth=auth,
        exclude_branch_name=exclude_branch_name,
    )
    result = author_duplicate_work_gate.classify_and_assess(
        ctx["issue"],
        stage=stage,
        comments=ctx["comments"],
        open_prs=ctx["open_prs"],
        branch_names=ctx["branch_names"],
        allow_stale_takeover=allow_stale_takeover,
    )
    assessment = result.get("duplicate_work") or {}
    return {
        "success": True,
        "performed": False,
        "issue_number": issue_number,
        "stage": stage,
        "allowed": assessment.get("allowed"),
        "block": assessment.get("block"),
        "eligibility_class": assessment.get("eligibility_class"),
        "outcome": assessment.get("outcome"),
        "linked_open_pr": assessment.get("linked_open_pr"),
        "matching_branches": assessment.get("matching_branches"),
        "claim_status": assessment.get("claim_status"),
        "reasons": assessment.get("reasons"),
        "safe_next_action": assessment.get("safe_next_action"),
        "claim": result.get("claim"),
    }
@mcp.tool()
 def gitea_reconcile_issue_claims(
    state: str = "open",
@@ -3622,33 +3622,18 @@ def assess_work_issue_mode_isolation(report_text: str) -> dict:
    }
 def assess_work_issue_duplicate_prevention_report(report_text, **kwargs):
    """#400: work-issue reports must classify duplicate-work prevention."""
    from author_duplicate_work_gate import (
        assess_work_issue_duplicate_prevention_report as _assess,
    )
    return _assess(report_text, **kwargs)
 def assess_work_issue_final_report(report_text: str) -> dict:
    """#139: composite verifier for work-issue final reports."""
    checks = {
        "workflow_source": assess_work_issue_workflow_source(report_text),
        "mode_isolation": assess_work_issue_mode_isolation(report_text),
        "duplicate_prevention": assess_work_issue_duplicate_prevention_report(
            report_text
        ),
    }
    reasons = []
    downgraded = False
    for name, result in checks.items():
        verdict = result.get("verdict")
-        if result.get("block"):
+        if verdict in ("missing", "incomplete"):
            downgraded = True
            reasons.extend(result.get("reasons") or [])
        elif verdict in ("missing", "incomplete"):
            downgraded = True
            reasons.extend(result.get("reasons") or [])
        elif result.get("downgraded") or not result.get("complete", True):
@@ -3656,9 +3641,6 @@ def assess_work_issue_final_report(report_text: str) -> dict:
            reasons.extend(
                f"{name}: {r}" for r in (result.get("reasons") or [])
            )
        elif result.get("proven") is False:
            downgraded = True
            reasons.extend(result.get("reasons") or [])
    grade = "A" if not downgraded else "downgraded"
    return {
@@ -5616,3 +5598,12 @@ def assess_proof_backed_handoff_report(report_text, **kwargs):
    from reviewer_proof_backed_handoff import assess_proof_backed_handoff_report as _assess
    return _assess(report_text, **kwargs)
 def assess_mutation_capability_proof(report_text, **kwargs):
    """#405: exact per-mutation capability proof in reviewer final reports."""
    from reviewer_mutation_capability_proof import (
        assess_mutation_capability_proof as _assess,
    )
    return _assess(report_text, **kwargs)
@@ -0,0 +1,129 @@
 """Exact per-mutation capability proof verifier for reviewer reports (#405).
 A reviewer final report may prove ``review_pr`` capability and then also merge
 a PR or delete a remote branch. Merge and branch deletion are separate
 mutations that require their own exact capability proof — a nearby capability
 must never authorize a different operation. This verifier requires a
 mutation-capability table pairing every performed mutation with the exact
 task/permission resolved *before* that mutation.
 """
 from __future__ import annotations
 import re
 # Mutations this verifier tracks, with the exact capability tokens that
 # authorize each. A row for the mutation must cite one of its own tokens;
 # tokens from a different mutation (a "nearby capability") never count.
 _REVIEW_TOKENS = ("review_pr", "gitea.pr.review", "gitea.pr.approve",
                  "gitea.pr.request_changes", "request_changes_pr", "approve_pr")
 _MERGE_TOKENS = ("merge_pr", "gitea.pr.merge")
 _DELETE_TOKENS = ("delete_branch", "gitea.branch.delete")
 # Detect that a mutation was actually performed (not merely mentioned as a
 # non-goal or skipped).
 _MERGE_PERFORMED = re.compile(
    r"(?:gitea_merge_pr\b(?![^\n]*\b(?:not called|skipped|blocked)\b)|"
    r"^\s*[-*]?\s*merge result\s*:\s*merged\b|"
    r"\bpr merged\b|\bmerge commit\s*(?:sha)?\s*[:=]?\s*[0-9a-f]{7,})",
    re.IGNORECASE | re.MULTILINE,
 )
 _DELETE_PERFORMED = re.compile(
    r"(?:gitea_delete_branch\b(?![^\n]*\b(?:not called|skipped|blocked)\b)|"
    r"^\s*[-*]?\s*(?:remote )?branch deleted\s*:|"
    r"\bdeleted (?:the )?(?:remote )?branch\b|"
    r"^\s*[-*]?\s*branch deletion\s*:\s*(?!skipped|none|not)\S)",
    re.IGNORECASE | re.MULTILINE,
 )
 _REVIEW_PERFORMED = re.compile(
    r"(?:gitea_submit_pr_review\b|gitea_mark_final_review_decision\b|"
    r"^\s*[-*]?\s*review (?:decision|verdict|mutation)\s*:\s*"
    r"(?:approved|request[_ ]changes)\b|\breview submitted\b)",
    re.IGNORECASE | re.MULTILINE,
 )
 # Post-hoc proof: capability resolved *after* the mutation is never valid.
 _POST_HOC = re.compile(
    r"capabilit(?:y|ies)\s+(?:resolved|proven|checked)\s+(?:after|post[- ])\s*"
    r"(?:the\s+)?(?:merge|deletion|delete|mutation|review)",
    re.IGNORECASE,
 )
 # The report must carry an explicit mutation-capability table.
 _TABLE_MARKER = re.compile(
    r"mutation[- ]capability(?:\s+table)?|capability[- ]per[- ]mutation",
    re.IGNORECASE,
 )
 def _tokens_present(text: str, tokens: tuple[str, ...]) -> bool:
    low = text.lower()
    return any(tok.lower() in low for tok in tokens)
 def assess_mutation_capability_proof(report_text: str) -> dict:
    """Validate exact per-mutation capability proof in a reviewer report.
    Returns ``{proven, block, reasons, safe_next_action}``. A report that
    performs no mutation beyond an ordinary review passes only when its
    review capability is cited; merge/delete each demand their own exact
    capability row. Fail closed on nearby-capability substitution, a
    missing table, missing rows, or post-hoc proof.
    """
    text = report_text or ""
    reasons: list[str] = []
    merged = bool(_MERGE_PERFORMED.search(text))
    deleted = bool(_DELETE_PERFORMED.search(text))
    reviewed = bool(_REVIEW_PERFORMED.search(text))
    extra_mutation = merged or deleted
    if _POST_HOC.search(text):
        reasons.append(
            "capability proof recorded after the mutation; exact capability "
            "must be resolved before each mutation"
        )
    # A review-only report needs its review capability cited; no table required.
    if reviewed and not _tokens_present(text, _REVIEW_TOKENS):
        reasons.append(
            "review mutation performed without exact review capability proof "
            "(review_pr / gitea.pr.review)"
        )
    if extra_mutation and not _TABLE_MARKER.search(text):
        reasons.append(
            "mutation beyond review performed without a mutation-capability "
            "table (mutation, exact task/capability, result, order-before)"
        )
    if merged:
        if not _tokens_present(text, _MERGE_TOKENS):
            reasons.append(
                "merge performed without exact merge capability proof "
                "(merge_pr / gitea.pr.merge); nearby review_pr does not "
                "authorize merge"
            )
    if deleted:
        if not _tokens_present(text, _DELETE_TOKENS):
            reasons.append(
                "branch deletion performed without exact delete capability "
                "proof (delete_branch / gitea.branch.delete); nearby "
                "merge_pr does not authorize branch deletion"
            )
    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "safe_next_action": (
            "proceed"
            if proven
            else "add a mutation-capability table with the exact resolved "
            "task/permission and pre-mutation order for every mutation; "
            "skip any mutation whose exact capability is unproven"
        ),
    }
@@ -895,6 +895,26 @@ Use precise wording:
 Do not collapse review, merge, cleanup, or external-state mutations into vague wording.
 ## 31B. Mutation-capability table (#405)
 Every performed mutation requires exact capability proof resolved **before** that
 mutation executes. Nearby capabilities never authorize a different operation —
 `review_pr` does not authorize `merge_pr`, and `merge_pr` does not authorize
 `delete_branch` / `gitea.branch.delete`.
 When any mutation beyond a bare review occurs (merge, branch delete, issue
 close/comment, etc.), the final report must include a **mutation-capability table**
 with one row per performed mutation:
 * mutation (tool/action name)
 * exact task/capability resolved (for example `merge_pr` / `gitea.pr.merge`)
 * result
 * order/timestamp proof that capability was resolved before the mutation
 If exact capability proof is missing, skip the mutation or stop the workflow —
 never claim a performed mutation without its row. Post-hoc capability proof after
 the mutation fails validation.
 ## 31A. Local artifact and report consistency rule
 Do not create local walkthrough, notes, markdown, JSON, or report artifacts during reviewer runs unless the canonical workflow or operator explicitly requires it.
@@ -277,24 +277,6 @@ Do not select an issue based only on memory from a previous session.
 Before claiming or working on an issue, check whether there is already an open PR, branch, or active claim for that issue.
 Run `gitea_assess_author_duplicate_work` at these stages and stop when `block` is true:
 * `claim` — before `gitea_mark_issue`
 * `lock` — before `gitea_lock_issue`
 * `worktree` / `edit` — before creating a worktree or editing files
 * `commit` — immediately before `git commit`
 * `push` — immediately before `git push`
 * `create_pr` — immediately before `gitea_create_pr` (also enforced server-side)
 `gitea_mark_issue`, `gitea_lock_issue`, and `gitea_create_pr` enforce the same gate server-side and fail closed.
 If a concurrent open PR appears after work begins:
 * before commit or push — stop and preserve local work without pushing
 * after push but before PR creation — produce a reconciliation handoff instead of opening a PR
 Final reports must name the duplicate-work outcome (`duplicate PR prevented`, `duplicate branch prevented`, `duplicate commit prevented`, `duplicate push prevented`, `duplicate work not prevented`, or `reconciliation handoff`).
 If an open PR already exists for the issue, do not implement duplicate work.
 Classify the issue as:
@@ -1,102 +0,0 @@
 """Tests for early author duplicate-work gate (#400)."""
 import sys
 import unittest
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 from author_duplicate_work_gate import (  # noqa: E402
    ELIGIBILITY_OPEN_PR_EXISTS,
    assess_author_duplicate_work,
    assess_work_issue_duplicate_prevention_report,
 )
 def _open_pr(number: int = 397, issue: int = 395) -> dict:
    return {
        "number": number,
        "head": {"ref": f"feat/issue-{issue}-example"},
        "title": f"feat: example (Closes #{issue})",
        "body": f"Closes #{issue}",
    }
 class TestAuthorDuplicateWorkGate(unittest.TestCase):
    def test_clear_when_no_duplicates(self):
        result = assess_author_duplicate_work(
            400,
            stage="claim",
            open_prs=[],
            branch_names=["master", "feat/issue-399-other"],
        )
        self.assertTrue(result["allowed"])
        self.assertFalse(result["block"])
    def test_open_pr_blocks_claim(self):
        result = assess_author_duplicate_work(
            395,
            stage="claim",
            open_prs=[_open_pr()],
            branch_names=[],
        )
        self.assertFalse(result["allowed"])
        self.assertEqual(result["eligibility_class"], ELIGIBILITY_OPEN_PR_EXISTS)
    def test_matching_branch_blocks_lock_not_create_pr(self):
        branches = ["feat/issue-400-early-duplicate-work-gate"]
        lock = assess_author_duplicate_work(
            400,
            stage="lock",
            open_prs=[],
            branch_names=branches,
        )
        self.assertFalse(lock["allowed"])
        create_pr = assess_author_duplicate_work(
            400,
            stage="create_pr",
            open_prs=[],
            branch_names=branches,
            matching_branches=[],
        )
        self.assertTrue(create_pr["allowed"])
    def test_open_pr_blocks_create_pr_stage(self):
        result = assess_author_duplicate_work(
            395,
            stage="create_pr",
            open_prs=[_open_pr()],
            branch_names=["feat/issue-395-proof-backed-review-handoff"],
        )
        self.assertFalse(result["allowed"])
        self.assertEqual(result["outcome"], "duplicate_pr_prevented")
    def test_push_stage_blocks_on_concurrent_pr(self):
        result = assess_author_duplicate_work(
            395,
            stage="push",
            open_prs=[_open_pr()],
            branch_names=[],
        )
        self.assertFalse(result["allowed"])
        self.assertEqual(result["outcome"], "duplicate_push_prevented")
    def test_duplicate_prevention_report_requires_outcome(self):
        bad = assess_work_issue_duplicate_prevention_report(
            "Duplicate work detected for issue #395."
        )
        self.assertFalse(bad["proven"])
        good = assess_work_issue_duplicate_prevention_report(
            "Duplicate PR prevented; reconciliation handoff produced."
        )
        self.assertTrue(good["proven"])
    def test_exported_from_review_proofs(self):
        from review_proofs import assess_work_issue_duplicate_prevention_report as exported
        self.assertTrue(callable(exported))
 if __name__ == "__main__":
    unittest.main()
@@ -95,12 +95,6 @@ def test_create_issue_workflow_contract():
    assert "## 9. Duplicate search before mutation" in text
 def test_author_duplicate_work_gate_exported():
    from review_proofs import assess_work_issue_duplicate_prevention_report
    assert callable(assess_work_issue_duplicate_prevention_report)
 def test_work_issue_workflow_contract():
    text = (SKILL_DIR / "workflows" / "work-issue.md").read_text(encoding="utf-8")
    assert "canonical: true" in text
@@ -0,0 +1,141 @@
 """Tests for exact per-mutation capability proof in reviewer reports (#405)."""
 import sys
 import unittest
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 from reviewer_mutation_capability_proof import assess_mutation_capability_proof
 from review_proofs import assess_mutation_capability_proof as proofs_assess
 from final_report_validator import assess_final_report_validator
 REVIEW_ONLY = """
 Review decision: approved
 Mutation capability table:
 - gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | resolved before review
 """
 MERGE_PROVEN = """
 Review decision: approved
 Merge result: merged 0123456789ab
 Mutation capability table:
 - gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
 - gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | resolved before merge
 """
 MERGE_AND_DELETE_PROVEN = """
 Review decision: approved
 Merge result: merged 0123456789ab
 Remote branch deleted: feat/issue-x
 Mutation capability table:
 - gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
 - gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | before merge
 - gitea_delete_branch | delete_branch (gitea.branch.delete) | deleted | before delete
 """
 class TestModule(unittest.TestCase):
    def test_review_only_with_capability_passes(self):
        r = assess_mutation_capability_proof(REVIEW_ONLY)
        self.assertTrue(r["proven"], r["reasons"])
    def test_merge_with_exact_capability_passes(self):
        r = assess_mutation_capability_proof(MERGE_PROVEN)
        self.assertTrue(r["proven"], r["reasons"])
    def test_merge_and_delete_fully_proven_passes(self):
        r = assess_mutation_capability_proof(MERGE_AND_DELETE_PROVEN)
        self.assertTrue(r["proven"], r["reasons"])
    def test_review_pr_does_not_authorize_merge(self):
        report = """
 Review decision: approved
 Merge result: merged 0123456789ab
 Mutation capability table:
 - gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
 """
        r = assess_mutation_capability_proof(report)
        self.assertFalse(r["proven"])
        self.assertTrue(any("merge" in x.lower() for x in r["reasons"]), r["reasons"])
    def test_merge_pr_does_not_authorize_branch_deletion(self):
        report = """
 Review decision: approved
 Merge result: merged 0123456789ab
 Remote branch deleted: feat/issue-x
 Mutation capability table:
 - gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
 - gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | before merge
 """
        r = assess_mutation_capability_proof(report)
        self.assertFalse(r["proven"])
        self.assertTrue(any("delet" in x.lower() for x in r["reasons"]), r["reasons"])
    def test_delete_skipped_when_capability_missing_passes(self):
        report = """
 Review decision: approved
 Merge result: merged 0123456789ab
 Branch deletion: skipped — delete_branch capability not available
 Mutation capability table:
 - gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
 - gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | before merge
 """
        r = assess_mutation_capability_proof(report)
        self.assertTrue(r["proven"], r["reasons"])
    def test_missing_table_when_merging_blocks(self):
        report = """
 Review decision: approved
 Merge result: merged 0123456789ab
 merge_pr gitea.pr.merge resolved
 """
        r = assess_mutation_capability_proof(report)
        self.assertFalse(r["proven"])
        self.assertTrue(any("table" in x.lower() for x in r["reasons"]), r["reasons"])
    def test_post_hoc_proof_blocks(self):
        report = """
 Review decision: approved
 Merge result: merged 0123456789ab
 Mutation capability table:
 - gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | capability resolved after merge
 """
        r = assess_mutation_capability_proof(report)
        self.assertFalse(r["proven"])
        self.assertTrue(any("after" in x.lower() for x in r["reasons"]), r["reasons"])
    def test_review_without_capability_blocks(self):
        report = "Review decision: approved\nreview submitted\n"
        r = assess_mutation_capability_proof(report)
        self.assertFalse(r["proven"])
    def test_no_mutation_no_requirement(self):
        r = assess_mutation_capability_proof("Selected PR: #1\nSkipped, no action.")
        self.assertTrue(r["proven"], r["reasons"])
 class TestWiring(unittest.TestCase):
    def test_review_proofs_wrapper_matches_module(self):
        self.assertEqual(
            proofs_assess(MERGE_PROVEN)["proven"],
            assess_mutation_capability_proof(MERGE_PROVEN)["proven"],
        )
    def test_final_report_validator_flags_nearby_capability_merge(self):
        report = """
 ## Controller Handoff
 - Task: review_pr
 Review decision: approved
 Merge result: merged 0123456789ab
 Mutation capability table:
 - gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
 """
        result = assess_final_report_validator(report, "review_pr")
        rule_ids = [f["rule_id"] for f in result.get("findings", [])]
        self.assertIn("reviewer.mutation_capability_proof", rule_ids)
 if __name__ == "__main__":
    unittest.main()