fix: hide empty-state copy on queue credential fail-closed (Closes #458 )

When fetch_error is set, render only the unavailable banner and an explicit not-fetched inventory note. Suppress queue tables, pagination proof, and "No open items." copy that implied a successful empty fetch. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 16:07:20 -04:00
21 changed files with 99 additions and 2113 deletions
@@ -1,344 +0,0 @@
 """Audit vs cleanup phase gates for reconciliation workflows (#419).
 Audit/reconciliation tasks are read-only unless a separate cleanup phase is
 explicitly authorized with exact capability proof, safety proof, and
 before/after snapshots. Cleanup mutations must be classified in final reports;
 audit reports must not claim ``no mutations`` when cleanup occurred.
 """
 from __future__ import annotations
 import re
 from typing import Any
 RECONCILE_WORKFLOW_PATH = "workflows/reconcile-landed-pr.md"
 PHASE_AUDIT = "audit"
 PHASE_CLEANUP = "cleanup"
 # Tasks that enter audit phase on capability resolution (read-only default).
 AUDIT_PHASE_TASKS = frozenset({
    "reconcile-landed-pr",
    "reconcile_landed_pr",
    "reconcile_issue_claims",
    "reconcile_merged_cleanups",
 })
 # Mutation tasks forbidden during audit phase (fail closed).
 AUDIT_FORBIDDEN_TASKS = frozenset({
    "delete_branch",
    "create_branch",
    "push_branch",
    "create_pr",
    "commit_files",
    "gitea_commit_files",
    "mark_issue",
    "lock_issue",
    "claim_issue",
    "close_pr",
    "close_issue",
    "create_issue",
    "merge_pr",
    "review_pr",
    "submit_pr_review",
    "comment_pr",
    "comment_issue",
    "set_issue_labels",
 })
 # Shell/git commands audit phase must not run.
 AUDIT_FORBIDDEN_COMMAND_RE = re.compile(
    r"(?:^|\s)(?:git\s+(?:push|branch\s+-D|worktree\s+remove)|"
    r"gitea_delete_branch|delete_remote_branch)",
    re.IGNORECASE,
 )
 _NO_MUTATIONS_RE = re.compile(
    r"(?:no\s+mutations|mutations\s*:\s*none|no\s+unsafe\s+mutation)",
    re.IGNORECASE,
 )
 _CLEANUP_OCCURRED_RE = re.compile(
    r"(?:delete_remote_branch|remove_local_worktree|git\s+branch\s+-D|"
    r"git\s+worktree\s+remove|remote branch.*deleted|worktree.*removed|"
    r"cleanup\s+phase\s*:\s*(?!none\b)\S)",
    re.IGNORECASE,
 )
 _EXTERNAL_STATE_RE = re.compile(
    r"^\s*[-*]?\s*external[- ]state mutations\s*:",
    re.IGNORECASE | re.MULTILINE,
 )
 _GIT_REF_RE = re.compile(
    r"^\s*[-*]?\s*git ref mutations\s*:",
    re.IGNORECASE | re.MULTILINE,
 )
 _CLEANUP_MUTATIONS_RE = re.compile(
    r"^\s*[-*]?\s*cleanup mutations\s*:",
    re.IGNORECASE | re.MULTILINE,
 )
 _CLEANUP_PHASE_AUTH_RE = re.compile(
    r"^\s*[-*]?\s*cleanup phase (?:authorized|authorization)\s*:\s*true",
    re.IGNORECASE | re.MULTILINE,
 )
 _DELETE_CAPABILITY_RE = re.compile(
    r"^\s*[-*]?\s*delete.?branch capability(?: proven)?\s*:\s*true",
    re.IGNORECASE | re.MULTILINE,
 )
 _BEFORE_AFTER_RE = re.compile(
    r"^\s*[-*]?\s*before/after (?:state )?snapshot\s*:",
    re.IGNORECASE | re.MULTILINE,
 )
 _SAFETY_PROOF_RE = re.compile(
    r"^\s*[-*]?\s*(?:branch|worktree) safe to remove\s*:\s*true",
    re.IGNORECASE | re.MULTILINE,
 )
 _session: dict[str, Any] | None = None
 def _blank_session() -> dict[str, Any]:
    return {
        "phase": PHASE_AUDIT,
        "entered_from_task": None,
        "cleanup_authorized": False,
        "cleanup_authorization": {},
    }
 def current_phase() -> str | None:
    """Return active reconciliation phase or None when unset."""
    if not _session:
        return None
    return _session.get("phase")
 def active_record() -> dict[str, Any] | None:
    """Return a copy of the session record, if any."""
    return dict(_session) if _session else None
 def clear_phase() -> None:
    """Clear reconciliation phase state."""
    global _session
    _session = None
 def enter_audit_phase(task: str) -> dict[str, Any]:
    """Enter read-only audit phase for a reconciliation task."""
    global _session
    normalized = (task or "").strip().lower()
    _session = _blank_session()
    _session["entered_from_task"] = normalized
    return dict(_session)
 def authorize_cleanup_phase(
    *,
    operator_approved: bool = False,
    workflow_authorized: bool = False,
    delete_capability_proven: bool = False,
    safety_proof: dict[str, Any] | None = None,
    before_after_snapshot: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
    """Authorize cleanup phase after explicit approval and safety proofs."""
    reasons: list[str] = []
    if not (operator_approved or workflow_authorized):
        reasons.append(
            "cleanup phase requires operator approval or explicit workflow "
            "authorization"
        )
    if not delete_capability_proven:
        reasons.append(
            "cleanup phase requires exact delete_branch capability proof "
            "(gitea.branch.delete)"
        )
    safety = dict(safety_proof or {})
    if not safety.get("safe_to_delete_remote") and not safety.get(
        "safe_to_remove_worktree"
    ):
        reasons.append(
            "cleanup phase requires proof that branch/worktree is safe to remove"
        )
    snapshot = dict(before_after_snapshot or {})
    if not snapshot.get("before") or not snapshot.get("after"):
        reasons.append(
            "cleanup phase requires before/after state snapshot"
        )
    if reasons:
        return {
            "authorized": False,
            "phase": current_phase() or PHASE_AUDIT,
            "reasons": reasons,
            "safe_next_action": (
                "remain in audit-only mode or supply operator approval, "
                "delete_branch capability proof, safety proof, and "
                "before/after snapshot before cleanup"
            ),
        }
    global _session
    if _session is None:
        _session = _blank_session()
    _session["phase"] = PHASE_CLEANUP
    _session["cleanup_authorized"] = True
    _session["cleanup_authorization"] = {
        "operator_approved": operator_approved,
        "workflow_authorized": workflow_authorized,
        "delete_capability_proven": delete_capability_proven,
        "safety_proof": safety,
        "before_after_snapshot": snapshot,
    }
    return {
        "authorized": True,
        "phase": PHASE_CLEANUP,
        "reasons": [],
        "cleanup_authorization": dict(_session["cleanup_authorization"]),
        "safe_next_action": "proceed with authorized cleanup mutations only",
    }
 def check_audit_task_enters_phase(task: str) -> bool:
    """Return whether resolving *task* should enter audit phase."""
    return (task or "").strip().lower() in AUDIT_PHASE_TASKS
 def check_audit_mutation_allowed(task: str) -> tuple[bool, list[str]]:
    """Fail closed when a mutation task runs during audit phase."""
    normalized = (task or "").strip().lower()
    phase = current_phase()
    if phase != PHASE_AUDIT:
        return True, []
    if normalized in AUDIT_FORBIDDEN_TASKS:
        return False, [
            f"task '{normalized}' is forbidden in audit-only reconciliation "
            "mode: switch to an explicit cleanup phase with operator approval "
            "and exact delete_branch capability proof before cleanup mutations"
        ]
    return True, []
 def check_cleanup_execution_allowed() -> tuple[bool, list[str]]:
    """Fail closed when cleanup execution is attempted without authorization."""
    phase = current_phase()
    if phase == PHASE_CLEANUP and (_session or {}).get("cleanup_authorized"):
        return True, []
    if phase is None:
        return False, [
            "cleanup execution requires an active reconciliation session; "
            "resolve a reconciliation audit task first"
        ]
    return False, [
        "cleanup execution forbidden in audit-only reconciliation mode; "
        "call gitea_authorize_reconciliation_cleanup_phase with operator "
        "approval, delete_branch capability proof, safety proof, and "
        "before/after snapshot"
    ]
 def classify_cleanup_mutation(action: str) -> str:
    """Map a cleanup action to the required mutation ledger category (#419)."""
    normalized = (action or "").strip().lower()
    if "delete_remote" in normalized or normalized in {
        "delete_branch",
        "gitea_delete_branch",
    }:
        return "external-state"
    if "branch" in normalized and "delete" in normalized:
        return "git-ref"
    if "worktree" in normalized or "remove_local" in normalized:
        return "cleanup"
    return "cleanup"
 def assess_audit_reconciliation_report(report_text: str) -> dict[str, Any]:
    """Validate audit/cleanup reconciliation reports (fail closed)."""
    text = report_text or ""
    reasons: list[str] = []
    cleanup_occurred = bool(_CLEANUP_OCCURRED_RE.search(text))
    claims_no_mutations = bool(_NO_MUTATIONS_RE.search(text))
    if cleanup_occurred and claims_no_mutations:
        reasons.append(
            "report claims no mutations but documents cleanup mutations; "
            "audit-only reports must not perform cleanup and cleanup reports "
            "must not claim no mutations"
        )
    if cleanup_occurred:
        if not _CLEANUP_PHASE_AUTH_RE.search(text):
            reasons.append(
                "cleanup mutations reported without "
                "'Cleanup phase authorized: true'"
            )
        if not _DELETE_CAPABILITY_RE.search(text):
            reasons.append(
                "cleanup mutations reported without delete_branch capability "
                "proof"
            )
        if not _BEFORE_AFTER_RE.search(text):
            reasons.append(
                "cleanup mutations reported without before/after state snapshot"
            )
        if not _SAFETY_PROOF_RE.search(text):
            reasons.append(
                "cleanup mutations reported without branch/worktree safety proof"
            )
        if re.search(r"delete_remote|remote branch.*delet", text, re.I):
            if not _EXTERNAL_STATE_RE.search(text):
                reasons.append(
                    "remote branch deletion must be classified under "
                    "External-state mutations"
                )
        if re.search(r"git\s+branch\s+-D|local branch.*delet", text, re.I):
            if not _GIT_REF_RE.search(text):
                reasons.append(
                    "local branch deletion must be classified under "
                    "Git ref mutations"
                )
        if re.search(r"worktree.*remov|remove_local_worktree", text, re.I):
            if not _CLEANUP_MUTATIONS_RE.search(text):
                reasons.append(
                    "worktree removal must be classified under Cleanup mutations"
                )
    if (
        RECONCILE_WORKFLOW_PATH.replace("workflows/", "") in text
        or "reconcile-landed-pr" in text.lower()
    ):
        if cleanup_occurred and "audit phase" in text.lower():
            if "cleanup phase" not in text.lower():
                reasons.append(
                    "report mixes audit phase with cleanup mutations without "
                    "documenting cleanup phase transition"
                )
    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "cleanup_occurred": cleanup_occurred,
        "claims_no_mutations": claims_no_mutations,
        "safe_next_action": (
            "proceed"
            if proven
            else "fix audit/cleanup report: separate audit from cleanup phase, "
            "classify mutations, and do not claim no mutations after cleanup"
        ),
    }
 def assess_audit_command_allowed(command: str) -> tuple[bool, list[str]]:
    """Block shell commands that perform cleanup during audit phase."""
    phase = current_phase()
    if phase != PHASE_AUDIT:
        return True, []
    cmd = (command or "").strip()
    if AUDIT_FORBIDDEN_COMMAND_RE.search(cmd):
        return False, [
            f"command forbidden in audit-only reconciliation mode: {cmd!r}; "
            "authorize cleanup phase before branch/worktree deletion or push"
        ]
    return True, []
@@ -281,21 +281,6 @@ same-issue/same-operation lease blocks duplicate work. An expired lease still
 blocks takeover until a recovery review records why the prior work is abandoned,
 completed, or unsafe to continue.
 **Issue-lock recovery (#447):** Do not manually seed, restore, or delete
 `/tmp/gitea_issue_lock.json` as a normal recovery path. That file is global
 shared state and manual writes can clobber another session's live lease. Use
 `sanctioned recovery` instead:
 1. `gitea_lock_issue` on a clean `branches/` worktree (normal path).
 2. Own-branch adoption via #442 when the issue's exact branch is already pushed.
 3. Operator override only when explicitly authorized — record
   `External-state mutations` and `operator override proof` in the final report.
 `gitea_create_pr` rejects lock files that lack sanctioned `lock_provenance`
 metadata. Final-report validation blocks handoffs that hide lock read/write/delete
 under `External-state mutations: none` or mix author PR creation with reviewer
 approval in one run. See also #438 (global lock redesign).
 Remote branches matching the issue number are also treated as active work unless
 the recovery review proves the branch is abandoned or superseded. Never delete
 or clean up a branch when it has an active lease, dirty worktree, open PR, or is
@@ -11,7 +11,6 @@ import inspect
 import re
 from typing import Any, Callable
 import issue_lock_provenance
 from review_proofs import (
    HANDOFF_HEADING,
    assess_controller_handoff,
@@ -871,70 +870,6 @@ def _rule_reviewer_mutation_ledger(
    )
 def _rule_shared_issue_lock_external_state(report_text: str) -> list[dict[str, str]]:
    result = issue_lock_provenance.assess_issue_lock_external_state_report(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "shared.issue_lock_external_state",
        result.get("reasons") or [],
        field="External-state mutations",
        severity="block",
        safe_next_action=(
            "disclose gitea_issue_lock.json read/write/delete under "
            "External-state mutations; never claim none after lock seeding"
        ),
    )
 def _rule_shared_manual_lock_pr_override(report_text: str) -> list[dict[str, str]]:
    result = issue_lock_provenance.assess_manual_lock_pr_without_override(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "shared.manual_lock_pr_override",
        result.get("reasons") or [],
        field="External-state mutations",
        severity="block",
        safe_next_action=(
            "use gitea_lock_issue or #442 adoption instead of manual lock seeding; "
            "if operator override was authorized, cite override proof"
        ),
    )
 def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, str]]:
    result = issue_lock_provenance.assess_author_reviewer_same_run_report(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "shared.author_reviewer_same_run",
        result.get("reasons") or [],
        field="Review mutations",
        severity="block",
        safe_next_action=(
            "split author PR creation and reviewer approval across separate "
            "sessions and handoffs"
        ),
    )
 def _rule_audit_reconciliation_boundary(report_text: str) -> list[dict[str, str]]:
    from audit_reconciliation_mode import assess_audit_reconciliation_report
    result = assess_audit_reconciliation_report(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "reconcile.audit_cleanup_boundary",
        result.get("reasons") or [],
        field="Audit/cleanup phase",
        severity="block",
        safe_next_action=result.get("safe_next_action")
        or "separate audit from authorized cleanup and classify mutations",
    )
 def _rule_reviewer_review_mutation(
    report_text: str,
    *,
@@ -954,17 +889,10 @@ def _rule_reviewer_review_mutation(
    )
 _SHARED_ISSUE_LOCK_RULES = (
    _rule_shared_issue_lock_external_state,
    _rule_shared_manual_lock_pr_override,
    _rule_shared_author_reviewer_same_run,
 )
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
    "review_pr": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reviewer_legacy_workspace_mutations,
        _rule_reviewer_vague_mutations_none,
        _rule_reviewer_mutation_categories,
@@ -985,7 +913,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
    "reconcile_already_landed": [
        _rule_reconcile_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reconcile_stale_author_fields,
        _rule_reconcile_eligible_reviewed,
        _rule_reconcile_linked_issue_live,
@@ -993,35 +920,29 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
        _rule_reviewer_git_fetch_readonly,
        _rule_reviewer_legacy_workspace_mutations,
        _rule_reviewer_vague_mutations_none,
        _rule_audit_reconciliation_boundary,
    ],
    "author_issue": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reviewer_vague_mutations_none,
    ],
    "work_issue": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reviewer_vague_mutations_none,
    ],
    "issue_filing": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
    ],
    "inventory": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reconcile_pagination_proof,
    ],
    "issue_selection": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
    ],
 }
@@ -538,15 +538,12 @@ import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
 import issue_lock_provenance  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
 import audit_reconciliation_mode  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import native_mcp_preference  # noqa: E402
@@ -588,7 +585,7 @@ def _load_existing_issue_lock() -> dict | None:
    if not os.path.exists(ISSUE_LOCK_FILE):
        return None
    try:
-        with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
+        with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
            data = json.load(f)
        return data if isinstance(data, dict) else None
    except Exception:
@@ -682,119 +679,6 @@ def _branch_entry_name(branch: dict | str) -> str:
    return str(branch.get("name") or branch.get("ref") or "")
 def _live_fetch_issue_duplicate_context(
    h: str,
    o: str,
    r: str,
    auth: str,
    issue_number: int,
 ) -> tuple[list[dict], list[str], dict]:
    """Live open PRs, remote branch names, and claim state for one issue."""
    base = repo_api_url(h, o, r)
    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
    branches = api_get_all(f"{base}/branches", auth)
    branch_names = [_branch_entry_name(b) for b in branches]
    issue = api_request("GET", f"{base}/issues/{issue_number}", auth) or {}
    comments = api_request(
        "GET", f"{base}/issues/{issue_number}/comments", auth
    ) or []
    claim_entry = issue_claim_heartbeat.classify_issue_claim(
        issue=issue,
        comments=comments,
        open_prs=open_prs,
        branch_names=branch_names,
    )
    return open_prs, branch_names, claim_entry
 # Injectable duplicate-work context fetcher (#400). Production uses the live
 # Gitea API path above; unit tests patch this symbol instead of hitting the
 # network.
 issue_duplicate_context_fetcher = _live_fetch_issue_duplicate_context
 def _collect_issue_duplicate_context(
    h: str,
    o: str,
    r: str,
    auth: str,
    issue_number: int,
 ) -> tuple[list[dict], list[str], dict]:
    return issue_duplicate_context_fetcher(h, o, r, auth, issue_number)
 def _assess_issue_duplicate_gate(
    issue_number: int,
    *,
    h: str,
    o: str,
    r: str,
    auth: str,
    locked_branch: str | None = None,
    phase: str,
 ) -> dict:
    open_prs, branch_names, claim_entry = _collect_issue_duplicate_context(
        h, o, r, auth, issue_number
    )
    return issue_work_duplicate_gate.assess_work_issue_duplicate_gate(
        issue_number,
        open_prs=open_prs,
        branch_names=branch_names,
        claim_entry=claim_entry,
        locked_branch=locked_branch,
        phase=phase,
    )
 def _duplicate_gate_block_response(gate: dict, **extra) -> dict:
    out = {
        "success": False,
        "performed": False,
        "reasons": list(gate.get("reasons") or []),
        "duplicate_gate": gate,
        "safe_next_action": gate.get("safe_next_action"),
    }
    out.update(extra)
    return out
 def _enforce_locked_issue_duplicate_recheck(
    remote: str,
    phase: str,
    *,
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
 ) -> dict | None:
    """Re-check duplicate-work gates for the locked issue (#400)."""
    lock_data = _load_existing_issue_lock()
    if not lock_data:
        return None
    issue_number = int(lock_data.get("issue_number") or 0)
    locked_branch = lock_data.get("branch_name")
    if not issue_number:
        return None
    h, o, r = _resolve(
        remote or lock_data.get("remote") or "dadeschools",
        host or lock_data.get("host"),
        org or lock_data.get("org"),
        repo or lock_data.get("repo"),
    )
    auth = _auth(h)
    gate = _assess_issue_duplicate_gate(
        issue_number,
        h=h,
        o=o,
        r=r,
        auth=auth,
        locked_branch=locked_branch,
        phase=phase,
    )
    if gate.get("block"):
        return gate
    return None
 def _reveal_endpoints() -> bool:
    """Admin/debug opt-in (#120): include endpoint URLs and token source
    names in tool output. Off by default so normal LLM-facing responses
@@ -1267,21 +1151,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)
-    duplicate_gate = _assess_issue_duplicate_gate(
+    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
-        issue_number,
+
-        h=h,
+    try:
-        o=o,
+        prs = api_get_all(url, auth)
-        r=r,
+    except Exception as e:
-        auth=auth,
+        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
-        locked_branch=branch_name,
+
-        phase=issue_work_duplicate_gate.PHASE_LOCK,
+    for pr in prs:
-    )
+        pr_head = pr.get("head", {}).get("ref", "")
-    if duplicate_gate.get("block"):
+        pr_title = pr.get("title", "")
-        raise ValueError("; ".join(duplicate_gate.get("reasons") or [
+        pr_body = pr.get("body", "")
-            f"duplicate work gate blocked issue #{issue_number} (fail closed)"
+
-        ]))
+        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,
@@ -1297,10 +1208,6 @@ def gitea_lock_issue(
        "repo": r,
        "worktree_path": resolved_worktree,
        "work_lease": work_lease,
        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
            tool="gitea_lock_issue",
            claimant=work_lease.get("claimant"),
        ),
    }
    try:
@@ -1331,39 +1238,6 @@ def gitea_lock_issue(
    return result
@mcp.tool()
 def gitea_assess_work_issue_duplicate(
    issue_number: int,
    branch_name: str | None = None,
    phase: str = issue_work_duplicate_gate.PHASE_LOCK,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
 ) -> dict:
    """Read-only duplicate-work gate for author sessions before mutations (#400)."""
    read_block = _profile_operation_gate("gitea.read")
    if read_block:
        return {
            "success": False,
            "performed": False,
            "reasons": read_block,
            "permission_report": _permission_block_report("gitea.read"),
        }
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    gate = _assess_issue_duplicate_gate(
        issue_number,
        h=h,
        o=o,
        r=r,
        auth=auth,
        locked_branch=branch_name,
        phase=phase,
    )
    return {"success": not gate.get("block"), **gate}
@mcp.tool()
 def gitea_create_pr(
    title: str,
@@ -1424,14 +1298,6 @@ def gitea_create_pr(
    except Exception as e:
        raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)")
    lock_provenance_check = issue_lock_provenance.assess_lock_file_for_create_pr(
        lock_data
    )
    if lock_provenance_check["block"]:
        raise RuntimeError(
            issue_lock_provenance.format_lock_provenance_error(lock_provenance_check)
        )
    locked_issue = lock_data.get("issue_number")
    locked_branch = lock_data.get("branch_name")
    locked_worktree = lock_data.get("worktree_path")
@@ -1463,21 +1329,6 @@ 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)"
        )
    duplicate_block = _enforce_locked_issue_duplicate_recheck(
        remote,
        issue_work_duplicate_gate.PHASE_CREATE_PR,
        host=host,
        org=org,
        repo=repo,
    )
    if duplicate_block:
        return _duplicate_gate_block_response(
            duplicate_block,
            number=None,
            issue_number=locked_issue,
            branch_name=locked_branch,
        )
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/pulls"
    payload = {"title": title, "body": body, "head": head, "base": base}
@@ -3082,20 +2933,6 @@ def gitea_commit_files(
    if blocked:
        return blocked
    duplicate_block = _enforce_locked_issue_duplicate_recheck(
        remote,
        issue_work_duplicate_gate.PHASE_COMMIT,
        host=host,
        org=org,
        repo=repo,
    )
    if duplicate_block:
        return _duplicate_gate_block_response(
            duplicate_block,
            commit="",
            branch="",
        )
    verify_preflight_purity(remote)
    processed_files, source_proofs = _prepare_commit_payload_files(files)
@@ -3681,18 +3518,6 @@ def gitea_delete_branch(
            "permission_report": _permission_block_report("gitea.branch.delete"),
        }
    audit_allowed, audit_reasons = (
        audit_reconciliation_mode.check_audit_mutation_allowed("delete_branch")
    )
    if not audit_allowed:
        return {
            "success": False,
            "performed": False,
            "required_permission": "gitea.branch.delete",
            "reasons": audit_reasons,
            "audit_phase": audit_reconciliation_mode.current_phase(),
        }
    verify_preflight_purity(remote)
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
@@ -3762,18 +3587,6 @@ def gitea_reconcile_merged_cleanups(
            "execute_confirmed must be True when dry_run=False (fail closed)"
        )
    if not dry_run:
        exec_allowed, exec_reasons = (
            audit_reconciliation_mode.check_cleanup_execution_allowed()
        )
        if not exec_allowed:
            return {
                "success": False,
                "performed": False,
                "reasons": exec_reasons,
                "audit_phase": audit_reconciliation_mode.current_phase(),
            }
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    base = repo_api_url(h, o, r)
@@ -3856,53 +3669,6 @@ def gitea_reconcile_merged_cleanups(
    return {"success": True, "performed": True, **report}
@mcp.tool()
 def gitea_authorize_reconciliation_cleanup_phase(
    operator_approved: bool = False,
    workflow_authorized: bool = False,
    delete_capability_proven: bool = False,
    safe_to_delete_remote: bool = False,
    safe_to_remove_worktree: bool = False,
    before_state: str = "",
    after_state: str = "",
 ) -> dict:
    """Authorize cleanup phase after audit-only reconciliation (#419).
    Requires operator or workflow approval, exact delete_branch capability proof,
    branch/worktree safety proof, and before/after state snapshots. Audit phase
    forbids branch deletion, worktree removal, pushes, and issue/PR mutations.
    """
    delete_gate = _profile_operation_gate("gitea.branch.delete")
    capability_ok = not bool(delete_gate)
    if delete_capability_proven and delete_gate:
        return {
            "authorized": False,
            "performed": False,
            "delete_capability_verified": False,
            "reasons": [
                "delete_capability_proven=true but active profile lacks "
                "gitea.branch.delete",
            ] + delete_gate,
            "audit_phase": audit_reconciliation_mode.current_phase(),
        }
    result = audit_reconciliation_mode.authorize_cleanup_phase(
        operator_approved=operator_approved,
        workflow_authorized=workflow_authorized,
        delete_capability_proven=delete_capability_proven and capability_ok,
        safety_proof={
            "safe_to_delete_remote": safe_to_delete_remote,
            "safe_to_remove_worktree": safe_to_remove_worktree,
        },
        before_after_snapshot={
            "before": (before_state or "").strip(),
            "after": (after_state or "").strip(),
        },
    )
    result["performed"] = bool(result.get("authorized"))
    result["delete_capability_verified"] = capability_ok
    return result
@mcp.tool()
 def gitea_assess_already_landed_reconciliation(
    pr_number: int,
@@ -7009,9 +6775,6 @@ def gitea_resolve_task_capability(
    if reason_msg:
        result["reason"] = reason_msg
    role_session_router.sync_route_from_capability(result)
    if audit_reconciliation_mode.check_audit_task_enters_phase(task):
        phase_record = audit_reconciliation_mode.enter_audit_phase(task)
        result["reconciliation_phase"] = phase_record.get("phase")
    was_terminal = capability_stop_terminal.is_active()
    terminal = capability_stop_terminal.sync_from_capability_result(result)
    if terminal:
@@ -1,269 +0,0 @@
 """Issue-lock provenance and external-state disclosure (#447).
 Sanctioned locks are written only by ``gitea_lock_issue`` (or adoption recovery
 #442). Manual seeding of ``/tmp/gitea_issue_lock.json`` is unsafe and must be
 blocked at PR creation unless explicit operator override proof is recorded.
 """
 from __future__ import annotations
 import os
 import re
 from datetime import datetime, timezone
 ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json")
 SOURCE_LOCK_ISSUE = "gitea_lock_issue"
 SOURCE_LOCK_ADOPTION = "gitea_lock_issue_adoption"
 SOURCE_OPERATOR_OVERRIDE = "operator_override"
 SANCTIONED_LOCK_SOURCES = frozenset({
    SOURCE_LOCK_ISSUE,
    SOURCE_LOCK_ADOPTION,
    SOURCE_OPERATOR_OVERRIDE,
 })
 _OPERATOR_OVERRIDE_ENV = "GITEA_ISSUE_LOCK_OPERATOR_OVERRIDE"
 _ISSUE_LOCK_PATH_RE = re.compile(
    r"(?:/tmp/)?gitea_issue_lock\.json",
    re.IGNORECASE,
 )
 _LOCK_SEED_RE = re.compile(
    r"(?:seed(?:ed|ing)?|restor(?:e|ed|ing)|wrote|written|write|programmatically|"
    r"hand[- ]forg|manual(?:ly)?).{0,80}gitea_issue_lock",
    re.IGNORECASE | re.DOTALL,
 )
 _LOCK_REMOVE_RE = re.compile(
    r"(?:\brm\b|remove|deleted?|unlink).{0,80}gitea_issue_lock",
    re.IGNORECASE | re.DOTALL,
 )
 _LOCK_READ_RE = re.compile(
    r"(?:read|loaded?|parsed?).{0,80}gitea_issue_lock",
    re.IGNORECASE | re.DOTALL,
 )
 _EXTERNAL_NONE_RE = re.compile(
    r"external[- ]state mutations\s*:\s*none\b",
    re.IGNORECASE,
 )
 _EXTERNAL_FIELD_RE = re.compile(
    r"external[- ]state mutations\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
 )
 _CLEANUP_ONLY_RE = re.compile(
    r"cleanup mutations\s*:\s*(?:none|lock removed|removed issue lock)",
    re.IGNORECASE,
 )
 _PR_CREATED_RE = re.compile(
    r"(?:\bgitea_create_pr\b|PR\s*#\s*\d+\s+created|created\s+PR\s*#|opened\s+PR\s*#|"
    r"PR\s+creation\s+(?:succeeded|complete))",
    re.IGNORECASE,
 )
 _REVIEW_APPROVE_RE = re.compile(
    r"(?:submitted\s+(?:['\"]approve['\"]|approve\s+review)|"
    r"review decision\s*:\s*approve|approved\s+PR\s*#|gitea_review_pr.*approve)",
    re.IGNORECASE,
 )
 _OVERRIDE_PROOF_RE = re.compile(
    r"operator[- ]override\s+proof\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
 )
 def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
 def build_sanctioned_lock_provenance(
    *,
    tool: str,
    source: str = SOURCE_LOCK_ISSUE,
    claimant: dict | None = None,
    adoption: dict | None = None,
 ) -> dict:
    """Return provenance metadata stored with a sanctioned lock write."""
    entry = {
        "source": source,
        "written_at": _utc_now_iso(),
        "written_by_tool": tool,
        "lock_file_path": ISSUE_LOCK_FILE,
    }
    if claimant:
        entry["claimant"] = claimant
    if adoption:
        entry["adoption"] = adoption
    return entry
 def operator_override_requested() -> bool:
    return os.environ.get(_OPERATOR_OVERRIDE_ENV, "").strip().lower() in {
        "1",
        "true",
        "yes",
    }
 def build_operator_override_provenance(*, reason: str, claimant: dict | None = None) -> dict:
    text = (reason or "").strip()
    if not text:
        raise ValueError(
            "operator override requires a non-empty override reason (fail closed)"
        )
    entry = build_sanctioned_lock_provenance(
        tool="operator_override",
        source=SOURCE_OPERATOR_OVERRIDE,
        claimant=claimant,
    )
    entry["override_reason"] = text
    return entry
 def assess_lock_file_for_create_pr(lock_data: dict | None) -> dict:
    """Fail closed when lock file lacks sanctioned provenance (#447)."""
    data = lock_data if isinstance(lock_data, dict) else {}
    reasons: list[str] = []
    provenance = data.get("lock_provenance")
    if not isinstance(provenance, dict):
        reasons.append(
            "issue lock file lacks sanctioned lock_provenance; manual seeding is "
            "not a normal recovery path — call gitea_lock_issue or use #442 adoption"
        )
        return _provenance_result(False, reasons, provenance)
    source = str(provenance.get("source") or "").strip()
    if source not in SANCTIONED_LOCK_SOURCES:
        reasons.append(
            f"issue lock provenance source '{source or '(missing)'}' is not sanctioned"
        )
    if source == SOURCE_OPERATOR_OVERRIDE and not str(
        provenance.get("override_reason") or ""
    ).strip():
        reasons.append(
            "operator_override lock provenance requires override_reason proof"
        )
    if not data.get("work_lease"):
        reasons.append("issue lock file missing work_lease metadata")
    if not str(provenance.get("written_by_tool") or "").strip():
        reasons.append("issue lock provenance missing written_by_tool")
    proven = not reasons
    return _provenance_result(proven, reasons, provenance)
 def _provenance_result(proven: bool, reasons: list[str], provenance: dict | None) -> dict:
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "lock_provenance": provenance,
    }
 def format_lock_provenance_error(assessment: dict) -> str:
    reasons = "; ".join(assessment.get("reasons") or ["unknown lock provenance violation"])
    return f"Issue lock provenance guard (#447): {reasons} (fail closed)"
 def _lock_activity_detected(text: str) -> dict[str, bool]:
    body = text or ""
    return {
        "seed_or_restore": bool(_LOCK_SEED_RE.search(body)),
        "remove": bool(_LOCK_REMOVE_RE.search(body)),
        "read": bool(_LOCK_READ_RE.search(body)),
    }
 def _external_state_discloses_lock(text: str) -> bool:
    match = _EXTERNAL_FIELD_RE.search(text or "")
    if not match:
        return False
    value = (match.group(1) or "").strip().lower()
    if value in {"", "none", "n/a"}:
        return False
    return "lock" in value or "gitea_issue_lock" in value or "issue-lock" in value
 def assess_issue_lock_external_state_report(report_text: str) -> dict:
    """Require explicit external-state disclosure for issue-lock mutations (#447)."""
    text = report_text or ""
    activity = _lock_activity_detected(text)
    if not any(activity.values()):
        return {"proven": True, "block": False, "reasons": [], "activity": activity}
    reasons: list[str] = []
    disclosed = _external_state_discloses_lock(text)
    if activity["seed_or_restore"] and _EXTERNAL_NONE_RE.search(text):
        reasons.append(
            "report mentions seeding/restoring gitea_issue_lock.json but claims "
            "External-state mutations: none"
        )
    elif activity["seed_or_restore"] and not disclosed:
        reasons.append(
            "report mentions issue-lock file activity but External-state mutations "
            "does not disclose read/write of gitea_issue_lock.json"
        )
    if activity["remove"]:
        if _EXTERNAL_NONE_RE.search(text):
            reasons.append(
                "report mentions removing gitea_issue_lock.json but claims "
                "External-state mutations: none"
            )
        elif not disclosed and _CLEANUP_ONLY_RE.search(text):
            reasons.append(
                "report removes issue lock but classifies it as cleanup only; "
                "record under External-state mutations"
            )
        elif not disclosed:
            reasons.append(
                "report mentions deleting issue lock without External-state "
                "mutation disclosure"
            )
    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "activity": activity,
    }
 def assess_manual_lock_pr_without_override(report_text: str) -> dict:
    """Block reports that created a PR via manual lock seed without override proof."""
    text = report_text or ""
    seeded = bool(_LOCK_SEED_RE.search(text))
    created = bool(_PR_CREATED_RE.search(text))
    if not (seeded and created):
        return {"proven": True, "block": False, "reasons": []}
    if _OVERRIDE_PROOF_RE.search(text):
        return {"proven": True, "block": False, "reasons": []}
    return {
        "proven": False,
        "block": True,
        "reasons": [
            "report created/opened a PR after manual issue-lock seeding without "
            "operator override proof"
        ],
    }
 def assess_author_reviewer_same_run_report(report_text: str) -> dict:
    """Reviewer handoff must not create and approve the same PR in one run (#447)."""
    text = report_text or ""
    if not (_PR_CREATED_RE.search(text) and _REVIEW_APPROVE_RE.search(text)):
        return {"proven": True, "block": False, "reasons": []}
    return {
        "proven": False,
        "block": True,
        "reasons": [
            "report mixes author-side PR creation and reviewer approval in one "
            "final handoff; split author and reviewer sessions"
        ],
    }
@@ -1,180 +0,0 @@
 """Early duplicate-work detection for author work-issue sessions (#400)."""
 from __future__ import annotations
 from typing import Any
 import issue_claim_heartbeat as claim_hb
 PHASE_LOCK = "lock_issue"
 PHASE_COMMIT = "commit"
 PHASE_PUSH = "push"
 PHASE_CREATE_PR = "create_pr"
 OUTCOME_DUPLICATE_PR_PREVENTED = "duplicate_pr_prevented"
 OUTCOME_DUPLICATE_BRANCH_PREVENTED = "duplicate_branch_prevented"
 OUTCOME_DUPLICATE_COMMIT_PREVENTED = "duplicate_commit_prevented"
 OUTCOME_DUPLICATE_WORK_NOT_PREVENTED = "duplicate_work_not_prevented"
 _ACTIVE_CLAIM_STATUSES = frozenset({"active", "awaiting_review"})
 def _issue_pattern(issue_number: int) -> str:
    return f"issue-{int(issue_number)}"
 def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None:
    return claim_hb._linked_open_pr(issue_number, open_prs)
 def _matching_branches(
    issue_number: int,
    branch_names: list[str],
    *,
    locked_branch: str | None = None,
 ) -> list[str]:
    pattern = _issue_pattern(issue_number)
    matches = [
        name for name in (branch_names or [])
        if pattern in (name or "").lower()
    ]
    if locked_branch:
        locked = locked_branch.strip()
        matches = [name for name in matches if name != locked]
    return matches
 def assess_work_issue_duplicate_gate(
    issue_number: int,
    *,
    open_prs: list[dict] | None = None,
    branch_names: list[str] | None = None,
    claim_entry: dict | None = None,
    locked_branch: str | None = None,
    phase: str = PHASE_LOCK,
 ) -> dict[str, Any]:
    """Fail closed when duplicate work is already in flight for an issue."""
    reasons: list[str] = []
    outcome = OUTCOME_DUPLICATE_WORK_NOT_PREVENTED
    prs = list(open_prs or [])
    branches = list(branch_names or [])
    pattern = _issue_pattern(issue_number)
    linked = _linked_open_pr(issue_number, prs)
    if linked:
        reasons.append(
            f"open PR #{linked.get('number')} already covers issue "
            f"#{issue_number} (fail closed)"
        )
        outcome = OUTCOME_DUPLICATE_PR_PREVENTED
    conflicting_branches = _matching_branches(
        issue_number, branches, locked_branch=locked_branch
    )
    if conflicting_branches:
        names = ", ".join(conflicting_branches[:5])
        reasons.append(
            f"remote branch(es) already match issue pattern '{pattern}': "
            f"{names} (fail closed)"
        )
        if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
            outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED
    entry = claim_entry or {}
    if entry.get("linked_open_pr") and not linked:
        reasons.append(
            f"claim inventory reports open PR #{entry['linked_open_pr']} "
            f"for issue #{issue_number} (fail closed)"
        )
        outcome = OUTCOME_DUPLICATE_PR_PREVENTED
    status = (entry.get("status") or "").strip().lower()
    if status in _ACTIVE_CLAIM_STATUSES and not linked:
        heartbeat = entry.get("latest_heartbeat") or {}
        claim_branch = (heartbeat.get("branch") or "").strip()
        if locked_branch and claim_branch and claim_branch != locked_branch:
            reasons.append(
                f"active claim lease on branch '{claim_branch}' blocks "
                f"work on '{locked_branch}' for issue #{issue_number} "
                "(fail closed)"
            )
            if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
                outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED
        elif not locked_branch and status == "active":
            reasons.append(
                f"issue #{issue_number} has an active claim lease "
                "(fail closed)"
            )
            if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
                outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED
    if phase in {PHASE_COMMIT, PHASE_PUSH} and reasons:
        if outcome == OUTCOME_DUPLICATE_PR_PREVENTED:
            outcome = OUTCOME_DUPLICATE_COMMIT_PREVENTED
        elif outcome == OUTCOME_DUPLICATE_BRANCH_PREVENTED:
            outcome = OUTCOME_DUPLICATE_COMMIT_PREVENTED
    block = bool(reasons)
    return {
        "block": block,
        "performed": not block,
        "issue_number": issue_number,
        "phase": phase,
        "outcome": outcome,
        "linked_open_pr": linked.get("number") if linked else entry.get("linked_open_pr"),
        "conflicting_branches": conflicting_branches,
        "claim_status": status or None,
        "reasons": reasons,
        "safe_next_action": (
            "stop before mutating; preserve local work and produce a "
            "reconciliation handoff if a concurrent PR appeared after push"
            if block and phase == PHASE_CREATE_PR
            else "stop before mutating; do not commit or push duplicate work"
            if block
            else "proceed"
        ),
    }
 def assess_work_issue_duplicate_report(report_text: str) -> dict[str, Any]:
    """Require explicit duplicate-work outcome wording in work-issue reports."""
    text = (report_text or "").lower()
    markers = {
        OUTCOME_DUPLICATE_PR_PREVENTED: (
            "duplicate pr prevented",
            "duplicate_pr_prevented",
        ),
        OUTCOME_DUPLICATE_BRANCH_PREVENTED: (
            "duplicate branch prevented",
            "duplicate_branch_prevented",
        ),
        OUTCOME_DUPLICATE_COMMIT_PREVENTED: (
            "duplicate commit prevented",
            "duplicate_commit_prevented",
        ),
        OUTCOME_DUPLICATE_WORK_NOT_PREVENTED: (
            "duplicate work not prevented",
            "duplicate_work_not_prevented",
            "no duplicate work",
        ),
    }
    matched = [
        key for key, phrases in markers.items()
        if any(phrase in text for phrase in phrases)
    ]
    if len(matched) != 1:
        return {
            "complete": False,
            "downgraded": True,
            "reasons": [
                "work-issue report must state exactly one duplicate-work "
                "outcome (duplicate PR/branch/commit prevented, or "
                "duplicate work not prevented)"
            ],
        }
    return {
        "complete": True,
        "downgraded": False,
        "outcome": matched[0],
        "reasons": [],
    }
@@ -3624,12 +3624,9 @@ def assess_work_issue_mode_isolation(report_text: str) -> dict:
 def assess_work_issue_final_report(report_text: str) -> dict:
    """#139: composite verifier for work-issue final reports."""
    from issue_work_duplicate_gate import assess_work_issue_duplicate_report
    checks = {
        "workflow_source": assess_work_issue_workflow_source(report_text),
        "mode_isolation": assess_work_issue_mode_isolation(report_text),
        "duplicate_work_outcome": assess_work_issue_duplicate_report(report_text),
    }
    reasons = []
@@ -5115,15 +5112,6 @@ def assess_pr_queue_cleanup_report(report_text: str | None) -> dict:
    return _assess(report_text or "")
 def assess_audit_reconciliation_report(report_text: str | None) -> dict:
    """#419: validate audit vs cleanup reconciliation report boundaries."""
    from audit_reconciliation_mode import (
        assess_audit_reconciliation_report as _assess,
    )
    return _assess(report_text or "")
 _GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
 _NOT_APPLICABLE_VALUE = re.compile(
@@ -304,40 +304,6 @@ If any required mutation capability is missing:
 * include safe next action (profile switch, human close, or dedicated reconciler
  profile)
 ## 15A. Audit vs cleanup phase (#419)
 Reconciliation audits are **read-only** unless a separate cleanup phase is
 explicitly authorized.
 **Audit phase forbids** (``audit_reconciliation_mode.check_audit_mutation_allowed``
 fails closed):
 * ``gitea_delete_branch``
 * ``git branch -D``
 * ``git worktree remove``
 * pushes
 * issue/PR mutations
 * file edits
 Dry-run merged-cleanup reconciliation (``gitea_reconcile_merged_cleanups`` with
 ``dry_run=True``) stays in audit phase. Execution requires:
 1. Operator approval or workflow authorization
 2. Exact ``delete_branch`` capability proof (``gitea.branch.delete``)
 3. Proof branch/worktree is safe to remove
 4. Before/after state snapshot
 Call ``gitea_authorize_reconciliation_cleanup_phase`` before any cleanup
 mutation. Final reports must not claim ``no mutations`` if cleanup occurred.
 Classify cleanup mutations as:
 * remote branch deletion → **External-state mutations**
 * local branch deletion → **Git ref mutations**
 * worktree removal → **Cleanup mutations**
 ``audit_reconciliation_mode.assess_audit_reconciliation_report`` validates
 these boundaries in final reports.
 ## 16. Mutation classification
 Use precise mutation categories in the final report:
@@ -297,36 +297,6 @@ Report:
 Do not create another branch/PR for the same issue unless the project explicitly allows taking over or updating existing work and exact capability is proven.
 ### 10A. Duplicate-work gate phases (#400)
 Before any file edits, prove duplicate-work clearance with
 `gitea_assess_work_issue_duplicate` or `gitea_lock_issue` (which runs the same
 gate). The gate checks live:
 * open PRs linked to the issue (head branch or Closes/Fixes reference),
 * remote branches matching `issue-<number>`,
 * active claim leases from structured heartbeats.
 Re-check immediately before:
 * `gitea_commit_files` (commit),
 * branch push,
 * `gitea_create_pr` (PR creation).
 If a concurrent open PR appears after work begins:
 * before commit/push → stop and preserve local work without pushing,
 * after commit but before push → stop without pushing,
 * after push but before PR creation → stop and produce a reconciliation
  handoff instead of opening a PR.
 Final reports must state exactly one duplicate-work outcome:
 * `duplicate PR prevented`
 * `duplicate branch prevented`
 * `duplicate commit prevented`
 * `duplicate work not prevented`
 ## 11. Claim or lock the issue before implementation
 Claim/lock the issue before implementation if the project provides a claim/lock mechanism.
@@ -747,12 +717,6 @@ Use only precise categories:
 * External-state mutations:
 * Read-only diagnostics:
 Issue-lock file (`/tmp/gitea_issue_lock.json`) read/write/delete is always an
 external-state mutation. Never claim `External-state mutations: none` after
 seeding, restoring, or removing that file. Manual lock seeding is not a normal
 recovery path (#447); use `gitea_lock_issue` or the #442 adoption recovery path
 instead. Link broader redesign: #438.
 `git fetch`, `git remote update`, and any command that updates refs must be listed under `Git ref mutations`, not read-only diagnostics.
 If `git reset --hard`, checkout, clean, worktree add/remove, merge simulation, merge abort, or similar commands occurred, report them under `Worktree/index mutations`.
@@ -104,10 +104,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
        "permission": "gitea.read",
        "role": "author",
    },
    "reconciliation_cleanup": {
        "permission": "gitea.branch.delete",
        "role": "author",
    },
    "work_issue": {
        "permission": "gitea.pr.create",
        "role": "author",
@@ -80,10 +80,7 @@ class TestIssueLockArtifactWarning(unittest.TestCase):
    def tearDown(self):
        self._env_patcher.stop()
-    @patch(
+    @patch("mcp_server.api_get_all", return_value=[])
        "mcp_server.issue_duplicate_context_fetcher",
        return_value=([], [], {"status": "not_claimed"}),
    )
    @patch("mcp_server._auth", return_value="token x")
    @patch("mcp_server._resolve", return_value=("h", "o", "r"))
    @patch("mcp_server.ISSUE_LOCK_FILE", new_callable=lambda: tempfile.mktemp())
@@ -1,291 +0,0 @@
 """Tests for audit vs cleanup reconciliation mode (#419)."""
 import os
 import sys
 import unittest
 from unittest.mock import patch
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 import audit_reconciliation_mode as arm
 import mcp_server
 from audit_reconciliation_mode import (
    AUDIT_FORBIDDEN_TASKS,
    PHASE_AUDIT,
    PHASE_CLEANUP,
    assess_audit_command_allowed,
    assess_audit_reconciliation_report,
    authorize_cleanup_phase,
    check_audit_mutation_allowed,
    check_cleanup_execution_allowed,
    classify_cleanup_mutation,
    clear_phase,
    enter_audit_phase,
 )
 from final_report_validator import assess_final_report_validator
 from review_proofs import assess_audit_reconciliation_report as proofs_assess
 from task_capability_map import required_permission, required_role
 DELETE_PROFILE = {
    "profile_name": "prgs-author-delete",
    "allowed_operations": ["gitea.read", "gitea.branch.delete"],
    "forbidden_operations": [],
    "audit_label": "prgs-author-delete",
 }
 READ_PROFILE = {
    "profile_name": "prgs-author",
    "allowed_operations": ["gitea.read", "gitea.issue.comment"],
    "forbidden_operations": ["gitea.branch.delete"],
    "audit_label": "prgs-author",
 }
 READ_ENV = {
    "GITEA_MCP_CONFIG": os.path.join(
        os.path.dirname(__file__), "..", "profiles.json"
    ),
    "GITEA_MCP_PROFILE": "prgs-author",
 }
 def _authorize_cleanup(**kwargs):
    defaults = {
        "operator_approved": True,
        "delete_capability_proven": True,
        "safety_proof": {"safe_to_delete_remote": True},
        "before_after_snapshot": {
            "before": "remote branch exists",
            "after": "remote branch absent",
        },
    }
    defaults.update(kwargs)
    return authorize_cleanup_phase(**defaults)
 def _cleanup_report(**overrides):
    base = (
        "Task mode: reconcile-landed-pr\n"
        "Workflow source: workflows/reconcile-landed-pr.md\n"
        "Audit phase: read-only assessment\n"
        "Cleanup phase authorized: true\n"
        "Delete-branch capability proven: true\n"
        "Branch safe to remove: true\n"
        "Before/after state snapshot: remote branch feat/x present → absent\n"
        "External-state mutations: deleted remote branch feat/x\n"
        "Git ref mutations: none\n"
        "Cleanup mutations: removed worktree branches/feat-x\n"
    )
    for key, value in overrides.items():
        base = base.replace(key, value)
    return base
 class TestAuditPhaseGates(unittest.TestCase):
    def setUp(self):
        clear_phase()
        enter_audit_phase("reconcile-landed-pr")
    def tearDown(self):
        clear_phase()
    def test_audit_blocks_delete_branch_task(self):
        allowed, reasons = check_audit_mutation_allowed("delete_branch")
        self.assertFalse(allowed)
        self.assertTrue(reasons)
    def test_audit_blocks_worktree_shell_commands(self):
        allowed, reasons = assess_audit_command_allowed(
            "git worktree remove branches/feat-x"
        )
        self.assertFalse(allowed)
        self.assertTrue(reasons)
    def test_audit_blocks_local_branch_delete_command(self):
        allowed, reasons = assess_audit_command_allowed("git branch -D feat/x")
        self.assertFalse(allowed)
    def test_audit_allows_read_tasks(self):
        allowed, _ = check_audit_mutation_allowed("reconcile_landed_pr")
        self.assertTrue(allowed)
    def test_forbidden_set_covers_issue_and_pr_mutations(self):
        for task in ("close_pr", "comment_issue", "commit_files", "push_branch"):
            self.assertIn(task, AUDIT_FORBIDDEN_TASKS)
 class TestCleanupAuthorization(unittest.TestCase):
    def setUp(self):
        clear_phase()
        enter_audit_phase("reconcile_merged_cleanups")
    def tearDown(self):
        clear_phase()
    def test_cleanup_without_approval_blocked(self):
        result = authorize_cleanup_phase(
            delete_capability_proven=True,
            safety_proof={"safe_to_delete_remote": True},
            before_after_snapshot={"before": "a", "after": "b"},
        )
        self.assertFalse(result["authorized"])
    def test_cleanup_without_capability_proof_blocked(self):
        result = _authorize_cleanup(delete_capability_proven=False)
        self.assertFalse(result["authorized"])
    def test_cleanup_without_snapshot_blocked(self):
        result = _authorize_cleanup(before_after_snapshot={"before": "", "after": ""})
        self.assertFalse(result["authorized"])
    def test_authorized_cleanup_switches_phase(self):
        result = _authorize_cleanup()
        self.assertTrue(result["authorized"])
        self.assertEqual(result["phase"], PHASE_CLEANUP)
    def test_cleanup_execution_allowed_only_after_authorization(self):
        self.assertFalse(check_cleanup_execution_allowed()[0])
        _authorize_cleanup()
        self.assertTrue(check_cleanup_execution_allowed()[0])
 class TestReportVerifier(unittest.TestCase):
    def test_false_no_mutations_after_cleanup_blocked(self):
        report = (
            "Task mode: reconcile-landed-pr\n"
            "No mutations performed.\n"
            "delete_remote_branch feat/dup\n"
        )
        result = assess_audit_reconciliation_report(report)
        self.assertFalse(result["proven"])
        self.assertIn("no mutations", result["reasons"][0].lower())
    def test_cleanup_without_authorization_fields_blocked(self):
        report = (
            "Task mode: reconcile-landed-pr\n"
            "remove_local_worktree branches/feat-x\n"
        )
        result = assess_audit_reconciliation_report(report)
        self.assertFalse(result["proven"])
    def test_authorized_cleanup_report_passes(self):
        result = assess_audit_reconciliation_report(_cleanup_report())
        self.assertTrue(result["proven"])
    def test_mutation_classification_enforced(self):
        report = (
            "Task mode: reconcile-landed-pr\n"
            "Cleanup phase authorized: true\n"
            "Delete-branch capability proven: true\n"
            "Branch safe to remove: true\n"
            "Before/after state snapshot: present\n"
            "remove_local_worktree branches/feat-x\n"
        )
        result = assess_audit_reconciliation_report(report)
        self.assertFalse(result["proven"])
    def test_proofs_export_matches_module(self):
        report = "No mutations performed.\ndelete_remote_branch feat/dup"
        self.assertEqual(
            proofs_assess(report)["proven"],
            assess_audit_reconciliation_report(report)["proven"],
        )
    def test_final_report_validator_includes_boundary_rule(self):
        report = "No mutations performed.\ndelete_remote_branch feat/dup"
        result = assess_final_report_validator(
            report_text=report,
            task_kind="reconcile_already_landed",
        )
        self.assertTrue(result["blocked"])
        rule_ids = [f["rule_id"] for f in result["findings"]]
        self.assertIn("reconcile.audit_cleanup_boundary", rule_ids)
 class TestMutationClassification(unittest.TestCase):
    def test_remote_delete_is_external_state(self):
        self.assertEqual(
            classify_cleanup_mutation("delete_remote_branch"),
            "external-state",
        )
    def test_worktree_remove_is_cleanup(self):
        self.assertEqual(
            classify_cleanup_mutation("remove_local_worktree"),
            "cleanup",
        )
 class TestMcpGates(unittest.TestCase):
    def setUp(self):
        clear_phase()
        enter_audit_phase("reconcile_merged_cleanups")
        self.mock_api = patch("mcp_server.api_request").start()
        self.mock_auth = patch(
            "mcp_server.get_auth_header", return_value="token test"
        ).start()
    def tearDown(self):
        patch.stopall()
        clear_phase()
        mcp_server._IDENTITY_CACHE.clear()
    @patch.dict(os.environ, READ_ENV, clear=True)
    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
    def test_delete_branch_blocked_in_audit_phase(self, _profile):
        mcp_server.record_preflight_check("whoami")
        mcp_server.record_preflight_check("capability", resolved_role="author")
        result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
        self.assertFalse(result["success"])
        self.assertEqual(result["audit_phase"], PHASE_AUDIT)
        self.mock_api.assert_not_called()
    @patch.dict(os.environ, READ_ENV, clear=True)
    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
    def test_reconcile_execute_blocked_without_cleanup_auth(self, _profile):
        with self.assertRaises(ValueError):
            mcp_server.gitea_reconcile_merged_cleanups(
                dry_run=False,
                execute_confirmed=False,
                remote="prgs",
            )
    @patch.dict(os.environ, READ_ENV, clear=True)
    @patch("mcp_server.get_profile", return_value=READ_PROFILE)
    def test_cleanup_auth_fails_without_delete_capability(self, _profile):
        result = mcp_server.gitea_authorize_reconciliation_cleanup_phase(
            operator_approved=True,
            delete_capability_proven=True,
            safe_to_delete_remote=True,
            before_state="exists",
            after_state="gone",
        )
        self.assertFalse(result["authorized"])
        self.assertFalse(result["delete_capability_verified"])
    @patch.dict(os.environ, READ_ENV, clear=True)
    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
    def test_delete_branch_allowed_after_cleanup_authorization(self, _profile):
        mcp_server.gitea_authorize_reconciliation_cleanup_phase(
            operator_approved=True,
            delete_capability_proven=True,
            safe_to_delete_remote=True,
            before_state="exists",
            after_state="gone",
        )
        mcp_server.record_preflight_check("whoami")
        mcp_server.record_preflight_check("capability", resolved_role="author")
        self.mock_api.return_value = {}
        result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
        self.assertTrue(result["success"])
 class TestTaskCapabilityMap(unittest.TestCase):
    def test_reconciliation_cleanup_maps_delete_permission(self):
        self.assertEqual(
            required_permission("reconciliation_cleanup"),
            "gitea.branch.delete",
        )
        self.assertEqual(required_role("reconciliation_cleanup"), "author")
 if __name__ == "__main__":
    unittest.main()
@@ -76,14 +76,7 @@ class CommitFilesCapabilityBase(unittest.TestCase):
        with open(self.config_path, "w", encoding="utf-8") as fh:
            fh.write(json.dumps(CONFIG))
        self._dup_fetcher_patcher = patch(
            "mcp_server.issue_duplicate_context_fetcher",
            return_value=([], [], {"status": "not_claimed"}),
        )
        self._dup_fetcher_patcher.start()
    def tearDown(self):
        self._dup_fetcher_patcher.stop()
        self._remotes.stop()
        mcp_server._IDENTITY_CACHE.clear()
        mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (
@@ -67,15 +67,6 @@ class TestCommitPayloads(unittest.TestCase):
        self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
        self.lock_file_path = "/tmp/gitea_issue_lock.json"
        import issue_lock_provenance
        work_lease = {
            "operation_type": "author_issue_work",
            "issue_number": 263,
            "branch": "feat/issue-263-native-commit-payloads",
            "claimant": {"username": "test-user", "profile": "test-author"},
            "expires_at": "2999-01-01T00:00:00Z",
        }
        self.lock_data = {
            "issue_number": 263,
            "branch_name": "feat/issue-263-native-commit-payloads",
@@ -83,11 +74,6 @@ class TestCommitPayloads(unittest.TestCase):
            "org": "Example-Org",
            "repo": "Example-Repo",
            "worktree_path": self.locked_worktree_path,
            "work_lease": work_lease,
            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
                tool="gitea_lock_issue",
                claimant=work_lease.get("claimant"),
            ),
        }
        with open(self.lock_file_path, "w", encoding="utf-8") as fh:
            fh.write(json.dumps(self.lock_data))
@@ -98,14 +84,7 @@ class TestCommitPayloads(unittest.TestCase):
        mcp_server._preflight_whoami_called = True
        mcp_server._preflight_capability_called = True
        self._dup_fetcher_patcher = patch(
            "mcp_server.issue_duplicate_context_fetcher",
            return_value=([], [], {"status": "not_claimed"}),
        )
        self._dup_fetcher_patcher.start()
    def tearDown(self):
        self._dup_fetcher_patcher.stop()
        self._remotes.stop()
        mcp_server._IDENTITY_CACHE.clear()
@@ -1,130 +0,0 @@
 """Tests for issue-lock provenance and external-state disclosure (#447)."""
 from __future__ import annotations
 import sys
 import unittest
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 import issue_lock_provenance as ilp  # noqa: E402
 from final_report_validator import assess_final_report_validator  # noqa: E402
 def _sanctioned_lock(**overrides):
    work_lease = {
        "operation_type": "author_issue_work",
        "issue_number": 447,
        "branch": "feat/issue-447-lock-provenance",
        "claimant": {"username": "jcwalker3", "profile": "prgs-author"},
        "expires_at": "2999-01-01T00:00:00Z",
    }
    data = {
        "issue_number": 447,
        "branch_name": "feat/issue-447-lock-provenance",
        "work_lease": work_lease,
        "lock_provenance": ilp.build_sanctioned_lock_provenance(
            tool="gitea_lock_issue",
            claimant=work_lease["claimant"],
        ),
    }
    data.update(overrides)
    return data
 class TestLockProvenanceForCreatePr(unittest.TestCase):
    def test_sanctioned_lock_passes(self):
        result = ilp.assess_lock_file_for_create_pr(_sanctioned_lock())
        self.assertTrue(result["proven"])
        self.assertFalse(result["block"])
    def test_manual_seed_without_provenance_blocked(self):
        result = ilp.assess_lock_file_for_create_pr(
            {"issue_number": 420, "branch_name": "feat/x", "work_lease": {}}
        )
        self.assertTrue(result["block"])
        self.assertIn("lock_provenance", result["reasons"][0])
    def test_operator_override_requires_reason(self):
        result = ilp.assess_lock_file_for_create_pr(
            _sanctioned_lock(
                lock_provenance=ilp.build_sanctioned_lock_provenance(
                    tool="operator_override",
                    source=ilp.SOURCE_OPERATOR_OVERRIDE,
                )
            )
        )
        self.assertTrue(result["block"])
 class TestExternalStateReportRules(unittest.TestCase):
    def test_seed_with_external_none_blocked(self):
        report = (
            "Restored /tmp/gitea_issue_lock.json to unblock PR creation.\n"
            "- External-state mutations: none\n"
        )
        result = ilp.assess_issue_lock_external_state_report(report)
        self.assertTrue(result["block"])
    def test_seed_with_disclosure_passes(self):
        report = (
            "Restored /tmp/gitea_issue_lock.json after MCP restart.\n"
            "- External-state mutations: wrote /tmp/gitea_issue_lock.json\n"
        )
        result = ilp.assess_issue_lock_external_state_report(report)
        self.assertTrue(result["proven"])
    def test_remove_claimed_as_cleanup_only_blocked(self):
        report = (
            "rm /tmp/gitea_issue_lock.json after PR creation.\n"
            "- Cleanup mutations: lock removed\n"
            "- External-state mutations: none\n"
        )
        result = ilp.assess_issue_lock_external_state_report(report)
        self.assertTrue(result["block"])
    def test_manual_lock_pr_without_override_blocked(self):
        report = (
            "Programmatically seeded gitea_issue_lock.json then gitea_create_pr.\n"
            "PR #444 created.\n"
        )
        result = ilp.assess_manual_lock_pr_without_override(report)
        self.assertTrue(result["block"])
    def test_author_reviewer_same_run_blocked(self):
        report = (
            "gitea_create_pr opened PR #444.\n"
            "Submitted approve review on PR #444.\n"
        )
        result = ilp.assess_author_reviewer_same_run_report(report)
        self.assertTrue(result["block"])
 class TestFinalReportValidatorIntegration(unittest.TestCase):
    def test_work_issue_blocks_hidden_lock_mutation(self):
        report = (
            "## Controller Handoff\n"
            "- Task: work issue #420\n"
            "- External-state mutations: none\n"
            "Restored /tmp/gitea_issue_lock.json before PR creation.\n"
        )
        result = assess_final_report_validator(report, "work_issue")
        rule_ids = {f["rule_id"] for f in result["findings"]}
        self.assertIn("shared.issue_lock_external_state", rule_ids)
        self.assertTrue(result["blocked"])
    def test_review_pr_blocks_create_and_approve(self):
        report = (
            "## Controller Handoff\n"
            "- Task: review PR #444\n"
            "- Review decision: approve\n"
            "Created PR #444 via gitea_create_pr earlier in this run.\n"
        )
        result = assess_final_report_validator(report, "review_pr")
        rule_ids = {f["rule_id"] for f in result["findings"]}
        self.assertIn("shared.author_reviewer_same_run", rule_ids)
 if __name__ == "__main__":
    unittest.main()
@@ -1,251 +0,0 @@
 """Tests for early duplicate-work detection (#400)."""
 import json
 import os
 import sys
 import tempfile
 import unittest
 from pathlib import Path
 from unittest.mock import patch
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 import issue_work_duplicate_gate as dup_gate
 import mcp_server
 from issue_work_duplicate_gate import (
    OUTCOME_DUPLICATE_BRANCH_PREVENTED,
    OUTCOME_DUPLICATE_COMMIT_PREVENTED,
    OUTCOME_DUPLICATE_PR_PREVENTED,
    OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
    PHASE_COMMIT,
    PHASE_CREATE_PR,
    PHASE_LOCK,
    assess_work_issue_duplicate_gate,
    assess_work_issue_duplicate_report,
 )
 class TestDuplicateGateAssessment(unittest.TestCase):
    def test_clear_issue_passes(self):
        result = assess_work_issue_duplicate_gate(
            400,
            open_prs=[],
            branch_names=["feat/other-issue-99"],
            claim_entry={"status": "not_claimed"},
            locked_branch="feat/issue-400-duplicate-work-preflight",
            phase=PHASE_LOCK,
        )
        self.assertFalse(result["block"])
        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
    def test_open_pr_blocks(self):
        prs = [{
            "number": 397,
            "title": "feat: handoff",
            "body": "Closes #395",
            "head": {"ref": "feat/issue-395-proof-backed-review-handoff"},
        }]
        result = assess_work_issue_duplicate_gate(
            395,
            open_prs=prs,
            branch_names=[],
            phase=PHASE_LOCK,
        )
        self.assertTrue(result["block"])
        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED)
    def test_conflicting_remote_branch_blocks(self):
        result = assess_work_issue_duplicate_gate(
            395,
            open_prs=[],
            branch_names=["feat/issue-395-proof-backed-handoff-claims"],
            locked_branch="feat/issue-395-new-attempt",
            phase=PHASE_LOCK,
        )
        self.assertTrue(result["block"])
        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_BRANCH_PREVENTED)
    def test_active_claim_on_other_branch_blocks(self):
        result = assess_work_issue_duplicate_gate(
            398,
            open_prs=[],
            branch_names=[],
            claim_entry={
                "status": "active",
                "latest_heartbeat": {
                    "branch": "feat/issue-398-validation-cwd-proof",
                },
            },
            locked_branch="feat/issue-398-other-branch",
            phase=PHASE_LOCK,
        )
        self.assertTrue(result["block"])
    def test_commit_phase_maps_to_commit_outcome(self):
        prs = [{
            "number": 411,
            "title": "x",
            "body": "Closes #398",
            "head": {"ref": "feat/issue-398-validation-cwd-proof"},
        }]
        result = assess_work_issue_duplicate_gate(
            398,
            open_prs=prs,
            branch_names=[],
            locked_branch="feat/issue-398-alt",
            phase=PHASE_COMMIT,
        )
        self.assertTrue(result["block"])
        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_COMMIT_PREVENTED)
    def test_stale_claim_does_not_block_by_status_alone(self):
        result = assess_work_issue_duplicate_gate(
            400,
            open_prs=[],
            branch_names=[],
            claim_entry={"status": "reclaimable", "reasons": ["stale"]},
            locked_branch="feat/issue-400-duplicate-work-preflight",
            phase=PHASE_LOCK,
        )
        self.assertFalse(result["block"])
 class TestDuplicateReportOutcome(unittest.TestCase):
    def test_requires_exactly_one_outcome(self):
        bad = assess_work_issue_duplicate_report("work finished")
        self.assertFalse(bad["complete"])
        good = assess_work_issue_duplicate_report(
            "Duplicate work not prevented for issue #400."
        )
        self.assertTrue(good["complete"])
        self.assertEqual(good["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
 class TestInjectableDuplicateFetcher(unittest.TestCase):
    @patch("mcp_server.get_auth_header", return_value="token x")
    def test_lock_issue_uses_injected_fetcher(self, _auth):
        seen = {}
        def fetcher(h, o, r, auth, issue_number):
            seen["issue_number"] = issue_number
            return [], [], {"status": "not_claimed"}
        with patch(
            "mcp_server.issue_duplicate_context_fetcher",
            side_effect=fetcher,
        ), patch(
            "mcp_server.issue_lock_worktree.read_worktree_git_state",
            return_value={
                "current_branch": "master",
                "porcelain_status": "",
                "base_equivalent": True,
            },
        ), patch.dict(os.environ, {
            "GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment",
        }, clear=True):
            with patch.object(mcp_server, "ISSUE_LOCK_FILE", tempfile.mktemp()):
                mcp_server.gitea_lock_issue(
                    issue_number=400,
                    branch_name="feat/issue-400-duplicate-work-preflight",
                    remote="prgs",
                )
        self.assertEqual(seen["issue_number"], 400)
 class TestMcpDuplicateRecheck(unittest.TestCase):
    def setUp(self):
        self._dir = tempfile.TemporaryDirectory()
        self.lock_path = os.path.join(self._dir.name, "gitea_issue_lock.json")
        self._lock_patch = patch.object(
            mcp_server, "ISSUE_LOCK_FILE", self.lock_path
        )
        self._lock_patch.start()
        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()
    def tearDown(self):
        patch.stopall()
        self._dir.cleanup()
    def _write_lock(self, issue_number=400, branch="feat/issue-400-x"):
        with open(self.lock_path, "w", encoding="utf-8") as fh:
            json.dump({
                "issue_number": issue_number,
                "branch_name": branch,
                "remote": "prgs",
            }, fh)
    @patch("mcp_server._assess_issue_duplicate_gate")
    @patch("mcp_server.get_profile", return_value={
        "profile_name": "test-author",
        "allowed_operations": ["gitea.read", "gitea.repo.commit"],
        "forbidden_operations": [],
        "audit_label": "test-author",
    })
    @patch("mcp_server.get_auth_header", return_value="token x")
    def test_commit_files_blocked_on_recheck(self, _auth, _profile, mock_gate):
        self._write_lock()
        mock_gate.return_value = {
            "block": True,
            "reasons": ["open PR #412 already covers issue #400"],
            "outcome": OUTCOME_DUPLICATE_COMMIT_PREVENTED,
            "safe_next_action": "stop",
        }
        mcp_server.record_preflight_check("whoami")
        mcp_server.record_preflight_check("capability", resolved_role="author")
        with patch(
            "mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            return_value=(True, []),
        ):
            result = mcp_server.gitea_commit_files(
                files=[{
                    "operation": "create",
                    "path": "a.txt",
                    "content_plain": "hi",
                }],
                message="test",
                remote="prgs",
            )
        self.assertFalse(result["success"])
        self.assertIn("duplicate_gate", result)
    @patch("mcp_server._assess_issue_duplicate_gate")
    @patch("mcp_server.get_profile", return_value={
        "profile_name": "test-author",
        "allowed_operations": ["gitea.read", "gitea.pr.create"],
        "forbidden_operations": [],
        "audit_label": "test-author",
    })
    @patch("mcp_server.get_auth_header", return_value="token x")
    def test_create_pr_returns_handoff_on_duplicate(self, _auth, _profile, mock_gate):
        self._write_lock()
        mock_gate.return_value = {
            "block": True,
            "reasons": ["open PR #412 already covers issue #400"],
            "outcome": OUTCOME_DUPLICATE_PR_PREVENTED,
            "safe_next_action": "reconciliation handoff",
        }
        mcp_server.record_preflight_check("whoami")
        mcp_server.record_preflight_check("capability", resolved_role="author")
        with patch(
            "mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            return_value=(True, []),
        ):
            result = mcp_server.gitea_create_pr(
                title="feat: x (Closes #400)",
                head="feat/issue-400-x",
                base="master",
                body="Closes #400",
                remote="prgs",
            )
        self.assertFalse(result["success"])
        self.assertIsNone(result.get("number"))
        self.assertIn("duplicate_gate", result)
 if __name__ == "__main__":
    unittest.main()
@@ -87,14 +87,6 @@ def test_reconcile_landed_workflow_contract():
    assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text
    assert "RECOVERY_HANDOFF_ONLY" in text
    assert "resolve_partial_reconciliation_plan" in text
    assert "check_audit_mutation_allowed" in text
    assert "gitea_authorize_reconciliation_cleanup_phase" in text
 def test_audit_reconciliation_verifier_exported():
    from review_proofs import assess_audit_reconciliation_report
    assert callable(assess_audit_reconciliation_report)
 def test_create_issue_workflow_contract():
@@ -101,36 +101,17 @@ ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
 def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
    import issue_lock_provenance
    work_lease = {
        "operation_type": "author_issue_work",
        "issue_number": issue_number,
        "branch": branch_name,
        "claimant": {"username": "test-user", "profile": "test-author"},
        "expires_at": "2999-01-01T00:00:00Z",
    }
    record = {
        "issue_number": issue_number,
        "branch_name": branch_name,
        "remote": "dadeschools",
        "org": "Scaled-Tech-Consulting",
        "repo": "Gitea-Tools",
        "work_lease": work_lease,
        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
            tool="gitea_lock_issue",
            claimant=work_lease.get("claimant"),
        ),
    }
    record.update(overrides)
    return record
 def _clear_duplicate_context_fetcher(*_args, **_kwargs):
    """Default injectable duplicate-work context for lock/create_pr tests."""
    return [], [], {"status": "not_claimed"}
 # ---------------------------------------------------------------------------
 # Create Issue
 # ---------------------------------------------------------------------------
@@ -185,17 +166,13 @@ class TestCreateIssue(unittest.TestCase):
 # ---------------------------------------------------------------------------
 class TestCreatePR(unittest.TestCase):
    @patch(
        "mcp_server.issue_duplicate_context_fetcher",
        return_value=([], [], {"status": "not_claimed"}),
    )
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("os.path.exists", return_value=True)
    @patch("builtins.open")
-    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
+    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
        mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
@@ -210,17 +187,13 @@ class TestCreatePR(unittest.TestCase):
        self.assertEqual(payload["base"], "main")
        self.assertIn("Closes #123", payload["title"])
    @patch(
        "mcp_server.issue_duplicate_context_fetcher",
        return_value=([], [], {"status": "not_claimed"}),
    )
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("os.path.exists", return_value=True)
    @patch("builtins.open")
-    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
+    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
        mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
@@ -3071,14 +3044,8 @@ class TestIssueLocking(unittest.TestCase):
    def setUp(self):
        self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
        self._env_patcher.start()
        self._dup_fetcher_patcher = patch(
            "mcp_server.issue_duplicate_context_fetcher",
            return_value=([], [], {"status": "not_claimed"}),
        )
        self.mock_dup_fetcher = self._dup_fetcher_patcher.start()
    def tearDown(self):
        self._dup_fetcher_patcher.stop()
        self._env_patcher.stop()
        if os.path.exists(ISSUE_LOCK_FILE):
            os.remove(ISSUE_LOCK_FILE)
@@ -3087,8 +3054,10 @@ class TestIssueLocking(unittest.TestCase):
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_success(self, _auth, _git_state):
+    def test_lock_issue_success(self, _auth, mock_api, _git_state):
        mock_api.return_value = [] # no open PRs
        res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
        self.assertTrue(res["success"])
        self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
@@ -3113,48 +3082,50 @@ class TestIssueLocking(unittest.TestCase):
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_reused_by_open_pr_branch(self, _auth, _git_state):
+    def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api, _git_state):
-        self.mock_dup_fetcher.return_value = ([{
+        mock_api.return_value = [{
            "number": 200,
            "head": {"ref": "feat/issue-196-boundary"},
            "title": "Some PR",
-            "body": "No closes ref",
+            "body": "No closes ref"
-        }], [], {"status": "not_claimed"})
+        }]
        with self.assertRaises(ValueError) as ctx:
            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("open PR #200 already covers issue", str(ctx.exception))
+        self.assertIn("already tied to an open PR", str(ctx.exception))
    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, _git_state):
+    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api, _git_state):
-        self.mock_dup_fetcher.return_value = ([{
+        mock_api.return_value = [{
            "number": 200,
            "head": {"ref": "feat/other-branch"},
            "title": "Some PR",
-            "body": "fixes #196",
+            "body": "fixes #196"
-        }], [], {"status": "not_claimed"})
+        }]
        with self.assertRaises(ValueError) as ctx:
            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("open PR #200 already covers issue", str(ctx.exception))
+        self.assertIn("already tied to an open PR", str(ctx.exception))
    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_reused_by_remote_branch(self, _auth, _git_state):
+    def test_lock_issue_reused_by_remote_branch(self, _auth, mock_api, _git_state):
-        self.mock_dup_fetcher.return_value = (
+        mock_api.side_effect = [
            [],
-            ["feat/issue-196-existing-work"],
+            [{"name": "feat/issue-196-existing-work"}],
-            {"status": "not_claimed"},
+        ]
        )
        with self.assertRaises(ValueError) as ctx:
            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("remote branch(es) already match issue pattern", str(ctx.exception))
+        self.assertIn("already has matching branch", str(ctx.exception))
    def test_lock_issue_blocks_active_same_operation_lease(self):
        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
@@ -3313,28 +3284,6 @@ class TestIssueLocking(unittest.TestCase):
                )
        self.assertIn("does not match locked worktree", str(ctx.exception))
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_manual_lock_seed_blocked(self, _auth, _role):
        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
            json.dump(
                _sample_issue_lock(
                    issue_number=447,
                    branch_name="feat/issue-447-lock-provenance",
                    lock_provenance=None,
                ),
                f,
            )
        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
            with self.assertRaises(RuntimeError) as ctx:
                gitea_create_pr(
                    title="feat: lock provenance Closes #447",
                    head="feat/issue-447-lock-provenance",
                    remote="prgs",
                )
        self.assertIn("lock provenance", str(ctx.exception).lower())
    @patch("mcp_server.api_request")
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
@@ -2346,7 +2346,6 @@ class TestWorkIssueFinalReport(unittest.TestCase):
            "- Safe next action: open PR",
            "- Next: open PR",
            "- Safety statement: no review/merge",
            "- Duplicate work outcome: duplicate work not prevented",
        ])
    def test_complete_work_issue_report_earns_a(self):
@@ -12,14 +12,12 @@ from starlette.testclient import TestClient
 from webui.app import create_app
 from webui.queue_loader import (
    PaginationMeta,
    QueueSnapshot,
    _classify_issue,
    _classify_pr,
    _extract_linked_issue,
    load_queue_snapshot,
    snapshot_to_dict,
 )
 from webui.queue_views import render_queue_page
 _RECENT = datetime.now(timezone.utc).isoformat()
 _STALE = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
@@ -211,77 +209,36 @@ class TestQueueRoutes(unittest.TestCase):
        self.assertIsNotNone(snapshot.fetch_error)
        self.assertEqual(len(snapshot.prs), 0)
-
+    def test_queue_fail_closed_hides_empty_state_copy(self):
-def _empty_pagination() -> PaginationMeta:
+        with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
-    return PaginationMeta(
+            snapshot = load_queue_snapshot()
        page=1,
        per_page=50,
        returned_count=0,
        has_more=False,
        is_final_page=True,
        inventory_complete=True,
        pages_fetched=1,
    )
 def _empty_fetch(*_args, **_kwargs):
    return [], _empty_pagination()
 class TestQueueFailClosedUx(unittest.TestCase):
    """Regression tests for #458 fail-closed empty-state copy."""
    def test_fail_closed_view_suppresses_empty_queue_copy(self):
        snapshot = QueueSnapshot(
            project_id="gitea-tools",
            repo_label="Scaled-Tech-Consulting/Gitea-Tools",
            prs=(),
            issues=(),
            pr_pagination=None,
            issue_pagination=None,
            fetch_error="Gitea credentials unavailable for gitea.prgs.cc",
        )
        html = render_queue_page(snapshot)
        self.assertIn("Queue unavailable", html)
        self.assertIn("Not loaded", html)
        self.assertNotIn("No open items.", html)
        self.assertIn("pagination:</strong> unavailable", html)
    def test_fail_closed_route_does_not_show_empty_queue_copy(self):
        client = TestClient(create_app())
        snapshot = load_queue_snapshot(
            fetch_prs=_empty_fetch,
            fetch_issues=_empty_fetch,
        )
        snapshot = QueueSnapshot(
            project_id=snapshot.project_id,
            repo_label=snapshot.repo_label,
            prs=(),
            issues=(),
            pr_pagination=None,
            issue_pagination=None,
            fetch_error="Gitea credentials unavailable for gitea.prgs.cc",
        )
        with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
-            response = client.get("/queue")
+            response = self.client.get("/queue")
        self.assertEqual(response.status_code, 200)
        self.assertIn("Queue unavailable", response.text)
        self.assertIn("not fetched", response.text.lower())
        self.assertNotIn("No open items.", response.text)
        self.assertNotIn("Open pull requests", response.text)
        self.assertNotIn("pages_fetched", response.text)
    def test_queue_successful_empty_inventory_shows_empty_state(self):
        def _empty_prs(*_args, **_kwargs):
            return [], _mock_pagination(0)
        def _empty_issues(*_args, **_kwargs):
            return [], _mock_pagination(0)
    def test_successful_empty_inventory_shows_empty_copy(self):
        snapshot = load_queue_snapshot(
-            fetch_prs=_empty_fetch,
+            fetch_prs=_empty_prs,
-            fetch_issues=_empty_fetch,
+            fetch_issues=_empty_issues,
        )
        with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
            response = self.client.get("/queue")
        self.assertEqual(response.status_code, 200)
        self.assertIsNone(snapshot.fetch_error)
-        self.assertEqual(len(snapshot.prs), 0)
+        self.assertEqual(response.text.count("No open items."), 2)
-        self.assertEqual(len(snapshot.issues), 0)
+        self.assertIn("pages_fetched", response.text)
-        self.assertTrue(snapshot.pr_pagination.inventory_complete)
+        self.assertIn("pagination (complete)", response.text)
        html = render_queue_page(snapshot)
        self.assertNotIn("Queue unavailable", html)
        self.assertEqual(html.count("No open items."), 2)
        self.assertIn("pagination (complete)", html)
 if __name__ == "__main__":
@@ -38,13 +38,7 @@ def _queue_table(
    title: str,
    items: tuple[QueueItem, ...],
    columns: tuple[tuple[str, str], ...],
    fetch_failed: bool = False,
 ) -> str:
    if fetch_failed:
        return (
            f"<h3>{html.escape(title)}</h3>"
            "<p class='muted'>Not loaded — queue fetch did not complete.</p>"
        )
    if not items:
        return f"<h3>{html.escape(title)}</h3><p class='muted'>No open items.</p>"
@@ -78,12 +72,22 @@ def render_queue_page(snapshot: QueueSnapshot) -> str:
            f'<div class="stub"><p><strong>Queue unavailable:</strong> '
            f"{html.escape(snapshot.fetch_error)}</p></div>"
        )
        inventory_block = (
            "<p class='muted'><strong>Inventory:</strong> not fetched "
            "(fail closed — queue tables omitted).</p>"
        )
        return (
            "<h2>Live queue</h2>"
            f"<p class='meta'>Repository: <code>{html.escape(snapshot.repo_label)}</code> "
            f"· project <code>{html.escape(snapshot.project_id)}</code></p>"
            f"{error_block}"
            f"{inventory_block}"
            "<p class='muted'>Read-only MVP — claims, reviews, and merges stay in Gitea/MCP tools.</p>"
        )
    fetch_failed = bool(snapshot.fetch_error)
    pr_section = _queue_table(
        title="Open pull requests",
        items=snapshot.prs,
        fetch_failed=fetch_failed,
        columns=(
            ("number", "#"),
            ("title", "Title"),
@@ -96,7 +100,6 @@ def render_queue_page(snapshot: QueueSnapshot) -> str:
    issue_section = _queue_table(
        title="Open issues",
        items=snapshot.issues,
        fetch_failed=fetch_failed,
        columns=(
            ("number", "#"),
            ("title", "Title"),
@@ -110,7 +113,6 @@ def render_queue_page(snapshot: QueueSnapshot) -> str:
        "<h2>Live queue</h2>"
        f"<p class='meta'>Repository: <code>{html.escape(snapshot.repo_label)}</code> "
        f"· project <code>{html.escape(snapshot.project_id)}</code></p>"
        f"{error_block}"
        f"{_pagination_html('PR', snapshot.pr_pagination)}"
        f"{pr_section}"
        f"{_pagination_html('Issue', snapshot.issue_pagination)}"