feat: require post-merge cleanup safety-gate proof in reviewer reports (Closes #402)

Adds post_merge_cleanup_proof validation so cleanup claims must carry the
branch deletion and worktree removal checklists, or report CLEANUP_SKIPPED
with an exact blocker.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

author_duplicate_work_gate.py

@@ -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,
-    }

final_report_validator.py

@@ -11,6 +11,7 @@ import inspect
 import re
 from typing import Any, Callable
 
+from post_merge_cleanup_proof import assess_post_merge_cleanup_proof
 from review_proofs import (
     HANDOFF_HEADING,
     assess_controller_handoff,
@@ -456,40 +457,6 @@ def _rule_reviewer_git_fetch_readonly(
     return []
 
 
-def _rule_reviewer_validation_failure_history(
-    report_text: str,
-    *,
-    validation_session: dict | None = None,
-) -> list[dict[str, str]]:
-    from reviewer_validation_failure_history import (
-        assess_validation_failure_history_report,
-    )
-
-    session = validation_session or {}
-    observed = session.get("observed_failures") or []
-    if not observed:
-        return []
-
-    result = assess_validation_failure_history_report(
-        report_text,
-        validation_session=session,
-    )
-    if result.get("proven"):
-        return []
-    severity = "block" if result.get("violations") else "downgrade"
-    return [
-        validator_finding(
-            "reviewer.validation_failure_history",
-            severity,
-            "Validation failure history",
-            reason,
-            result.get("safe_next_action")
-            or "document every observed validation failure before claiming pass",
-        )
-        for reason in (result.get("violations") or result.get("reasons") or ["incomplete"])
-    ]
-
-
 def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
     text = report_text or ""
     if not _BARE_PYTEST_RE.search(text):
@@ -889,6 +856,20 @@ def _rule_reviewer_review_mutation(
     )
 
 
+def _rule_reviewer_post_merge_cleanup_proof(report_text: str) -> list[dict[str, str]]:
+    result = assess_post_merge_cleanup_proof(report_text)
+    if not result.get("block"):
+        return []
+    return _findings_from_reasons(
+        "reviewer.post_merge_cleanup_proof",
+        result.get("reasons") or [],
+        field="Cleanup status",
+        severity="block",
+        safe_next_action=result.get("safe_next_action")
+        or "report CLEANUP_SKIPPED with blocker or full cleanup checklist",
+    )
+
+
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "review_pr": [
         _rule_shared_controller_handoff,
@@ -898,7 +879,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_mutation_categories,
         _rule_reviewer_git_fetch_readonly,
         _rule_reviewer_validation_command,
-        _rule_reviewer_validation_failure_history,
         _rule_reviewer_validation_structured,
         _rule_reviewer_linked_issue,
         _rule_reviewer_baseline_on_failure,
@@ -909,6 +889,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_post_merge_cleanup_proof,
     ],
     "reconcile_already_landed": [
         _rule_reconcile_controller_handoff,
@@ -1005,7 +986,6 @@ def assess_final_report_validator(
     local_edits: bool = False,
     issue_filing_lock: dict | None = None,
     session_pr_opened: bool = False,
-    validation_session: dict | None = None,
 ) -> dict[str, Any]:
     """Validate final-report text against task-specific proof rules (#327).
 
@@ -1061,7 +1041,6 @@ def assess_final_report_validator(
         "mutations_observed": mutations_observed,
         "local_edits": local_edits,
         "session_pr_opened": session_pr_opened,
-        "validation_session": validation_session,
     }
 
     for rule in _RULES_BY_TASK.get(normalized_kind, ()):

gitea_mcp_server.py

@@ -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(
-        issue_number,
-        "lock",
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=branch_name,
-    )
+    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
+
+    try:
+        prs = api_get_all(url, auth)
+    except Exception as e:
+        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
+
+    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,
@@ -1285,7 +1245,7 @@ def gitea_create_pr(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196) ──
@@ -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}
@@ -3514,32 +3464,16 @@ def gitea_delete_branch(
         repo: Override the repository name.
 
     Returns:
-        dict with 'success' and 'message'; on a permission block,
-        'success'/'performed' False, 'reasons', and a structured
-        'permission_report' (#142) with no API call made.
+        dict with 'success' and 'message'.
     """
-    gate_reasons = _profile_operation_gate("gitea.branch.delete")
-    if gate_reasons:
-        return {
-            "success": False,
-            "performed": False,
-            "required_permission": "gitea.branch.delete",
-            "reasons": gate_reasons,
-            "permission_report": _permission_block_report("gitea.branch.delete"),
-        }
-
     verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     import urllib.parse
     encoded_branch = urllib.parse.quote(branch, safe="")
     url = f"{repo_api_url(h, o, r)}/branches/{encoded_branch}"
-    request_metadata = {
-        "branch": branch,
-        "required_permission": "gitea.branch.delete",
-    }
     with _audited("delete_branch", host=h, remote=remote, org=o, repo=r,
-                  target_branch=branch, request_metadata=request_metadata):
+                  target_branch=branch, request_metadata={"branch": branch}):
         api_request("DELETE", url, auth)
     return {"success": True, "message": f"Remote branch '{branch}' deleted."}
 
@@ -5576,7 +5510,6 @@ def gitea_validate_review_final_report(
     action_log: list[dict] | None = None,
     mutations_observed: bool = False,
     local_edits: bool = False,
-    validation_session: dict | None = None,
 ) -> dict:
     """Read-only: machine-validate reviewer final report schema before output (#391).
 
@@ -5594,7 +5527,6 @@ def gitea_validate_review_final_report(
         action_log=action_log,
         mutations_observed=mutations_observed,
         local_edits=local_edits,
-        validation_session=validation_session,
     )
 
 
@@ -6053,14 +5985,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 +6066,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",

post_merge_cleanup_proof.py

@@ -0,0 +1,239 @@
+"""Post-merge cleanup proof verifier for reviewer final reports (#402)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+CLEANUP_SKIPPED = "CLEANUP_SKIPPED"
+CLEANUP_PERFORMED = "CLEANUP_PERFORMED"
+
+_CLEANUP_SECTION_HINT = re.compile(
+    r"(?:cleanup (?:status|result|mutations)|post-merge cleanup|"
+    r"gitea_delete_branch|remote branch.*deleted|worktree remove)",
+    re.IGNORECASE,
+)
+_CLEANUP_SKIPPED_RE = re.compile(r"\bCLEANUP_SKIPPED\b", re.IGNORECASE)
+_CLEANUP_BLOCKER_RE = re.compile(
+    r"(?:cleanup blocker|cleanup skip(?:ped)? reason)\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+_REMOTE_DELETE_CLAIM_RE = re.compile(
+    r"(?:gitea_delete_branch|remote (?:head )?branch (?:was )?deleted|"
+    r"deleted remote branch|delete_branch)",
+    re.IGNORECASE,
+)
+_WORKTREE_REMOVE_CLAIM_RE = re.compile(
+    r"(?:git worktree remove|worktree (?:was )?removed|removed (?:local )?worktree|"
+    r"worktree cleanup performed)",
+    re.IGNORECASE,
+)
+_DELETE_CAPABILITY_RE = re.compile(
+    r"(?:delete[- ]branch capability resolved|gitea\.branch\.delete)\s*:\s*"
+    r".*(?:gitea\.branch\.delete|delete_branch).*(?:resolved|allowed|proven)|"
+    r"gitea\.branch\.delete\s+(?:resolved|allowed|proven)",
+    re.IGNORECASE,
+)
+_DELETE_TASK_RE = re.compile(
+    r"(?:delete_branch|cleanup_branch|reconcile_merged_cleanups)",
+    re.IGNORECASE,
+)
+_MERGE_RESULT_RE = re.compile(
+    r"merge result\s*:\s*(?:merged|success|performed)",
+    re.IGNORECASE,
+)
+_MERGE_COMMIT_SHA_RE = re.compile(
+    r"(?:merge commit sha|merged commit sha|merge commit)\s*:\s*([0-9a-f]{7,40})",
+    re.IGNORECASE,
+)
+_PR_HEAD_BRANCH_RE = re.compile(
+    r"(?:merged pr head branch|pr head branch|deleted branch)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_BRANCH_NOT_PROTECTED_RE = re.compile(
+    r"branch (?:is )?not protected|branch protection\s*:\s*(?:none|false|no)",
+    re.IGNORECASE,
+)
+_OPEN_PR_INVENTORY_RE = re.compile(
+    r"(?:no other open pr(?:\s+references)?(?:\s+\S+)?|open pr inventory proof|"
+    r"open pr references).*(?:none|zero|0|clear|inventory complete)|"
+    r"no other open pr references branch",
+    re.IGNORECASE,
+)
+_ACTIVE_CLAIM_LEASE_RE = re.compile(
+    r"(?:no active (?:heartbeat|claim|lease)|"
+    r"(?:active )?(?:heartbeat|claim|lease)(?:/(?:claim|lease))*\s*:\s*none)",
+    re.IGNORECASE,
+)
+_SESSION_OWNED_WORKTREE_RE = re.compile(
+    r"(?:removed worktree path|cleanup worktree path|session-owned worktree)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_BRANCHES_PATH_RE = re.compile(r"\bbranches/", re.IGNORECASE)
+_CLEAN_TRACKED_RE = re.compile(
+    r"(?:pre-removal tracked state|tracked state before removal)\s*:\s*clean",
+    re.IGNORECASE,
+)
+_CLEAN_UNTRACKED_RE = re.compile(
+    r"(?:pre-removal untracked state|untracked state before removal)\s*:\s*clean",
+    re.IGNORECASE,
+)
+_WORKTREE_LIST_AFTER_RE = re.compile(
+    r"(?:git worktree list after|post-removal worktree list|worktree list after)",
+    re.IGNORECASE,
+)
+_WRONG_BRANCH_RE = re.compile(
+    r"deleted branch (?:does not match|!=|differs from) (?:merged )?pr head",
+    re.IGNORECASE,
+)
+
+
+def _claims_remote_delete(text: str) -> bool:
+    return bool(_REMOTE_DELETE_CLAIM_RE.search(text))
+
+
+def _claims_worktree_remove(text: str) -> bool:
+    return bool(_WORKTREE_REMOVE_CLAIM_RE.search(text))
+
+
+def _branch_safety_fields_present(text: str) -> list[str]:
+    missing: list[str] = []
+    if not _DELETE_CAPABILITY_RE.search(text):
+        missing.append("delete-branch capability resolved (gitea.branch.delete)")
+    if not _DELETE_TASK_RE.search(text):
+        missing.append("delete-branch task named (delete_branch or cleanup)")
+    if not _MERGE_RESULT_RE.search(text):
+        missing.append("merge result: merged")
+    if not _MERGE_COMMIT_SHA_RE.search(text):
+        missing.append("merge commit SHA")
+    if not _PR_HEAD_BRANCH_RE.search(text):
+        missing.append("merged PR head branch / deleted branch name")
+    if not _BRANCH_NOT_PROTECTED_RE.search(text):
+        missing.append("branch not protected proof")
+    if not _OPEN_PR_INVENTORY_RE.search(text):
+        missing.append("open PR inventory proof (no other PR references branch)")
+    if not _ACTIVE_CLAIM_LEASE_RE.search(text):
+        missing.append("no active heartbeat/claim/lease proof")
+    return missing
+
+
+def _worktree_cleanup_fields_present(text: str) -> list[str]:
+    missing: list[str] = []
+    match = _SESSION_OWNED_WORKTREE_RE.search(text)
+    path = match.group(1).strip() if match else ""
+    if not path:
+        missing.append("session-owned worktree path")
+    elif not _BRANCHES_PATH_RE.search(path.replace("\\", "/")):
+        missing.append("worktree path under branches/")
+    if not _CLEAN_TRACKED_RE.search(text):
+        missing.append("pre-removal tracked state: clean")
+    if not _CLEAN_UNTRACKED_RE.search(text):
+        missing.append("pre-removal untracked state: clean")
+    if not _WORKTREE_LIST_AFTER_RE.search(text):
+        missing.append("git worktree list after removal")
+    return missing
+
+
+def assess_post_merge_cleanup_proof(
+    report_text: str,
+    *,
+    cleanup_session: dict | None = None,
+) -> dict[str, Any]:
+    """Validate post-merge cleanup claims carry safety-gate proof (#402)."""
+    text = report_text or ""
+    session = dict(cleanup_session or {})
+    reasons: list[str] = []
+
+    if _CLEANUP_SKIPPED_RE.search(text) or session.get("outcome") == CLEANUP_SKIPPED:
+        blocker = (session.get("blocker") or "").strip()
+        if not blocker:
+            match = _CLEANUP_BLOCKER_RE.search(text)
+            blocker = match.group(1).strip() if match else ""
+        if blocker.upper() == CLEANUP_SKIPPED:
+            blocker = ""
+        if not blocker:
+            reasons.append(
+                "CLEANUP_SKIPPED requires exact cleanup blocker reason (#402)"
+            )
+        return {
+            "block": bool(reasons),
+            "proven": not reasons,
+            "outcome": CLEANUP_SKIPPED,
+            "remote_delete_claimed": False,
+            "worktree_remove_claimed": False,
+            "reasons": reasons,
+            "safe_next_action": (
+                "report CLEANUP_SKIPPED with exact blocker; do not claim performed cleanup"
+                if reasons
+                else "proceed"
+            ),
+        }
+
+    if not _CLEANUP_SECTION_HINT.search(text) and not session.get("cleanup_claimed"):
+        return {
+            "block": False,
+            "proven": True,
+            "outcome": None,
+            "remote_delete_claimed": False,
+            "worktree_remove_claimed": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+
+    remote_delete = bool(
+        session.get("remote_delete_claimed") or _claims_remote_delete(text)
+    )
+    worktree_remove = bool(
+        session.get("worktree_remove_claimed") or _claims_worktree_remove(text)
+    )
+
+    if _WRONG_BRANCH_RE.search(text):
+        reasons.append(
+            "cleanup report claims deleted branch that is not the merged PR head branch"
+        )
+
+    if remote_delete:
+        reasons.extend(
+            f"remote branch deletion missing {field}"
+            for field in _branch_safety_fields_present(text)
+        )
+
+    if worktree_remove:
+        reasons.extend(
+            f"worktree removal missing {field}"
+            for field in _worktree_cleanup_fields_present(text)
+        )
+
+    if (remote_delete or worktree_remove) and not (remote_delete or worktree_remove):
+        pass
+
+    if not remote_delete and not worktree_remove:
+        cleanup_mutations = re.search(
+            r"cleanup mutations\s*:\s*(?!none\b)(.+)",
+            text,
+            re.IGNORECASE,
+        )
+        if cleanup_mutations:
+            reasons.append(
+                "cleanup mutations reported without post-merge cleanup proof checklist"
+            )
+
+    outcome = CLEANUP_PERFORMED if (remote_delete or worktree_remove) and not reasons else None
+    if remote_delete or worktree_remove:
+        outcome = CLEANUP_PERFORMED if not reasons else "CLEANUP_CLAIMED_UNPROVEN"
+
+    block = bool(reasons)
+    return {
+        "block": block,
+        "proven": not block,
+        "outcome": outcome,
+        "remote_delete_claimed": remote_delete,
+        "worktree_remove_claimed": worktree_remove,
+        "reasons": reasons,
+        "safe_next_action": (
+            "report CLEANUP_SKIPPED with exact blocker or include the full cleanup "
+            "checklist before claiming remote delete or worktree removal"
+            if reasons
+            else "proceed"
+        ),
+    }

review_final_report_schema.py

@@ -301,7 +301,6 @@ def assess_review_final_report_schema(
     action_log: list[dict] | None = None,
     mutations_observed: bool = False,
     local_edits: bool = False,
-    validation_session: dict | None = None,
 ) -> dict[str, Any]:
     """Validate reviewer final report text before session completion (#391)."""
     base = assess_final_report_validator(
@@ -313,7 +312,6 @@ def assess_review_final_report_schema(
         action_log=action_log,
         mutations_observed=mutations_observed,
         local_edits=local_edits,
-        validation_session=validation_session,
     )
     extra: list[dict[str, str]] = []
     for rule in _SCHEMA_RULES:

review_proofs.py

@@ -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"):
-            downgraded = True
-            reasons.extend(result.get("reasons") or [])
-        elif verdict in ("missing", "incomplete"):
+        if 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 {
@@ -5521,15 +5503,6 @@ def assess_validation_integrity_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
-def assess_validation_failure_history_report(report_text, **kwargs):
-    """#396: final reports must account for transient validation failures."""
-    from reviewer_validation_failure_history import (
-        assess_validation_failure_history_report as _assess,
-    )
-
-    return _assess(report_text, **kwargs)
-
-
 def assess_already_landed_classification_report(report_text, **kwargs):
     """#295: already-landed PRs are reconciliation-only, not review eligible."""
     from reviewer_already_landed_classification import (

reviewer_validation_failure_history.py

@@ -1,198 +0,0 @@
-"""Transient validation failure history verifier (#396).
-
-Reviewer sessions may observe validation failures that later pass on rerun.
-Final reports must document every failure observed during the session, not
-only the last passing result.
-"""
-
-from __future__ import annotations
-
-import re
-from typing import Any
-
-_SECTION_RE = re.compile(
-    r"validation failure history",
-    re.IGNORECASE,
-)
-_FAILURE_ENTRY_RE = re.compile(
-    r"(?:failure\s*(?:#|entry)?\s*\d+|validation failure)\s*:",
-    re.IGNORECASE,
-)
-_COMMAND_RE = re.compile(
-    r"(?:command|validation command)\s*:\s*(.+)",
-    re.IGNORECASE,
-)
-_FAILING_TEST_RE = re.compile(
-    r"(?:failing test|failure|error)\s*:\s*(.+)",
-    re.IGNORECASE,
-)
-_CAUSE_RE = re.compile(
-    r"(?:suspected cause|cause)\s*:\s*(.+)",
-    re.IGNORECASE,
-)
-_REPRODUCED_RE = re.compile(
-    r"(?:reproduced|reproduces)\s*:\s*(yes|no|unknown)",
-    re.IGNORECASE,
-)
-_BASELINE_RE = re.compile(
-    r"(?:on baseline master|baseline master|exists on baseline)\s*:\s*(yes|no|unknown|not checked)",
-    re.IGNORECASE,
-)
-_PR_CAUSED_RE = re.compile(
-    r"(?:pr[- ]caused|pr caused)\s*:\s*(yes|no|unknown|not proven)",
-    re.IGNORECASE,
-)
-_TRANSIENT_STATUS_RE = re.compile(
-    r"(?:passed after transient failure investigation|transient failure investigation)",
-    re.IGNORECASE,
-)
-_PLAIN_PASS_RE = re.compile(
-    r"validation\s*:\s*(?:pass|passed|strong|ok|green)\b",
-    re.IGNORECASE,
-)
-_ENV_CLEANUP_RE = re.compile(
-    r"(?:environmental cleanup|state cleaned|what changed between runs)\s*:",
-    re.IGNORECASE,
-)
-_UNKNOWN_CAUSE_RE = re.compile(
-    r"(?:suspected cause|cause)\s*:\s*(?:unknown|unexplained|not determined)",
-    re.IGNORECASE,
-)
-
-
-def _failure_documented_in_text(text: str, failure: dict[str, Any]) -> bool:
-    """Return True when *failure* appears documented in free-form report text."""
-    command = (failure.get("command") or "").strip()
-    failing = (failure.get("failing_test") or failure.get("error") or "").strip()
-    if command and command not in text:
-        return False
-    if failing and failing not in text:
-        return False
-    return bool(command or failing)
-
-
-def _section_has_structured_fields(text: str) -> bool:
-    if not _SECTION_RE.search(text):
-        return False
-    section_start = _SECTION_RE.search(text).start()
-    section = text[section_start:]
-    has_command = bool(_COMMAND_RE.search(section))
-    has_failure = bool(_FAILING_TEST_RE.search(section))
-    return has_command and has_failure
-
-
-def assess_validation_failure_history_report(
-    report_text: str,
-    *,
-    validation_session: dict | None = None,
-) -> dict[str, Any]:
-    """Require final reports to account for transient validation failures (#396)."""
-    text = report_text or ""
-    session = dict(validation_session or {})
-    reasons: list[str] = []
-    violations: list[str] = []
-
-    observed = list(session.get("observed_failures") or [])
-    final_status = (session.get("final_validation_status") or "").strip().lower()
-    cause_unknown = any(
-        (f.get("suspected_cause") or "").strip().lower() in {
-            "unknown", "unexplained", "not determined", ""
-        }
-        and not (f.get("pr_caused") or "").strip().lower() in {"yes", "no"}
-        for f in observed
-    ) or bool(session.get("cause_unknown"))
-
-    if not observed:
-        return {
-            "proven": True,
-            "block": False,
-            "reasons": [],
-            "violations": [],
-            "observed_failure_count": 0,
-            "safe_next_action": "proceed",
-        }
-
-    documented = _section_has_structured_fields(text)
-    if not documented:
-        for failure in observed:
-            if _failure_documented_in_text(text, failure):
-                documented = True
-                break
-
-    if not documented:
-        violations.append(
-            "session observed validation failure(s) but final report omits "
-            "Validation failure history"
-        )
-        reasons.append(
-            "every validation failure observed during the session must appear "
-            "in a Validation failure history section"
-        )
-
-    if documented and not _SECTION_RE.search(text):
-        reasons.append(
-            "failure details must appear under an explicit "
-            "'Validation failure history' heading"
-        )
-
-    for idx, failure in enumerate(observed, start=1):
-        prefix = f"failure #{idx}"
-        if not _failure_documented_in_text(text, failure) and documented:
-            reasons.append(
-                f"{prefix}: command and failing test/error not documented in report"
-            )
-        command = (failure.get("command") or "").strip()
-        failing = (failure.get("failing_test") or failure.get("error") or "").strip()
-        if not command:
-            reasons.append(f"{prefix}: missing command in session failure record")
-        if not failing:
-            reasons.append(
-                f"{prefix}: missing failing test or error in session failure record"
-            )
-
-    plain_pass = bool(_PLAIN_PASS_RE.search(text))
-    transient_wording = bool(_TRANSIENT_STATUS_RE.search(text))
-    final_passed = final_status in {
-        "passed",
-        "pass",
-        "passed_after_transient_failure_investigation",
-    }
-
-    if (final_passed or plain_pass) and observed:
-        if cause_unknown and not transient_wording:
-            violations.append(
-                "unknown transient failure cause cannot be erased as plain 'passed'"
-            )
-            reasons.append(
-                "when cause is unknown, use status "
-                "'passed after transient failure investigation'"
-            )
-        elif not transient_wording and final_status != "passed_after_transient_failure_investigation":
-            if not _ENV_CLEANUP_RE.search(text):
-                reasons.append(
-                    "later passing rerun must document what changed between runs "
-                    "or environmental cleanup performed"
-                )
-
-    if session.get("environmental_contamination") and not _ENV_CLEANUP_RE.search(text):
-        reasons.append(
-            "environmental /tmp state contamination must document cleanup or "
-            "what changed before the passing rerun"
-        )
-
-    proven = not reasons and not violations
-    return {
-        "proven": proven,
-        "block": bool(violations) or not proven,
-        "reasons": reasons,
-        "violations": violations,
-        "observed_failure_count": len(observed),
-        "documented": documented,
-        "safe_next_action": (
-            "add Validation failure history with command, failing test, cause, "
-            "reproduction, baseline comparison, and PR-caused evidence; use "
-            "'passed after transient failure investigation' when appropriate"
-            if not proven
-            else "proceed"
-        ),
-    }

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -538,36 +538,6 @@ Official validation integrity status must be one of:
 
 Do not invent a softer status.
 
-## 21A. Validation failure history rule
-
-Every validation failure observed during the review session must appear in the
-final report, even if a later rerun passes.
-
-Before claiming `Validation: passed`, list every failed validation command from
-the session under `Validation failure history`.
-
-For each failure, report:
-
-* command
-* failing test or error
-* suspected cause
-* whether it reproduced
-* whether it exists on baseline master
-* evidence for whether it is or is not PR-caused
-
-If a later rerun passes, also report:
-
-* what changed between runs
-* whether environmental state was cleaned (for example `/tmp` lock files)
-* why the pass is trustworthy
-
-If the cause is unknown, do not erase the earlier failure with plain
-`Validation: passed`. Use a status such as
-`passed after transient failure investigation`.
-
-`gitea_validate_review_final_report` rejects reports that omit known earlier
-validation failures when `validation_session.observed_failures` is supplied.
-
 ## 22. Baseline validation rule
 
 Do not run tests in the main checkout.
@@ -797,6 +767,40 @@ Do not update the main checkout if merge failed, was blocked, or produced reconc
 
 If any local artifact is created after final cleanup, run and report a new final status check.
 
+## 28A. Post-merge cleanup proof checklist (#402)
+
+Successful tool execution is not proof that cleanup was authorized. Before claiming remote branch deletion or local worktree removal, the final report must carry the full safety checklist below. If any gate is missing, report `CLEANUP_SKIPPED` with the exact blocker — never perform cleanup and never claim it was performed.
+
+### Remote branch deletion checklist
+
+When `gitea_delete_branch` (or equivalent) deletes the merged PR head branch, report:
+
+* Delete-branch capability resolved: name the task (`delete_branch` / `cleanup_branch` / `reconcile_merged_cleanups`) and permission (`gitea.branch.delete`) with resolver proof before the delete call
+* Merge result: merged
+* Merge commit SHA: full 40-character SHA
+* Merged PR head branch / deleted branch: exact branch name (must match)
+* Branch protection: none / branch is not protected
+* Open PR inventory proof: no other open PR references the branch
+* Active heartbeat/claim/lease: none
+
+### Local worktree removal checklist
+
+When removing session-owned review/simulation worktrees under `branches/`, report:
+
+* Session-owned worktree path: exact path under `branches/`
+* Pre-removal tracked state: clean
+* Pre-removal untracked state: clean
+* Git worktree list after removal: command output or equivalent proof
+
+### Skipped cleanup
+
+If any gate fails, report:
+
+* Cleanup outcome: `CLEANUP_SKIPPED`
+* Cleanup blocker: exact missing gate (for example `gitea.branch.delete capability not resolved`)
+
+Skipped cleanup with an exact blocker passes validation. Performed-cleanup claims without the checklist fail validation.
+
 ## 29. Recovery handoff rules
 
 If blocked, produce a recovery handoff with:
@@ -1126,7 +1130,6 @@ Controller Handoff:
 * Baseline worktree path:
 * Files reviewed:
 * Validation:
-* Validation failure history:
 * Official validation integrity status:
 * Terminal review mutation:
 * Review decision:

skills/llm-project-workflow/workflows/work-issue.md

@@ -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:

tests/test_author_duplicate_work_gate.py

@@ -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()

tests/test_delete_branch_capability.py

@@ -1,217 +0,0 @@
-"""Tests for gitea_delete_branch capability gate (Issue #408).
-
-``gitea_delete_branch`` requires the exact ``gitea.branch.delete`` operation:
-without it the delete fails closed (no preflight, no auth lookup, no API call,
-structured permission report). With it, deletion proceeds through existing
-preflight and audit unchanged.
-"""
-import json
-import os
-import sys
-import tempfile
-import unittest
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-import mcp_server
-from mcp_server import gitea_delete_branch
-
-FAKE_AUTH = "token fake"
-
-AUTHOR_NO_DELETE = {
-    "profile_name": "prgs-author",
-    "allowed_operations": [
-        "gitea.read", "gitea.pr.create", "gitea.pr.comment",
-        "gitea.branch.push", "gitea.issue.comment",
-    ],
-    "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
-    "audit_label": "prgs-author",
-}
-
-AUTHOR_WITH_DELETE = {
-    "profile_name": "prgs-author-deleter",
-    "allowed_operations": AUTHOR_NO_DELETE["allowed_operations"] + [
-        "gitea.branch.delete",
-    ],
-    "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
-    "audit_label": "prgs-author-deleter",
-}
-
-CONFIG = {
-    "version": 2,
-    "contexts": {
-        "ctx": {
-            "enabled": True,
-            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
-        }
-    },
-    "profiles": {
-        "author-no-delete": {
-            "enabled": True,
-            "context": "ctx",
-            "role": "author",
-            "username": "author-user",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
-            "allowed_operations": AUTHOR_NO_DELETE["allowed_operations"],
-            "forbidden_operations": [],
-            "execution_profile": "author-no-delete",
-        },
-        "author-with-delete": {
-            "enabled": True,
-            "context": "ctx",
-            "role": "author",
-            "username": "deleter-user",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
-            "allowed_operations": AUTHOR_WITH_DELETE["allowed_operations"],
-            "forbidden_operations": [],
-            "execution_profile": "author-with-delete",
-        },
-        "reviewer-profile": {
-            "enabled": True,
-            "context": "ctx",
-            "role": "reviewer",
-            "username": "reviewer-user",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
-            "allowed_operations": [
-                "gitea.read", "gitea.pr.review", "gitea.pr.merge",
-            ],
-            "forbidden_operations": [
-                "gitea.branch.delete", "gitea.branch.push", "gitea.pr.create",
-            ],
-            "execution_profile": "reviewer-profile",
-        },
-    },
-    "rules": {"allow_runtime_switching": False},
-}
-
-
-class TestDeleteBranchToolGate(unittest.TestCase):
-
-    def setUp(self):
-        self._preflight_snapshot = (
-            mcp_server._preflight_whoami_called,
-            mcp_server._preflight_capability_called,
-        )
-        mcp_server._preflight_whoami_called = False
-        mcp_server._preflight_capability_called = False
-        self._remotes = patch.dict(mcp_server.REMOTES, {
-            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
-                     "repo": "Example-Repo"},
-        })
-        self._remotes.start()
-        mcp_server._IDENTITY_CACHE.clear()
-        patch("gitea_config.load_config", return_value={}).start()
-        patch(
-            "gitea_config.is_runtime_switching_enabled", return_value=False
-        ).start()
-        patch("gitea_audit.audit_enabled", return_value=False).start()
-        self.mock_api = patch("mcp_server.api_request").start()
-        self.mock_auth = patch(
-            "mcp_server.get_auth_header", return_value=FAKE_AUTH
-        ).start()
-
-    def tearDown(self):
-        patch.stopall()
-        mcp_server._IDENTITY_CACHE.clear()
-        mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (
-            self._preflight_snapshot
-        )
-
-    def _set_profile(self, profile):
-        patch("mcp_server.get_profile", return_value=profile).start()
-
-    def test_blocked_without_delete_capability(self):
-        self._set_profile(AUTHOR_NO_DELETE)
-        res = gitea_delete_branch(branch="feat/branch", remote="prgs")
-        self.assertFalse(res["success"])
-        self.assertFalse(res["performed"])
-        self.assertEqual(res["required_permission"], "gitea.branch.delete")
-        self.assertTrue(res["reasons"])
-        self.assertEqual(
-            res["permission_report"]["missing_permission"],
-            "gitea.branch.delete",
-        )
-        self.mock_api.assert_not_called()
-        self.mock_auth.assert_not_called()
-
-    def test_allowed_delete_proceeds(self):
-        self._set_profile(AUTHOR_WITH_DELETE)
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check("capability", resolved_role="author")
-        res = gitea_delete_branch(branch="feat/branch", remote="prgs")
-        self.assertTrue(res["success"])
-        self.assertIn("deleted", res["message"])
-        delete_calls = [
-            c for c in self.mock_api.call_args_list if c.args[0] == "DELETE"
-        ]
-        self.assertTrue(delete_calls)
-
-    def test_allowed_delete_audited_with_capability_proof(self):
-        self._set_profile(AUTHOR_WITH_DELETE)
-        patch("gitea_audit.audit_enabled", return_value=True).start()
-        mock_write = patch("gitea_audit.write_event").start()
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check("capability", resolved_role="author")
-        self.mock_api.return_value = {}
-        gitea_delete_branch(branch="feat/branch", remote="prgs")
-        mock_write.assert_called()
-        event = mock_write.call_args[0][0]
-        self.assertEqual(event["action"], "delete_branch")
-        self.assertEqual(
-            event["request_metadata"]["required_permission"],
-            "gitea.branch.delete",
-        )
-
-
-class TestDeleteBranchResolverParity(unittest.TestCase):
-    def setUp(self):
-        self._remotes = patch.dict(mcp_server.REMOTES, {
-            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
-                     "repo": "Example-Repo"},
-        })
-        self._remotes.start()
-        mcp_server._IDENTITY_CACHE.clear()
-        self._dir = tempfile.TemporaryDirectory()
-        self.config_path = os.path.join(self._dir.name, "profiles.json")
-        with open(self.config_path, "w", encoding="utf-8") as fh:
-            fh.write(json.dumps(CONFIG))
-        patch(
-            "gitea_config.is_runtime_switching_enabled", return_value=False
-        ).start()
-        patch("gitea_audit.audit_enabled", return_value=False).start()
-        self.mock_api = patch("mcp_server.api_request").start()
-        patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
-
-    def tearDown(self):
-        patch.stopall()
-        mcp_server._IDENTITY_CACHE.clear()
-        self._dir.cleanup()
-
-    def _env(self, profile: str) -> dict:
-        return {
-            "GITEA_MCP_CONFIG": self.config_path,
-            "GITEA_MCP_PROFILE": profile,
-            "GITEA_TOKEN_AUTHOR": "author-pass",
-            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
-        }
-
-    def test_reviewer_resolver_denial_blocks_raw_tool(self):
-        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
-            resolve = mcp_server.gitea_resolve_task_capability(
-                task="delete_branch", remote="prgs")
-            self.assertFalse(resolve["allowed_in_current_session"])
-            res = gitea_delete_branch(branch="feat/branch", remote="prgs")
-        self.assertFalse(res["success"])
-        self.assertEqual(
-            res["permission_report"]["missing_permission"],
-            resolve["required_operation_permission"],
-        )
-        delete_calls = [
-            c for c in self.mock_api.call_args_list if c.args[0] == "DELETE"
-        ]
-        self.assertFalse(delete_calls)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_llm_workflow_split.py

@@ -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
@@ -213,12 +207,6 @@ def test_validation_integrity_verifier_exported():
     assert callable(assess_validation_integrity_report)
 
 
-def test_validation_failure_history_verifier_exported():
-    from review_proofs import assess_validation_failure_history_report
-
-    assert callable(assess_validation_failure_history_report)
-
-
 def test_prior_blocker_skip_verifier_exported():
     from review_proofs import assess_prior_blocker_skip_proof
 

tests/test_mcp_server.py

@@ -1005,19 +1005,9 @@ class TestReviewPR(unittest.TestCase):
 # ---------------------------------------------------------------------------
 class TestDeleteBranch(unittest.TestCase):
 
-    DELETE_PROFILE = {
-        "profile_name": "test-deleter",
-        "allowed_operations": ["gitea.read", "gitea.branch.delete"],
-        "forbidden_operations": [],
-        "audit_label": "test-deleter",
-    }
-
-    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_delete_branch(self, _auth, mock_api, _profile):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check("capability", resolved_role="author")
+    def test_delete_branch(self, _auth, mock_api):
         mock_api.return_value = {}
         result = gitea_delete_branch(branch="feat/branch")
         self.assertTrue(result["success"])

tests/test_post_merge_cleanup_proof.py

@@ -0,0 +1,169 @@
+"""Tests for post-merge cleanup proof enforcement (#402)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from final_report_validator import assess_final_report_validator  # noqa: E402
+from post_merge_cleanup_proof import (  # noqa: E402
+    CLEANUP_SKIPPED,
+    assess_post_merge_cleanup_proof,
+)
+
+MERGE_SHA = "a" * 40
+HEAD_BRANCH = "feat/issue-274-branches-only-worktrees"
+WORKTREE = "branches/review-pr374-conflicts"
+
+
+def _full_remote_cleanup_report(**overrides):
+    fields = {
+        "Task": "review PR #374",
+        "Merge result": "merged",
+        "Merge commit SHA": MERGE_SHA,
+        "Cleanup status": "remote branch deleted",
+        "Delete-branch capability resolved": "gitea.branch.delete allowed via delete_branch task",
+        "Merged PR head branch": HEAD_BRANCH,
+        "Deleted branch": HEAD_BRANCH,
+        "Branch protection": "none",
+        "Open PR inventory proof": "no other open PR references branch (inventory complete)",
+        "Active heartbeat/claim/lease": "none",
+        "Cleanup mutations": "gitea_delete_branch on remote head branch",
+    }
+    fields.update(overrides)
+    lines = ["## Controller Handoff", ""]
+    lines.extend(f"- {key}: {value}" for key, value in fields.items())
+    return "\n".join(lines)
+
+
+def _full_worktree_cleanup_report(**overrides):
+    fields = {
+        "Task": "review PR #374",
+        "Merge result": "merged",
+        "Cleanup status": "local worktree removed",
+        "Removed worktree path": WORKTREE,
+        "Pre-removal tracked state": "clean",
+        "Pre-removal untracked state": "clean",
+        "Git worktree list after removal": "only main checkout listed",
+        "Cleanup mutations": "git worktree remove on session-owned review worktree",
+    }
+    fields.update(overrides)
+    lines = ["## Controller Handoff", ""]
+    lines.extend(f"- {key}: {value}" for key, value in fields.items())
+    return "\n".join(lines)
+
+
+class TestPostMergeCleanupProof(unittest.TestCase):
+    def test_no_cleanup_claim_passes(self):
+        report = "## Controller Handoff\n- Task: review PR #1\n- Merge result: merged"
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertFalse(result["block"])
+
+    def test_missing_capability_proof_blocked(self):
+        report = _full_remote_cleanup_report(
+            **{"Delete-branch capability resolved": "delete succeeded"}
+        )
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("capability" in r.lower() for r in result["reasons"]))
+
+    def test_unmerged_pr_blocked(self):
+        report = _full_remote_cleanup_report(**{"Merge result": "not merged"})
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("merge result" in r.lower() for r in result["reasons"]))
+
+    def test_wrong_branch_blocked(self):
+        report = _full_remote_cleanup_report(
+            **{"Deleted branch": "feat/other-branch"}
+        )
+        report += "\nDeleted branch does not match merged PR head branch"
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+
+    def test_protected_branch_blocked(self):
+        report = _full_remote_cleanup_report(**{"Branch protection": "enabled"})
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+
+    def test_open_pr_reference_blocked(self):
+        report = _full_remote_cleanup_report(
+            **{"Open PR inventory proof": "PR #999 still references branch"}
+        )
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+
+    def test_active_claim_blocked(self):
+        report = _full_remote_cleanup_report(
+            **{"Active heartbeat/claim/lease": "status:in-progress on linked issue"}
+        )
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+
+    def test_dirty_worktree_blocked(self):
+        report = _full_worktree_cleanup_report(
+            **{"Pre-removal tracked state": "dirty"}
+        )
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("tracked" in r.lower() for r in result["reasons"]))
+
+    def test_foreign_worktree_blocked(self):
+        report = _full_worktree_cleanup_report(
+            **{"Removed worktree path": "/tmp/foreign-worktree"}
+        )
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("branches/" in r.lower() for r in result["reasons"]))
+
+    def test_cleanup_skipped_with_blocker_passes(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup outcome: CLEANUP_SKIPPED",
+            "- Cleanup blocker: gitea.branch.delete capability not resolved in active profile",
+        ])
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertFalse(result["block"])
+        self.assertEqual(result["outcome"], CLEANUP_SKIPPED)
+
+    def test_cleanup_skipped_without_blocker_blocked(self):
+        report = "## Controller Handoff\n- Cleanup outcome: CLEANUP_SKIPPED"
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+
+    def test_full_remote_cleanup_passes(self):
+        result = assess_post_merge_cleanup_proof(_full_remote_cleanup_report())
+        self.assertFalse(result["block"])
+        self.assertTrue(result["remote_delete_claimed"])
+
+    def test_full_worktree_cleanup_passes(self):
+        result = assess_post_merge_cleanup_proof(_full_worktree_cleanup_report())
+        self.assertFalse(result["block"])
+        self.assertTrue(result["worktree_remove_claimed"])
+
+    def test_validator_integration_blocks_unproven_delete(self):
+        report = (
+            "## Controller Handoff\n"
+            "- Cleanup mutations: gitea_delete_branch deleted remote branch\n"
+        )
+        result = assess_final_report_validator(report, task_kind="review_pr")
+        self.assertTrue(result["blocked"])
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.post_merge_cleanup_proof", rule_ids)
+
+    def test_validator_integration_allows_skipped(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup outcome: CLEANUP_SKIPPED",
+            "- Cleanup blocker: branch still referenced by open PR #414",
+        ])
+        result = assess_final_report_validator(report, task_kind="review_pr")
+        cleanup_findings = [
+            f for f in result["findings"]
+            if f["rule_id"] == "reviewer.post_merge_cleanup_proof"
+        ]
+        self.assertEqual(cleanup_findings, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_validation_failure_history.py

@@ -1,147 +0,0 @@
-"""Tests for transient validation failure history verifier (#396)."""
-import sys
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from final_report_validator import assess_final_report_validator  # noqa: E402
-from reviewer_validation_failure_history import (  # noqa: E402
-    assess_validation_failure_history_report,
-)
-
-
-def _full_history_report() -> str:
-    return "\n".join([
-        "Validation failure history",
-        "Failure #1:",
-        "Command: venv/bin/python -m pytest tests/test_worktrees.py -q",
-        "Failing test: tests/test_worktrees.py::TestWorktreeStart::test_rejects_untraceable_branches",
-        "Suspected cause: stale /tmp/gitea_issue_lock.json from prior session",
-        "Reproduced: no",
-        "On baseline master: no",
-        "PR-caused: no",
-        "What changed between runs: removed /tmp/gitea_issue_lock.json",
-        "Environmental cleanup: lock file removed before rerun",
-        "Validation: passed after transient failure investigation",
-        "Final validation command: venv/bin/python -m pytest tests/ -q",
-        "Final result: 1497 passed, 6 skipped",
-    ])
-
-
-class TestValidationFailureHistory(unittest.TestCase):
-    def test_no_failures_observed_passes(self):
-        result = assess_validation_failure_history_report(
-            "Validation: passed",
-            validation_session={"observed_failures": []},
-        )
-        self.assertTrue(result["proven"])
-
-    def test_omitted_failure_blocks(self):
-        result = assess_validation_failure_history_report(
-            "Validation: passed\n1497 passed, 6 skipped",
-            validation_session={
-                "observed_failures": [{
-                    "command": "pytest tests/test_worktrees.py -q",
-                    "failing_test": (
-                        "tests/test_worktrees.py::TestWorktreeStart::"
-                        "test_rejects_untraceable_branches"
-                    ),
-                }],
-                "final_validation_status": "passed",
-            },
-        )
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["block"])
-
-    def test_full_history_with_transient_investigation_passes(self):
-        result = assess_validation_failure_history_report(
-            _full_history_report(),
-            validation_session={
-                "observed_failures": [{
-                    "command": (
-                        "venv/bin/python -m pytest tests/test_worktrees.py -q"
-                    ),
-                    "failing_test": (
-                        "tests/test_worktrees.py::TestWorktreeStart::"
-                        "test_rejects_untraceable_branches"
-                    ),
-                    "suspected_cause": "stale /tmp/gitea_issue_lock.json",
-                    "reproduced": "no",
-                    "on_baseline_master": "no",
-                    "pr_caused": "no",
-                }],
-                "final_validation_status": "passed_after_transient_failure_investigation",
-                "environmental_contamination": True,
-            },
-        )
-        self.assertTrue(result["proven"], result["reasons"])
-
-    def test_baseline_equivalent_failure_documented(self):
-        report = "\n".join([
-            "Validation failure history",
-            "Command: pytest tests/test_example.py -q",
-            "Failing test: tests/test_example.py::test_flaky",
-            "Suspected cause: pre-existing master failure",
-            "On baseline master: yes",
-            "PR-caused: no",
-            "What changed between runs: none; failure matches baseline",
-            "Validation: passed after transient failure investigation",
-        ])
-        result = assess_validation_failure_history_report(
-            report,
-            validation_session={
-                "observed_failures": [{
-                    "command": "pytest tests/test_example.py -q",
-                    "failing_test": "tests/test_example.py::test_flaky",
-                    "on_baseline_master": "yes",
-                    "pr_caused": "no",
-                }],
-                "final_validation_status": "passed_after_transient_failure_investigation",
-            },
-        )
-        self.assertTrue(result["proven"], result["reasons"])
-
-    def test_unknown_cause_plain_pass_blocks(self):
-        result = assess_validation_failure_history_report(
-            "Validation: passed",
-            validation_session={
-                "observed_failures": [{
-                    "command": "pytest tests/ -q",
-                    "failing_test": "tests/test_foo.py::test_bar",
-                    "suspected_cause": "unknown",
-                }],
-                "final_validation_status": "passed",
-                "cause_unknown": True,
-            },
-        )
-        self.assertFalse(result["proven"])
-        self.assertTrue(any("transient" in r.lower() for r in result["reasons"]))
-
-    def test_final_report_validator_rejects_omitted_failure(self):
-        result = assess_final_report_validator(
-            "Validation: passed",
-            "review_pr",
-            validation_session={
-                "observed_failures": [{
-                    "command": "pytest tests/ -q",
-                    "failing_test": "tests/test_foo.py::test_bar",
-                }],
-            },
-        )
-        self.assertTrue(result["blocked"] or result["downgraded"])
-        self.assertTrue(
-            any(
-                f.get("rule_id") == "reviewer.validation_failure_history"
-                for f in result.get("findings") or []
-            )
-        )
-
-    def test_exported_from_review_proofs(self):
-        from review_proofs import assess_validation_failure_history_report as exported
-
-        self.assertTrue(callable(exported))
-
-
-if __name__ == "__main__":
-    unittest.main()