Merge branch 'prgs/master' into feat/issue-503-reviewer-active-worktree

fix: resolve conflicts for PR #508
Merge prgs/master into feat/issue-503-reviewer-active-worktree. Preserve reviewer mutation workspace binding (#503) and fold in master's review_pr task preflight checks via _verify_reviewer_mutation_workspace.
2026-07-08 04:43:59 -04:00 · 2026-07-08 04:10:57 -04:00 · 2026-07-08 03:42:07 -04:00
31 changed files with 15 additions and 3817 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, []
@@ -377,37 +377,6 @@ explicit control-checkout repair.
 Portable wording: [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
 ### Root checkout guard (#475)
 The MCP server enforces a fail-closed **root checkout guard** before author,
 reviewer, and merger mutations when the active workspace is not an isolated
 `branches/...` worktree (reconciler close paths remain exempt per #468).
 The guard blocks when the **control checkout** (repository root) is:
 - on a non-stable branch (`master` / `main` / `dev` expected),
 - detached HEAD,
 - dirty (tracked edits),
 - or its `HEAD` does not match `prgs/master` when that ref is available.
 **Remediation (never auto-reset or stash):**
 > Root checkout is not on master. Preserve state, switch root back to master,
 > and use `scripts/worktree-review` or the sanctioned issue worktree flow.
 **Recovery after root hijack:**
 1. Preserve any in-progress edits (copy paths, note branch name, or commit on a
   rescue branch from a `branches/...` worktree).
 2. From the repository root: `git checkout master` (or `main` / `dev` per repo
   policy) and `git fetch prgs && git merge --ff-only prgs/master` when safe.
 3. Confirm `git status` is clean and `git branch --show-current` is `master`.
 4. Resume work only inside `branches/issue-<n>-<slug>` via `gitea_lock_issue` /
   `git worktree add`.
 `branches/...` directories are disposable role worktrees; the root checkout is
 the stable orchestration surface only.
 ## Shell Spawn Hard-Stop Rule
 Symptom: a shell tool call returns `exit_code: -1` with empty stdout/stderr.
@@ -118,22 +118,6 @@ _TARGET_BRANCH_SHA_RE = re.compile(
    r"target branch sha\s*:\s*[0-9a-f]{40}",
    re.IGNORECASE,
 )
 _WORKFLOW_LOAD_HELPER_RE = re.compile(
    r"workflow[- ]load helper result\s*:",
    re.IGNORECASE,
 )
 _WORKFLOW_LOAD_HASH_RE = re.compile(
    r"workflow[- ]load helper result[\s\S]{0,400}?workflow[_ ]hash\s*:\s*[0-9a-f]{12}",
    re.IGNORECASE,
 )
 _WORKFLOW_LOAD_BOUNDARY_RE = re.compile(
    r"workflow[- ]load helper result[\s\S]{0,400}?boundary[_ ]status\s*:\s*(?:clean|violation)",
    re.IGNORECASE,
 )
 _WORKFLOW_FILE_VIEW_NARRATIVE_RE = re.compile(
    r"(?:read|viewed|loaded)\s+(?:the\s+)?(?:canonical\s+)?(?:workflow|review-merge-pr\.md)",
    re.IGNORECASE,
 )
 _FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
 _RECONCILE_STALE_FIELDS = (
    "pr number opened",
@@ -1040,70 +1024,6 @@ def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, st
    )
 def _rule_reviewer_workflow_load_boundary(report_text: str) -> list[dict[str, str]]:
    """#403: require structured workflow-load helper result, not file-view narrative."""
    if not report_text.strip():
        return []
    findings: list[dict[str, str]] = []
    has_helper = bool(_WORKFLOW_LOAD_HELPER_RE.search(report_text))
    has_hash = bool(_WORKFLOW_LOAD_HASH_RE.search(report_text))
    has_boundary = bool(_WORKFLOW_LOAD_BOUNDARY_RE.search(report_text))
    has_narrative_only = bool(_WORKFLOW_FILE_VIEW_NARRATIVE_RE.search(report_text))
    if has_narrative_only and not has_helper:
        findings.append(validator_finding(
            "reviewer.workflow_load_boundary",
            "block",
            "Workflow-load helper result",
            (
                "canonical workflow file-view narrative without structured "
                "gitea_load_review_workflow helper result"
            ),
            (
                "include Workflow-load helper result with workflow_hash and "
                "boundary_status from gitea_load_review_workflow"
            ),
        ))
        return findings
    if has_helper and (not has_hash or not has_boundary):
        missing = []
        if not has_hash:
            missing.append("workflow_hash")
        if not has_boundary:
            missing.append("boundary_status")
        findings.append(validator_finding(
            "reviewer.workflow_load_boundary",
            "block",
            "Workflow-load helper result",
            (
                "workflow-load helper result incomplete; missing "
                + ", ".join(missing)
            ),
            (
                "copy workflow_load_helper_result fields from "
                "gitea_load_review_workflow into the final report"
            ),
        ))
    return findings
 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,
    *,
@@ -1164,7 +1084,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
        _rule_reviewer_already_landed_eligible,
        _rule_reviewer_already_landed_state,
        _rule_reviewer_target_branch_freshness,
        _rule_reviewer_workflow_load_boundary,
        _rule_reviewer_mutation_ledger,
        _rule_reviewer_review_mutation,
        _rule_reviewer_post_merge_cleanup_proof,
@@ -1181,7 +1100,6 @@ _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,
@@ -651,7 +651,6 @@ def verify_preflight_purity(
                    f"{_format_preflight_files(reviewer_delta)}"
                )
    _enforce_root_checkout_guard(worktree_path)
    _enforce_branches_only_author_mutation(worktree_path)
    _clear_preflight_capability_state()
@@ -694,27 +693,6 @@ def _verify_role_mutation_workspace(
    verify_preflight_purity(remote, worktree_path=resolved, task=task)
    return resolved
 def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None:
    """#475: fail closed when the stable control checkout is contaminated."""
    ctx = _resolve_author_mutation_context(worktree_path)
    canonical_root = ctx["canonical_repo_root"]
    workspace = ctx["workspace_path"]
    git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
    remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
    assessment = root_checkout_guard.assess_root_checkout_guard(
        workspace_path=workspace,
        canonical_repo_root=canonical_root,
        current_branch=git_state.get("current_branch"),
        head_sha=git_state.get("head_sha"),
        porcelain_status=git_state.get("porcelain_status") or "",
        remote_master_sha=remote_master_sha,
        resolved_role=_preflight_resolved_role,
    )
    if assessment["block"]:
        raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment))
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 from gitea_auth import (  # noqa: E402
@@ -736,8 +714,6 @@ import role_session_router  # noqa: E402
 import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
 import review_workflow_boundary  # noqa: E402
 import review_workflow_load  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
 import issue_lock_provenance  # noqa: E402
@@ -747,18 +723,15 @@ import stacked_pr_support  # noqa: E402
 import merge_approval_gate  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import root_checkout_guard  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import reviewer_pr_lease  # 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 pr_work_lease  # noqa: E402
 import native_mcp_preference  # noqa: E402
 import worktree_cleanup_audit  # noqa: E402
 # Keyed issue-lock storage (#443): per remote/org/repo/issue files under
@@ -1473,6 +1446,7 @@ def gitea_create_issue(
    return _with_optional_url({"number": data["number"]}, data.get("html_url"))
@mcp.tool()
 def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]:
    """Fetch all OPEN pull requests for a repo (used for stacked-base proof, #484)."""
    try:
@@ -1483,7 +1457,6 @@ def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]:
        )
@mcp.tool()
 def gitea_lock_issue(
    issue_number: int,
    branch_name: str,
@@ -2411,7 +2384,6 @@ def init_review_decision_lock(remote: str | None, task: str | None):
    """Seed read-only-until-ready state for reviewer PR review tasks."""
    if task != "review_pr":
        return
    review_workflow_load.clear_review_workflow_load()
    profile = get_profile()
    profile_name = (profile.get("profile_name") or "").strip()
    session_lock = (
@@ -2438,11 +2410,6 @@ def init_review_decision_lock(remote: str | None, task: str | None):
    })
 def _review_workflow_load_gate_reasons() -> list[str]:
    """Fail closed when canonical review workflow was not loaded (#389)."""
    return review_workflow_load.review_workflow_load_blockers(PROJECT_ROOT)
 def check_review_decision_gate(
    pr_number: int,
    action: str,
@@ -2453,10 +2420,7 @@ def check_review_decision_gate(
    repo: str | None = None,
 ) -> list[str]:
    """Fail closed unless validation completed and the final decision is ready."""
-    reasons = list(_review_workflow_load_gate_reasons())
+    reasons = []
    if reasons:
        reasons.extend(review_workflow_load.recovery_handoff_without_replay())
        return reasons
    lock = _load_review_decision_lock()
    if lock is None:
        reasons.append(
@@ -2871,7 +2835,6 @@ def _evaluate_pr_review_submission(
        remote, worktree_path=worktree_path, task="review_pr"
    )
    action = (action or "").strip().lower()
    workflow_blockers = _review_workflow_load_gate_reasons() if live else []
    result = {
        "requested_action": action,
        "performed": False,
@@ -2887,10 +2850,6 @@ def _evaluate_pr_review_submission(
        "reasons": [],
    }
    reasons = result["reasons"]
    if workflow_blockers:
        reasons.extend(workflow_blockers)
        reasons.extend(review_workflow_load.recovery_handoff_without_replay())
        return result
    if action not in _REVIEW_ACTIONS:
        reasons.append(
@@ -3109,13 +3068,6 @@ def gitea_mark_final_review_decision(
        }
    org = resolved_org
    repo = resolved_repo
    workflow_blockers = _review_workflow_load_gate_reasons()
    if workflow_blockers:
        return {
            "marked_ready": False,
            "reasons": workflow_blockers + (
                review_workflow_load.recovery_handoff_without_replay()),
        }
    hard_stop = terminal_review_hard_stop_reasons(pr_number, "mark_ready")
    if hard_stop:
        return {"marked_ready": False, "reasons": hard_stop}
@@ -3751,7 +3703,6 @@ def gitea_merge_pr(
    _verify_role_mutation_workspace(
        remote, worktree_path=worktree_path, task="merge_pr"
    )
    workflow_blockers = _review_workflow_load_gate_reasons()
    do = (do or "").strip().lower()
    result = {
        "performed": False,
@@ -3769,10 +3720,6 @@ def gitea_merge_pr(
        "reasons": [],
    }
    reasons = result["reasons"]
    if workflow_blockers:
        reasons.extend(workflow_blockers)
        reasons.extend(review_workflow_load.recovery_handoff_without_replay())
        return result
    # Gate 1 — valid merge method (no API call on a bad method).
    if do not in _MERGE_METHODS:
@@ -4284,18 +4231,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, task="delete_branch")
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
@@ -4365,18 +4300,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)
@@ -4459,53 +4382,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,
@@ -4678,90 +4554,6 @@ def gitea_scan_already_landed_open_prs(
    }
@mcp.tool()
 def gitea_audit_worktree_cleanup(
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    ttl_hours: float = worktree_cleanup_audit.DEFAULT_TTL_HOURS,
 ) -> dict:
    """Read-only: classify every session-owned worktree under ``branches/`` (#401).
    Audits the local ``branches/`` directory, classifying each worktree as
    active open PR, active issue work, dirty local, clean stale removable,
    detached review leftover, or unsafe/unknown. Open PR branch heads are
    fetched live so a worktree tied to an open PR is never marked removable;
    the active issue-lock branch is read from the local lock file and treated
    as active work. Deletes nothing and mutates no Gitea state.
    Fails closed if the live open-PR list cannot be fetched: without it,
    removability cannot be proven, so no candidates are returned.
    Args:
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.
        ttl_hours: Age (hours) after which a clean issue/conflict-fix
            worktree becomes stale-removable (default from
            GITEA_WORKTREE_TTL_HOURS).
    Returns:
        dict with per-worktree classifications, counts, removable
        candidates, and the ``git worktree list`` verification proof.
    """
    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"),
        }
    try:
        h, o, r = _resolve(remote, host, org, repo)
        auth = _auth(h)
        open_prs = api_get_all(f"{repo_api_url(h, o, r)}/pulls?state=open", auth)
    except Exception as exc:
        return {
            "success": False,
            "performed": False,
            "open_pr_state_verified": False,
            "reasons": [
                "could not fetch live open PRs; removability unverified "
                f"(fail closed): {_redact(str(exc))}"
            ],
        }
    open_pr_branches = {
        str((pr.get("head") or {}).get("ref"))
        for pr in open_prs
        if (pr.get("head") or {}).get("ref")
    }
    active_issue_branches: set[str] = set()
    lock = merged_cleanup_reconcile.read_issue_lock(ISSUE_LOCK_FILE)
    if lock and lock.get("branch_name"):
        active_issue_branches.add(str(lock["branch_name"]).strip())
    report = worktree_cleanup_audit.audit_branches_directory(
        PROJECT_ROOT,
        open_pr_branches=open_pr_branches,
        active_issue_branches=active_issue_branches,
        now=datetime.now(timezone.utc),
        ttl_hours=ttl_hours,
    )
    return {
        "success": True,
        "performed": False,
        "open_pr_state_verified": True,
        "task_mode": "work-issue",
        **report,
    }
@mcp.tool()
 def gitea_reconcile_already_landed_pr(
    pr_number: int,
@@ -5403,13 +5195,6 @@ def gitea_acquire_reviewer_pr_lease(
    comments = _fetch_pr_comments(
        pr_number, remote=remote, host=host, org=org, repo=repo)
    # Refuse merge-oriented lease acquisition/adoption on an already-merged or
    # closed PR: the lease is moot and adopting it for merge work is unsafe (#515).
    pr_live = api_request(
        "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
    pr_merged_or_closed = bool(
        pr_live.get("merged") or pr_live.get("merged_at")
    ) or (str(pr_live.get("state") or "").strip().lower() == "closed")
    assessment = reviewer_pr_lease.assess_acquire_lease(
        comments,
        pr_number=pr_number,
@@ -5422,7 +5207,6 @@ def gitea_acquire_reviewer_pr_lease(
        candidate_head=candidate_head,
        target_branch=target_branch,
        target_branch_sha=target_branch_sha,
        pr_merged_or_closed=pr_merged_or_closed,
    )
    if not assessment.get("acquire_allowed"):
        return {
@@ -5430,7 +5214,6 @@ def gitea_acquire_reviewer_pr_lease(
            "acquired": False,
            "reasons": assessment.get("reasons") or [],
            "existing_lease": assessment.get("existing_lease"),
            "post_merge_moot": assessment.get("post_merge_moot", False),
        }
    body = assessment["lease_body"]
@@ -5579,126 +5362,6 @@ def gitea_assess_reviewer_pr_lease(
    }
@mcp.tool()
 def gitea_cleanup_post_merge_moot_lease(
    pr_number: int,
    apply: bool = False,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
 ) -> dict:
    """Safely resolve a moot reviewer lease left on an ALREADY-MERGED/closed PR (#515).
    Read-first and fail-safe. This tool never merges and never adopts a lease.
    It only acts when the live PR state is merged/closed, and it never steals or
    force-cleans an active *foreign* lease while the PR is still open. When
    ``apply`` is true and a lease is still active on a merged/closed PR, it posts
    a terminal ``phase: released`` lease marker (``blocker: post-merge-moot``) —
    an append-only comment that neutralises the moot lease without deleting any
    other session's comment.
    Args:
        pr_number: The PR whose lingering lease to assess/clean.
        apply: When false (default) report only (read-only). When true, post the
            terminal released marker if — and only if — cleanup is allowed.
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.
    Returns:
        dict reporting PR merged/closed state, merge_commit_sha, linked-issue
        closure state, whether the lease is moot, whether cleanup was performed
        or skipped (and why), and ``no_merge_or_adoption`` True — this path never
        merges or adopts.
    """
    read_block = _profile_operation_gate("gitea.read")
    if read_block:
        return {
            "success": False,
            "cleanup_performed": False,
            "no_merge_or_adoption": True,
            "reasons": read_block,
            "permission_report": _permission_block_report("gitea.read"),
        }
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    pr_live = api_request(
        "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
    comments = _fetch_pr_comments(
        pr_number, remote=remote, host=host, org=org, repo=repo)
    assessment = reviewer_pr_lease.assess_post_merge_moot_lease(
        comments,
        pr_number=pr_number,
        pr_merged=bool(pr_live.get("merged") or pr_live.get("merged_at")),
        pr_state=pr_live.get("state"),
        merge_commit_sha=pr_live.get("merge_commit_sha"),
    )
    # Best-effort linked-issue closure state for the report.
    active = assessment.get("active_lease") or {}
    issue_no = active.get("issue_number")
    linked_issue_state = None
    if issue_no:
        try:
            issue = api_request(
                "GET", f"{repo_api_url(h, o, r)}/issues/{issue_no}", auth) or {}
            linked_issue_state = issue.get("state")
        except Exception:
            linked_issue_state = None
    report = {
        "success": True,
        "pr_number": pr_number,
        "pr_state": assessment.get("pr_state"),
        "pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
        "merge_commit_sha": assessment.get("merge_commit_sha"),
        "linked_issue_number": issue_no,
        "linked_issue_state": linked_issue_state,
        "lease_moot": assessment.get("is_moot"),
        "active_lease": assessment.get("active_lease"),
        "cleanup_allowed": assessment.get("cleanup_allowed"),
        "cleanup_performed": False,
        "no_merge_or_adoption": True,
        "mode": "apply" if apply else "read_only",
        "reasons": assessment.get("reasons") or [],
    }
    if not apply:
        return report
    if not assessment.get("cleanup_allowed"):
        report["cleanup_skipped_reason"] = (
            assessment.get("reasons") or ["cleanup not allowed"]
        )
        return report
    comment_block = _profile_operation_gate("gitea.pr.comment")
    if comment_block:
        report["success"] = False
        report["reasons"] = comment_block
        report["permission_report"] = _permission_block_report("gitea.pr.comment")
        return report
    verify_preflight_purity(remote)
    body = assessment["release_body"]
    comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
    with _audited(
        "comment_pr",
        host=h,
        remote=remote,
        org=o,
        repo=r,
        pr_number=pr_number,
        request_metadata={"source": "cleanup_post_merge_moot_lease"},
    ):
        posted = api_request("POST", comment_url, auth, {"body": body})
    report["cleanup_performed"] = True
    report["released_comment_id"] = posted.get("id")
    return report
@mcp.tool()
 def gitea_list_issue_comments(
    issue_number: int,
@@ -6098,8 +5761,6 @@ _PROJECT_SKILLS = {
        "steps": [
            "Resolve task first: gitea_resolve_task_capability(task='review_pr') "
            "to confirm reviewer namespace and avoid author-profile blocks.",
            "Load canonical workflow proof with gitea_load_review_workflow "
            "before any review/merge mutation (#389).",
            "Verify reviewer identity with gitea_whoami; the PR author "
            "must be a different user.",
            "Reconcile live queue state FIRST (do not trust prior handoffs): "
@@ -7023,8 +6684,6 @@ def gitea_get_runtime_context(
        ),
        "role_kind": _role_kind(allowed, forbidden),
        "shell_health": native_mcp_preference.shell_health_status(),
        "workflow_load_proof": review_workflow_load.workflow_load_status(
            PROJECT_ROOT),
    }
    if reveal and h:
@@ -7033,80 +6692,6 @@ def gitea_get_runtime_context(
    return result
@mcp.tool()
 def gitea_record_pre_review_command(
    command: str,
    cwd: str | None = None,
    classification: str | None = None,
 ) -> dict:
    """Classify and record a command executed before workflow load (#403).
    Read-only with respect to Gitea API. Pre-review inventory/diagnostic commands
    may be recorded as allowed; boundary violations block reviewer mutations.
    """
    recorded = review_workflow_boundary.record_pre_review_command(
        command,
        cwd=cwd,
        project_root=PROJECT_ROOT,
        classification=classification,
    )
    boundary_state = review_workflow_boundary.assess_boundary_status(PROJECT_ROOT)
    return {
        "success": True,
        "recorded": recorded,
        "boundary_status": boundary_state.get("boundary_status"),
        "boundary_clean": boundary_state.get("boundary_clean"),
        "reasons": list(boundary_state.get("reasons") or []),
    }
@mcp.tool()
 def gitea_load_review_workflow(
    prompt_text: str | None = None,
 ) -> dict:
    """Load and record canonical review-merge workflow proof for this session (#389, #403).
    Read-only with respect to Gitea API; records in-process workflow source/hash
    proof and session boundary state required before reviewer review or merge
    mutations.
    """
    try:
        recorded = review_workflow_load.record_review_workflow_load(
            PROJECT_ROOT, prompt_text=prompt_text)
    except OSError as exc:
        return {
            "success": False,
            "loaded": False,
            "reasons": [str(exc)],
            "recovery_handoff": review_workflow_load.recovery_handoff_without_replay(),
        }
    boundary_reasons = review_workflow_boundary.boundary_blockers(PROJECT_ROOT)
    helper = review_workflow_boundary.workflow_load_helper_result(
        recorded, PROJECT_ROOT)
    return {
        "success": not boundary_reasons,
        "loaded": True,
        "workflow_source": recorded["workflow_source"],
        "task_mode": recorded["task_mode"],
        "workflow_hash": recorded["workflow_hash"],
        "workflow_version": recorded["workflow_version"],
        "final_report_schema_path": recorded["final_report_schema_path"],
        "final_report_schema_hash": recorded["final_report_schema_hash"],
        "prompt_conflicts_with_workflow": recorded[
            "prompt_conflicts_with_workflow"],
        "prompt_conflict_reasons": recorded.get("prompt_conflict_reasons") or [],
        "workflow_load_proof_present": True,
        "boundary_status": recorded.get("boundary_status"),
        "boundary_clean": recorded.get("boundary_clean"),
        "workflow_load_helper_result": helper,
        "reasons": boundary_reasons,
        "recovery_handoff": (
            review_workflow_load.recovery_handoff_without_replay()
            if boundary_reasons else []
        ),
    }
@mcp.tool()
 def gitea_list_profiles() -> dict:
    """Read-only: list all Gitea MCP profiles with redacted metadata.
@@ -8275,20 +7860,7 @@ def gitea_resolve_task_capability(
    }
    if reason_msg:
        result["reason"] = reason_msg
    if task in ("review_pr", "merge_pr"):
        result["workflow_load_proof"] = review_workflow_load.workflow_load_status(
            PROJECT_ROOT)
        if not result["workflow_load_proof"].get("workflow_load_valid"):
            guidance = (
                "Call gitea_load_review_workflow before any reviewer review "
                "or merge mutation."
            )
            if guidance not in task_role_guidance:
                task_role_guidance.append(guidance)
    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:
@@ -5115,15 +5115,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(
@@ -1,259 +0,0 @@
 """Reviewer session boundary tracking for workflow-load gate (#403).
 Pre-review commands executed before ``gitea_load_review_workflow`` must be
 classified. Boundary violations block downstream reviewer mutations even when
 workflow hash proof is present.
 """
 from __future__ import annotations
 import os
 import re
 from typing import Any
 CLASSIFICATION_READ_ONLY_INVENTORY = "read_only_inventory"
 CLASSIFICATION_DIAGNOSTIC = "diagnostic"
 CLASSIFICATION_BOUNDARY_VIOLATION = "boundary_violation"
 CLASSIFICATION_UNCLASSIFIED = "unclassified"
 ALLOWED_CLASSIFICATIONS = frozenset({
    CLASSIFICATION_READ_ONLY_INVENTORY,
    CLASSIFICATION_DIAGNOSTIC,
    CLASSIFICATION_BOUNDARY_VIOLATION,
    CLASSIFICATION_UNCLASSIFIED,
 })
 _PRE_REVIEW_COMMANDS: list[dict[str, Any]] = []
 _READ_ONLY_INVENTORY_PATTERNS = (
    re.compile(
        r"\bgitea[_-](?:list|view|whoami|get[-_]|resolve[-_]task|check[-_]pr|route[-_]task)",
        re.I,
    ),
    re.compile(r"\bgit\s+(?:fetch|remote\s+update|branch\s+-a|log|show|rev-parse)\b", re.I),
    re.compile(r"\bgit\s+status\b", re.I),
    re.compile(r"\bgit\s+worktree\s+list\b", re.I),
 )
 _DIAGNOSTIC_PATTERNS = (
    re.compile(r"\bgit\s+diff(?:\s+--stat)?\b", re.I),
    re.compile(r"\bwhich\s+pytest\b", re.I),
    re.compile(r"\bpytest\s+--version\b", re.I),
 )
 _BOUNDARY_VIOLATION_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
    (re.compile(r"\b(?:pytest|python\s+-m\s+pytest|python\s+-m\s+unittest)\b", re.I),
     "validation command before workflow load"),
    (re.compile(r"\bprofiles\.json\b", re.I), "local profile config inspection"),
    (re.compile(r"\bgitea-mcp(?:\.v2-contexts)?\.json\b", re.I),
     "local Gitea MCP config inspection"),
    (re.compile(r"\b\.env(?:\.|$|\b)", re.I), "credential file inspection"),
    (re.compile(r"\bkeychain\b", re.I), "credential store inspection"),
    (re.compile(r"\bpkill\b", re.I), "MCP repair activity"),
    (re.compile(r"\b(?:edit|write|modify).{0,40}\bmcp\b", re.I),
     "MCP config exploration"),
    (re.compile(r"\bgit\s+(?:add|commit|reset|clean|checkout|merge|rebase|push)\b", re.I),
     "git mutation before workflow load"),
 )
 def clear_pre_review_commands() -> None:
    """Test helper and session reset."""
    global _PRE_REVIEW_COMMANDS
    _PRE_REVIEW_COMMANDS = []
 def pre_review_commands() -> list[dict[str, Any]]:
    """Return a shallow copy of recorded pre-review commands."""
    return [dict(entry) for entry in _PRE_REVIEW_COMMANDS]
 def _normalize_path(path: str | None) -> str:
    return os.path.realpath(os.path.abspath((path or "").strip() or os.getcwd()))
 def is_main_checkout_path(cwd: str | None, project_root: str | None) -> bool:
    """True when *cwd* is the stable control checkout (not under branches/)."""
    if not project_root:
        return False
    root = _normalize_path(project_root)
    path = _normalize_path(cwd)
    if path != root:
        return False
    marker = f"{os.sep}branches{os.sep}"
    return marker not in path
 def classify_pre_review_command(
    command: str,
    *,
    cwd: str | None = None,
    project_root: str | None = None,
 ) -> dict[str, Any]:
    """Classify a command executed before workflow load."""
    text = (command or "").strip()
    path = _normalize_path(cwd)
    root = _normalize_path(project_root) if project_root else None
    reasons: list[str] = []
    for pattern, label in _BOUNDARY_VIOLATION_PATTERNS:
        if pattern.search(text):
            if label.startswith("validation") and root and not is_main_checkout_path(path, root):
                continue
            if label.startswith("git mutation") and root and not is_main_checkout_path(path, root):
                continue
            reasons.append(label)
            return {
                "command": text,
                "cwd": path,
                "classification": CLASSIFICATION_BOUNDARY_VIOLATION,
                "reasons": reasons,
            }
    for pattern in _READ_ONLY_INVENTORY_PATTERNS:
        if pattern.search(text):
            return {
                "command": text,
                "cwd": path,
                "classification": CLASSIFICATION_READ_ONLY_INVENTORY,
                "reasons": [],
            }
    for pattern in _DIAGNOSTIC_PATTERNS:
        if pattern.search(text):
            return {
                "command": text,
                "cwd": path,
                "classification": CLASSIFICATION_DIAGNOSTIC,
                "reasons": [],
            }
    if root and is_main_checkout_path(path, root):
        if re.search(r"\b(?:cat|head|less|read)\b", text, re.I):
            if re.search(r"workflow|skill|runbook", text, re.I):
                return {
                    "command": text,
                    "cwd": path,
                    "classification": CLASSIFICATION_BOUNDARY_VIOLATION,
                    "reasons": [
                        "canonical workflow viewed as local file without "
                        "gitea_load_review_workflow (narrative load is not proof)"
                    ],
                }
    return {
        "command": text,
        "cwd": path,
        "classification": CLASSIFICATION_UNCLASSIFIED,
        "reasons": [
            "pre-review command not classified; record via "
            "gitea_record_pre_review_command before workflow load"
        ],
    }
 def record_pre_review_command(
    command: str,
    *,
    cwd: str | None = None,
    project_root: str | None = None,
    classification: str | None = None,
 ) -> dict[str, Any]:
    """Record and classify a pre-review command for the current session."""
    assessed = classify_pre_review_command(
        command, cwd=cwd, project_root=project_root)
    if classification:
        if classification not in ALLOWED_CLASSIFICATIONS:
            assessed["classification"] = CLASSIFICATION_UNCLASSIFIED
            assessed["reasons"] = [
                f"unknown classification '{classification}'; fail closed"
            ]
        else:
            assessed["classification"] = classification
            assessed["reasons"] = []
    entry = {
        **assessed,
        "session_pid": os.getpid(),
    }
    _PRE_REVIEW_COMMANDS.append(entry)
    return dict(entry)
 def assess_boundary_status(project_root: str | None = None) -> dict[str, Any]:
    """Summarize pre-review boundary state for session proof and reports."""
    violations = [
        entry for entry in _PRE_REVIEW_COMMANDS
        if entry.get("classification") == CLASSIFICATION_BOUNDARY_VIOLATION
    ]
    unclassified = [
        entry for entry in _PRE_REVIEW_COMMANDS
        if entry.get("classification") == CLASSIFICATION_UNCLASSIFIED
    ]
    reasons: list[str] = []
    for entry in violations:
        reasons.extend(entry.get("reasons") or [
            f"boundary violation: {entry.get('command', '')[:80]}"
        ])
    for entry in unclassified:
        reasons.extend(entry.get("reasons") or [
            "unclassified pre-review command blocks reviewer mutations"
        ])
    clean = not reasons
    return {
        "boundary_status": "clean" if clean else "violation",
        "boundary_clean": clean,
        "pre_review_command_count": len(_PRE_REVIEW_COMMANDS),
        "boundary_violation_count": len(violations),
        "unclassified_command_count": len(unclassified),
        "violations": [
            {
                "command": v.get("command"),
                "cwd": v.get("cwd"),
                "reasons": list(v.get("reasons") or []),
            }
            for v in violations
        ],
        "reasons": reasons,
    }
 def boundary_blockers(project_root: str | None = None) -> list[str]:
    """Reasons reviewer mutations must fail closed due to boundary state."""
    status = assess_boundary_status(project_root)
    if status.get("boundary_clean"):
        return []
    return list(status.get("reasons") or [
        "reviewer session boundary violation before workflow load"
    ])
 def workflow_load_helper_result(
    load: dict | None,
    project_root: str | None = None,
 ) -> dict[str, Any]:
    """Structured helper result for final reports (#403)."""
    boundary = assess_boundary_status(project_root)
    if load is None:
        return {
            "workflow_load_proof_present": False,
            "workflow_source": None,
            "workflow_hash": None,
            "final_report_schema_hash": None,
            "boundary_status": boundary.get("boundary_status"),
            "boundary_clean": False,
            "reasons": [
                "gitea_load_review_workflow helper result missing from report"
            ],
        }
    return {
        "workflow_load_proof_present": True,
        "workflow_source": load.get("workflow_source"),
        "workflow_hash": load.get("workflow_hash"),
        "final_report_schema_path": load.get("final_report_schema_path"),
        "final_report_schema_hash": load.get("final_report_schema_hash"),
        "boundary_status": load.get("boundary_status", boundary.get("boundary_status")),
        "boundary_clean": bool(load.get("boundary_clean", boundary.get("boundary_clean"))),
        "pre_review_command_count": boundary.get("pre_review_command_count"),
        "reasons": [],
    }
@@ -1,217 +0,0 @@
 """Canonical review-merge workflow load proof for reviewer mutations (#389, #403)."""
 from __future__ import annotations
 import hashlib
 import os
 import re
 from pathlib import Path
 import review_workflow_boundary as boundary
 WORKFLOW_REL_PATH = (
    "skills/llm-project-workflow/workflows/review-merge-pr.md"
 )
 SCHEMA_REL_PATH = (
    "skills/llm-project-workflow/schemas/review-merge-final-report.md"
 )
 TASK_MODE = "review-merge-pr"
 LOAD_TOOL_NAME = "gitea_load_review_workflow"
 _REVIEW_WORKFLOW_LOAD: dict | None = None
 def compute_content_hash(text: str) -> str:
    """Short deterministic hash for workflow/schema version proof."""
    return hashlib.sha256((text or "").encode("utf-8")).hexdigest()[:12]
 def _read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")
 def _canonical_paths(project_root: str) -> tuple[Path, Path]:
    root = Path(project_root)
    workflow = root / WORKFLOW_REL_PATH
    schema = root / SCHEMA_REL_PATH
    if not workflow.is_file():
        raise FileNotFoundError(f"canonical workflow missing: {workflow}")
    if not schema.is_file():
        raise FileNotFoundError(f"final report schema missing: {schema}")
    return workflow, schema
 def build_canonical_workflow_metadata(
    project_root: str,
    *,
    prompt_text: str | None = None,
 ) -> dict:
    """Load workflow + schema from disk and compute proof metadata."""
    workflow_path, schema_path = _canonical_paths(project_root)
    workflow_text = _read_text(workflow_path)
    schema_text = _read_text(schema_path)
    workflow_hash = compute_content_hash(workflow_text)
    schema_hash = compute_content_hash(schema_text)
    conflict, conflict_reasons = assess_prompt_conflict(prompt_text)
    return {
        "workflow_source": WORKFLOW_REL_PATH,
        "workflow_path": str(workflow_path),
        "task_mode": TASK_MODE,
        "workflow_hash": workflow_hash,
        "workflow_version": workflow_hash,
        "final_report_schema_path": SCHEMA_REL_PATH,
        "final_report_schema_hash": schema_hash,
        "prompt_conflicts_with_workflow": conflict,
        "prompt_conflict_reasons": conflict_reasons,
        "load_tool": LOAD_TOOL_NAME,
    }
 def assess_prompt_conflict(prompt_text: str | None) -> tuple[bool, list[str]]:
    """Detect obvious task-mode conflicts between prompt and review workflow."""
    if not (prompt_text or "").strip():
        return False, []
    text = prompt_text.lower()
    reasons: list[str] = []
    conflicting = (
        (r"\bwork[- ]issue\b", "work-issue author mode"),
        (r"\bcreate[- ]issue\b", "create-issue mode"),
        (r"\bauthor/coder\b", "author/coder mode"),
        (r"\breconcile[- ]landed\b", "reconcile-landed mode"),
    )
    for pattern, label in conflicting:
        if re.search(pattern, text):
            reasons.append(
                f"active prompt appears to request {label} while loading "
                f"{TASK_MODE} workflow"
            )
    return bool(reasons), reasons
 def record_review_workflow_load(
    project_root: str,
    *,
    prompt_text: str | None = None,
 ) -> dict:
    """Record in-process workflow load proof for the current MCP session."""
    global _REVIEW_WORKFLOW_LOAD
    meta = build_canonical_workflow_metadata(
        project_root, prompt_text=prompt_text)
    boundary_state = boundary.assess_boundary_status(project_root)
    _REVIEW_WORKFLOW_LOAD = {
        **meta,
        "session_pid": os.getpid(),
        "loaded": True,
        "boundary_status": boundary_state.get("boundary_status"),
        "boundary_clean": boundary_state.get("boundary_clean"),
        "pre_review_command_count": boundary_state.get("pre_review_command_count"),
        "boundary_violation_count": boundary_state.get("boundary_violation_count"),
        "boundary_reasons": list(boundary_state.get("reasons") or []),
    }
    return dict(_REVIEW_WORKFLOW_LOAD)
 def clear_review_workflow_load() -> None:
    """Test helper and review_pr session reset."""
    global _REVIEW_WORKFLOW_LOAD
    _REVIEW_WORKFLOW_LOAD = None
    boundary.clear_pre_review_commands()
 def workflow_load_status(project_root: str | None = None) -> dict:
    """Non-throwing status for capability/runtime reports."""
    load = _REVIEW_WORKFLOW_LOAD
    if load is None:
        return {
            "workflow_load_proof_present": False,
            "workflow_load_valid": False,
            "workflow_source": None,
            "workflow_hash": None,
            "final_report_schema_path": SCHEMA_REL_PATH,
            "reasons": [
                f"{LOAD_TOOL_NAME} has not been called in this session "
                "(fail closed for reviewer mutations)"
            ],
        }
    reasons = _session_validation_reasons(load, project_root)
    boundary_reasons = boundary.boundary_blockers(project_root)
    if boundary_reasons:
        reasons = list(reasons) + boundary_reasons
    return {
        "workflow_load_proof_present": True,
        "workflow_load_valid": not reasons,
        "workflow_source": load.get("workflow_source"),
        "workflow_hash": load.get("workflow_hash"),
        "task_mode": load.get("task_mode"),
        "final_report_schema_path": load.get("final_report_schema_path"),
        "final_report_schema_hash": load.get("final_report_schema_hash"),
        "prompt_conflicts_with_workflow": load.get(
            "prompt_conflicts_with_workflow"),
        "session_pid": load.get("session_pid"),
        "boundary_status": load.get("boundary_status"),
        "boundary_clean": load.get("boundary_clean"),
        "workflow_load_helper_result": boundary.workflow_load_helper_result(
            load, project_root),
        "reasons": reasons,
    }
 def _session_validation_reasons(
    load: dict,
    project_root: str | None,
 ) -> list[str]:
    reasons: list[str] = []
    if load.get("session_pid") != os.getpid():
        reasons.append(
            "workflow load proof was recorded in a different process "
            "(fail closed)"
        )
        return reasons
    if load.get("prompt_conflicts_with_workflow"):
        reasons.extend(load.get("prompt_conflict_reasons") or [
            "active prompt conflicts with loaded review-merge workflow"
        ])
    if project_root:
        try:
            current = build_canonical_workflow_metadata(project_root)
        except OSError as exc:
            reasons.append(f"cannot re-verify workflow hash: {exc}")
            return reasons
        if current["workflow_hash"] != load.get("workflow_hash"):
            reasons.append(
                "stored workflow hash is stale; reload via "
                f"{LOAD_TOOL_NAME} (fail closed)"
            )
        if current["final_report_schema_hash"] != load.get(
                "final_report_schema_hash"):
            reasons.append(
                "stored final-report schema hash is stale; reload via "
                f"{LOAD_TOOL_NAME} (fail closed)"
            )
    return reasons
 def review_workflow_load_blockers(
    project_root: str | None = None,
 ) -> list[str]:
    """Reasons reviewer mutations must fail closed."""
    boundary_reasons = boundary.boundary_blockers(project_root)
    if boundary_reasons and _REVIEW_WORKFLOW_LOAD is None:
        return boundary_reasons
    status = workflow_load_status(project_root)
    if not status.get("workflow_load_proof_present"):
        return list(status.get("reasons") or []) + boundary_reasons
    if not status.get("workflow_load_valid"):
        return list(status.get("reasons") or [])
    return []
 def recovery_handoff_without_replay() -> list[str]:
    """Safe next-step lines that must not include approve/merge replay."""
    return [
        "Reload the canonical workflow via gitea_load_review_workflow, then "
        "rerun the full review-merge workflow from inventory.",
        "Do not call gitea_submit_pr_review, gitea_mark_final_review_decision, "
        "or gitea_merge_pr until workflow-load proof is present.",
        "Do not include approve/merge replay commands in the recovery handoff.",
    ]
@@ -217,24 +217,12 @@ def assess_acquire_lease(
    candidate_head: str | None,
    target_branch: str,
    target_branch_sha: str | None,
    pr_merged_or_closed: bool = False,
    now: datetime | None = None,
 ) -> dict[str, Any]:
-    """Fail closed when another session holds an active lease.
+    """Fail closed when another session holds an active lease."""
    When *pr_merged_or_closed* is true the PR has already merged/closed, so any
    reviewer-lease acquisition or adoption for merge work is moot: fail closed
    with a ``post_merge_moot`` reason and never mint a lease body (#515).
    """
    now = now or datetime.now(timezone.utc)
    reasons: list[str] = []
    existing = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
    post_merge_moot = bool(pr_merged_or_closed)
    if post_merge_moot:
        reasons.append(
            f"post_merge_moot: PR #{pr_number} is already merged/closed; reviewer "
            "lease adoption for merge is moot (fail closed)"
        )
    if existing:
        owner_session = (existing.get("session_id") or "").strip()
        freshness = existing.get("freshness") or classify_lease_freshness(existing, now=now)
@@ -282,101 +270,6 @@ def assess_acquire_lease(
        "existing_lease": existing,
        "lease_body": body,
        "session_id": session_id,
        "post_merge_moot": post_merge_moot,
    }
 def assess_post_merge_moot_lease(
    comments: list[dict],
    *,
    pr_number: int,
    pr_merged: bool = False,
    pr_state: str | None = None,
    merge_commit_sha: str | None = None,
    now: datetime | None = None,
 ) -> dict[str, Any]:
    """Assess a reviewer lease left lingering on an already-merged/closed PR (#515).
    Read-first and fail-safe:
    - Only treats a lease as moot when the live PR state is merged/closed.
    - Never proposes touching an *active* lease while the PR is still open
      (``cleanup_allowed`` stays false and a refusal reason is returned).
    - When the PR is merged/closed and a lease is still active, ``cleanup_allowed``
      is true and a terminal ``phase: released`` lease body (``blocker:
      post-merge-moot``) is provided so the moot lease can be neutralised by an
      append-only comment — never by deleting a foreign session's comment, and
      never by adopting or merging.
    Posting the released body makes that lease terminal, so a subsequent call
    finds no active lease and reports nothing left to clean (idempotent).
    """
    now = now or datetime.now(timezone.utc)
    merged_or_closed = bool(pr_merged) or (
        str(pr_state or "").strip().lower() == "closed"
    )
    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
    # The newest lease comment is authoritative: once a terminal marker
    # (released/done/blocked) is the latest entry, the lease is resolved even if
    # an earlier non-terminal comment from the same session still lingers. This
    # keeps post-merge cleanup idempotent.
    entries = _lease_entries(comments, pr_number=pr_number)
    newest = entries[-1] if entries else None
    newest_terminal = bool(newest) and (
        (newest.get("phase") or "").strip().lower() in _TERMINAL_PHASES
    )
    reasons: list[str] = []
    cleanup_allowed = False
    release_body: str | None = None
    is_moot = bool(active) and merged_or_closed and not newest_terminal
    if not merged_or_closed:
        if active:
            reasons.append(
                f"PR #{pr_number} is still open; refusing to touch active reviewer "
                "lease (fail closed)"
            )
        else:
            reasons.append(
                f"PR #{pr_number} is still open; no post-merge lease cleanup applicable"
            )
    elif newest_terminal:
        reasons.append(
            f"PR #{pr_number} reviewer lease already released/terminal; nothing to clean"
        )
    elif active:
        cleanup_allowed = True
        release_body = format_lease_body(
            repo=active.get("repo") or "",
            pr_number=pr_number,
            issue_number=active.get("issue_number"),
            reviewer_identity=active.get("reviewer_identity") or "",
            profile=active.get("profile") or "unknown",
            session_id=active.get("session_id") or "",
            worktree=active.get("worktree") or "",
            phase="released",
            candidate_head=active.get("candidate_head"),
            target_branch=active.get("target_branch") or "master",
            target_branch_sha=active.get("target_branch_sha"),
            last_activity=now,
            blocker="post-merge-moot",
        )
    else:
        reasons.append(
            f"PR #{pr_number} is merged/closed but no active reviewer lease remains; "
            "nothing to clean"
        )
    return {
        "pr_number": pr_number,
        "pr_state": pr_state,
        "pr_merged_or_closed": merged_or_closed,
        "merge_commit_sha": merge_commit_sha,
        "active_lease": active,
        "is_moot": is_moot,
        "cleanup_allowed": cleanup_allowed,
        "release_body": release_body,
        "reasons": reasons,
    }
@@ -1,133 +0,0 @@
 """Root checkout guard (#475).
 The project root checkout is the stable control checkout on master/prgs/master.
 Author/reviewer/merge flows must fail closed when the control checkout is
 contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of
 prgs/master). Isolated ``branches/...`` worktrees remain allowed.
 """
 from __future__ import annotations
 import os
 import subprocess
 from author_mutation_worktree import is_path_under_branches
 from reviewer_worktree import parse_dirty_tracked_files
 REMEDIATION = (
    "Root checkout is not on master. Preserve state, switch root back to master, "
    "and use scripts/worktree-review or the sanctioned issue worktree flow."
 )
 BASE_BRANCHES = frozenset({"master", "main", "dev"})
 REMOTE_MASTER_REFS = ("prgs/master", "refs/remotes/prgs/master")
 def resolve_remote_master_sha(
    canonical_repo_root: str,
    *,
    remote_refs: tuple[str, ...] | None = None,
 ) -> str | None:
    """Return the commit SHA for the tracking master ref when available."""
    root = (canonical_repo_root or "").strip()
    if not root:
        return None
    for ref in remote_refs or REMOTE_MASTER_REFS:
        res = subprocess.run(
            ["git", "-C", root, "rev-parse", "--verify", ref],
            capture_output=True,
            text=True,
            check=False,
        )
        if res.returncode == 0:
            sha = (res.stdout or "").strip()
            if sha:
                return sha
    return None
 resolve_tracking_master_sha = resolve_remote_master_sha
 def assess_root_checkout_guard(
    *,
    workspace_path: str,
    canonical_repo_root: str,
    current_branch: str | None,
    head_sha: str | None,
    porcelain_status: str,
    remote_master_sha: str | None,
    resolved_role: str | None = None,
 ) -> dict:
    """Fail closed when the control checkout is not clean master/prgs/master."""
    reasons: list[str] = []
    root = os.path.realpath(canonical_repo_root)
    workspace = os.path.realpath(workspace_path)
    branch = (current_branch or "").strip()
    dirty_files = parse_dirty_tracked_files(porcelain_status)
    if resolved_role == "reconciler":
        return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)
    if resolved_role != "merger" and is_path_under_branches(workspace, root):
        return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)
    if dirty_files:
        reasons.append(
            "control checkout has tracked local edits before role work "
            f"(dirty files: {', '.join(dirty_files)})"
        )
    if not branch:
        reasons.append("control checkout is detached HEAD; expected branch 'master'")
    elif branch not in BASE_BRANCHES:
        reasons.append(
            f"control checkout branch '{branch}' is not a stable base branch "
            f"({'/'.join(sorted(BASE_BRANCHES))})"
        )
    if remote_master_sha and head_sha and head_sha != remote_master_sha:
        reasons.append(
            "control checkout HEAD does not match prgs/master "
            f"(HEAD {head_sha[:12]}, prgs/master {remote_master_sha[:12]})"
        )
    proven = not reasons
    return _assessment(proven, reasons, root, workspace, branch or None, head_sha, dirty_files)
 def format_root_checkout_guard_error(assessment: dict) -> str:
    """Single RuntimeError message for MCP preflight gates."""
    root = assessment.get("canonical_repo_root") or "(unknown)"
    workspace = assessment.get("workspace_path") or "(unknown)"
    reasons = "; ".join(assessment.get("reasons") or ["unknown root checkout violation"])
    return (
        f"Root checkout guard (#475): {reasons}. "
        f"canonical repository root: {root}; workspace: {workspace}. "
        f"{REMEDIATION}"
    )
 def _assessment(
    proven: bool,
    reasons: list[str],
    canonical_repo_root: str,
    workspace_path: str,
    current_branch: str | None,
    head_sha: str | None,
    dirty_files: list[str],
 ) -> dict:
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "canonical_repo_root": canonical_repo_root,
        "workspace_path": workspace_path,
        "current_branch": current_branch,
        "head_sha": head_sha,
        "dirty_files": dirty_files,
        "remediation": REMEDIATION,
    }
 assess_root_checkout = assess_root_checkout_guard
@@ -63,14 +63,8 @@ Do not use legacy fields: `Pinned reviewed head`, `Scratch worktree used`,
 - Current status:
 - Safe next action:
 - Safety statement:
 - Workflow-load helper result:
 ```
 The **Workflow-load helper result** field must carry structured output from
 `gitea_load_review_workflow` (workflow_hash, final_report_schema_hash,
 boundary_status). Narrative claims that workflow files were viewed locally are
 not sufficient (#403).
 ### Already-landed handoff overrides
 When eligibility class is `ALREADY_LANDED_RECONCILE_REQUIRED`:
@@ -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:
@@ -36,44 +36,6 @@ If available, load it first and report:
 If the canonical workflow cannot be loaded and the project requires it, stop and produce a recovery handoff only.
 ## 0A. Workflow-load and session boundary anchor (#403)
 The MCP gate is the authority — not local file viewing.
 Before any reviewer mutation:
 1. Record pre-review commands with `gitea_record_pre_review_command` when they
   are not automatically classified (inventory/diagnostic commands may be
   recorded explicitly for proof).
 2. Call `gitea_load_review_workflow` to establish workflow hash proof **and**
   session boundary state in the same in-process session proof.
 3. Do not claim the workflow was loaded by reading
   `skills/llm-project-workflow/workflows/review-merge-pr.md` as a local file;
   that narrative does not satisfy the validator.
 Allowed before workflow load (classify as `read_only_inventory` or
 `diagnostic`):
 * `gitea_whoami`, `gitea_resolve_task_capability`, `gitea_list_prs`,
  `gitea_view_pr`, `gitea_get_runtime_context`
 * `git fetch` / `git remote update` for inventory
 * `git status`, `git worktree list` (read-only)
 Boundary violations (block downstream reviewer mutations even after load):
 * validation commands (`pytest`, `python -m unittest`) in the main checkout
 * local profile/credential/config inspection (`profiles.json`, `gitea-mcp.json`,
  `.env`, keychain dumps)
 * MCP repair (`pkill`, MCP config edits)
 * git mutations before workflow load
 Final reports must include a structured **Workflow-load helper result** block
 copied from `gitea_load_review_workflow`, including at minimum:
 * `workflow_hash`
 * `final_report_schema_hash`
 * `boundary_status` (`clean` or `violation`)
 ## 1. Start with live identity, profile, runtime, and capability checks
 Prove:
@@ -909,16 +871,6 @@ Confirm:
 Clean only the session-owned `branches/` review worktree if the project workflow explicitly allows cleanup.
 Review, baseline, and merge-simulation worktrees created during this run are
 transient and are removed automatically at successful completion once they are
 clean, carry no open PR, and hold no active lease (#401). Use
 `gitea_audit_worktree_cleanup` (read-only) to classify `branches/` entries; only
 `clean_stale_removable` and `detached_review_leftover` may be removed, one-by-one,
 after `git worktree list` proof plus per-worktree proof of path, branch/HEAD,
 clean/dirty status, no active PR/lease, and the removal result. Dirty,
 active-PR, active-issue, and leased worktrees are never deleted automatically; a
 failed removal must be reported with the leftover path and reason.
 Do not delete or mutate unrelated branches/worktrees.
 Do not touch the main checkout except to update the stable branch after merge if explicitly allowed by the workflow.
@@ -699,39 +699,6 @@ Do not update the main checkout unless the canonical workflow explicitly allows
 Any cleanup is a mutation and must be reported.
 ### 22A. Session-owned worktree cleanup and TTL (#401)
 Every session-owned worktree created under `branches/` has ownership metadata:
 path, workflow type, issue number, PR number, branch/head SHA, creator
 identity/profile, created timestamp, last-used timestamp, and cleanup
 eligibility.
 Cleanup is classification-driven. `gitea_audit_worktree_cleanup` (read-only)
 classifies every `branches/` entry as exactly one of:
 * `active_open_pr` — branch has an open PR; never auto-removed.
 * `active_issue_work` — active claim/lease or fresh issue worktree; never
  auto-removed.
 * `dirty_local_worktree` — uncommitted changes; never auto-removed.
 * `clean_stale_removable` — clean, no PR, no lease; removable.
 * `detached_review_leftover` — clean detached review/baseline/merge-simulation
  worktree; removable.
 * `unsafe_unknown` — protected base checkout or unknown workflow type; never
  auto-removed.
 Only `clean_stale_removable` and `detached_review_leftover` may be removed, and
 only one-by-one after `git worktree list` proof plus per-worktree proof of:
 worktree path, branch/HEAD, clean/dirty status, no active PR/lease, and the
 removal result. Review, baseline, and merge-simulation worktrees are removed
 automatically at successful workflow completion; issue/conflict-fix worktrees
 are removed only after their TTL (`GITEA_WORKTREE_TTL_HOURS`, default 24h)
 expires and no lock/lease is held.
 Dirty, active-PR, active-issue, and leased worktrees are never deleted
 automatically. If a removal fails, the final report must list the leftover
 worktree path and the reason. Include the `git worktree list` output as final
 cleanup verification.
 ## 23. Recovery handoff rules
 If blocked, produce a recovery handoff with:
@@ -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",
@@ -157,7 +157,6 @@ class _AuditWiringBase(unittest.TestCase):
    def tearDown(self):
        mcp_server._IDENTITY_CACHE.clear()
        mcp_server.review_workflow_load.clear_review_workflow_load()
        self._dir.cleanup()
    def _env(self, **extra):
@@ -292,15 +291,12 @@ class TestGatedToolAudit(_AuditWiringBase):
    def setUp(self):
        super().setUp()
-        from tests.test_mcp_server import _init_reviewer_session, _install_owned_reviewer_lease
+        from mcp_server import init_review_decision_lock
        from tests.test_mcp_server import _install_owned_reviewer_lease
        import reviewer_pr_lease
        import review_workflow_boundary
        import review_workflow_load
-        # Session init clears any prior session lease (#407) and loads workflow (#389).
+        # init_review_decision_lock clears any prior session lease (#407).
-        _init_reviewer_session("prgs")
+        init_review_decision_lock("prgs", "review_pr")
        self.addCleanup(review_workflow_load.clear_review_workflow_load)
        self.addCleanup(review_workflow_boundary.clear_pre_review_commands)
        self._lease_patch = _install_owned_reviewer_lease(8)
        self._lease_patch.start()
        self._auth_identity_patch = patch(
@@ -341,7 +337,6 @@ class TestGatedToolAudit(_AuditWiringBase):
        env = self._env(GITEA_PROFILE_NAME="gitea-merger",
                        GITEA_ALLOWED_OPERATIONS="read,merge")
        with patch.dict(os.environ, env, clear=True):
            mcp_server.gitea_load_review_workflow()
            r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8",
                               expected_head_sha="abc123", remote="prgs")
        self.assertTrue(r["performed"])
@@ -361,7 +356,6 @@ class TestGatedToolAudit(_AuditWiringBase):
        env = self._env(GITEA_PROFILE_NAME="gitea-merger",
                        GITEA_ALLOWED_OPERATIONS="read,merge")
        with patch.dict(os.environ, env, clear=True):
            mcp_server.gitea_load_review_workflow()
            r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8", remote="prgs")
        self.assertFalse(r["performed"])
        recs = self._records()
@@ -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()
@@ -38,18 +38,8 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
    @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
    @patch("gitea_mcp_server.api_request")
    @patch("gitea_mcp_server.api_get_all", return_value=[])
-    @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
-    @patch(
+    def test_create_issue_stable_checkout_rejected(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value={
            "current_branch": "master",
            "head_sha": "a" * 40,
            "porcelain_status": "",
        },
    )
    def test_create_issue_stable_checkout_rejected(
        self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth,
    ):
        # Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
        # path is the stable control checkout (not under branches/), mutation must fail.
        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
@@ -22,7 +22,6 @@ from unittest.mock import patch
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 import mcp_server  # noqa: E402
 from mcp_server import (  # noqa: E402
    gitea_check_pr_eligibility,
    gitea_merge_pr,
@@ -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():
@@ -1,127 +0,0 @@
 """Regression tests for gitea_lock_issue MCP tool registration (#521)."""
 from __future__ import annotations
 import os
 import tempfile
 import unittest
 from unittest.mock import patch
 import issue_lock_store
 import mcp_server
 from mcp_server import gitea_create_pr, gitea_lock_issue
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 ISSUE_WRITE_ENV = {
    "GITEA_ALLOWED_OPERATIONS": (
        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
    ),
 }
 CREATE_PR_ENV = {
    "GITEA_PROFILE_NAME": "author-test",
    "GITEA_ALLOWED_OPERATIONS": (
        "gitea.read,gitea.pr.create,gitea.branch.push,"
        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
    ),
    "GITEA_FORBIDDEN_OPERATIONS": (
        "gitea.pr.approve,gitea.pr.merge,gitea.pr.review"
    ),
 }
 def _clean_master_git_state_for_lock():
    return {
        "current_branch": "master",
        "porcelain_status": "",
        "base_equivalent": True,
        "inspected_git_root": "/scratch/wt",
        "base_branch": "origin/master",
    }
 def _registered_tool_names() -> set[str]:
    manager = mcp_server.mcp._tool_manager
    tools = getattr(manager, "_tools", None) or {}
    return set(tools.keys())
 class TestLockIssueMcpRegistration(unittest.TestCase):
    def test_gitea_lock_issue_registered_as_public_mcp_tool(self):
        names = _registered_tool_names()
        self.assertIn("gitea_lock_issue", names)
    def test_internal_list_open_pulls_not_exposed_as_mcp_tool(self):
        names = _registered_tool_names()
        self.assertNotIn("_list_open_pulls", names)
 class TestCreatePrLockRegistrationFlow(unittest.TestCase):
    def setUp(self):
        self._lock_dir = tempfile.TemporaryDirectory()
        self._env_patcher = patch.dict(
            os.environ,
            {
                **CREATE_PR_ENV,
                "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
            },
            clear=True,
        )
        self._env_patcher.start()
        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._env_patcher.stop()
        self._lock_dir.cleanup()
    @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_still_fails_closed_without_issue_lock(self, _auth, _role):
        with self.assertRaises(RuntimeError) as ctx:
            gitea_create_pr(
                title="feat: X Closes #521",
                head="feat/issue-521-lock-issue-registration",
                remote="prgs",
            )
        self.assertIn("Issue lock is missing", str(ctx.exception))
    @patch("mcp_server.api_request")
    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all", return_value=[])
    @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_proceeds_after_valid_issue_lock(
        self, _auth, _role, _api, _git_state, mock_api_request
    ):
        worktree = os.path.realpath(os.getcwd())
        mock_api_request.return_value = {"number": 521, "html_url": "https://example/pr/521"}
        lock_res = gitea_lock_issue(
            issue_number=521,
            branch_name="feat/issue-521-lock-issue-registration",
            remote="prgs",
            worktree_path=worktree,
        )
        self.assertTrue(lock_res["success"])
        lock_path = lock_res["lock_file_path"]
        self.assertTrue(os.path.exists(lock_path))
        lock = issue_lock_store.read_lock_file(lock_path)
        self.assertEqual(lock["issue_number"], 521)
        res = gitea_create_pr(
            title="fix: restore lock tool registration Closes #521",
            head="feat/issue-521-lock-issue-registration",
            remote="prgs",
            worktree_path=worktree,
        )
        self.assertEqual(res["number"], 521)
@@ -58,12 +58,6 @@ _NO_BLOCKER_FEEDBACK = {
 }
 def _init_reviewer_session(remote="prgs"):
    """Seed review decision lock and required workflow-load proof (#389)."""
    init_review_decision_lock(remote, "review_pr")
    mcp_server.gitea_load_review_workflow()
 def _mark_request_changes_ready(pr_number=8, **kwargs):
    """Mark a request_changes decision ready with the #332 duplicate-
    suppression feedback fetch stubbed to 'no existing blocker'."""
@@ -185,7 +179,6 @@ def _seed_ready_review_decision(
        "correction_authorized": False,
        "correction_reason": None,
    })
    _m.gitea_load_review_workflow()
 # Issue-write tools are profile-gated (#69).
@@ -678,7 +671,6 @@ class TestMergePR(unittest.TestCase):
    def setUp(self):
        import reviewer_pr_lease
        mcp_server.gitea_load_review_workflow()
        self._lease_patch = _install_owned_reviewer_lease(8)
        self._lease_patch.start()
        self._auth_identity_patch = patch(
@@ -1057,8 +1049,7 @@ class TestMergePR(unittest.TestCase):
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=True):
            r = gitea_merge_pr(
-                pr_number=8, confirmation=self._confirm(8), remote="prgs",
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
                expected_head_sha=new_sha)
        self.assertFalse(r["performed"])
        self.assertTrue(r.get("approval_visible"))
        self.assertFalse(r.get("approval_at_current_head"))
@@ -1885,7 +1876,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
    def setUp(self):
        import reviewer_pr_lease
-        _init_reviewer_session("prgs")
+        init_review_decision_lock("prgs", "review_pr")
        self._lease_patch = _install_owned_reviewer_lease(
            self.PR, head_sha=self.SHA,
        )
@@ -2005,7 +1996,7 @@ class TestSubmitPrReview(unittest.TestCase):
    def setUp(self):
        import reviewer_pr_lease
-        _init_reviewer_session("prgs")
+        init_review_decision_lock("prgs", "review_pr")
        self._lease_patch = _install_owned_reviewer_lease(8)
        self._lease_patch.start()
        self._auth_identity_patch = patch(
@@ -2425,7 +2416,7 @@ class TestSubmitPrReview(unittest.TestCase):
                os.remove(spoof_path)
    def test_mark_final_decision_rejects_remote_mismatch(self):
-        _init_reviewer_session("prgs")
+        init_review_decision_lock("prgs", "review_pr")
        r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools", expected_head_sha="abc123")
        self.assertFalse(r["marked_ready"])
        self.assertTrue(any("does not match locked remote" in x for x in r["reasons"]))
@@ -2509,7 +2500,6 @@ if __name__ == "__main__":
 class TestTrackerHygieneCleanup(unittest.TestCase):
    def setUp(self):
        mcp_server.gitea_load_review_workflow()
        self.mock_api = patch("mcp_server.api_request").start()
        self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
        patch("gitea_audit.audit_enabled", return_value=True).start()
@@ -95,15 +95,10 @@ class PermissionReportBase(unittest.TestCase):
        self._dir = tempfile.TemporaryDirectory()
        self.config_path = os.path.join(self._dir.name, "profiles.json")
        self._write_config(CONFIG)
        import review_workflow_load
        review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
    def tearDown(self):
        import review_workflow_load
        review_workflow_load.clear_review_workflow_load()
        self._remotes.stop()
        mcp_server._IDENTITY_CACHE.clear()
        mcp_server.review_workflow_load.clear_review_workflow_load()
        gitea_config._active_profile_override = None
        self._dir.cleanup()
@@ -255,8 +250,6 @@ class TestEligibilityDenialReport(PermissionReportBase):
                return {"login": "author-user"}
            return PR_PAYLOAD
        mock_api.side_effect = fake_api
        mcp_server.init_review_decision_lock("prgs", "review_pr")
        mcp_server.gitea_load_review_workflow()
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_merge_pr(
                pr_number=42, confirmation="MERGE PR 42",
@@ -1,257 +0,0 @@
 """Post-merge moot reviewer-lease handling (#515).
 Covers:
  a. Reviewer-lease acquisition/adoption is refused on an already-merged/closed
     PR (fail closed, no mutation).
  b. The post-merge moot cleanup path is safe and idempotent.
  c. An active foreign lease on an *open* PR is never force-cleaned.
 """
 import os
 import sys
 import unittest
 from datetime import datetime, timezone
 from unittest.mock import patch
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 import reviewer_pr_lease as leases  # noqa: E402
 from mcp_server import (  # noqa: E402
    gitea_acquire_reviewer_pr_lease,
    gitea_cleanup_post_merge_moot_lease,
 )
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 MERGER_ENV = {
    "GITEA_PROFILE_NAME": "prgs-merger",
    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment",
 }
 PR = 487
 ISSUE = 485
 SESSION = "97274-676d20a825c4"
 def _lease_comment(pr_number=PR, session_id=SESSION, *, phase="claimed",
                   candidate_head="a" * 40):
    body = leases.format_lease_body(
        repo="Scaled-Tech-Consulting/Gitea-Tools",
        pr_number=pr_number,
        issue_number=ISSUE,
        reviewer_identity="sysadmin",
        profile="prgs-reviewer",
        session_id=session_id,
        worktree="branches/review-pr487",
        phase=phase,
        candidate_head=candidate_head,
        target_branch="master",
        target_branch_sha="b" * 40,
        last_activity=datetime.now(timezone.utc),
    )
    return {"id": 6603, "body": body, "user": {"login": "sysadmin"}}
 # --------------------------------------------------------------------------- #
 # Pure logic
 # --------------------------------------------------------------------------- #
 class TestAcquireRefusedOnMergedPR(unittest.TestCase):
    def test_acquire_refused_when_pr_merged_or_closed(self):
        result = leases.assess_acquire_lease(
            [_lease_comment()],
            pr_number=PR,
            reviewer_identity="sysadmin",
            profile="prgs-merger",
            session_id="new-session",
            repo="Scaled-Tech-Consulting/Gitea-Tools",
            issue_number=ISSUE,
            worktree="branches/merge-pr487",
            candidate_head="a" * 40,
            target_branch="master",
            target_branch_sha="b" * 40,
            pr_merged_or_closed=True,
        )
        self.assertFalse(result["acquire_allowed"])
        self.assertTrue(result["post_merge_moot"])
        self.assertIsNone(result["lease_body"])
        self.assertTrue(any("post_merge_moot" in r for r in result["reasons"]))
    def test_acquire_still_allowed_on_open_pr_without_flag(self):
        result = leases.assess_acquire_lease(
            [],
            pr_number=PR,
            reviewer_identity="sysadmin",
            profile="prgs-reviewer",
            session_id="s1",
            repo="Scaled-Tech-Consulting/Gitea-Tools",
            issue_number=ISSUE,
            worktree="branches/review-pr487",
            candidate_head="a" * 40,
            target_branch="master",
            target_branch_sha="b" * 40,
        )
        self.assertTrue(result["acquire_allowed"])
        self.assertFalse(result["post_merge_moot"])
 class TestPostMergeMootAssessment(unittest.TestCase):
    def test_merged_pr_with_active_lease_is_moot_and_cleanable(self):
        a = leases.assess_post_merge_moot_lease(
            [_lease_comment()],
            pr_number=PR,
            pr_merged=True,
            pr_state="closed",
            merge_commit_sha="c" * 40,
        )
        self.assertTrue(a["pr_merged_or_closed"])
        self.assertTrue(a["is_moot"])
        self.assertTrue(a["cleanup_allowed"])
        self.assertIsNotNone(a["release_body"])
        self.assertIn("phase: released", a["release_body"])
        self.assertIn("blocker: post-merge-moot", a["release_body"])
    def test_open_pr_active_lease_never_cleaned(self):
        a = leases.assess_post_merge_moot_lease(
            [_lease_comment()],
            pr_number=PR,
            pr_merged=False,
            pr_state="open",
        )
        self.assertFalse(a["pr_merged_or_closed"])
        self.assertFalse(a["is_moot"])
        self.assertFalse(a["cleanup_allowed"])
        self.assertIsNone(a["release_body"])
        self.assertTrue(any("still open" in r for r in a["reasons"]))
    def test_merged_pr_without_lease_nothing_to_clean(self):
        a = leases.assess_post_merge_moot_lease(
            [], pr_number=PR, pr_merged=True, pr_state="closed")
        self.assertTrue(a["pr_merged_or_closed"])
        self.assertFalse(a["is_moot"])
        self.assertFalse(a["cleanup_allowed"])
        self.assertTrue(any("nothing to clean" in r for r in a["reasons"]))
    def test_cleanup_is_idempotent(self):
        """After the released marker is posted, a re-assess finds nothing to clean."""
        first = leases.assess_post_merge_moot_lease(
            [_lease_comment()], pr_number=PR, pr_merged=True, pr_state="closed")
        self.assertTrue(first["cleanup_allowed"])
        released_comment = {
            "id": 7000, "body": first["release_body"], "user": {"login": "sysadmin"}}
        second = leases.assess_post_merge_moot_lease(
            [_lease_comment(), released_comment],
            pr_number=PR, pr_merged=True, pr_state="closed")
        self.assertFalse(second["is_moot"])
        self.assertFalse(second["cleanup_allowed"])
 # --------------------------------------------------------------------------- #
 # Server tools
 # --------------------------------------------------------------------------- #
 def _api_side_effect(*, pr_state, pr_merged, comments, posted_id=9999):
    """Build an api_request side effect keyed on method + url."""
    calls = {"post": []}
    def _side(method, url, auth=None, payload=None, *a, **k):
        m = (method or "").upper()
        if m == "POST":
            calls["post"].append({"url": url, "payload": payload})
            return {"id": posted_id}
        if "/comments" in url:
            return list(comments)
        if "/pulls/" in url:
            pr = {"state": pr_state, "number": PR, "merge_commit_sha": "c" * 40}
            if pr_merged:
                pr["merged"] = True
                pr["merged_at"] = "2026-07-08T07:46:04Z"
            return pr
        if "/issues/" in url:
            return {"state": "closed" if pr_merged else "open", "number": ISSUE}
        return {}
    return _side, calls
 class TestAcquireToolRefusesMergedPR(unittest.TestCase):
    def setUp(self):
        leases.clear_session_lease()
    @patch("mcp_server.verify_preflight_purity", return_value=None)
    @patch("mcp_server._authenticated_username", return_value="sysadmin")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("mcp_server.api_request")
    def test_acquire_tool_fails_closed_on_merged_pr_without_posting(
            self, mock_api, _auth, _user, _purity):
        side, calls = _api_side_effect(
            pr_state="closed", pr_merged=True, comments=[])
        mock_api.side_effect = side
        with patch.dict(os.environ, MERGER_ENV, clear=True):
            result = gitea_acquire_reviewer_pr_lease(
                pr_number=PR,
                worktree="branches/merge-pr487",
                candidate_head="a" * 40,
                issue_number=ISSUE,
                remote="prgs",
            )
        self.assertFalse(result["success"])
        self.assertFalse(result["acquired"])
        self.assertTrue(result.get("post_merge_moot"))
        self.assertEqual(calls["post"], [], "must not post a lease comment")
 class TestCleanupTool(unittest.TestCase):
    def setUp(self):
        leases.clear_session_lease()
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("mcp_server.api_request")
    def test_read_only_reports_moot_without_mutating(self, mock_api, _auth):
        side, calls = _api_side_effect(
            pr_state="closed", pr_merged=True, comments=[_lease_comment()])
        mock_api.side_effect = side
        with patch.dict(os.environ, MERGER_ENV, clear=True):
            result = gitea_cleanup_post_merge_moot_lease(
                pr_number=PR, apply=False, remote="prgs")
        self.assertTrue(result["success"])
        self.assertTrue(result["pr_merged_or_closed"])
        self.assertTrue(result["lease_moot"])
        self.assertFalse(result["cleanup_performed"])
        self.assertTrue(result["no_merge_or_adoption"])
        self.assertEqual(result["mode"], "read_only")
        self.assertEqual(calls["post"], [])
    @patch("mcp_server.verify_preflight_purity", return_value=None)
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("mcp_server.api_request")
    def test_apply_posts_released_marker_on_merged_pr(
            self, mock_api, _auth, _purity):
        side, calls = _api_side_effect(
            pr_state="closed", pr_merged=True, comments=[_lease_comment()])
        mock_api.side_effect = side
        with patch.dict(os.environ, MERGER_ENV, clear=True):
            result = gitea_cleanup_post_merge_moot_lease(
                pr_number=PR, apply=True, remote="prgs")
        self.assertTrue(result["success"])
        self.assertTrue(result["cleanup_performed"])
        self.assertEqual(result["released_comment_id"], 9999)
        self.assertEqual(len(calls["post"]), 1)
        self.assertIn("phase: released", calls["post"][0]["payload"]["body"])
        self.assertIn("post-merge-moot", calls["post"][0]["payload"]["body"])
    @patch("mcp_server.verify_preflight_purity", return_value=None)
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("mcp_server.api_request")
    def test_apply_refuses_to_clean_open_pr(self, mock_api, _auth, _purity):
        side, calls = _api_side_effect(
            pr_state="open", pr_merged=False, comments=[_lease_comment()])
        mock_api.side_effect = side
        with patch.dict(os.environ, MERGER_ENV, clear=True):
            result = gitea_cleanup_post_merge_moot_lease(
                pr_number=PR, apply=True, remote="prgs")
        self.assertFalse(result["cleanup_performed"])
        self.assertFalse(result["pr_merged_or_closed"])
        self.assertEqual(calls["post"], [], "never force-clean an open PR lease")
        self.assertTrue(
            any("still open" in r for r in result.get("cleanup_skipped_reason", [])))
 if __name__ == "__main__":
    unittest.main()
@@ -11,7 +11,6 @@ from mcp_server import (
    gitea_view_pr,
    gitea_review_pr,
    gitea_check_pr_eligibility,
    gitea_load_review_workflow,
 )
 import gitea_config
@@ -1,138 +0,0 @@
 """Tests for workflow-load session boundary tracking (#403)."""
 import os
 import sys
 import unittest
 from unittest.mock import patch
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 import final_report_validator
 import review_workflow_boundary
 import review_workflow_load
 import mcp_server
 class TestPreReviewClassification(unittest.TestCase):
    def setUp(self):
        review_workflow_boundary.clear_pre_review_commands()
        review_workflow_load.clear_review_workflow_load()
    def test_inventory_command_allowed(self):
        result = review_workflow_boundary.classify_pre_review_command(
            "gitea_list_prs remote=prgs",
            cwd="/tmp",
            project_root="/repo/Gitea-Tools",
        )
        self.assertEqual(
            result["classification"],
            review_workflow_boundary.CLASSIFICATION_READ_ONLY_INVENTORY,
        )
    def test_main_checkout_pytest_is_boundary_violation(self):
        root = "/repo/Gitea-Tools"
        result = review_workflow_boundary.classify_pre_review_command(
            "python -m pytest tests/",
            cwd=root,
            project_root=root,
        )
        self.assertEqual(
            result["classification"],
            review_workflow_boundary.CLASSIFICATION_BOUNDARY_VIOLATION,
        )
    def test_profiles_json_inspection_is_boundary_violation(self):
        result = review_workflow_boundary.classify_pre_review_command(
            "cat profiles.json",
            cwd="/repo/Gitea-Tools",
            project_root="/repo/Gitea-Tools",
        )
        self.assertEqual(
            result["classification"],
            review_workflow_boundary.CLASSIFICATION_BOUNDARY_VIOLATION,
        )
 class TestWorkflowLoadBoundaryGate(unittest.TestCase):
    def setUp(self):
        review_workflow_boundary.clear_pre_review_commands()
        review_workflow_load.clear_review_workflow_load()
        mcp_server.init_review_decision_lock("prgs", "review_pr")
        mcp_server.record_preflight_check("whoami")
        mcp_server.record_preflight_check("capability", "reviewer")
    def _root(self) -> str:
        return str(__import__("pathlib").Path(__file__).resolve().parent.parent)
    def test_boundary_violation_blocks_mutation_after_load(self):
        root = self._root()
        review_workflow_boundary.record_pre_review_command(
            "python -m pytest tests/",
            cwd=root,
            project_root=root,
        )
        res = mcp_server.gitea_load_review_workflow()
        self.assertFalse(res["success"])
        self.assertEqual(res["boundary_status"], "violation")
        blocked = mcp_server.gitea_mark_final_review_decision(
            42, "approve", remote="prgs")
        self.assertFalse(blocked["marked_ready"])
        joined = " ".join(blocked["reasons"]).lower()
        self.assertTrue(
            "validation" in joined or "boundary" in joined or "workflow" in joined
        )
    def test_clean_inventory_then_load_passes(self):
        root = self._root()
        review_workflow_boundary.record_pre_review_command(
            "gitea_list_prs remote=prgs",
            cwd=root,
            project_root=root,
        )
        res = mcp_server.gitea_load_review_workflow()
        self.assertTrue(res["success"])
        self.assertEqual(res["boundary_status"], "clean")
        blockers = review_workflow_load.review_workflow_load_blockers(root)
        self.assertEqual(blockers, [])
    def test_file_view_narrative_fails_validator_without_helper(self):
        report = (
            "## Controller Handoff\n"
            "- Task: review-merge-pr\n"
            "- I read the canonical workflow review-merge-pr.md before review.\n"
        )
        findings = final_report_validator.assess_final_report_validator(
            report,
            task_kind="review_pr",
        )
        self.assertTrue(any(
            f["rule_id"] == "reviewer.workflow_load_boundary"
            for f in findings.get("findings") or []
        ))
    def test_helper_result_passes_validator(self):
        root = self._root()
        review_workflow_load.record_review_workflow_load(root)
        helper = review_workflow_boundary.workflow_load_helper_result(
            review_workflow_load._REVIEW_WORKFLOW_LOAD,
            root,
        )
        report = (
            "## Controller Handoff\n"
            "- Task: review-merge-pr\n"
            f"- Workflow-load helper result: workflow_hash: {helper['workflow_hash']}; "
            f"boundary_status: {helper['boundary_status']}\n"
        )
        findings = final_report_validator.assess_final_report_validator(
            report,
            task_kind="review_pr",
        )
        boundary_findings = [
            f for f in (findings.get("findings") or [])
            if f.get("rule_id") == "reviewer.workflow_load_boundary"
        ]
        self.assertEqual(boundary_findings, [])
 if __name__ == "__main__":
    unittest.main()
@@ -1,153 +0,0 @@
 """Tests for canonical review workflow load proof (#389)."""
 import os
 import sys
 import unittest
 from unittest.mock import patch
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 import review_workflow_load
 import mcp_server
 class TestReviewWorkflowLoadModule(unittest.TestCase):
    def setUp(self):
        review_workflow_load.clear_review_workflow_load()
        mcp_server._save_review_decision_lock(None)
    def test_load_records_hash_and_schema(self):
        root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
        recorded = review_workflow_load.record_review_workflow_load(root)
        self.assertEqual(
            recorded["workflow_source"],
            review_workflow_load.WORKFLOW_REL_PATH,
        )
        self.assertEqual(recorded["task_mode"], "review-merge-pr")
        self.assertRegex(recorded["workflow_hash"], r"^[0-9a-f]{12}$")
        self.assertEqual(
            recorded["final_report_schema_path"],
            review_workflow_load.SCHEMA_REL_PATH,
        )
        status = review_workflow_load.workflow_load_status(root)
        self.assertTrue(status["workflow_load_proof_present"])
        self.assertTrue(status["workflow_load_valid"])
    def test_stale_session_pid_blocks(self):
        root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
        review_workflow_load.record_review_workflow_load(root)
        review_workflow_load._REVIEW_WORKFLOW_LOAD["session_pid"] = 0
        blockers = review_workflow_load.review_workflow_load_blockers(root)
        self.assertTrue(any("different process" in b for b in blockers))
    def test_prompt_conflict_detected(self):
        conflict, reasons = review_workflow_load.assess_prompt_conflict(
            "Run work-issue author implementation only")
        self.assertTrue(conflict)
        self.assertTrue(reasons)
 class TestReviewWorkflowLoadGates(unittest.TestCase):
    def setUp(self):
        review_workflow_load.clear_review_workflow_load()
        mcp_server.init_review_decision_lock("prgs", "review_pr")
        mcp_server.record_preflight_check("whoami")
        mcp_server.record_preflight_check("capability", "reviewer")
    def _load_workflow(self):
        return mcp_server.gitea_load_review_workflow()
    def test_mcp_helper_returns_required_fields(self):
        res = self._load_workflow()
        self.assertTrue(res["success"])
        self.assertTrue(res["loaded"])
        self.assertIn("workflow_source", res)
        self.assertIn("workflow_hash", res)
        self.assertIn("final_report_schema_path", res)
        self.assertIn("final_report_schema_hash", res)
    def test_mark_final_blocked_without_load(self):
        res = mcp_server.gitea_mark_final_review_decision(
            42, "approve", remote="prgs")
        self.assertFalse(res["marked_ready"])
        self.assertTrue(any(
            "gitea_load_review_workflow" in r for r in res["reasons"]))
        self.assertTrue(any(
            "approve/merge replay" in r.lower() or "Do not call" in r
            for r in res["reasons"]))
    def test_submit_review_blocked_without_load(self):
        with patch("mcp_server.gitea_check_pr_eligibility") as elig:
            elig.return_value = {
                "eligible": True,
                "authenticated_user": "rev",
                "profile_name": "prgs-reviewer",
                "pr_author": "author",
                "head_sha": "abc123",
                "reasons": [],
            }
            res = mcp_server.gitea_submit_pr_review(
                42,
                "approve",
                remote="prgs",
                final_review_decision_ready=True,
            )
        self.assertFalse(res["performed"])
        self.assertTrue(any(
            "gitea_load_review_workflow" in r for r in res["reasons"]))
    def test_merge_blocked_without_load(self):
        res = mcp_server.gitea_merge_pr(
            42,
            confirmation="MERGE PR 42",
            remote="prgs",
        )
        self.assertFalse(res["performed"])
        self.assertTrue(any(
            "gitea_load_review_workflow" in r for r in res["reasons"]))
    def test_resolve_capability_reports_missing_load(self):
        with patch.object(mcp_server, "_ensure_matching_profile"):
            with patch.object(
                    mcp_server.gitea_config, "is_runtime_switching_enabled",
                    return_value=False):
                with patch.object(
                        mcp_server, "_authenticated_username",
                        return_value="rev"):
                    res = mcp_server.gitea_resolve_task_capability(
                        "review_pr", remote="prgs")
        proof = res.get("workflow_load_proof") or {}
        self.assertFalse(proof.get("workflow_load_valid"))
        self.assertTrue(any(
            "gitea_load_review_workflow" in g
            for g in res.get("task_role_guidance") or []))
    def test_init_review_lock_clears_prior_load(self):
        self._load_workflow()
        mcp_server.init_review_decision_lock("prgs", "review_pr")
        blockers = review_workflow_load.review_workflow_load_blockers(
            str(__import__("pathlib").Path(__file__).resolve().parent.parent))
        self.assertTrue(blockers)
    def test_dry_run_allowed_without_load(self):
        with patch("mcp_server.get_auth_header", return_value="Basic dGVzdA=="), \
                patch("mcp_server._list_pr_lease_comments", return_value=[]), \
                patch("mcp_server.gitea_check_pr_eligibility") as elig:
            elig.return_value = {
                "eligible": True,
                "authenticated_user": "rev",
                "profile_name": "prgs-reviewer",
                "pr_author": "author",
                "head_sha": "abc123",
                "reasons": [],
            }
            res = mcp_server.gitea_dry_run_pr_review(
                42, "approve", remote="prgs")
        self.assertNotIn(
            "gitea_load_review_workflow",
            " ".join(res.get("reasons") or []),
        )
 if __name__ == "__main__":
    unittest.main()
@@ -1,161 +0,0 @@
 """Tests for root checkout guard (#475)."""
 from __future__ import annotations
 import os
 import sys
 import unittest
 from pathlib import Path
 from unittest.mock import patch
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 import gitea_mcp_server as srv  # noqa: E402
 import root_checkout_guard as rcg  # noqa: E402
 CONTROL_ROOT = str(Path(__file__).resolve().parents[3])
 BRANCHES_WORKTREE = str(Path(__file__).resolve().parents[1])
 MASTER_SHA = "a" * 40
 OTHER_SHA = "b" * 40
 class TestAssessRootCheckoutGuard(unittest.TestCase):
    def _assess(self, **kwargs):
        defaults = {
            "workspace_path": CONTROL_ROOT,
            "canonical_repo_root": CONTROL_ROOT,
            "current_branch": "master",
            "head_sha": MASTER_SHA,
            "porcelain_status": "",
            "remote_master_sha": MASTER_SHA,
            "resolved_role": "author",
        }
        defaults.update(kwargs)
        return rcg.assess_root_checkout_guard(**defaults)
    def test_clean_master_control_checkout_allowed(self):
        result = self._assess()
        self.assertTrue(result["proven"])
        self.assertFalse(result["block"])
    def test_branches_worktree_allowed_for_author(self):
        result = self._assess(
            workspace_path=BRANCHES_WORKTREE,
            current_branch="feat/issue-475-root-checkout-guard",
            head_sha=OTHER_SHA,
            resolved_role="author",
        )
        self.assertTrue(result["proven"])
    def test_branches_worktree_allowed_for_reviewer(self):
        result = self._assess(
            workspace_path=f"{CONTROL_ROOT}/branches/review-pr-1",
            current_branch="review-pr-1",
            resolved_role="reviewer",
        )
        self.assertTrue(result["proven"])
    def test_reconciler_always_allowed(self):
        result = self._assess(
            current_branch="feat/some-branch",
            head_sha=OTHER_SHA,
            porcelain_status=" M gitea_mcp_server.py\n",
            resolved_role="reconciler",
        )
        self.assertTrue(result["proven"])
    def test_feature_branch_on_control_checkout_blocked(self):
        result = self._assess(
            current_branch="feat/issue-99-example",
            head_sha=OTHER_SHA,
        )
        self.assertTrue(result["block"])
        self.assertIn("not a stable base branch", result["reasons"][0])
    def test_detached_head_blocked(self):
        result = self._assess(current_branch=None)
        self.assertTrue(result["block"])
        self.assertIn("detached HEAD", result["reasons"][0])
    def test_dirty_control_checkout_blocked(self):
        result = self._assess(porcelain_status=" M gitea_mcp_server.py\n")
        self.assertTrue(result["block"])
        self.assertIn("tracked local edits", result["reasons"][0])
    def test_head_behind_prgs_master_blocked(self):
        result = self._assess(
            head_sha=OTHER_SHA,
            remote_master_sha=MASTER_SHA,
        )
        self.assertTrue(result["block"])
        self.assertIn("does not match prgs/master", result["reasons"][0])
    def test_merger_requires_clean_control_checkout(self):
        result = self._assess(
            workspace_path=BRANCHES_WORKTREE,
            current_branch="feat/issue-475-root-checkout-guard",
            head_sha=OTHER_SHA,
            resolved_role="merger",
        )
        self.assertTrue(result["block"])
 class TestVerifyPreflightRootGuardIntegration(unittest.TestCase):
    def setUp(self):
        srv._preflight_whoami_called = True
        srv._preflight_capability_called = True
        srv._preflight_resolved_role = "reviewer"
        srv._preflight_whoami_violation = False
        srv._preflight_capability_violation = False
        self._orig_in_test = srv._preflight_in_test_mode
        srv._preflight_in_test_mode = lambda: False
    def tearDown(self):
        srv._preflight_in_test_mode = self._orig_in_test
    @patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
    @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value=MASTER_SHA)
    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state")
    @patch("gitea_mcp_server._resolve_author_mutation_context")
    def test_reviewer_from_contaminated_root_blocked(
        self, mock_ctx, mock_git, _remote_sha, _porcelain,
    ):
        srv._preflight_capability_baseline_porcelain = ""
        mock_ctx.return_value = {
            "workspace_path": CONTROL_ROOT,
            "canonical_repo_root": CONTROL_ROOT,
            "process_project_root": CONTROL_ROOT,
        }
        mock_git.return_value = {
            "current_branch": "feat/hijacked-root",
            "head_sha": OTHER_SHA,
            "porcelain_status": "",
        }
        with self.assertRaises(RuntimeError) as ctx:
            srv.verify_preflight_purity("prgs", worktree_path=CONTROL_ROOT)
        self.assertIn("Root checkout guard (#475)", str(ctx.exception))
        self.assertIn(rcg.REMEDIATION, str(ctx.exception))
    @patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
    @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value=MASTER_SHA)
    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state")
    @patch("gitea_mcp_server._resolve_author_mutation_context")
    def test_reviewer_from_branches_worktree_allowed(
        self, mock_ctx, mock_git, _remote_sha, _porcelain,
    ):
        srv._preflight_capability_baseline_porcelain = ""
        mock_ctx.return_value = {
            "workspace_path": BRANCHES_WORKTREE,
            "canonical_repo_root": CONTROL_ROOT,
            "process_project_root": BRANCHES_WORKTREE,
        }
        mock_git.return_value = {
            "current_branch": "feat/issue-475-root-checkout-guard",
            "head_sha": OTHER_SHA,
            "porcelain_status": "",
        }
        srv.verify_preflight_purity("prgs", worktree_path=BRANCHES_WORKTREE)
 if __name__ == "__main__":
    unittest.main()
@@ -36,11 +36,7 @@ def _lock(mutations=None, correction=False):
 def _seed(mutations=None, correction=False):
    import review_workflow_load
    review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
    mcp_server._save_review_decision_lock(_lock(mutations, correction))
    mcp_server.gitea_load_review_workflow()
 APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1,
@@ -52,7 +48,6 @@ RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2,
 class TestTerminalHardStopReasons(unittest.TestCase):
    def tearDown(self):
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()
    def test_no_lock_no_reasons(self):
        mcp_server._save_review_decision_lock(None)
@@ -97,10 +92,7 @@ class TestTerminalHardStopReasons(unittest.TestCase):
 class TestMergeHardStopWiring(unittest.TestCase):
    def tearDown(self):
        import review_workflow_load
        review_workflow_load.clear_review_workflow_load()
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()
    def test_merge_blocked_after_request_changes(self):
        _seed([RC_A])
@@ -121,10 +113,7 @@ class TestMergeHardStopWiring(unittest.TestCase):
 class TestMarkFinalHardStopWiring(unittest.TestCase):
    def tearDown(self):
        import review_workflow_load
        review_workflow_load.clear_review_workflow_load()
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()
    def test_mark_ready_blocked_after_terminal_mutation(self):
        _seed([RC_A])
@@ -156,10 +145,7 @@ def _mark(action, pr_number=6, **kwargs):
 class TestDuplicateRequestChangesSuppression(unittest.TestCase):
    def tearDown(self):
        import review_workflow_load
        review_workflow_load.clear_review_workflow_load()
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()
    def test_duplicate_request_changes_blocked_at_same_head(self):
        _seed()
@@ -126,37 +126,17 @@ class TestRuntimeContextGuardAlignment(unittest.TestCase):
            with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
                srv.verify_preflight_purity(worktree_path=BRANCHES_WORKTREE)
-    @mock.patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
+    def test_stable_checkout_still_rejected(self):
    @mock.patch(
        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value={
            "current_branch": "master",
            "head_sha": "a" * 40,
            "porcelain_status": "",
        },
    )
    def test_stable_checkout_still_rejected(self, _git, _remote_sha):
        with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT):
            with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
                with self.assertRaises(RuntimeError) as ctx:
                    srv.verify_preflight_purity()
        self.assertIn("stable control checkout", str(ctx.exception))
    @mock.patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
    @mock.patch(
        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value={
            "current_branch": "master",
            "head_sha": "a" * 40,
            "porcelain_status": "",
        },
    )
    @mock.patch("os.path.isdir", return_value=True)
    @mock.patch("os.path.exists", return_value=True)
    @mock.patch("subprocess.run")
-    def test_non_branches_worktree_rejected(
+    def test_non_branches_worktree_rejected(self, mock_run, *_exists):
        self, mock_run, mock_exists, mock_isdir, _git, _remote_sha,
    ):
        outside = "/tmp/outside-repo-checkout"
        mock_run.return_value = MagicMock(
            returncode=0,
@@ -1,383 +0,0 @@
 """Tests for session-owned worktree cleanup audit and TTL enforcement (#401)."""
 import sys
 import unittest
 from datetime import datetime, timedelta, timezone
 from unittest.mock import patch
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 import worktree_cleanup_audit as wca  # noqa: E402
 NOW = datetime(2026, 7, 7, 12, 0, 0, tzinfo=timezone.utc)
 def _iso(dt):
    return dt.isoformat().replace("+00:00", "Z")
 class TestWorkflowTypeInference(unittest.TestCase):
    def test_review_pr_path(self):
        self.assertEqual(
            wca.infer_workflow_type("branches/review-pr376", "review-pr376"),
            wca.WORKFLOW_REVIEW,
        )
    def test_baseline_path(self):
        self.assertEqual(
            wca.infer_workflow_type("branches/baseline-master-issue-401"),
            wca.WORKFLOW_BASELINE,
        )
    def test_merge_simulation_path(self):
        self.assertEqual(
            wca.infer_workflow_type("branches/merge-sim-pr380"),
            wca.WORKFLOW_MERGE_SIMULATION,
        )
    def test_issue_work_branch(self):
        self.assertEqual(
            wca.infer_workflow_type(
                "branches/issue-401-worktree", "feat/issue-401-worktree"
            ),
            wca.WORKFLOW_ISSUE_WORK,
        )
    def test_conflict_fix_path(self):
        self.assertEqual(
            wca.infer_workflow_type("branches/conflict-fix-pr376"),
            wca.WORKFLOW_CONFLICT_FIX,
        )
    def test_unknown_path(self):
        self.assertEqual(
            wca.infer_workflow_type("branches/scratchpad"), wca.WORKFLOW_UNKNOWN
        )
 class TestMetadata(unittest.TestCase):
    def test_metadata_fields_present(self):
        meta = wca.build_worktree_metadata(
            path="branches/issue-401-worktree",
            branch="feat/issue-401-worktree",
            head_sha="abc123",
            creator="jcwalker3",
            profile="prgs-author",
            created_at=_iso(NOW),
            last_used_at=_iso(NOW),
        )
        for field in (
            "path",
            "workflow_type",
            "issue_number",
            "pr_number",
            "branch",
            "head_sha",
            "creator",
            "profile",
            "created_at",
            "last_used_at",
            "cleanup_eligibility",
        ):
            self.assertIn(field, meta)
        self.assertEqual(meta["issue_number"], 401)
        self.assertEqual(meta["workflow_type"], wca.WORKFLOW_ISSUE_WORK)
    def test_review_metadata_auto_removable_flag(self):
        meta = wca.build_worktree_metadata(path="branches/review-pr42")
        self.assertTrue(meta["auto_remove_on_success"])
 class TestTTL(unittest.TestCase):
    def test_expired(self):
        old = _iso(NOW - timedelta(hours=48))
        self.assertTrue(wca.is_ttl_expired(last_used_at=old, now=NOW, ttl_hours=24))
    def test_not_expired(self):
        recent = _iso(NOW - timedelta(hours=1))
        self.assertFalse(
            wca.is_ttl_expired(last_used_at=recent, now=NOW, ttl_hours=24)
        )
    def test_unknown_timestamp_fails_safe(self):
        self.assertFalse(wca.is_ttl_expired(last_used_at=None, now=NOW))
        self.assertFalse(
            wca.is_ttl_expired(last_used_at="not-a-date", now=NOW)
        )
 class TestClassification(unittest.TestCase):
    def test_successful_review_cleanup(self):
        # Scenario 1: clean review worktree -> removable.
        cls = wca.classify_worktree(
            workflow_type=wca.WORKFLOW_REVIEW, is_dirty=False
        )
        self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
        self.assertTrue(wca.is_removable(cls))
    def test_dirty_worktree_preserved(self):
        # Scenario 3: dirty worktree is never removable.
        cls = wca.classify_worktree(
            workflow_type=wca.WORKFLOW_REVIEW, is_dirty=True
        )
        self.assertEqual(cls, wca.CLASS_DIRTY_LOCAL)
        self.assertFalse(wca.is_removable(cls))
    def test_active_pr_worktree_preserved(self):
        # Scenario 4: open PR wins over an otherwise-removable review worktree.
        cls = wca.classify_worktree(
            workflow_type=wca.WORKFLOW_REVIEW,
            is_dirty=False,
            has_open_pr=True,
        )
        self.assertEqual(cls, wca.CLASS_ACTIVE_OPEN_PR)
        self.assertFalse(wca.is_removable(cls))
    def test_stale_clean_issue_worktree_removable(self):
        # Scenario 5: clean issue worktree, TTL expired, no lock -> removable.
        cls = wca.classify_worktree(
            workflow_type=wca.WORKFLOW_ISSUE_WORK,
            is_dirty=False,
            ttl_expired=True,
        )
        self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
        self.assertTrue(wca.is_removable(cls))
    def test_fresh_clean_issue_worktree_preserved(self):
        # Clean issue worktree not yet TTL-expired stays active.
        cls = wca.classify_worktree(
            workflow_type=wca.WORKFLOW_ISSUE_WORK,
            is_dirty=False,
            ttl_expired=False,
        )
        self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
        self.assertFalse(wca.is_removable(cls))
    def test_detached_review_worktree_classified(self):
        # Scenario 6: detached review worktree -> detached_review_leftover.
        cls = wca.classify_worktree(
            workflow_type=wca.WORKFLOW_REVIEW,
            is_dirty=False,
            is_detached=True,
        )
        self.assertEqual(cls, wca.CLASS_DETACHED_REVIEW_LEFTOVER)
        self.assertTrue(wca.is_removable(cls))
    def test_ttl_expired_but_dirty_preserved(self):
        # Scenario 7: dirty wins over TTL expiry.
        cls = wca.classify_worktree(
            workflow_type=wca.WORKFLOW_ISSUE_WORK,
            is_dirty=True,
            ttl_expired=True,
        )
        self.assertEqual(cls, wca.CLASS_DIRTY_LOCAL)
        self.assertFalse(wca.is_removable(cls))
    def test_lease_protected_worktree_preserved(self):
        # Scenario 8: active lease is never removable.
        cls = wca.classify_worktree(
            workflow_type=wca.WORKFLOW_REVIEW,
            is_dirty=False,
            has_active_lease=True,
        )
        self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
        self.assertFalse(wca.is_removable(cls))
    def test_active_issue_lock_preserved(self):
        cls = wca.classify_worktree(
            workflow_type=wca.WORKFLOW_ISSUE_WORK,
            is_dirty=False,
            has_active_issue_lock=True,
            ttl_expired=True,
        )
        self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
        self.assertFalse(wca.is_removable(cls))
    def test_protected_base_worktree_never_removable(self):
        cls = wca.classify_worktree(
            workflow_type=wca.WORKFLOW_ISSUE_WORK,
            is_dirty=False,
            is_protected=True,
            ttl_expired=True,
        )
        self.assertEqual(cls, wca.CLASS_UNSAFE_UNKNOWN)
        self.assertFalse(wca.is_removable(cls))
    def test_unknown_workflow_type_unsafe(self):
        cls = wca.classify_worktree(
            workflow_type=wca.WORKFLOW_UNKNOWN, is_dirty=False, ttl_expired=True
        )
        self.assertEqual(cls, wca.CLASS_UNSAFE_UNKNOWN)
        self.assertFalse(wca.is_removable(cls))
 class TestRemovalDecision(unittest.TestCase):
    def test_safe_removal_proof(self):
        decision = wca.assess_worktree_removal(
            path="branches/review-pr42",
            branch="review-pr42",
            head_sha="abc123",
            is_dirty=False,
            has_open_pr=False,
            has_active_lease=False,
            classification=wca.CLASS_CLEAN_STALE_REMOVABLE,
        )
        self.assertTrue(decision["safe_to_remove"])
        self.assertEqual(decision["block_reasons"], [])
        self.assertTrue(decision["clean"])
        self.assertTrue(decision["no_active_pr"])
        self.assertTrue(decision["no_active_lease"])
    def test_dirty_blocks_removal(self):
        decision = wca.assess_worktree_removal(
            path="branches/review-pr42",
            branch="review-pr42",
            head_sha="abc123",
            is_dirty=True,
            has_open_pr=False,
            has_active_lease=False,
            classification=wca.CLASS_DIRTY_LOCAL,
        )
        self.assertFalse(decision["safe_to_remove"])
        self.assertIn("worktree has uncommitted changes", decision["block_reasons"])
    def test_open_pr_and_lease_block_removal(self):
        decision = wca.assess_worktree_removal(
            path="branches/review-pr42",
            branch="review-pr42",
            head_sha="abc123",
            is_dirty=False,
            has_open_pr=True,
            has_active_lease=True,
            classification=wca.CLASS_ACTIVE_OPEN_PR,
        )
        self.assertFalse(decision["safe_to_remove"])
        self.assertIn("worktree branch has an open PR", decision["block_reasons"])
        self.assertIn("worktree has an active lease", decision["block_reasons"])
 class TestSuccessCleanupPlan(unittest.TestCase):
    def test_review_worktree_removed_at_success(self):
        # Scenario 1 end-to-end: clean review worktree removed at success.
        meta = wca.build_worktree_metadata(path="branches/review-pr42")
        plan = wca.plan_success_cleanup(
            metadata=meta, is_dirty=False, has_open_pr=False, has_active_lease=False
        )
        self.assertTrue(plan["remove"])
    def test_failed_review_leaves_worktree_reported(self):
        # Scenario 2: dirty review worktree preserved and reported at failure.
        meta = wca.build_worktree_metadata(path="branches/review-pr42")
        plan = wca.plan_success_cleanup(
            metadata=meta, is_dirty=True, has_open_pr=False, has_active_lease=False
        )
        self.assertFalse(plan["remove"])
        self.assertIn("uncommitted", plan["reason"])
        report = wca.cleanup_failure_report(meta["path"], plan["reason"])
        self.assertFalse(report["removed"])
        self.assertEqual(report["path"], "branches/review-pr42")
    def test_issue_worktree_preserved_by_policy(self):
        meta = wca.build_worktree_metadata(
            path="branches/issue-401-worktree", branch="feat/issue-401-worktree"
        )
        plan = wca.plan_success_cleanup(
            metadata=meta, is_dirty=False, has_open_pr=False, has_active_lease=False
        )
        self.assertFalse(plan["remove"])
        self.assertIn("preserved by policy", plan["reason"])
 class TestPorcelainParser(unittest.TestCase):
    def test_parse_branch_and_detached(self):
        text = (
            "worktree /repo\n"
            "HEAD 1111111111111111111111111111111111111111\n"
            "branch refs/heads/master\n"
            "\n"
            "worktree /repo/branches/review-pr42\n"
            "HEAD 2222222222222222222222222222222222222222\n"
            "detached\n"
        )
        entries = wca.parse_worktree_porcelain(text)
        self.assertEqual(len(entries), 2)
        self.assertEqual(entries[0]["branch"], "master")
        self.assertFalse(entries[0]["detached"])
        self.assertIsNone(entries[1]["branch"])
        self.assertTrue(entries[1]["detached"])
 class TestAuditReportAccuracy(unittest.TestCase):
    """Scenario 9: cleanup report accuracy over a mixed branches/ directory."""
    PORCELAIN = (
        "worktree /repo\n"
        "HEAD 1111111111111111111111111111111111111111\n"
        "branch refs/heads/master\n"
        "\n"
        "worktree /repo/branches/review-pr42\n"
        "HEAD 2222222222222222222222222222222222222222\n"
        "branch refs/heads/review-pr42\n"
        "\n"
        "worktree /repo/branches/issue-400-open-pr\n"
        "HEAD 3333333333333333333333333333333333333333\n"
        "branch refs/heads/feat/issue-400-open-pr\n"
        "\n"
        "worktree /repo/branches/issue-401-dirty\n"
        "HEAD 4444444444444444444444444444444444444444\n"
        "branch refs/heads/feat/issue-401-dirty\n"
    )
    def _fake_dirty(self, path):
        if path.endswith("issue-401-dirty"):
            return {"exists": True, "dirty": True, "dirty_files": [" M x.py"]}
        return {"exists": True, "dirty": False, "dirty_files": []}
    def test_mixed_directory_classified(self):
        with patch.object(
            wca,
            "list_worktrees",
            return_value=wca.parse_worktree_porcelain(self.PORCELAIN),
        ), patch.object(
            wca, "read_worktree_dirty", side_effect=self._fake_dirty
        ), patch.object(
            wca, "git_worktree_list", return_value="(mocked)"
        ):
            report = wca.audit_branches_directory(
                "/repo",
                open_pr_branches={"feat/issue-400-open-pr"},
            )
        by_path = {wt["path"]: wt for wt in report["worktrees"]}
        # main checkout on master -> protected -> unsafe/unknown, not removable
        self.assertEqual(
            by_path["/repo"]["classification"], wca.CLASS_UNSAFE_UNKNOWN
        )
        # clean review worktree -> removable
        self.assertEqual(
            by_path["/repo/branches/review-pr42"]["classification"],
            wca.CLASS_CLEAN_STALE_REMOVABLE,
        )
        # open PR branch -> preserved
        self.assertEqual(
            by_path["/repo/branches/issue-400-open-pr"]["classification"],
            wca.CLASS_ACTIVE_OPEN_PR,
        )
        # dirty worktree -> preserved
        self.assertEqual(
            by_path["/repo/branches/issue-401-dirty"]["classification"],
            wca.CLASS_DIRTY_LOCAL,
        )
        # exactly one removable candidate (the clean review worktree)
        self.assertEqual(report["removable_count"], 1)
        self.assertEqual(
            report["removable_candidates"][0]["path"],
            "/repo/branches/review-pr42",
        )
        self.assertEqual(report["total"], 4)
        self.assertEqual(report["git_worktree_list"], "(mocked)")
 if __name__ == "__main__":
    unittest.main()
@@ -1,480 +0,0 @@
 """Session-owned worktree cleanup audit and TTL enforcement (#401).
 LLM workflows create many session-owned worktrees under ``branches/``
 (review, baseline, merge-simulation, issue, and conflict-fix worktrees).
 When a run stops early, races a sibling session, hits a validation failure,
 or loses shell/cwd state, those worktrees are left behind and later workflow
 decisions get harder and riskier.
 This module provides:
 * ``build_worktree_metadata`` — ownership/purpose metadata for a worktree.
 * ``classify_worktree`` — safety-first classification into the audit
  vocabulary (active open PR, active issue work, dirty, clean stale
  removable, detached review leftover, unsafe/unknown).
 * ``assess_worktree_removal`` — a per-worktree removal decision with an
  explicit proof and block reasons.
 * git-shelling helpers (``list_worktrees``, ``read_worktree_dirty``,
  ``git_worktree_list``, ``remove_worktree``) and ``audit_branches_directory``
  that classify every entry under ``branches/``.
 Pure assessment functions take explicit state so they are unit-testable
 without a filesystem or network. Only the thin git helpers shell out, and
 removal is only ever executed after ``assess_worktree_removal`` proves the
 worktree is safe to delete.
 """
 from __future__ import annotations
 import os
 import re
 import subprocess
 from datetime import datetime, timezone
 from typing import Any
 PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
 DEFAULT_TTL_HOURS = float(os.environ.get("GITEA_WORKTREE_TTL_HOURS", "24") or 24)
 # Workflow types that can create session-owned worktrees.
 WORKFLOW_REVIEW = "review"
 WORKFLOW_BASELINE = "baseline"
 WORKFLOW_MERGE_SIMULATION = "merge_simulation"
 WORKFLOW_ISSUE_WORK = "issue_work"
 WORKFLOW_CONFLICT_FIX = "conflict_fix"
 WORKFLOW_UNKNOWN = "unknown"
 # Workflow types whose worktrees are transient and removed automatically at
 # successful workflow completion (acceptance criterion 2).
 AUTO_REMOVE_ON_SUCCESS = frozenset(
    {WORKFLOW_REVIEW, WORKFLOW_BASELINE, WORKFLOW_MERGE_SIMULATION}
 )
 # Cleanup-audit classification vocabulary (acceptance criterion 4).
 CLASS_ACTIVE_OPEN_PR = "active_open_pr"
 CLASS_ACTIVE_ISSUE_WORK = "active_issue_work"
 CLASS_DIRTY_LOCAL = "dirty_local_worktree"
 CLASS_CLEAN_STALE_REMOVABLE = "clean_stale_removable"
 CLASS_DETACHED_REVIEW_LEFTOVER = "detached_review_leftover"
 CLASS_UNSAFE_UNKNOWN = "unsafe_unknown"
 # Only these two classifications may ever be removed automatically.
 REMOVABLE_CLASSES = frozenset(
    {CLASS_CLEAN_STALE_REMOVABLE, CLASS_DETACHED_REVIEW_LEFTOVER}
 )
 _ISSUE_REF_RE = re.compile(r"issue-(\d+)", re.IGNORECASE)
 _ISSUE_BRANCH_PREFIXES = ("feat/", "fix/", "docs/", "chore/")
 def infer_workflow_type(path: str | None, branch: str | None = None) -> str:
    """Infer the creating workflow type from a worktree path or branch name."""
    text = f"{path or ''} {branch or ''}".lower()
    if "baseline" in text:
        return WORKFLOW_BASELINE
    if "merge-sim" in text or "merge_sim" in text or "mergesim" in text:
        return WORKFLOW_MERGE_SIMULATION
    if "review" in text or "review-pr" in text:
        return WORKFLOW_REVIEW
    if "conflict" in text:
        return WORKFLOW_CONFLICT_FIX
    if "issue-" in text or (branch or "").startswith(_ISSUE_BRANCH_PREFIXES):
        return WORKFLOW_ISSUE_WORK
    return WORKFLOW_UNKNOWN
 def _extract_issue_number(path: str | None, branch: str | None) -> int | None:
    match = _ISSUE_REF_RE.search(f"{path or ''} {branch or ''}")
    return int(match.group(1)) if match else None
 def build_worktree_metadata(
    *,
    path: str,
    branch: str | None = None,
    head_sha: str | None = None,
    workflow_type: str | None = None,
    issue_number: int | None = None,
    pr_number: int | None = None,
    creator: str | None = None,
    profile: str | None = None,
    created_at: str | None = None,
    last_used_at: str | None = None,
    cleanup_eligibility: str | None = None,
 ) -> dict[str, Any]:
    """Return ownership/purpose metadata for a session-owned worktree.
    Covers acceptance criterion 1: path, workflow type, issue/PR number,
    branch/head SHA, creator identity/profile, created and last-used
    timestamps, and cleanup eligibility.
    """
    wt = workflow_type or infer_workflow_type(path, branch)
    issue = issue_number
    if issue is None:
        issue = _extract_issue_number(path, branch)
    return {
        "path": path,
        "workflow_type": wt,
        "issue_number": issue,
        "pr_number": pr_number,
        "branch": branch,
        "head_sha": head_sha,
        "creator": creator,
        "profile": profile,
        "created_at": created_at,
        "last_used_at": last_used_at,
        "auto_remove_on_success": wt in AUTO_REMOVE_ON_SUCCESS,
        "cleanup_eligibility": cleanup_eligibility,
    }
 def _parse_timestamp(value: str | None) -> datetime | None:
    if not value:
        return None
    text = str(value).strip()
    if not text:
        return None
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=timezone.utc)
    return parsed
 def is_ttl_expired(
    *,
    last_used_at: str | None,
    now: datetime | str | None,
    ttl_hours: float = DEFAULT_TTL_HOURS,
 ) -> bool:
    """Return True only when the age is known and exceeds ``ttl_hours``.
    Unknown or unparseable timestamps fail safe (not expired) so a worktree
    is never treated as removable merely because its age is unknown.
    """
    last = _parse_timestamp(last_used_at)
    now_dt = now if isinstance(now, datetime) else _parse_timestamp(now)
    if last is None or now_dt is None:
        return False
    if now_dt.tzinfo is None:
        now_dt = now_dt.replace(tzinfo=timezone.utc)
    return (now_dt - last).total_seconds() > ttl_hours * 3600.0
 def classify_worktree(
    *,
    workflow_type: str,
    is_dirty: bool,
    has_open_pr: bool = False,
    has_active_lease: bool = False,
    has_active_issue_lock: bool = False,
    is_detached: bool = False,
    branch_gone: bool = False,
    ttl_expired: bool = False,
    is_protected: bool = False,
    metadata_known: bool = True,
 ) -> str:
    """Classify a worktree, safety-first: any preservation signal wins.
    Dirty, open-PR, leased, active-lock, protected, and unknown states are
    all non-removable and are checked before any removable classification,
    so nothing removable can shadow a preservation signal (criteria 6-8).
    """
    if is_protected:
        # The main checkout / a protected base branch is never removable.
        return CLASS_UNSAFE_UNKNOWN
    if is_dirty:
        return CLASS_DIRTY_LOCAL  # never auto-deleted (criterion 6)
    if has_open_pr:
        return CLASS_ACTIVE_OPEN_PR  # never auto-deleted (criterion 7)
    if has_active_lease:
        return CLASS_ACTIVE_ISSUE_WORK  # never auto-deleted (criterion 8)
    if has_active_issue_lock:
        return CLASS_ACTIVE_ISSUE_WORK
    if not metadata_known or workflow_type == WORKFLOW_UNKNOWN:
        return CLASS_UNSAFE_UNKNOWN  # never auto-deleted without proof
    # Clean, no PR, no lease, no lock, known workflow type.
    if workflow_type in AUTO_REMOVE_ON_SUCCESS:
        if is_detached or branch_gone:
            return CLASS_DETACHED_REVIEW_LEFTOVER
        return CLASS_CLEAN_STALE_REMOVABLE
    # issue_work / conflict_fix: only removable once the TTL has expired.
    if ttl_expired:
        return CLASS_CLEAN_STALE_REMOVABLE
    return CLASS_ACTIVE_ISSUE_WORK
 def is_removable(classification: str) -> bool:
    """Return True only for the two auto-removable classifications."""
    return classification in REMOVABLE_CLASSES
 def assess_worktree_removal(
    *,
    path: str,
    branch: str | None,
    head_sha: str | None,
    is_dirty: bool,
    has_open_pr: bool,
    has_active_lease: bool,
    classification: str,
 ) -> dict[str, Any]:
    """Return a per-worktree removal decision with proof (criterion 9).
    A worktree is only safe to remove when it is clean, has no active PR,
    has no active lease, and its classification is auto-removable.
    """
    block_reasons: list[str] = []
    if is_dirty:
        block_reasons.append("worktree has uncommitted changes")
    if has_open_pr:
        block_reasons.append("worktree branch has an open PR")
    if has_active_lease:
        block_reasons.append("worktree has an active lease")
    if not is_removable(classification):
        block_reasons.append(
            f"classification '{classification}' is not auto-removable"
        )
    return {
        "path": path,
        "branch": branch,
        "head_sha": head_sha,
        "classification": classification,
        "clean": not is_dirty,
        "no_active_pr": not has_open_pr,
        "no_active_lease": not has_active_lease,
        "safe_to_remove": not block_reasons,
        "block_reasons": block_reasons,
    }
 def plan_success_cleanup(
    *,
    metadata: dict[str, Any],
    is_dirty: bool,
    has_open_pr: bool,
    has_active_lease: bool,
 ) -> dict[str, Any]:
    """Decide whether a just-completed worktree is removed at success.
    Review/baseline/merge-simulation worktrees are removed automatically at
    successful completion (criterion 2); everything else is preserved and
    reported. Dirty/PR/leased worktrees are always preserved (criteria 6-8).
    """
    workflow_type = metadata.get("workflow_type", WORKFLOW_UNKNOWN)
    if not metadata.get("auto_remove_on_success"):
        return {
            "remove": False,
            "reason": f"workflow type '{workflow_type}' is preserved by policy",
        }
    classification = classify_worktree(
        workflow_type=workflow_type,
        is_dirty=is_dirty,
        has_open_pr=has_open_pr,
        has_active_lease=has_active_lease,
    )
    decision = assess_worktree_removal(
        path=metadata.get("path", ""),
        branch=metadata.get("branch"),
        head_sha=metadata.get("head_sha"),
        is_dirty=is_dirty,
        has_open_pr=has_open_pr,
        has_active_lease=has_active_lease,
        classification=classification,
    )
    return {
        "remove": decision["safe_to_remove"],
        "reason": "clean transient worktree removable at success completion"
        if decision["safe_to_remove"]
        else "; ".join(decision["block_reasons"]),
        "classification": classification,
        "decision": decision,
    }
 def cleanup_failure_report(path: str, reason: str) -> dict[str, Any]:
    """Structured leftover-worktree record for the final report (criterion 3)."""
    return {"path": path, "removed": False, "reason": reason}
 # --------------------------------------------------------------------------
 # git-shelling helpers (only these touch the filesystem)
 # --------------------------------------------------------------------------
 def parse_worktree_porcelain(text: str) -> list[dict[str, Any]]:
    """Parse ``git worktree list --porcelain`` output into entries."""
    entries: list[dict[str, Any]] = []
    current: dict[str, Any] = {}
    for raw in (text or "").splitlines():
        line = raw.rstrip("\n")
        if not line:
            if current:
                entries.append(current)
                current = {}
            continue
        if line.startswith("worktree "):
            if current:
                entries.append(current)
            current = {
                "path": line[len("worktree ") :].strip(),
                "head": None,
                "branch": None,
                "detached": False,
                "bare": False,
            }
        elif line.startswith("HEAD "):
            current["head"] = line[len("HEAD ") :].strip()
        elif line.startswith("branch "):
            ref = line[len("branch ") :].strip()
            current["branch"] = ref.replace("refs/heads/", "", 1)
        elif line == "detached":
            current["detached"] = True
        elif line == "bare":
            current["bare"] = True
    if current:
        entries.append(current)
    return entries
 def list_worktrees(project_root: str) -> list[dict[str, Any]]:
    """Return parsed ``git worktree list`` entries for ``project_root``."""
    result = subprocess.run(
        ["git", "-C", project_root, "worktree", "list", "--porcelain"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        return []
    return parse_worktree_porcelain(result.stdout)
 def git_worktree_list(project_root: str) -> str:
    """Return plain ``git worktree list`` output for final verification (criterion 10)."""
    result = subprocess.run(
        ["git", "-C", project_root, "worktree", "list"],
        capture_output=True,
        text=True,
        check=False,
    )
    return (result.stdout or "").strip()
 def read_worktree_dirty(path: str) -> dict[str, Any]:
    """Return dirty state for a worktree path via ``git status --porcelain``."""
    if not path or not os.path.isdir(path):
        return {"exists": False, "dirty": None, "dirty_files": []}
    result = subprocess.run(
        ["git", "-C", path, "status", "--porcelain"],
        capture_output=True,
        text=True,
        check=False,
    )
    files = [ln for ln in (result.stdout or "").splitlines() if ln.strip()]
    return {"exists": True, "dirty": bool(files), "dirty_files": files}
 def remove_worktree(project_root: str, path: str) -> dict[str, Any]:
    """Remove a single worktree via ``git worktree remove`` (no ``--force``)."""
    result = subprocess.run(
        ["git", "-C", project_root, "worktree", "remove", path],
        capture_output=True,
        text=True,
        check=False,
    )
    ok = result.returncode == 0
    return {
        "path": path,
        "removed": ok,
        "reason": None if ok else (result.stderr or "git worktree remove failed").strip(),
    }
 def _is_under_branches(project_root: str, path: str) -> bool:
    branches_root = os.path.join(os.path.abspath(project_root), "branches")
    return os.path.abspath(path or "").startswith(branches_root + os.sep)
 def audit_branches_directory(
    project_root: str,
    *,
    open_pr_branches: set[str] | None = None,
    leased_branches: set[str] | None = None,
    active_issue_branches: set[str] | None = None,
    now: datetime | str | None = None,
    ttl_hours: float = DEFAULT_TTL_HOURS,
 ) -> dict[str, Any]:
    """Classify every session-owned worktree under ``branches/``.
    Read-only: shells out to git for discovery and dirty state, then applies
    the pure classifier. Returns per-worktree classifications, counts, the
    list of removable candidates, and the ``git worktree list`` proof.
    """
    open_pr_branches = open_pr_branches or set()
    leased_branches = leased_branches or set()
    active_issue_branches = active_issue_branches or set()
    worktrees: list[dict[str, Any]] = []
    for entry in list_worktrees(project_root):
        path = entry.get("path") or ""
        branch = entry.get("branch")
        is_protected = (branch in PROTECTED_BRANCHES) or not _is_under_branches(
            project_root, path
        )
        dirty_state = read_worktree_dirty(path)
        is_dirty = bool(dirty_state.get("dirty"))
        metadata = build_worktree_metadata(
            path=path, branch=branch, head_sha=entry.get("head")
        )
        has_open_pr = bool(branch) and branch in open_pr_branches
        has_active_lease = bool(branch) and branch in leased_branches
        has_active_lock = bool(branch) and branch in active_issue_branches
        ttl_expired = is_ttl_expired(
            last_used_at=metadata.get("last_used_at"), now=now, ttl_hours=ttl_hours
        )
        classification = classify_worktree(
            workflow_type=metadata["workflow_type"],
            is_dirty=is_dirty,
            has_open_pr=has_open_pr,
            has_active_lease=has_active_lease,
            has_active_issue_lock=has_active_lock,
            is_detached=bool(entry.get("detached")),
            branch_gone=branch is None and not entry.get("detached"),
            ttl_expired=ttl_expired,
            is_protected=is_protected,
        )
        metadata["cleanup_eligibility"] = classification
        worktrees.append(
            {
                **metadata,
                "detached": bool(entry.get("detached")),
                "dirty": is_dirty,
                "dirty_files": dirty_state.get("dirty_files", []),
                "has_open_pr": has_open_pr,
                "has_active_lease": has_active_lease,
                "has_active_issue_lock": has_active_lock,
                "is_protected": is_protected,
                "classification": classification,
                "removable": is_removable(classification),
            }
        )
    counts: dict[str, int] = {}
    for wt in worktrees:
        counts[wt["classification"]] = counts.get(wt["classification"], 0) + 1
    removable = [wt for wt in worktrees if wt["removable"]]
    return {
        "project_root": project_root,
        "worktrees": worktrees,
        "counts": counts,
        "removable_candidates": removable,
        "removable_count": len(removable),
        "total": len(worktrees),
        "git_worktree_list": git_worktree_list(project_root),
    }