docs(wiki): refresh Gitea-Tools wiki proof head to d1f0693 (#232)

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

capability_stop_terminal.py

@@ -53,28 +53,6 @@ def enter_from_capability_result(capability: dict) -> dict | None:
     return dict(record)
 
 
-def _is_reviewer_denial(capability: dict) -> bool:
-    task = (capability or {}).get("requested_task", "")
-    required_role = (capability or {}).get("required_role_kind")
-    return (
-        required_role == "reviewer"
-        or task in REVIEWER_CAPABILITY_TASKS
-    )
-
-
-def sync_from_capability_result(capability: dict) -> dict | None:
-    """Enter or clear terminal mode from a capability resolution (#238).
-
-    Reviewer denials activate terminal mode for the denied operation only.
-    A later allowed task route clears stale denial state so author read-only
-    tools (e.g. ``list_prs``) are not permanently blocked.
-    """
-    if (capability or {}).get("stop_required") and _is_reviewer_denial(capability):
-        return enter_from_capability_result(capability)
-    clear()
-    return None
-
-
 def enter_from_route_result(route: dict) -> dict | None:
     """Enter terminal mode from a role router wrong_role_stop (#206 compat)."""
     if (route or {}).get("route_result") != "wrong_role_stop":
@@ -112,13 +90,11 @@ def check_reviewer_queue_tool(tool_name: str) -> tuple[bool, list[str]]:
         return True, []
     name = (tool_name or "").strip().lower().removeprefix("gitea_")
     if name in BLOCKED_QUEUE_TOOLS:
-        denied_task = (_session_terminal or {}).get("requested_task") or "unknown"
         return False, [
             TERMINAL_REPORT_HEADING,
-            f"Reviewer queue tool '{tool_name}' is blocked by the current "
-            f"capability denial for task '{denied_task}' (fail closed).",
-            "Resolve or route an allowed author task to clear stale denial "
-            "state, or relaunch a reviewer MCP namespace for reviewer work.",
+            f"Reviewer queue tool '{tool_name}' is blocked after "
+            "capability denial (fail closed).",
+            "Relaunch a reviewer MCP namespace to perform reviewer work.",
         ]
     return True, []
 
@@ -172,14 +148,8 @@ def assess_capability_stop_report(
         r"inventory empty",
         re.I,
     )
-    parsed_status = None
-    for line in text.splitlines():
-        if "pr_inventory_trust_gate.status:" in line.lower():
-            parsed_status = line.split(":", 1)[1].strip()
-            break
-    effective_status = trust_gate_status or parsed_status
     if empty_queue_patterns.search(text):
-        if effective_status != "trusted_empty":
+        if trust_gate_status != "trusted_empty":
             violations.append(
                 "empty-queue claim after capability stop without "
                 "pr_inventory_trust_gate.status == trusted_empty"

docs/wiki/Repositories.md

@@ -32,7 +32,6 @@ Wiki-related issues cannot be closed until the live Wiki is verified — see
 | Repository | `docs/wiki/` source | Gitea Wiki published | Proof |
 |---|---|---|---|
 | `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `d1f0693` |
-| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home); 10 pages (History, Home, Identity-and-Profiles, MCP-Tools, Open-Decisions, Operator-Guide, Repositories, Runbooks, Safety-and-Gates, Workflow); wiki git log head `ef3dec2` |
-
+| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home) |
 
 Update this table whenever a wiki is published, re-synced, or found stale.

gitea_mcp_server.py

@@ -104,7 +104,7 @@ def verify_mutation_authority(remote: str | None, host: str | None = None,
         )
 
     session_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
-    if session_lock and session_lock != active_profile and not gitea_config.is_runtime_switching_enabled():
+    if session_lock and session_lock != active_profile:
         raise RuntimeError(
             f"Active profile '{active_profile}' does not match the session "
             f"profile lock '{session_lock}' — profile side-channel override "
@@ -259,8 +259,6 @@ import issue_duplicate_gate  # noqa: E402
 import role_session_router  # noqa: E402
 import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
-import review_proofs  # noqa: E402
-import issue_lock_worktree  # noqa: E402
 
 
 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
@@ -416,67 +414,6 @@ def _authenticated_username(host: str):
     return user
 
 
-def _ensure_matching_profile(required_permission: str, required_role: str, remote: str | None, host: str | None = None) -> str | None:
-    """Check if the active profile is allowed to perform *required_permission*.
-    If not, automatically switch to the first matching usable configured profile.
-    """
-    try:
-        profile = get_profile()
-    except Exception:
-        return None
-    active_profile = profile.get("profile_name")
-    active_allowed = profile.get("allowed_operations") or []
-    active_forbidden = profile.get("forbidden_operations") or []
-    allowed, _ = gitea_config.check_operation(required_permission, active_allowed, active_forbidden)
-    if allowed:
-        return active_profile
-
-    # Try to find a matching usable profile in config
-    if gitea_config.is_runtime_switching_enabled():
-        config = gitea_config.load_config()
-        if config and "profiles" in config:
-            for p_name, p_data in config["profiles"].items():
-                p_allowed = p_data.get("allowed_operations") or []
-                p_forbidden = p_data.get("forbidden_operations") or []
-                p_role = p_data.get("role") or _role_kind(p_allowed, p_forbidden)
-                if required_role and p_role != required_role:
-                    continue
-                p_allowed_n = []
-                for op in p_allowed:
-                    try:
-                        p_allowed_n.append(gitea_config.normalize_operation(op))
-                    except Exception:
-                        pass
-                p_forbidden_n = []
-                for op in p_forbidden:
-                    try:
-                        p_forbidden_n.append(gitea_config.normalize_operation(op))
-                    except Exception:
-                        pass
-                ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n)
-                if ok:
-                    # Verify credentials/token are available
-                    try:
-                        tok = gitea_config.resolve_token(p_data)
-                        if tok:
-                            # Perform automatic switch
-                            gitea_config._active_profile_override = p_name
-                            h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
-                            if h:
-                                _IDENTITY_CACHE.pop(h, None)
-                            username = _authenticated_username(h) if h else None
-                            # Update mutation authority
-                            global _MUTATION_AUTHORITY
-                            if _MUTATION_AUTHORITY is not None:
-                                _MUTATION_AUTHORITY["current_profile"] = p_name
-                                _MUTATION_AUTHORITY["current_identity"] = username
-                                _MUTATION_AUTHORITY["role_pivot_authorized"] = True
-                            return p_name
-                    except Exception:
-                        pass
-    return None
-
-
 def _audit(action: str, *, host, remote, result, org=None, repo=None,
            reason=None, request_metadata=None, issue_number=None,
            pr_number=None, target_branch=None, head_sha=None, username=_UNSET,
@@ -687,7 +624,6 @@ def gitea_lock_issue(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
-    worktree_path: str | None = None,
 ) -> dict:
     """Lock exactly one Gitea issue and its branch name to ensure durable tracking.
 
@@ -698,8 +634,6 @@ def gitea_lock_issue(
         host: Override Gitea host.
         org: Override Org.
         repo: Override Repo.
-        worktree_path: Author scratch-clone path to validate (defaults to
-            GITEA_AUTHOR_WORKTREE or the MCP server project root).
     """
     # 1. Enforce branch name includes issue number
     expected_pattern = f"issue-{issue_number}"
@@ -708,20 +642,6 @@ def gitea_lock_issue(
             f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
         )
 
-    resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
-        worktree_path, PROJECT_ROOT
-    )
-    git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
-    lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
-        worktree_path=resolved_worktree,
-        current_branch=git_state.get("current_branch"),
-        porcelain_status=git_state.get("porcelain_status") or "",
-    )
-    if lock_assessment["block"]:
-        raise RuntimeError(
-            issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
-        )
-
     # 2. Check if the issue already has an open PR (reuse protection)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -758,7 +678,6 @@ def gitea_lock_issue(
         "remote": remote,
         "org": o,
         "repo": r,
-        "worktree_path": resolved_worktree,
     }
 
     try:
@@ -769,13 +688,9 @@ def gitea_lock_issue(
 
     return {
         "success": True,
-        "message": (
-            f"Successfully locked issue #{issue_number} to branch '{branch_name}' "
-            f"from worktree '{resolved_worktree}' (fail-closed check complete)."
-        ),
+        "message": f"Successfully locked issue #{issue_number} to branch '{branch_name}' (fail-closed check complete).",
         "issue_number": issue_number,
         "branch_name": branch_name,
-        "worktree_path": resolved_worktree,
     }
 
 
@@ -789,7 +704,6 @@ def gitea_create_pr(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
-    worktree_path: str | None = None,
 ) -> dict:
     """Create a pull request on a Gitea repository.
 
@@ -802,7 +716,6 @@ def gitea_create_pr(
         host: Override the Gitea host.
         org: Override the owner/organization.
         repo: Override the repository name.
-        worktree_path: Author worktree path; must match the path stored at lock time.
 
     Returns:
         dict with 'number' of the created PR ('url' only with the reveal opt-in).
@@ -841,13 +754,6 @@ def gitea_create_pr(
 
     locked_issue = lock_data.get("issue_number")
     locked_branch = lock_data.get("branch_name")
-    locked_worktree = lock_data.get("worktree_path")
-
-    worktree_check = issue_lock_worktree.verify_pr_worktree_matches_lock(
-        locked_worktree, worktree_path, PROJECT_ROOT
-    )
-    if worktree_check["block"]:
-        raise ValueError(worktree_check["reasons"][0])
 
     if head != locked_branch:
         raise ValueError(
@@ -1273,8 +1179,7 @@ _REVIEW_ACTIONS = {
     # 'comment' posts review findings without an approval/rejection state.
     # #14 names this eligibility category 'review'.
     "comment": ("review", "COMMENT"),
-    # Gitea ReviewStateType uses APPROVED, not APPROVE — wrong event leaves PENDING (#244).
-    "approve": ("approve", "APPROVED"),
+    "approve": ("approve", "APPROVE"),
     "request_changes": ("request_changes", "REQUEST_CHANGES"),
 }
 
@@ -1484,42 +1389,6 @@ def _redact(text: str) -> str:
 # neither may drive the blocking/approval summaries.
 _VERDICT_STATES = ("APPROVED", "REQUEST_CHANGES", "COMMENT")
 
-# Terminal review events that must be visible after live submission (#244).
-_SUBMIT_VISIBLE_EVENTS = frozenset({"APPROVED", "REQUEST_CHANGES"})
-
-
-def _latest_review_state_for_reviewer(raw_reviews: list, reviewer: str) -> str | None:
-    """Return the latest non-COMMENT verdict for *reviewer*, or None."""
-    ordered = sorted(
-        raw_reviews or [],
-        key=lambda rv: ((rv.get("submitted_at") or ""), rv.get("id") or 0),
-    )
-    latest = None
-    for rv in ordered:
-        state = (rv.get("state") or "").upper()
-        login = (rv.get("user") or {}).get("login", "")
-        if login != reviewer or state not in _VERDICT_STATES or state == "COMMENT":
-            continue
-        latest = state
-    return latest
-
-
-def _submit_pending_pull_review(
-    h: str,
-    o: str,
-    r: str,
-    pr_number: int,
-    review_id: int,
-    event: str,
-    body: str,
-    auth,
-) -> dict | None:
-    """Submit a PENDING draft review via Gitea's pending-review endpoint."""
-    submit_url = (
-        f"{repo_api_url(h, o, r)}/pulls/{pr_number}/reviews/{review_id}"
-    )
-    return api_request("POST", submit_url, auth, {"body": body, "event": event})
-
 
 @mcp.tool()
 def gitea_get_pr_review_feedback(
@@ -1760,7 +1629,6 @@ def _evaluate_pr_review_submission(
 
     h, o, r = _resolve(remote, host, org, repo)
     review_id = None
-    submitted_state = None
     try:
         auth = _auth(h)
         review_url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}/reviews"
@@ -1768,43 +1636,10 @@ def _evaluate_pr_review_submission(
         resp = api_request("POST", review_url, auth, payload)
         if isinstance(resp, dict):
             review_id = resp.get("id")
-            submitted_state = (resp.get("state") or "").upper() or None
-        if (
-            submitted_state == "PENDING"
-            and review_id
-            and event in _SUBMIT_VISIBLE_EVENTS
-        ):
-            resp = _submit_pending_pull_review(
-                h, o, r, pr_number, review_id, event, body, auth,
-            )
-            if isinstance(resp, dict):
-                submitted_state = (resp.get("state") or "").upper() or None
     except Exception as exc:  # noqa: BLE001 — redact before surfacing
         reasons.append(f"review submission failed: {_redact(str(exc))}")
         return result
 
-    if action in _TERMINAL_REVIEW_ACTIONS:
-        try:
-            auth = _auth(h)
-            raw_reviews = (
-                api_request("GET", f"{review_url}", auth) or []
-            )
-            visible = _latest_review_state_for_reviewer(raw_reviews, auth_user)
-        except Exception as exc:  # noqa: BLE001 — redact before surfacing
-            reasons.append(
-                f"could not verify submitted review verdict (fail closed): "
-                f"{_redact(str(exc))}"
-            )
-            return result
-        result["submitted_verdict"] = visible
-        result["review_verdict_visible"] = visible == event
-        if visible != event:
-            reasons.append(
-                f"review submission left verdict '{submitted_state or visible or 'PENDING'}'; "
-                f"expected visible '{event}' for reviewer '{auth_user}' (fail closed)"
-            )
-            return result
-
     record_live_review_mutation(pr_number, action, review_id)
     result["performed"] = True
     reasons.append(f"all gates passed; submitted '{event}' review on PR #{pr_number}")
@@ -2259,9 +2094,6 @@ def gitea_merge_pr(
     5. If ``expected_changed_files`` is given and the PR's changed file set
        differs → refuse.
     6. Redundant self-merge block (authenticated user == PR author).
-    7. Re-read formal review feedback (#167): refuse when
-       ``approval_visible`` is false or undismissed REQUEST_CHANGES block
-       merge — PENDING draft approvals do not count (#244).
 
     No force / ignore-checks option is exposed. Gitea's own ``mergeable`` signal
     (which reflects branch-protection required reviews and status checks) must
@@ -2386,33 +2218,7 @@ def gitea_merge_pr(
         reasons.append("self-merge blocked (authenticated user is PR author)")
         return result
 
-    # Gate 7 — visible formal approval required (#244). PENDING drafts and
-    # absent verdicts do not satisfy this gate even when Gitea mergeable is true.
-    feedback = gitea_get_pr_review_feedback(
-        pr_number=pr_number, remote=remote, host=host, org=org, repo=repo,
-    )
-    if not feedback.get("success"):
-        reasons.append("PR review feedback unavailable before merge (fail closed)")
-        reasons.extend(feedback.get("reasons", []))
-        if feedback.get("permission_report"):
-            result["permission_report"] = feedback["permission_report"]
-        return result
-    result["approval_visible"] = feedback.get("approval_visible")
-    result["has_blocking_change_requests"] = feedback.get(
-        "has_blocking_change_requests")
-    if feedback.get("has_blocking_change_requests"):
-        reasons.append(
-            "undismissed REQUEST_CHANGES review blocks merge (fail closed)"
-        )
-        return result
-    if not feedback.get("approval_visible"):
-        reasons.append(
-            "no visible APPROVED review on PR; verify review submission "
-            "completed before merge (fail closed)"
-        )
-        return result
-
-    # Gate 8 — in-process mutation authority (#199): the last check before
+    # Gate 7 — in-process mutation authority (#199): the last check before
     # the merge mutation, using the identity the eligibility gate proved.
     # A profile/identity flip or side-channel override between preflight
     # and merge fails closed here.
@@ -2468,23 +2274,6 @@ def gitea_merge_pr(
     return result
 
 
-def _local_git_remote_url(remote_name: str) -> str | None:
-    """Best-effort local ``git remote get-url`` for trust-gate corroboration."""
-    try:
-        proc = subprocess.run(
-            ["git", "remote", "get-url", remote_name],
-            capture_output=True,
-            text=True,
-            cwd=PROJECT_ROOT,
-        )
-        if proc.returncode != 0:
-            return None
-        url = (proc.stdout or "").strip()
-        return url or None
-    except Exception:
-        return None
-
-
 @mcp.tool()
 def gitea_review_pr(
     pr_number: int,
@@ -2552,8 +2341,6 @@ def gitea_review_pr(
     prs_found_count = 0
     pr_details_list = []
     inventory_msg = ""
-    inventory_trust_gate = None
-    prs: list = []
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = None
@@ -2638,28 +2425,7 @@ def gitea_review_pr(
             if inventory_msg:
                 report_lines.append(inventory_msg)
             else:
-                inventory_trust_gate = review_proofs.pr_inventory_trust_gate(
-                    prs if inventory_attempted else None,
-                    remote=remote,
-                    org=o,
-                    repo=r,
-                    state="open",
-                    authenticated_profile=profile,
-                    local_remote_url=_local_git_remote_url(remote),
-                    has_finality_metadata=True,
-                )
-                report_lines.extend(
-                    review_proofs.format_pr_inventory_trust_gate_report(
-                        inventory_trust_gate
-                    )
-                )
-                if inventory_trust_gate.get("status") == "trusted_empty":
-                    report_lines.append("Open PRs found: 0 (trusted_empty)")
-                else:
-                    report_lines.append(
-                        "Empty-queue claim blocked: "
-                        "pr_inventory_trust_gate did not return trusted_empty"
-                    )
+                report_lines.append("Open PRs found: 0")
     else:
         report_lines.append(inventory_msg)
 
@@ -2983,11 +2749,6 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
     Returns a block dict when the active profile forbids *required_operation*,
     or ``None`` when the gate passes. Never performs network I/O.
     """
-    req_role = "reviewer" if any(required_operation.startswith(p) for p in (
-        "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes", "gitea.pr.review"
-    )) else "author"
-    _ensure_matching_profile(required_operation, req_role, extra_fields.get("remote"))
-
     reasons = _profile_operation_gate(required_operation)
     if not reasons:
         return None
@@ -3003,10 +2764,6 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
 
 def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None:
     """Reviewer/author namespace alignment gate (#209)."""
-    required_permission = task_capability_map.required_permission(mutation_task)
-    required_role = task_capability_map.required_role(mutation_task)
-    _ensure_matching_profile(required_permission, required_role, extra_fields.get("remote"))
-
     try:
         profile = get_profile()
     except Exception as exc:
@@ -4613,9 +4370,6 @@ def gitea_resolve_task_capability(
 
     record_preflight_check("capability", required_role)
 
-    # Try automatic dispatch switching
-    _ensure_matching_profile(required_permission, required_role, remote, host)
-
     profile = get_profile()
     config = gitea_config.load_config()
 
@@ -4653,18 +4407,6 @@ def gitea_resolve_task_capability(
             if ok:
                 matching_profiles.append(p_name)
 
-    configured = len(matching_profiles) > 0
-    available_in_session = allowed_in_current_session
-    restart_required = False
-    reason = ""
-
-    if not allowed_in_current_session:
-        if configured:
-            restart_required = True
-            reason = f"Reviewer profile exists but MCP server was added after session startup and is not attached."
-        else:
-            reason = f"No profile configured with permission '{required_permission}'."
-
     switching = gitea_config.is_runtime_switching_enabled()
     different_namespace_required = False
     next_safe_action = "None; ready for operations."
@@ -4749,20 +4491,14 @@ def gitea_resolve_task_capability(
         "runtime_switching_supported": switching,
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
-        "available_in_session": available_in_session,
-        "configured": configured,
-        "restart_required": restart_required,
-        "reason": reason,
     }
-    was_terminal = capability_stop_terminal.is_active()
-    terminal = capability_stop_terminal.sync_from_capability_result(result)
-    if terminal:
-        result["terminal_mode"] = True
-        result["terminal_report"] = (
-            capability_stop_terminal.build_terminal_report(result)
-        )
-    elif was_terminal and not stop_required:
-        result["cleared_stale_denial"] = True
+    if stop_required:
+        terminal = capability_stop_terminal.enter_from_capability_result(result)
+        if terminal:
+            result["terminal_mode"] = True
+            result["terminal_report"] = (
+                capability_stop_terminal.build_terminal_report(result)
+            )
     return result
 
 

issue_lock_worktree.py

@@ -1,156 +0,0 @@
-"""Issue-lock worktree validation (#249).
-
-Author issue locks must validate the caller's own scratch clone (or declared
-worktree path), not the shared MCP server working directory. A clean scratch at
-``master``/``main`` must remain lockable while an unrelated session leaves the
-shared dev worktree dirty or on a feature branch.
-"""
-
-from __future__ import annotations
-
-import os
-import subprocess
-
-from reviewer_worktree import parse_dirty_tracked_files
-
-AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
-BASE_BRANCHES = frozenset({"master", "main"})
-
-
-def resolve_author_worktree_path(
-    explicit: str | None,
-    project_root: str,
-) -> str:
-    """Resolve the author worktree path for lock/PR gates."""
-    path = (explicit or "").strip()
-    if not path:
-        path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip()
-    if not path:
-        path = project_root
-    return os.path.realpath(os.path.abspath(path))
-
-
-def read_worktree_git_state(worktree_path: str) -> dict:
-    """Read branch name and porcelain status from a git worktree."""
-    path = (worktree_path or "").strip()
-    if not path:
-        return {"current_branch": None, "porcelain_status": ""}
-
-    branch_res = subprocess.run(
-        ["git", "-C", path, "branch", "--show-current"],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    current_branch = (branch_res.stdout or "").strip() or None
-
-    status_res = subprocess.run(
-        ["git", "-C", path, "status", "--porcelain"],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    return {
-        "current_branch": current_branch,
-        "porcelain_status": status_res.stdout or "",
-    }
-
-
-def assess_issue_lock_worktree(
-    *,
-    worktree_path: str,
-    current_branch: str | None,
-    porcelain_status: str,
-    base_branches: frozenset[str] | None = None,
-) -> dict:
-    """Fail closed when lock preconditions are not met on the declared worktree."""
-    bases = base_branches or BASE_BRANCHES
-    reasons: list[str] = []
-    path = (worktree_path or "").strip()
-    if not path:
-        reasons.append("worktree path not declared for issue lock; fail closed")
-        return _assessment(False, reasons, path, None, [])
-
-    branch = (current_branch or "").strip()
-    dirty_files = parse_dirty_tracked_files(porcelain_status)
-
-    if dirty_files:
-        reasons.append(
-            "tracked file edits exist before issue lock; "
-            "lock must precede implementation work"
-        )
-    if not branch:
-        reasons.append(
-            "current branch unknown (detached HEAD?); issue lock must be taken "
-            f"from base branch ({_base_list(bases)})"
-        )
-    elif branch not in bases:
-        reasons.append(
-            f"issue lock must be taken from base branch ({_base_list(bases)}), "
-            f"not '{branch}'"
-        )
-
-    proven = not reasons
-    return _assessment(proven, reasons, path, branch or None, dirty_files)
-
-
-def format_issue_lock_worktree_error(assessment: dict) -> str:
-    """Format a single fail-closed error for ``gitea_lock_issue``."""
-    reasons = list(assessment.get("reasons") or [])
-    if not reasons:
-        reasons = ["issue lock worktree validation failed"]
-    return "; ".join(reasons) + " (fail closed)"
-
-
-def verify_pr_worktree_matches_lock(
-    locked_worktree_path: str | None,
-    declared_worktree_path: str | None,
-    project_root: str,
-) -> dict:
-    """PR creation must use the same worktree the lock was validated against."""
-    locked = (locked_worktree_path or "").strip()
-    if not locked:
-        return {"proven": True, "block": False, "reasons": []}
-
-    declared = resolve_author_worktree_path(declared_worktree_path, project_root)
-    locked_real = os.path.realpath(locked)
-    declared_real = os.path.realpath(declared)
-    if locked_real != declared_real:
-        return {
-            "proven": False,
-            "block": True,
-            "reasons": [
-                f"PR worktree '{declared_real}' does not match locked worktree "
-                f"'{locked_real}' (fail closed)"
-            ],
-            "locked_worktree_path": locked_real,
-            "declared_worktree_path": declared_real,
-        }
-    return {
-        "proven": True,
-        "block": False,
-        "reasons": [],
-        "locked_worktree_path": locked_real,
-        "declared_worktree_path": declared_real,
-    }
-
-
-def _base_list(bases: frozenset[str]) -> str:
-    return "/".join(sorted(bases))
-
-
-def _assessment(
-    proven: bool,
-    reasons: list[str],
-    worktree_path: str,
-    current_branch: str | None,
-    dirty_files: list[str],
-) -> dict:
-    return {
-        "proven": proven,
-        "block": not proven,
-        "reasons": reasons,
-        "worktree_path": worktree_path or None,
-        "current_branch": current_branch,
-        "dirty_files": dirty_files,
-    }

review_proofs.py

@@ -17,7 +17,6 @@ here weakens or replaces them.
 import re
 
 import issue_duplicate_gate
-from reviewer_worktree import assess_reviewer_worktree_proof
 
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
 
@@ -69,251 +68,6 @@ def resolve_repos_from_user_reference(
     return list(configured)
 
 
-_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE)
-
-
-def _parse_pr_numbers(text):
-    """Extract PR numbers from operator context or backlog prose."""
-    if not text:
-        return []
-    seen = set()
-    ordered = []
-    for match in _PR_NUMBER_RE.finditer(text):
-        num = int(match.group(1))
-        if num not in seen:
-            seen.add(num)
-            ordered.append(num)
-    return ordered
-
-
-def _repo_hint_from_text(text, configured):
-    """Return a single configured repo named explicitly in *text*, if any."""
-    if not text:
-        return None
-    lower = text.lower()
-    for repo in configured:
-        if repo.lower() in lower:
-            return repo
-    resolved = resolve_repos_from_user_reference(text, configured)
-    if len(resolved) == 1:
-        return resolved[0]
-    return None
-
-
-def reconcile_queue_target(
-    *,
-    operator_context: str | None = None,
-    supplied_pr_backlog: list[dict] | None = None,
-    inventoried_repo: str | None = None,
-    project_context: str | None = None,
-    configured_repos: list[str] | None = None,
-) -> dict:
-    """Reconcile the inventory target repo before listing open PRs (#200).
-
-    Compares operator-supplied PR numbers/titles/backlog against the repo the
-    workflow is about to inventory. Returns a ``queue_target_lock`` dict whose
-    ``status`` must be ``resolved`` before an empty queue may stop cleanly.
-    """
-    if configured_repos is None:
-        configured_repos = [
-            "Scaled-Tech-Consulting/Gitea-Tools",
-            "Scaled-Tech-Consulting/mcp-control-plane",
-        ]
-
-    backlog_items = []
-    for item in supplied_pr_backlog or []:
-        number = item.get("number")
-        if number is None:
-            continue
-        repo = (item.get("repo") or "").strip() or None
-        backlog_items.append({
-            "number": int(number),
-            "repo": repo,
-            "title": (item.get("title") or "").strip() or None,
-        })
-
-    context_numbers = _parse_pr_numbers(operator_context or "")
-    backlog_numbers = [item["number"] for item in backlog_items]
-    supplied_pr_numbers = list(dict.fromkeys(backlog_numbers + context_numbers))
-
-    context_repo = _repo_hint_from_text(operator_context, configured_repos)
-    project_repo = _repo_hint_from_text(project_context, configured_repos)
-
-    resolution_source = None
-    resolved_repo = None
-    reasons = []
-
-    explicit_repos = {
-        item["repo"] for item in backlog_items if item.get("repo")
-    }
-    if len(explicit_repos) == 1:
-        resolved_repo = next(iter(explicit_repos))
-        resolution_source = "supplied_pr_backlog"
-    elif context_repo:
-        resolved_repo = context_repo
-        resolution_source = "operator_context"
-    elif project_repo and supplied_pr_numbers:
-        resolved_repo = project_repo
-        resolution_source = "project_context"
-
-    if (
-        context_repo
-        and explicit_repos
-        and context_repo not in explicit_repos
-    ):
-        return {
-            "status": "unresolved",
-            "resolved_repo": None,
-            "resolution_source": None,
-            "supplied_pr_numbers": supplied_pr_numbers,
-            "reconciliation": [],
-            "inventoried_repo": (inventoried_repo or "").strip() or None,
-            "reasons": [
-                "operator context repo conflicts with supplied PR backlog "
-                f"repos ({context_repo} vs {sorted(explicit_repos)})"
-            ],
-            "allow_clean_stop": False,
-            "allow_trusted_empty": False,
-        }
-
-    reconciliation = []
-    for number in supplied_pr_numbers:
-        expected_repo = None
-        for item in backlog_items:
-            if item["number"] == number and item.get("repo"):
-                expected_repo = item["repo"]
-                break
-        if expected_repo is None:
-            expected_repo = resolved_repo
-        reconciliation.append({
-            "pr_number": number,
-            "expected_repo": expected_repo,
-            "inventoried_repo": inventoried_repo,
-            "matches_inventoried_repo": (
-                expected_repo is not None
-                and inventoried_repo is not None
-                and expected_repo == inventoried_repo
-            ),
-        })
-
-    inventoried = (inventoried_repo or "").strip() or None
-
-    if supplied_pr_numbers and resolved_repo is None:
-        status = "unresolved"
-        reasons.append(
-            "operator supplied PR numbers but target repository could not "
-            "be resolved"
-        )
-    elif (
-        supplied_pr_numbers
-        and resolved_repo
-        and inventoried
-        and inventoried != resolved_repo
-    ):
-        status = "target_repo_mismatch"
-        reasons.append(
-            f"inventoried repository '{inventoried}' does not own the "
-            f"operator-supplied PR backlog (expected '{resolved_repo}')"
-        )
-    elif supplied_pr_numbers and resolved_repo and inventoried == resolved_repo:
-        status = "resolved"
-    elif not supplied_pr_numbers and inventoried:
-        status = "resolved"
-        resolution_source = resolution_source or "inventoried_repo_only"
-        resolved_repo = inventoried
-    elif not supplied_pr_numbers:
-        status = "unresolved"
-        reasons.append("no operator-supplied PR backlog to reconcile")
-    else:
-        status = "resolved"
-
-    allow_clean_stop = status == "resolved"
-    allow_trusted_empty = status == "resolved"
-
-    return {
-        "status": status,
-        "resolved_repo": resolved_repo,
-        "resolution_source": resolution_source,
-        "supplied_pr_numbers": supplied_pr_numbers,
-        "reconciliation": reconciliation,
-        "inventoried_repo": inventoried,
-        "reasons": reasons,
-        "allow_clean_stop": allow_clean_stop,
-        "allow_trusted_empty": allow_trusted_empty,
-    }
-
-
-resolve_pr_queue_target = reconcile_queue_target
-
-
-def assess_queue_target_final_report(report_text, queue_target_lock):
-    """Require final reports to document queue-target reconciliation."""
-    lock = queue_target_lock or {}
-    text = report_text or ""
-    lower = text.lower()
-    missing = []
-
-    if not lock.get("resolved_repo"):
-        missing.append("resolved repo")
-    elif lock["resolved_repo"].lower() not in lower:
-        missing.append("resolved repo")
-
-    source = lock.get("resolution_source")
-    if not source:
-        missing.append("resolution source")
-    else:
-        source_lower = str(source).lower()
-        if (
-            source_lower not in lower
-            and source_lower.replace("_", " ") not in lower
-        ):
-            missing.append("resolution source")
-
-    for number in lock.get("supplied_pr_numbers") or []:
-        if f"#{number}" not in lower and f"pr {number}" not in lower:
-            missing.append(f"supplied PR #{number} reconciliation")
-            break
-
-    if lock.get("status") and str(lock["status"]).lower() not in lower:
-        missing.append("queue_target_lock.status")
-
-    if missing:
-        return {
-            "complete": False,
-            "downgraded": True,
-            "missing_fields": missing,
-            "reasons": [
-                f"final report missing queue-target field: {field}"
-                for field in missing
-            ],
-        }
-    return {
-        "complete": True,
-        "downgraded": False,
-        "missing_fields": [],
-        "reasons": [],
-    }
-
-
-def format_queue_target_lock_report(lock: dict) -> list[str]:
-    """Render queue-target lock lines for inventory output."""
-    lines = [
-        f"queue_target_lock.status: {lock.get('status', 'unknown')}",
-    ]
-    if lock.get("resolved_repo"):
-        lines.append(f"queue_target_lock.resolved_repo: {lock['resolved_repo']}")
-    if lock.get("resolution_source"):
-        lines.append(
-            f"queue_target_lock.resolution_source: {lock['resolution_source']}"
-        )
-    if lock.get("supplied_pr_numbers"):
-        nums = ", ".join(f"#{n}" for n in lock["supplied_pr_numbers"])
-        lines.append(f"queue_target_lock.supplied_pr_numbers: {nums}")
-    for reason in lock.get("reasons") or []:
-        lines.append(f"queue_target_lock.reason: {reason}")
-    return lines
-
-
 SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION = (
     "evidence missing: report contamination as unknown and "
     "choose another PR or stop"
@@ -883,7 +637,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
                        role_boundary=None, review_mutation=None,
                        report_text=None, review_decision_lock=None,
                        controller_handoff=None, capability_proof=None,
-                       sweep_proof=None, worktree_proof=None):
+                       sweep_proof=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
 
     Combines the individual proof verdicts into the final-report fields the
@@ -909,8 +663,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             report_text, review_decision_lock
         )
 
-    empty_queue_report = assess_empty_queue_report(report_text)
-
     contamination_status = contamination.get("status", "unknown")
     checkout_proven = bool(checkout_proof.get("proven"))
     validation_claimable = bool(validation.get("claimable"))
@@ -955,14 +707,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "downgraded": True,
         "reasons": ["review mutation proof not provided (#211)"],
     }
-    if worktree_proof is not None:
-        worktree = assess_reviewer_worktree_proof(worktree_proof)
-    else:
-        worktree = {
-            "proven": False,
-            "block": True,
-            "reasons": ["reviewer worktree proof not provided (#233)"],
-        }
 
     capability_proven = bool(capability_evidence.get("proven"))
     sweep_proven = bool(sweep.get("proven"))
@@ -975,7 +719,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "reasons": ["review mutation proof missing"],
     }
     review_mutation_complete = bool(review_mutation.get("complete"))
-    worktree_proven = bool(worktree.get("proven"))
 
     downgrade_reasons = []
     if not identity_eligible:
@@ -1029,16 +772,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
     if not review_mutation_complete:
         downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
         downgrade_reasons.extend(review_mutation.get("reasons", []))
-    if not worktree_proven:
-        downgrade_reasons.append(
-            "reviewer worktree safety proof missing or failed (#233)"
-        )
-        downgrade_reasons.extend(worktree.get("reasons", []))
-    if empty_queue_report.get("claimed") and not empty_queue_report.get("proven"):
-        downgrade_reasons.append(
-            "empty-queue report missing or failed trust-gate proof (#198)"
-        )
-        downgrade_reasons.extend(empty_queue_report.get("reasons", []))
 
     merge_allowed = (
         identity_eligible
@@ -1049,7 +782,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         and validation.get("verdict") != "invalid"
         # #179: no merge without a proven final live-state recheck.
         and live_state_proven
-        and worktree_proven
     )
 
     violations = []
@@ -1093,17 +825,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "live_state_recheck_proven": live_state_proven,
         "role_boundary_clean": role_boundary_clean,
         "review_mutation_complete": review_mutation_complete,
-        "worktree_proof_proven": worktree_proven,
-        "worktree_scratch_used": bool(worktree.get("scratch_used")),
-        "unrelated_mutations_avoided": bool(
-            worktree.get("unrelated_mutations_avoided")
-        ),
-        "empty_queue_trust_gate_proven": (
-            empty_queue_report.get("proven")
-            if empty_queue_report.get("claimed")
-            else True
-        ),
-        "empty_queue_trust_gate_status": empty_queue_report.get("status"),
     }
 
 
@@ -1250,12 +971,6 @@ HANDOFF_ROLE_FIELDS = {
         ("Selected PR", ("selected pr",)),
         ("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
         ("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
-        ("Worktree path", ("worktree path", "starting worktree path")),
-        ("Worktree dirty", ("worktree dirty", "whether worktree was dirty")),
-        ("Scratch worktree used", ("scratch worktree used", "scratch clone used",
-                                   "scratch worktree")),
-        ("Unrelated local mutations", ("unrelated local mutations",
-                                       "unrelated files modified")),
         ("Review decision", ("review decision", "decision")),
         ("Merge result", ("merge result",)),
         ("Linked issue status", ("linked issue status", "linked issue")),
@@ -1263,7 +978,6 @@ HANDOFF_ROLE_FIELDS = {
     ),
     "author": (
         ("Selected issue", ("selected issue",)),
-        ("Issue lock proof", ("issue lock proof", "lock before diff")),
         ("Claim/comment status", ("claim/comment status", "claim status",
                                   "claim")),
         ("PR number opened", ("pr number opened", "pr opened", "pr number")),
@@ -1274,29 +988,11 @@ HANDOFF_ROLE_FIELDS = {
         ("Repositories checked", ("repositories checked", "repos checked")),
         ("Open PR counts", ("open pr counts", "open pr count",
                             "open prs per repo")),
-        ("PR inventory trust gate", ("pr inventory trust gate",
-                                     "pr_inventory_trust_gate.status",
-                                     "trust gate status")),
-        ("Trust gate reasons", ("trust gate reasons",
-                                "pr_inventory_trust_gate.reason")),
-        ("Trust gate corroborated", ("trust gate corroborated",
-                                     "pr_inventory_trust_gate.corroborated")),
-        ("Inventory profile", ("inventory profile", "inventory mcp profile")),
         ("Selected PR or reason", ("selected pr", "none selected",
                                    "reason none selected")),
         ("Inventory completeness", ("inventory complete", "inventory scoped",
                                     "inventory completeness")),
     ),
-    "continuation": (
-        ("Continuation mode", ("continuation mode", "continuation")),
-        ("Existing PR", ("existing pr", "pr number")),
-        ("PR author", ("pr author", "existing pr author")),
-        ("Branch", ("branch", "existing branch")),
-        ("Old PR head", ("old pr head", "old head")),
-        ("New PR head", ("new pr head", "new head")),
-        ("Session authored PR", ("session authored pr", "authored pr")),
-        ("Why continuation allowed", ("why continuation", "continuation allowed")),
-    ),
 }
 
 
@@ -1511,31 +1207,11 @@ def pr_inventory_trust_gate(
     user_context: str | None = None,
     corroboration_open_pr_counter: int | None = None,
     has_finality_metadata: bool = False,
-    queue_target_lock: dict | None = None,
 ) -> dict:
     """Evaluate whether an empty PR list is trusted or untrusted.
 
     Returns a dict with 'status', 'reasons', and 'corroborated'.
     """
-    lock = queue_target_lock or {}
-    lock_status = lock.get("status")
-    if lock_status == "target_repo_mismatch":
-        return {
-            "status": "target_repo_mismatch",
-            "reasons": list(lock.get("reasons") or []),
-            "corroborated": False,
-            "queue_target_lock": lock_status,
-        }
-    if lock and lock_status != "resolved":
-        return {
-            "status": "untrusted_empty",
-            "reasons": [
-                f"queue_target_lock.status is '{lock_status}', not 'resolved'"
-            ] + list(lock.get("reasons") or []),
-            "corroborated": False,
-            "queue_target_lock": lock_status,
-        }
-
     if list_prs_response is None or not isinstance(list_prs_response, list):
         return {
             "status": "inventory_error",
@@ -1605,533 +1281,6 @@ def pr_inventory_trust_gate(
         "status": "trusted_empty",
         "reasons": [],
         "corroborated": corroborated,
-        "queue_target_lock": lock_status or "resolved",
-    }
-
-
-def _split_repo_slug(full_repo: str) -> tuple[str | None, str | None]:
-    parts = (full_repo or "").split("/", 1)
-    if len(parts) == 2:
-        return parts[0].strip() or None, parts[1].strip() or None
-    return None, None
-
-
-def assess_reviewer_queue_inventory(
-    repo_reports: list[dict] | None,
-    required_repos: list[str] | None = None,
-    *,
-    user_context: str | None = None,
-    operator_context: str | None = None,
-    supplied_pr_backlog: list[dict] | None = None,
-    project_context: str | None = None,
-) -> dict:
-    """Canonical reviewer queue path: completeness plus per-repo trust gates (#196).
-
-    Any repository reporting ``open_pr_count == 0`` must pass
-    ``pr_inventory_trust_gate`` with ``trusted_empty`` before an empty-queue
-    claim is allowed. A bare ``[]`` from ``gitea_list_prs`` is never sufficient.
-    """
-    required = list(required_repos or [
-        "Scaled-Tech-Consulting/Gitea-Tools",
-        "Scaled-Tech-Consulting/mcp-control-plane",
-    ])
-    completeness = assess_inventory_completeness(repo_reports, required)
-
-    trust_gates: dict[str, dict] = {}
-    queue_target_locks: dict[str, dict] = {}
-    blockers: list[str] = []
-    can_claim_empty = bool(completeness.get("complete"))
-
-    context = operator_context or user_context
-
-    for report in repo_reports or []:
-        repo = (report.get("repo") or "").strip()
-        count = report.get("open_pr_count")
-        if not isinstance(count, int) or count != 0:
-            continue
-
-        org, repo_name = _split_repo_slug(repo)
-        list_response = report.get("list_prs_response")
-        if list_response is None:
-            list_response = []
-
-        queue_target_lock = reconcile_queue_target(
-            operator_context=context,
-            supplied_pr_backlog=supplied_pr_backlog,
-            inventoried_repo=repo,
-            project_context=project_context,
-            configured_repos=required,
-        )
-        queue_target_locks[repo] = queue_target_lock
-
-        gate = pr_inventory_trust_gate(
-            list_response,
-            remote=report.get("remote"),
-            org=org,
-            repo=repo_name,
-            state=report.get("state_filter"),
-            authenticated_profile=report.get("authenticated_profile"),
-            local_remote_url=report.get("local_remote_url"),
-            user_context=user_context or report.get("user_context"),
-            corroboration_open_pr_counter=report.get(
-                "corroboration_open_pr_counter"
-            ),
-            has_finality_metadata=report.get("pagination_complete") is True,
-            queue_target_lock=queue_target_lock,
-        )
-        trust_gates[repo] = gate
-        status = gate.get("status")
-        if status != "trusted_empty":
-            can_claim_empty = False
-            blockers.append(
-                f"repository '{repo}': empty PR list trust gate is "
-                f"'{status}'; cannot claim 'no open PRs'"
-            )
-            blockers.extend(gate.get("reasons") or [])
-
-    return {
-        "complete": bool(completeness.get("complete")),
-        "can_claim_empty_queue": can_claim_empty,
-        "can_claim_exhaustive": (
-            bool(completeness.get("can_claim_exhaustive")) and can_claim_empty
-        ),
-        "inventory_reasons": list(completeness.get("reasons") or []),
-        "trust_gates": trust_gates,
-        "queue_target_locks": queue_target_locks,
-        "blockers": blockers,
-        "reasons": list(completeness.get("reasons") or []) + blockers,
-    }
-
-
-def format_pr_inventory_trust_gate_report(
-    gate: dict,
-    queue_target_lock: dict | None = None,
-) -> list[str]:
-    """Render trust-gate lines for MCP inventory output."""
-    lines = []
-    if queue_target_lock:
-        lines.extend(format_queue_target_lock_report(queue_target_lock))
-    lines.append(f"pr_inventory_trust_gate.status: {gate.get('status', 'unknown')}")
-    if gate.get("corroborated"):
-        lines.append("pr_inventory_trust_gate.corroborated: true")
-    for reason in gate.get("reasons") or []:
-        lines.append(f"pr_inventory_trust_gate.reason: {reason}")
-    return lines
-
-
-_EMPTY_QUEUE_CLAIM = re.compile(
-    r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
-    r"nothing to review|queue cleared|inventory empty|"
-    r"open pr count:\s*0|workflow correctly stops with nothing",
-    re.I,
-)
-
-_WEAK_EMPTY_QUEUE_CORROBORATION = re.compile(
-    r"latest commit.*(?:merge|pr #)|merge of pr #|"
-    r"master latest commit|recent merge proves",
-    re.I,
-)
-
-_TRUST_GATE_STATUS_LINE = re.compile(
-    r"pr_inventory_trust_gate\.status:\s*(\S+)",
-    re.I,
-)
-
-
-def parse_trust_gate_status_from_report(report_text: str | None) -> str | None:
-    """Extract ``pr_inventory_trust_gate.status`` from report text, if present."""
-    match = _TRUST_GATE_STATUS_LINE.search(report_text or "")
-    return match.group(1).strip().lower() if match else None
-
-
-def assess_empty_queue_report(
-    report_text: str | None,
-    *,
-    trust_gate: dict | None = None,
-    task_role: str | None = None,
-    inventory_profile: str | None = None,
-) -> dict:
-    """Issue #198: empty-queue reports must cite the formal trust-gate result.
-
-    Blocks reports that claim an empty queue without
-    ``pr_inventory_trust_gate.status == trusted_empty``, required inventory
-    metadata, or that rely on weak corroboration (e.g. a recent merge commit).
-    """
-    text = report_text or ""
-    lower = text.lower()
-    reasons: list[str] = []
-    missing: list[str] = []
-
-    if not _EMPTY_QUEUE_CLAIM.search(text):
-        return {
-            "claimed": False,
-            "proven": True,
-            "block": False,
-            "status": None,
-            "missing_fields": [],
-            "reasons": [],
-        }
-
-    status = (
-        (trust_gate or {}).get("status")
-        or parse_trust_gate_status_from_report(text)
-    )
-    status_norm = (status or "").strip().lower() or None
-
-    if not status_norm:
-        missing.append("pr_inventory_trust_gate.status")
-        reasons.append(
-            "empty-queue report missing pr_inventory_trust_gate.status; "
-            "fail closed"
-        )
-    elif status_norm != "trusted_empty":
-        reasons.append(
-            f"empty-queue report has trust-gate status '{status_norm}', "
-            "not trusted_empty"
-        )
-
-    has_gate_reasons = (
-        "pr_inventory_trust_gate.reason" in lower
-        or bool((trust_gate or {}).get("reasons"))
-    )
-    if status_norm and status_norm != "trusted_empty" and not has_gate_reasons:
-        missing.append("pr_inventory_trust_gate.reasons")
-
-    if status_norm == "trusted_empty":
-        if "pr_inventory_trust_gate.corroborated" not in lower and (
-            trust_gate or {}
-        ).get("corroborated") is not True:
-            missing.append("pr_inventory_trust_gate.corroborated")
-
-    inventory_markers = (
-        "repository:",
-        "remote:",
-        "owner:",
-        "state filter:",
-        "state_filter:",
-    )
-    if not any(marker in lower for marker in inventory_markers):
-        missing.append("inventory remote/owner/repo/state filter")
-
-    profile_markers = (
-        "mcp profile:",
-        "mcp-profile:",
-        "inventory profile:",
-        "active profile:",
-    )
-    has_profile = (
-        any(marker in lower for marker in profile_markers)
-        or bool((inventory_profile or "").strip())
-    )
-    if not has_profile:
-        missing.append("inventory MCP profile")
-
-    if _WEAK_EMPTY_QUEUE_CORROBORATION.search(text):
-        if "pr_inventory_trust_gate.status: trusted_empty" not in lower:
-            reasons.append(
-                "weak corroboration (recent merge commit) cannot substitute "
-                "for pr_inventory_trust_gate.status == trusted_empty"
-            )
-
-    role = (task_role or "").strip().lower()
-    if role == "author" and re.search(
-        r"reviewer queue|nothing to review|review backlog empty",
-        text,
-        re.I,
-    ):
-        reasons.append(
-            "author-bound session presented reviewer queue inventory as a "
-            "reviewer decision"
-        )
-
-    if missing:
-        reasons.extend(
-            f"empty-queue report missing required field: {field}"
-            for field in missing
-        )
-
-    proven = not reasons and not missing
-    return {
-        "claimed": True,
-        "proven": proven,
-        "block": not proven,
-        "status": status_norm,
-        "missing_fields": missing,
-        "reasons": reasons,
-    }
-
-
-# ── Issue selection / continuation mode (#188) ───────────────────────────────
-
-ISSUE_SELECTION_UNCLAIMED_NO_PR = "unclaimed_no_pr"
-ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR = "represented_by_open_pr"
-ISSUE_SELECTION_IN_PROGRESS = "in_progress"
-ISSUE_SELECTION_CONTINUATION_EXPLICIT = "continuation_explicit"
-ISSUE_SELECTION_EXCLUDED = "excluded"
-
-_NO_OPEN_PR_CLAIM = re.compile(
-    r"no duplicate pr|no open pr|no pr open|no eligible pr|"
-    r"no existing pr|without an open pr",
-    re.IGNORECASE,
-)
-
-
-def classify_issue_for_selection(
-    issue_number: int,
-    *,
-    labels: list[str] | None = None,
-    open_prs: list[dict] | None = None,
-    operator_continuation_requested: bool = False,
-    continuation_issue_numbers: list[int] | None = None,
-    excluded: bool = False,
-) -> dict:
-    """Classify one issue for author queue selection (#188)."""
-    label_set = {str(l).lower() for l in (labels or [])}
-    prs = list(open_prs or [])
-    continuation_issues = set(continuation_issue_numbers or [])
-
-    if excluded:
-        status = ISSUE_SELECTION_EXCLUDED
-        selectable_for_fresh_work = False
-        reasons = ["issue explicitly excluded from selection"]
-    elif "status:in-progress" in label_set:
-        status = ISSUE_SELECTION_IN_PROGRESS
-        selectable_for_fresh_work = False
-        reasons = ["issue already marked status:in-progress"]
-    elif prs and (
-        operator_continuation_requested
-        or issue_number in continuation_issues
-    ):
-        status = ISSUE_SELECTION_CONTINUATION_EXPLICIT
-        selectable_for_fresh_work = False
-        reasons = ["operator requested continuation for issue with open PR"]
-    elif prs:
-        status = ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR
-        selectable_for_fresh_work = False
-        reasons = [
-            f"issue #{issue_number} already represented by open PR "
-            f"#{prs[0].get('number')}"
-        ]
-    else:
-        status = ISSUE_SELECTION_UNCLAIMED_NO_PR
-        selectable_for_fresh_work = True
-        reasons = []
-
-    return {
-        "issue_number": issue_number,
-        "status": status,
-        "selectable_for_fresh_work": selectable_for_fresh_work,
-        "open_prs": prs,
-        "reasons": reasons,
-    }
-
-
-def assess_fresh_issue_selection(classifications: list[dict] | None) -> dict:
-    """Fail closed when fresh selection picks an issue with an open PR."""
-    reasons = []
-    for item in classifications or []:
-        if item.get("selectable_for_fresh_work"):
-            continue
-        if item.get("status") == ISSUE_SELECTION_CONTINUATION_EXPLICIT:
-            continue
-        if item.get("status") == ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR:
-            reasons.append(
-                f"issue #{item.get('issue_number')} has open PR and was "
-                "selected for fresh work without continuation mode"
-            )
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_continuation_mode_report(
-    report_text: str,
-    *,
-    pr_number: int | None = None,
-    pr_author: str | None = None,
-    branch: str | None = None,
-    old_head_sha: str | None = None,
-    new_head_sha: str | None = None,
-    session_authored_pr: bool | None = None,
-    continuation_allowed_reason: str | None = None,
-) -> dict:
-    """Issue #188: continuation mode must disclose full PR evidence."""
-    text = report_text or ""
-    lower = text.lower()
-    reasons = []
-
-    if not any(p in lower for p in ("continuation", "continue pr", "continue issue")):
-        reasons.append("report does not declare continuation mode")
-
-    if pr_number is not None:
-        if f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower:
-            reasons.append(f"continuation report missing PR #{pr_number}")
-    if pr_author and pr_author.lower() not in lower:
-        reasons.append("continuation report missing PR author")
-    if branch and branch.lower() not in lower:
-        reasons.append("continuation report missing branch name")
-
-    for label, sha in (("old", old_head_sha), ("new", new_head_sha)):
-        if not sha:
-            reasons.append(f"continuation proof missing {label} head SHA")
-        elif not _FULL_SHA.match(sha.lower()):
-            reasons.append(
-                f"continuation {label} head SHA is not a full 40-hex SHA"
-            )
-        elif sha.lower() not in lower:
-            reasons.append(
-                f"continuation report missing {label} head SHA in evidence"
-            )
-
-    if session_authored_pr is not None:
-        authored_tokens = ("session authored", "authored pr", "own pr", "my pr")
-        if not any(t in lower for t in authored_tokens):
-            reasons.append(
-                "continuation report missing whether session authored the PR"
-            )
-
-    if continuation_allowed_reason:
-        reason_lower = continuation_allowed_reason.lower()
-        if (
-            reason_lower not in lower
-            and not any(w in lower for w in reason_lower.split()[:3])
-        ):
-            reasons.append("continuation report missing why continuation is allowed")
-
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_contradictory_no_pr_claim(
-    report_text: str,
-    *,
-    edited_pr_numbers: list[int] | None = None,
-    issue_open_pr_map: dict[int, int] | None = None,
-) -> dict:
-    """Downgrade when report claims no open PR but later edits one."""
-    text = report_text or ""
-    lower = text.lower()
-    reasons = []
-
-    if not _NO_OPEN_PR_CLAIM.search(lower):
-        return {"complete": True, "downgraded": False, "reasons": []}
-
-    edited = list(edited_pr_numbers or [])
-    for pr_num in edited:
-        if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
-            reasons.append(
-                f"report claims no open PR but edited PR #{pr_num}"
-            )
-
-    for issue_num, pr_num in (issue_open_pr_map or {}).items():
-        if f"#{issue_num}" in lower or f"issue #{issue_num}" in lower:
-            if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
-                reasons.append(
-                    f"report claims no open PR for issue #{issue_num} but "
-                    f"PR #{pr_num} exists and was referenced"
-                )
-
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_edited_pr_inventory_coverage(
-    report_text: str,
-    *,
-    edited_pr_numbers: list[int] | None = None,
-    inventoried_pr_numbers: list[int] | None = None,
-) -> dict:
-    """Open PR inventory must include PRs the run later edits (#188)."""
-    text = report_text or ""
-    lower = text.lower()
-    reasons = []
-    inventoried = set(inventoried_pr_numbers or [])
-
-    for pr_num in edited_pr_numbers or []:
-        if pr_num in inventoried:
-            continue
-        if f"#{pr_num}" not in lower and f"pr {pr_num}" not in lower:
-            reasons.append(
-                f"edited PR #{pr_num} missing from open PR inventory"
-            )
-
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_issue_selection_final_report(
-    report_text: str,
-    *,
-    mode: str = "fresh",
-    classifications: list[dict] | None = None,
-    continuation_proof: dict | None = None,
-    edited_pr_numbers: list[int] | None = None,
-    inventoried_pr_numbers: list[int] | None = None,
-    issue_open_pr_map: dict[int, int] | None = None,
-) -> dict:
-    """Issue #188: composite A-bar for author issue-selection runs."""
-    handoff_role = "continuation" if mode == "continuation" else "author"
-    checks = {
-        "controller_handoff": assess_controller_handoff(
-            report_text, role=handoff_role
-        ),
-        "contradictory_no_pr": assess_contradictory_no_pr_claim(
-            report_text,
-            edited_pr_numbers=edited_pr_numbers,
-            issue_open_pr_map=issue_open_pr_map,
-        ),
-        "edited_pr_inventory": assess_edited_pr_inventory_coverage(
-            report_text,
-            edited_pr_numbers=edited_pr_numbers,
-            inventoried_pr_numbers=inventoried_pr_numbers,
-        ),
-    }
-
-    if mode == "fresh":
-        checks["fresh_selection"] = assess_fresh_issue_selection(classifications)
-    else:
-        proof = continuation_proof or {}
-        checks["continuation_mode"] = assess_continuation_mode_report(
-            report_text,
-            pr_number=proof.get("pr_number"),
-            pr_author=proof.get("pr_author"),
-            branch=proof.get("branch"),
-            old_head_sha=proof.get("old_head_sha"),
-            new_head_sha=proof.get("new_head_sha"),
-            session_authored_pr=proof.get("session_authored_pr"),
-            continuation_allowed_reason=proof.get("continuation_allowed_reason"),
-        )
-
-    reasons = []
-    downgraded = False
-    for name, result in checks.items():
-        verdict = result.get("verdict")
-        if verdict in ("missing", "incomplete"):
-            downgraded = True
-            reasons.extend(result.get("reasons") or [])
-        elif result.get("downgraded") or not result.get("complete", True):
-            downgraded = True
-            reasons.extend(
-                f"{name}: {r}" for r in (result.get("reasons") or [])
-            )
-
-    return {
-        "grade": "A" if not downgraded else "downgraded",
-        "downgraded": downgraded,
-        "checks": checks,
-        "reasons": reasons,
-        "complete": not downgraded,
     }
 
 

reviewer_worktree.py

@@ -1,210 +0,0 @@
-"""Fail-closed reviewer worktree and local-git safety proofs (#233).
-
-Reviewer sessions must never stash, reset, or otherwise manipulate unrelated
-local changes from another session. When the active worktree has dirty tracked
-files outside the PR scope, the workflow must stop or switch to a disposable
-scratch worktree (``scripts/worktree-review``).
-
-Git command policy (#243): reviewers use an allowlist, not a blocklist.
-Any ``git`` invocation that does not match ``_READONLY_REVIEWER_GIT`` is
-forbidden — including ``checkout HEAD --``, ``checkout .``, ``switch``,
-and uncommon ``stash`` subcommands that older blocklists missed.
-"""
-
-from __future__ import annotations
-
-import re
-import shlex
-
-# Read-only git operations reviewers may use for validation (#243 allowlist).
-_READONLY_REVIEWER_GIT = re.compile(
-    r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
-    r"(?:fetch|status|diff|log|show|rev-parse|branch(?:\s+--show-current)?|"
-    r"worktree\s+list|worktree\s+add)",
-    re.IGNORECASE,
-)
-
-_GIT_INVOCATION = re.compile(r"\bgit\b", re.IGNORECASE)
-
-
-def parse_dirty_tracked_files(porcelain: str) -> list[str]:
-    """Return tracked paths with local modifications from ``git status --porcelain``.
-
-    Untracked entries (``??``) are ignored — they do not block reviewer work
-    when a scratch worktree is used, and authors may have unrelated untracked
-    files without implying reviewer interference.
-    """
-    paths: list[str] = []
-    for line in (porcelain or "").splitlines():
-        if not line or len(line) < 4:
-            continue
-        if line.startswith("??"):
-            continue
-        path = line[3:].strip()
-        if " -> " in path:
-            path = path.split(" -> ", 1)[1].strip()
-        if path:
-            paths.append(path)
-    return paths
-
-
-def files_outside_pr_scope(
-    dirty_files: list[str] | None,
-    pr_scope_files: list[str] | None,
-) -> list[str]:
-    """Dirty tracked files not explained by the PR diff file set."""
-    dirty = [p for p in (dirty_files or []) if p]
-    scope = {p for p in (pr_scope_files or []) if p}
-    if not dirty:
-        return []
-    if not scope:
-        return list(dirty)
-    return [path for path in dirty if path not in scope]
-
-
-def _is_git_command(command: str) -> bool:
-    return bool(_GIT_INVOCATION.search((command or "").strip()))
-
-
-def is_readonly_reviewer_git_command(command: str) -> bool:
-    """True when the command is an explicitly allowed read-only git operation."""
-    text = (command or "").strip()
-    if not text or not _is_git_command(text):
-        return False
-    return bool(_READONLY_REVIEWER_GIT.search(text))
-
-
-def is_forbidden_reviewer_git_command(command: str) -> bool:
-    """True when a git command is not on the reviewer readonly allowlist."""
-    text = (command or "").strip()
-    if not text or not _is_git_command(text):
-        return False
-    return not is_readonly_reviewer_git_command(text)
-
-
-def assess_reviewer_git_command_log(commands: list[str] | None) -> dict:
-    """Fail closed when reviewer shell history includes forbidden git mutations."""
-    forbidden = [
-        cmd for cmd in (commands or []) if is_forbidden_reviewer_git_command(cmd)
-    ]
-    if forbidden:
-        return {
-            "proven": False,
-            "block": True,
-            "forbidden_commands": forbidden,
-            "reasons": [
-                "reviewer workflow executed forbidden local git mutation: "
-                f"{cmd!r}"
-                for cmd in forbidden
-            ],
-            "safe_next_action": (
-                "stop; report worktree interference; do not stash/reset/checkout "
-                "unrelated files — use scripts/worktree-review instead"
-            ),
-        }
-    return {
-        "proven": True,
-        "block": False,
-        "forbidden_commands": [],
-        "reasons": [],
-        "safe_next_action": "proceed",
-    }
-
-
-def assess_reviewer_worktree_proof(proof: dict | None) -> dict:
-    """Evaluate reviewer worktree safety before checkout/diff/validation/review.
-
-    *proof* keys:
-    - ``worktree_path`` (required)
-    - ``porcelain_status`` or ``dirty_files``
-    - ``pr_scope_files`` (paths in the PR diff)
-    - ``scratch_used`` (bool)
-    - ``scratch_path`` (when scratch_used)
-    - ``git_commands`` (shell commands executed this session)
-    - ``unrelated_mutations_claimed`` (bool) — stash/reset/drop reported
-    """
-    proof = dict(proof or {})
-    reasons: list[str] = []
-    worktree_path = (proof.get("worktree_path") or "").strip()
-    if not worktree_path:
-        reasons.append("reviewer worktree path not reported; fail closed")
-
-    if proof.get("dirty_files") is not None:
-        dirty_files = list(proof.get("dirty_files") or [])
-    else:
-        dirty_files = parse_dirty_tracked_files(proof.get("porcelain_status") or "")
-
-    pr_scope = list(proof.get("pr_scope_files") or [])
-    unrelated = files_outside_pr_scope(dirty_files, pr_scope)
-    scratch_used = bool(proof.get("scratch_used"))
-    scratch_path = (proof.get("scratch_path") or "").strip()
-
-    is_dirty = bool(dirty_files)
-    unrelated_dirty = bool(unrelated)
-
-    if unrelated_dirty and not scratch_used:
-        reasons.append(
-            "worktree has dirty tracked files outside PR scope "
-            f"({', '.join(unrelated)}); stop or use a scratch worktree"
-        )
-    if scratch_used and not scratch_path:
-        reasons.append(
-            "scratch worktree was used but scratch_path was not reported"
-        )
-    if proof.get("unrelated_mutations_claimed"):
-        reasons.append(
-            "reviewer reported stash/reset/checkout cleanup of unrelated "
-            "local changes; this is forbidden"
-        )
-
-    command_assessment = assess_reviewer_git_command_log(
-        list(proof.get("git_commands") or [])
-    )
-    if command_assessment["block"]:
-        reasons.extend(command_assessment["reasons"])
-
-    proven = not reasons
-    return {
-        "proven": proven,
-        "block": not proven,
-        "reasons": reasons,
-        "worktree_path": worktree_path or None,
-        "is_dirty": is_dirty,
-        "dirty_files": dirty_files,
-        "unrelated_dirty_files": unrelated,
-        "scratch_used": scratch_used,
-        "scratch_path": scratch_path or None,
-        "unrelated_mutations_avoided": not bool(
-            proof.get("unrelated_mutations_claimed")
-            or command_assessment.get("forbidden_commands")
-        ),
-        "safe_next_action": (
-            "proceed"
-            if proven
-            else command_assessment.get("safe_next_action")
-            or "stop; use scripts/worktree-review or report dirty worktree"
-        ),
-        "forbidden_commands": command_assessment.get("forbidden_commands", []),
-    }
-
-
-def assess_author_worktree_continuity(proof: dict | None) -> dict:
-    """Authors may keep dirty feature worktrees; reviewers may not manipulate them.
-
-    This helper only proves the task role is author when dirty unrelated files
-    exist — it does not grant reviewers an exception.
-    """
-    proof = dict(proof or {})
-    role = (proof.get("task_role") or "").strip().lower()
-    dirty_files = list(proof.get("dirty_files") or [])
-    if role == "author" and dirty_files:
-        return {
-            "allowed": True,
-            "reasons": [
-                "author task may continue with dirty tracked files in its own "
-                "worktree; reviewer interference rules do not apply"
-            ],
-        }
-    if role == "reviewer" and dirty_files:
-        return assess_reviewer_worktree_proof(proof)
-    return {"allowed": True, "reasons": []}

skills/llm-project-workflow/SKILL.md

@@ -146,15 +146,6 @@ Worktree folder = branch with `/` replaced by `-`
 2. Fetch/prune: `git fetch <remote> --prune`.
 3. Confirm local `master` equals remote `master` (`git rev-list --left-right --count <remote>/master...master` → `0	0`).
 4. Create/claim the issue (§A).
-4b. **Issue lock from your scratch clone (#249):** when using
-   `gitea_lock_issue`, pass `worktree_path` pointing at your own clean
-   scratch clone (or set `GITEA_AUTHOR_WORKTREE`). The lock gate validates
-   *that* path — clean tree on `master`/`main`, no tracked edits yet —
-   not the shared MCP/orchestration checkout. Another session's dirty
-   feature branch in the shared dev worktree must not block your lock.
-   Never stash, reset, or checkout files in the shared worktree to satisfy
-   the gate. Pass the same `worktree_path` to `gitea_create_pr` so the PR
-   gate matches the lock record.
 5. Create the isolated worktree (§B) from latest remote `master`.
 6. Implement the narrow scope only — no unrelated refactors or formatting churn.
 7. Add/update focused tests when behavior changes.
@@ -225,10 +216,7 @@ Worktree folder = branch with `/` replaced by `-`
    the other.
    Both configured repos must be reported with state filter, pagination proof,
    and open-PR count (`review_proofs.assess_inventory_completeness` and
-   `resolve_repos_from_user_reference`). Before inventory, reconcile the
-   operator-supplied PR backlog against the target repo
-   (`review_proofs.reconcile_queue_target`); never report `trusted_empty`
-   for one repo while ignoring contradictory supplied PR numbers in another.
+   `resolve_repos_from_user_reference`).
 7. **Role-boundary proof (#175):** a reviewer queue task must not silently
    become author implementation. If no eligible PR exists, stop with the
    queue report. Do not claim issues, create branches, commit, push, or open
@@ -249,9 +237,6 @@ Worktree folder = branch with `/` replaced by `-`
    validation completes, call `gitea_mark_final_review_decision`, then submit
    exactly one live review via
    `gitea_submit_pr_review(..., final_review_decision_ready=True)`.
-   After submitting, re-read `gitea_get_pr_review_feedback` and confirm the
-   verdict is visible (`approval_visible` true for APPROVE; PENDING drafts do
-   not count — #244). Do not merge until a visible APPROVED review exists.
    Final reports must list exactly one review mutation
    (`review_proofs.assess_review_mutation_final_report`) unless an
    operator-approved correction flow was invoked and explained.
@@ -323,9 +308,7 @@ When in doubt, stop and surface the discrepancy; do not guess or work around a g
 ## I. Recovery patterns
 
 - **Dirty worktree from another issue:** do not touch it. Start your issue in its
-  own new worktree; unrelated dirty work must not block you. For Gitea-Tools
-  author flows, lock the issue from your scratch clone (`worktree_path` on
-  `gitea_lock_issue`) — do not manipulate the shared dev checkout.
+  own new worktree; unrelated dirty work must not block you.
 - **Local `master` ahead of remote unexpectedly:** do not push `master`. Confirm
   the commits are preserved on a feature branch (local + remote) first, then
   `git reset --hard <remote>/master` to realign. Never discard commits that are
@@ -396,12 +379,6 @@ Role-specific fields (append to the compact block):
   `Linked issue status:`, `Cleanup status:`
 - author tasks: `Selected issue:`, `Claim/comment status:`,
   `PR number opened:`, `No review/merge:` (explicit confirmation)
-- continuation tasks (#188): `Continuation mode:`, `Existing PR:`,
-  `PR author:`, `Branch:`, `Old PR head:`, `New PR head:`,
-  `Session authored PR:`, `Why continuation allowed:` — issues with open
-  PRs are excluded from fresh selection unless operator explicitly requests
-  continuation (`review_proofs.classify_issue_for_selection`,
-  `assess_issue_selection_final_report`)
 - queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
   `Selected PR or reason none selected:`, `Inventory completeness:`
 

skills/llm-project-workflow/templates/review-pr.md

@@ -17,32 +17,9 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
   configured repos were not checked. This is not a complete queue inventory."
 - A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
   if the other configured repo was not inventoried.
-- PR inventory trust gate (#196): before reporting "no open PRs" or "queue empty",
-  the workflow must run `pr_inventory_trust_gate` (via the live inventory path or
-  `review_proofs.assess_reviewer_queue_inventory`). Only `trusted_empty` allows a
-  clean empty-queue stop. Report `pr_inventory_trust_gate.status`, reasons, and
-  corroboration in the final report. A bare `[]` from `gitea_list_prs` is never
-  sufficient proof.
-- Empty-queue report wall (#198): if the final report claims "no open PRs",
-  "queue empty", or "nothing to review", it must include verbatim:
-  `pr_inventory_trust_gate.status`, trust-gate reasons, corroboration,
-  remote/owner/repo/state filter, and the inventory MCP profile. A recent merge
-  commit is not valid corroboration. Author-bound sessions must not present
-  reviewer queue inventory as a reviewer decision.
 
 Rules (llm-project-workflow):
 - Review in a SEPARATE detached review worktree, never the author's folder.
-- Worktree safety (#233): before checkout, diff, validation, review, or merge,
-  report the starting worktree path and whether it was dirty. If unrelated
-  tracked files exist outside the PR scope, STOP or run
-  `scripts/worktree-review <pr-head-branch>` and validate in the scratch path.
-  Scratch-clone validation is the norm; tests must not assume the shared
-  development worktree or a repo-local ``venv/`` (#245).
-  NEVER run `git stash`, `git stash pop/drop`, `git checkout --`, `git reset`,
-  or `git clean` to manage another session's dirty files.
-- Final report must state: Worktree path, Worktree dirty (yes/no),
-  Scratch worktree used (yes/no + path if yes), and confirm no unrelated local
-  files were modified, stashed, reset, or dropped.
 - You must NOT be the PR author. If the authenticated user == PR author, stop.
   A different LLM-Agent-SHA does NOT make you a different actor — only a
   different authenticated Gitea user does (docs/llm-agent-sha.md).

tests/test_audit.py

@@ -293,12 +293,9 @@ class TestGatedToolAudit(_AuditWiringBase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_merge_success_audited(self, _auth, mock_api):
-        # user, pr, feedback pr+reviews, merge POST, readback.
+        # user, pr, merge POST, readback — no extra identity call (uses result).
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
-            self._pr("author-bot"),
-            [{"id": 1, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
-              "submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}],
             {}, {"merged_commit_sha": "c1"},
         ]
         env = self._env(GITEA_PROFILE_NAME="gitea-merger",
@@ -335,10 +332,7 @@ class TestGatedToolAudit(_AuditWiringBase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_submit_review_success_audited(self, _auth, mock_api):
         mock_api.side_effect = [
-            {"login": "reviewer-bot"}, self._pr("author-bot"),
-            {"id": 7, "state": "APPROVED"},
-            [{"id": 7, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
-              "submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}],
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 7},
         ]
         env = self._env(GITEA_PROFILE_NAME="gitea-reviewer",
                         GITEA_ALLOWED_OPERATIONS="read,review,approve")

tests/test_capability_stop_terminal.py

@@ -29,8 +29,8 @@ CONFIG = {
             "username": "jcwalker3",
             "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
             "allowed_operations": [
-                "gitea.read", "gitea.issue.create", "gitea.issue.comment",
-                "gitea.pr.create", "gitea.branch.push",
+                "gitea.read", "gitea.issue.create", "gitea.pr.create",
+                "gitea.branch.push",
             ],
             "forbidden_operations": [
                 "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
@@ -97,30 +97,6 @@ class TestCapabilityStopTerminal(unittest.TestCase):
             with self.assertRaises(RuntimeError) as ctx:
                 mcp_server.gitea_list_prs(remote="prgs")
         self.assertIn("Cannot perform reviewer task", str(ctx.exception))
-        self.assertIn("review_pr", str(ctx.exception))
-
-    @patch("mcp_server.api_get_all", return_value=[])
-    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
-    @patch("mcp_server.get_auth_header", return_value="token author-pass")
-    def test_list_prs_allowed_after_author_task_clears_stale_denial(
-        self, _auth, _api, _get_all,
-    ):
-        with patch.dict(os.environ, self._env()):
-            denied = mcp_server.gitea_resolve_task_capability(
-                task="review_pr", remote="prgs"
-            )
-            self.assertTrue(denied["stop_required"])
-            self.assertTrue(capability_stop_terminal.is_active())
-
-            cleared = mcp_server.gitea_resolve_task_capability(
-                task="claim_issue", remote="prgs"
-            )
-            self.assertTrue(cleared["allowed_in_current_session"])
-            self.assertTrue(cleared.get("cleared_stale_denial"))
-            self.assertFalse(capability_stop_terminal.is_active())
-
-            prs = mcp_server.gitea_list_prs(remote="prgs")
-        self.assertEqual(prs, [])
 
     @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
@@ -172,19 +148,6 @@ class TestCapabilityStopTerminal(unittest.TestCase):
         )
         self.assertFalse(result["pure"])
 
-    def test_empty_queue_with_parsed_trusted_status_passes_gate_check(self):
-        report = (
-            "Cannot perform reviewer task under current profile. "
-            "No reviewer mutations performed.\n"
-            "Repository: Scaled-Tech-Consulting/Gitea-Tools\n"
-            "pr_inventory_trust_gate.status: trusted_empty\n"
-            "pr_inventory_trust_gate.corroborated: true\n"
-            "Inventory profile: prgs-reviewer\n"
-            "No open PRs in queue."
-        )
-        result = assess_capability_stop_terminal_report(report)
-        self.assertTrue(result["pure"])
-
     def test_pure_terminal_report_passes(self):
         report = (
             "Cannot perform reviewer task under current profile. "

tests/test_health.py

@@ -3,60 +3,9 @@ import subprocess
 import sys
 import unittest
 from unittest.mock import patch
-
 import role_session_router
 from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
 
-_HEALTH_SUBPROCESS_TIMEOUT_SEC = 30
-
-
-def _health_test_python():
-    """Portable interpreter for subprocess health checks (#245).
-
-    Prefer the active pytest interpreter, then ``GITEA_TOOLS_TEST_PYTHON``,
-    else skip — never hard-code ``<repo>/venv/bin/python``.
-    """
-    if (
-        sys.executable
-        and os.path.isfile(sys.executable)
-        and os.access(sys.executable, os.X_OK)
-    ):
-        return sys.executable
-    override = (os.environ.get("GITEA_TOOLS_TEST_PYTHON") or "").strip()
-    if override and os.path.isfile(override) and os.access(override, os.X_OK):
-        return override
-    raise unittest.SkipTest(
-        "repo venv not available; run under a python interpreter or set "
-        "GITEA_TOOLS_TEST_PYTHON"
-    )
-
-
-def _run_python_script(cwd, script_path, *, timeout=_HEALTH_SUBPROCESS_TIMEOUT_SEC):
-    """Run a script in a child process with timeout and guaranteed teardown."""
-    python = _health_test_python()
-    proc = subprocess.Popen(
-        [python, script_path],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        cwd=cwd,
-    )
-    try:
-        stdout, stderr = proc.communicate(timeout=timeout)
-    except subprocess.TimeoutExpired:
-        proc.kill()
-        stdout, stderr = proc.communicate()
-        raise
-    finally:
-        if proc.poll() is None:
-            proc.terminate()
-            try:
-                proc.wait(timeout=5)
-            except subprocess.TimeoutExpired:
-                proc.kill()
-                proc.wait()
-    return proc.returncode, stdout, stderr, proc
-
-
 class TestMCPHealth(unittest.TestCase):
 
     def setUp(self):
@@ -69,17 +18,6 @@ class TestMCPHealth(unittest.TestCase):
         if os.path.exists(self.temp_file):
             os.remove(self.temp_file)
 
-    def test_health_test_python_prefers_sys_executable(self):
-        resolved = _health_test_python()
-        self.assertEqual(resolved, sys.executable)
-
-    def test_health_test_python_skips_when_no_interpreter(self):
-        with patch.object(sys, "executable", ""), patch.dict(
-            os.environ, {}, clear=True
-        ):
-            with self.assertRaises(unittest.SkipTest):
-                _health_test_python()
-
     def test_startup_conflict_detection(self):
         # Create a Python file with conflict markers constructed dynamically
         with open(self.temp_file, "w") as f:
@@ -89,44 +27,32 @@ class TestMCPHealth(unittest.TestCase):
             f.write("print('world')\n")
             f.write(">" * 7 + " main\n")
 
+        # Run mcp_server.py
+        venv_python = os.path.join(self.project_root, "venv", "bin", "python")
         script_path = os.path.join(self.project_root, "mcp_server.py")
-        returncode, _stdout, stderr, _proc = _run_python_script(
-            self.project_root, script_path
+        res = subprocess.run(
+            [venv_python, script_path],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            cwd=self.project_root
         )
 
-        self.assertEqual(returncode, 1)
-        stderr_text = stderr.decode()
-        self.assertIn("infra_stop", stderr_text)
-        self.assertIn(
-            "Unresolved merge conflict detected in test_temp_conflict.py",
-            stderr_text,
-        )
+        self.assertEqual(res.returncode, 1)
+        stderr = res.stderr.decode()
+        self.assertIn("infra_stop", stderr)
+        self.assertIn("Unresolved merge conflict detected in test_temp_conflict.py", stderr)
 
     @patch("role_session_router.check_mid_merge", return_value=True)
-    @patch(
-        "mcp_server.get_profile",
-        return_value={
-            "profile_name": "prgs-reviewer",
-            "allowed_operations": ["gitea.pr.review"],
-        },
-    )
+    @patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
     def test_route_task_session_blocks_during_merge(self, mock_profile, mock_check):
         res = gitea_route_task_session("review_pr")
         self.assertEqual(res["route_result"], "infra_stop")
         self.assertIn("infra_stop", res["message"])
 
     @patch("role_session_router.check_mid_merge", return_value=True)
-    @patch(
-        "mcp_server.get_profile",
-        return_value={
-            "profile_name": "prgs-reviewer",
-            "allowed_operations": ["gitea.pr.review"],
-        },
-    )
-    def test_resolve_task_capability_blocks_during_merge(
-        self, mock_profile, mock_check
-    ):
+    @patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
+    def test_resolve_task_capability_blocks_during_merge(self, mock_profile, mock_check):
         res = gitea_resolve_task_capability("review_pr")
         self.assertTrue(res["infra_stop"])
         self.assertFalse(res["allowed_in_current_session"])
-        self.assertIn("infra_stop", res["exact_safe_next_action"])
+        self.assertIn("infra_stop", res["exact_safe_next_action"])

tests/test_issue_lock_worktree.py

@@ -1,99 +0,0 @@
-"""Tests for issue-lock worktree validation (#249)."""
-
-import os
-import sys
-import unittest
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-import issue_lock_worktree  # noqa: E402
-
-
-class TestIssueLockWorktreeAssessment(unittest.TestCase):
-    def test_clean_base_branch_passes(self):
-        result = issue_lock_worktree.assess_issue_lock_worktree(
-            worktree_path="/scratch/wt",
-            current_branch="master",
-            porcelain_status="",
-        )
-        self.assertTrue(result["proven"])
-        self.assertFalse(result["block"])
-
-    def test_dirty_tracked_files_fail(self):
-        result = issue_lock_worktree.assess_issue_lock_worktree(
-            worktree_path="/scratch/wt",
-            current_branch="master",
-            porcelain_status=" M gitea_mcp_server.py\n",
-        )
-        self.assertFalse(result["proven"])
-        self.assertIn("tracked file edits exist before issue lock", result["reasons"][0])
-
-    def test_feature_branch_fails(self):
-        result = issue_lock_worktree.assess_issue_lock_worktree(
-            worktree_path="/scratch/wt",
-            current_branch="feat/issue-243-forbidden-git-gaps",
-            porcelain_status="",
-        )
-        self.assertFalse(result["proven"])
-        self.assertIn("issue lock must be taken from base branch", result["reasons"][0])
-
-    def test_untracked_files_do_not_block(self):
-        result = issue_lock_worktree.assess_issue_lock_worktree(
-            worktree_path="/scratch/wt",
-            current_branch="main",
-            porcelain_status="?? notes.txt\n",
-        )
-        self.assertTrue(result["proven"])
-
-
-class TestIssueLockWorktreeResolution(unittest.TestCase):
-    def test_explicit_path_wins(self):
-        resolved = issue_lock_worktree.resolve_author_worktree_path(
-            "/tmp/scratch/wt", "/shared/dev"
-        )
-        self.assertEqual(resolved, os.path.realpath("/tmp/scratch/wt"))
-
-    def test_env_var_when_explicit_missing(self):
-        with patch.dict(os.environ, {"GITEA_AUTHOR_WORKTREE": "/tmp/from-env"}):
-            resolved = issue_lock_worktree.resolve_author_worktree_path(
-                None, "/shared/dev"
-            )
-        self.assertEqual(resolved, os.path.realpath("/tmp/from-env"))
-
-    def test_project_root_fallback(self):
-        resolved = issue_lock_worktree.resolve_author_worktree_path(
-            None, "/shared/dev"
-        )
-        self.assertEqual(resolved, os.path.realpath("/shared/dev"))
-
-
-class TestPrWorktreeMatch(unittest.TestCase):
-    def test_matching_paths_pass(self):
-        result = issue_lock_worktree.verify_pr_worktree_matches_lock(
-            "/tmp/scratch/wt",
-            "/tmp/scratch/wt",
-            "/shared/dev",
-        )
-        self.assertTrue(result["proven"])
-
-    def test_mismatch_fails(self):
-        result = issue_lock_worktree.verify_pr_worktree_matches_lock(
-            "/tmp/scratch/wt",
-            "/shared/dev",
-            "/shared/dev",
-        )
-        self.assertFalse(result["proven"])
-        self.assertIn("does not match locked worktree", result["reasons"][0])
-
-    def test_missing_locked_path_skips_check(self):
-        result = issue_lock_worktree.verify_pr_worktree_matches_lock(
-            None,
-            "/any/path",
-            "/shared/dev",
-        )
-        self.assertTrue(result["proven"])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_mcp_server.py

@@ -47,22 +47,6 @@ import mcp_server
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
-
-def _formal_review(reviewer, verdict, sha="abc123", review_id=1):
-    return {
-        "id": review_id,
-        "user": {"login": reviewer},
-        "state": verdict,
-        "commit_id": sha,
-        "submitted_at": "2026-07-06T10:00:00Z",
-        "dismissed": False,
-    }
-
-
-def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
-    return [_formal_review(reviewer, "APPROVED", sha=sha)]
-
-
 # Issue-write tools are profile-gated (#69).
 ISSUE_WRITE_ENV = {
     "GITEA_ALLOWED_OPERATIONS": (
@@ -81,20 +65,6 @@ CREATE_PR_ENV = {
     ),
 }
 
-ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
-
-
-def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
-    record = {
-        "issue_number": issue_number,
-        "branch_name": branch_name,
-        "remote": "dadeschools",
-        "org": "Scaled-Tech-Consulting",
-        "repo": "Gitea-Tools",
-    }
-    record.update(overrides)
-    return record
-
 
 # ---------------------------------------------------------------------------
 # Create Issue
@@ -157,19 +127,15 @@ class TestCreatePR(unittest.TestCase):
     @patch("os.path.exists", return_value=True)
     @patch("builtins.open")
     def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
-        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
-        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
+        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertEqual(result["number"], 3)
         self.assertNotIn("url", result)
-        mock_exists.assert_called_with(ISSUE_LOCK_FILE)
-        mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
         payload = mock_api.call_args[0][3]
         self.assertEqual(payload["head"], "feat/x")
         self.assertEqual(payload["base"], "main")
-        self.assertIn("Closes #123", payload["title"])
 
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            return_value=(True, []))
@@ -178,28 +144,13 @@ class TestCreatePR(unittest.TestCase):
     @patch("os.path.exists", return_value=True)
     @patch("builtins.open")
     def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
-        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
-        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
+        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
         with patch.dict(os.environ, env, clear=True):
             result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertIn("pulls/3", result["url"])
 
-    @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)
-    @patch("os.path.exists", return_value=True)
-    @patch("builtins.open")
-    def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
-        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
-        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
-            with self.assertRaises(ValueError) as ctx:
-                gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
-        self.assertIn("Closes #123", str(ctx.exception))
-        mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
-
 
 # ---------------------------------------------------------------------------
 # Close Issue
@@ -504,10 +455,6 @@ class TestMergePR(unittest.TestCase):
                 f"unexpected merge mutation: {method} {url}",
             )
 
-    def _feedback_reads(self, author="author-bot", sha="abc123"):
-        """PR + reviews GETs for gitea_get_pr_review_feedback during merge."""
-        return [self._pr(author, sha=sha), _visible_approval_reviews(sha=sha)]
-
     # -- success --------------------------------------------------------------
 
     @patch("mcp_server.api_request")
@@ -515,7 +462,6 @@ class TestMergePR(unittest.TestCase):
     def test_merge_succeeds_when_all_gates_pass(self, _auth, mock_api):
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
-            *self._feedback_reads(),
             {},  # merge POST
             {"merged_commit_sha": "mergecommit99"},  # read-back
         ]
@@ -532,8 +478,8 @@ class TestMergePR(unittest.TestCase):
         self.assertEqual(r["head_sha"], "abc123")
         self.assertEqual(r["merge_method"], "squash")
         self.assertEqual(r["merge_commit"], "mergecommit99")
-        # 5th call is the merge POST with the requested method/title/message.
-        merge_call = mock_api.call_args_list[4]
+        # 3rd call is the merge POST with the requested method/title/message.
+        merge_call = mock_api.call_args_list[2]
         self.assertEqual(merge_call.args[0], "POST")
         self.assertTrue(merge_call.args[1].endswith("/pulls/8/merge"))
         payload = merge_call.args[3]
@@ -548,7 +494,6 @@ class TestMergePR(unittest.TestCase):
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
             [{"filename": "a.py"}, {"filename": "b.py"}],  # files
-            *self._feedback_reads(),
             {},  # merge POST
             {"merged_commit_sha": "c1"},  # read-back
         ]
@@ -568,7 +513,6 @@ class TestMergePR(unittest.TestCase):
         """Merge OK + read-back GET failure => explicit cleanup skip, not silence."""
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
-            *self._feedback_reads(),
             {},  # merge POST
             RuntimeError("HTTP 502: Gitea upstream unavailable"),  # read-back fails
         ]
@@ -584,8 +528,8 @@ class TestMergePR(unittest.TestCase):
         # The skip is explicit, not silent.
         self.assertEqual(r["cleanup_status"], "skipped (merge read-back failed)")
         # No tracker-cleanup API traffic after the failed read-back:
-        # user, PR (eligibility), feedback PR+reviews, merge POST, read-back.
-        self.assertEqual(mock_api.call_count, 6)
+        # user, PR (eligibility), merge POST, read-back — and nothing more.
+        self.assertEqual(mock_api.call_count, 4)
         for c in mock_api.call_args_list:
             self.assertNotEqual(c.args[0], "DELETE")
 
@@ -597,7 +541,6 @@ class TestMergePR(unittest.TestCase):
         """Unexpected cleanup exception => merge still succeeds; error surfaced redacted."""
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
-            *self._feedback_reads(),
             {},  # merge POST
             {"merged_commit_sha": "c9"},  # read-back OK
         ]
@@ -785,7 +728,6 @@ class TestMergePR(unittest.TestCase):
     def test_output_redacts_secrets(self, _auth, mock_api):
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
-            *self._feedback_reads(),
             {}, {"merged_commit_sha": "c1"},
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
@@ -803,7 +745,6 @@ class TestMergePR(unittest.TestCase):
     def test_merge_error_message_redacts_credential(self, _auth, mock_api):
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
-            *self._feedback_reads(),
             RuntimeError("HTTP 500: token abc-secret-xyz rejected"),
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
@@ -816,45 +757,6 @@ class TestMergePR(unittest.TestCase):
         self.assertIn("[REDACTED]", blob)
         self.assertNotIn("abc-secret-xyz", blob)
 
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_merge_blocked_without_visible_approval(self, _auth, mock_api):
-        mock_api.side_effect = [
-            {"login": "merger-bot"}, self._pr("author-bot"),
-            self._pr("author-bot"),
-            [_formal_review("sysadmin", "PENDING")],
-        ]
-        env = {"GITEA_PROFILE_NAME": "gitea-merger",
-               "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")
-        self.assertFalse(r["performed"])
-        self.assertFalse(r.get("approval_visible"))
-        self.assertTrue(any("no visible APPROVED review" in x for x in r["reasons"]))
-        self._assert_no_merge_call(mock_api)
-
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_merge_blocked_on_request_changes(self, _auth, mock_api):
-        mock_api.side_effect = [
-            {"login": "merger-bot"}, self._pr("author-bot"),
-            self._pr("author-bot"),
-            [
-                _formal_review("reviewer-bot", "APPROVED"),
-                _formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=2),
-            ],
-        ]
-        env = {"GITEA_PROFILE_NAME": "gitea-merger",
-               "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")
-        self.assertFalse(r["performed"])
-        self.assertTrue(r.get("has_blocking_change_requests"))
-        self.assertTrue(any("REQUEST_CHANGES" in x for x in r["reasons"]))
-        self._assert_no_merge_call(mock_api)
-
 
 class TestNoUngatedMergePath(unittest.TestCase):
     """Prove no other exposed tool can merge (#16 surface audit)."""
@@ -1610,9 +1512,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_duplicate_terminal_decision_blocked(self, _auth, mock_api):
         mock_api.side_effect = [
-            {"login": "reviewer-bot"}, self._pr("author-bot"),
-            {"id": 1, "state": "APPROVED"},
-            [_formal_review("reviewer-bot", "APPROVED", review_id=1)],
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 1},
             {"login": "reviewer-bot"}, self._pr("author-bot"),
         ]
         gitea_mark_final_review_decision(
@@ -1695,9 +1595,7 @@ class TestSubmitPrReview(unittest.TestCase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_approve_succeeds_when_eligible(self, _auth, mock_api):
         mock_api.side_effect = [
-            {"login": "reviewer-bot"}, self._pr("author-bot"),
-            {"id": 7, "state": "APPROVED"},
-            [_formal_review("reviewer-bot", "APPROVED", review_id=7)],
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 7},
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
@@ -1707,63 +1605,16 @@ class TestSubmitPrReview(unittest.TestCase):
                 final_review_decision_ready=True,
             )
         self.assertTrue(r["performed"])
-        self.assertTrue(r.get("review_verdict_visible"))
         self.assertEqual(r["authenticated_user"], "reviewer-bot")
         self.assertEqual(r["pr_author"], "author-bot")
         self.assertEqual(r["head_sha"], "abc123")
-        post_calls = [
-            c for c in mock_api.call_args_list
-            if c.args[0] == "POST" and c.args[1].endswith("/pulls/8/reviews")
-        ]
-        self.assertEqual(len(post_calls), 1)
-        payload = post_calls[0].args[3]
-        self.assertEqual(payload["event"], "APPROVED")
+        method, url = mock_api.call_args.args[0], mock_api.call_args.args[1]
+        self.assertEqual(method, "POST")
+        self.assertTrue(url.endswith("/pulls/8/reviews"))
+        payload = mock_api.call_args.args[3]
+        self.assertEqual(payload["event"], "APPROVE")
         self.assertEqual(payload["commit_id"], "abc123")
 
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_approve_fails_when_verdict_stays_pending(self, _auth, mock_api):
-        mock_api.side_effect = [
-            {"login": "reviewer-bot"}, self._pr("author-bot"),
-            {"id": 7, "state": "PENDING"},
-            {"id": 7, "state": "PENDING"},
-            [_formal_review("reviewer-bot", "PENDING", review_id=7)],
-        ]
-        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
-               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=True):
-            r = gitea_submit_pr_review(
-                pr_number=8, action="approve", body="LGTM", remote="prgs",
-                final_review_decision_ready=True,
-            )
-        self.assertFalse(r["performed"])
-        self.assertFalse(r.get("review_verdict_visible"))
-        self.assertTrue(any("expected visible 'APPROVED'" in x for x in r["reasons"]))
-
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_approve_submits_pending_draft_when_api_returns_pending(self, _auth, mock_api):
-        mock_api.side_effect = [
-            {"login": "reviewer-bot"}, self._pr("author-bot"),
-            {"id": 7, "state": "PENDING"},
-            {"id": 7, "state": "APPROVED"},
-            [_formal_review("reviewer-bot", "APPROVED", review_id=7)],
-        ]
-        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
-               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=True):
-            r = gitea_submit_pr_review(
-                pr_number=8, action="approve", body="LGTM", remote="prgs",
-                final_review_decision_ready=True,
-            )
-        self.assertTrue(r["performed"])
-        submit_calls = [
-            c for c in mock_api.call_args_list
-            if c.args[0] == "POST" and "/reviews/7" in c.args[1]
-        ]
-        self.assertEqual(len(submit_calls), 1)
-        self.assertEqual(submit_calls[0].args[3]["event"], "APPROVED")
-
     # -- request_changes ------------------------------------------------------
 
     @patch("mcp_server.api_request")
@@ -1771,9 +1622,7 @@ class TestSubmitPrReview(unittest.TestCase):
     def test_request_changes_succeeds_when_eligible(self, _auth, mock_api):
         gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
         mock_api.side_effect = [
-            {"login": "reviewer-bot"}, self._pr("author-bot"),
-            {"id": 9, "state": "REQUEST_CHANGES"},
-            [_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=9)],
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 9},
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,request_changes"}
@@ -1784,12 +1633,7 @@ class TestSubmitPrReview(unittest.TestCase):
                 final_review_decision_ready=True,
             )
         self.assertTrue(r["performed"])
-        post_calls = [
-            c for c in mock_api.call_args_list
-            if c.args[0] == "POST" and c.args[1].endswith("/pulls/8/reviews")
-        ]
-        self.assertEqual(len(post_calls), 1)
-        self.assertEqual(post_calls[0].args[3]["event"], "REQUEST_CHANGES")
+        self.assertEqual(mock_api.call_args.args[3]["event"], "REQUEST_CHANGES")
 
     def test_request_changes_blocked_without_eligibility(self):
         gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
@@ -1900,8 +1744,7 @@ class TestSubmitPrReview(unittest.TestCase):
                 patch("mcp_server.api_request") as mock_api:
             mock_api.side_effect = [
                 {"login": "reviewer-bot"}, self._pr("author-bot", sha="abc123"),
-                {"id": 5, "state": "APPROVED"},
-                [_formal_review("reviewer-bot", "APPROVED", review_id=5)],
+                {"id": 5},
             ]
             env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                    "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
@@ -2053,12 +1896,8 @@ class TestSubmitPrReview(unittest.TestCase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_correction_flow_allows_second_terminal_review(self, _auth, mock_api):
         mock_api.side_effect = [
-            {"login": "reviewer-bot"}, self._pr("author-bot"),
-            {"id": 42, "state": "APPROVED"},
-            [_formal_review("reviewer-bot", "APPROVED", review_id=42)],
-            {"login": "reviewer-bot"}, self._pr("author-bot"),
-            {"id": 43, "state": "REQUEST_CHANGES"},
-            [_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=43)],
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 42},
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 43},
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
@@ -2131,7 +1970,7 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         patch("gitea_audit.audit_enabled", return_value=True).start()
         self.mock_audit = patch("gitea_audit.write_event").start()
         # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
-        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["read", "merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
+        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
 
     def tearDown(self):
         patch.stopall()
@@ -2179,8 +2018,6 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         def api_side_effect(method, url, auth, payload=None):
             if method == "GET" and "/user" in url:
                 return {"login": "merger"}
-            if method == "GET" and url.endswith("/reviews"):
-                return [_formal_review("reviewer", "APPROVED", sha="sha123")]
             if method == "GET" and "pulls/1" in url and "/files" not in url:
                 return {
                     "user": {"login": "author"},
@@ -2213,8 +2050,6 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         def api_side_effect(method, url, auth, payload=None):
             if method == "GET" and "/user" in url:
                 return {"login": "merger"}
-            if method == "GET" and url.endswith("/reviews"):
-                return [_formal_review("reviewer", "APPROVED", sha="sha123")]
             if method == "GET" and "pulls/1" in url and "/files" not in url:
                 return {
                     "user": {"login": "author"},
@@ -2934,41 +2769,26 @@ class TestVerifyMutationAuthority(unittest.TestCase):
 class TestIssueLocking(unittest.TestCase):
     """Test issue locking and PR gating constraints."""
 
-    @staticmethod
-    def _clean_master_git_state():
-        return {"current_branch": "master", "porcelain_status": ""}
-
     def tearDown(self):
-        if os.path.exists(ISSUE_LOCK_FILE):
-            os.remove(ISSUE_LOCK_FILE)
+        if os.path.exists("/tmp/gitea_issue_lock.json"):
+            os.remove("/tmp/gitea_issue_lock.json")
 
-    @patch(
-        "mcp_server.issue_lock_worktree.read_worktree_git_state",
-        return_value={"current_branch": "master", "porcelain_status": ""},
-    )
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_success(self, _auth, mock_api, _git_state):
+    def test_lock_issue_success(self, _auth, mock_api):
         mock_api.return_value = [] # no open PRs
         res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertTrue(res["success"])
-        self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
-        with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
-            lock = json.load(f)
-        self.assertIn("worktree_path", lock)
+        self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
 
     def test_lock_issue_mismatch_branch_fails(self):
         with self.assertRaises(ValueError) as ctx:
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
         self.assertIn("must contain locked issue pattern", str(ctx.exception))
 
-    @patch(
-        "mcp_server.issue_lock_worktree.read_worktree_git_state",
-        return_value={"current_branch": "master", "porcelain_status": ""},
-    )
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api, _git_state):
+    def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api):
         mock_api.return_value = [{
             "number": 200,
             "head": {"ref": "feat/issue-196-boundary"},
@@ -2979,13 +2799,9 @@ class TestIssueLocking(unittest.TestCase):
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("already tied to an open PR", str(ctx.exception))
 
-    @patch(
-        "mcp_server.issue_lock_worktree.read_worktree_git_state",
-        return_value={"current_branch": "master", "porcelain_status": ""},
-    )
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api, _git_state):
+    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api):
         mock_api.return_value = [{
             "number": 200,
             "head": {"ref": "feat/other-branch"},
@@ -2996,68 +2812,12 @@ class TestIssueLocking(unittest.TestCase):
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("already tied to an open PR", str(ctx.exception))
 
-    @patch("mcp_server.api_get_all", return_value=[])
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_from_clean_scratch_worktree(self, _auth, _api):
-        scratch = "/tmp/gitea-tools-author-scratch/issue-249-clean"
-        with patch(
-            "mcp_server.issue_lock_worktree.read_worktree_git_state",
-            return_value={"current_branch": "master", "porcelain_status": ""},
-        ) as mock_git:
-            res = gitea_lock_issue(
-                issue_number=249,
-                branch_name="feat/issue-249-issue-lock-scratch-worktree",
-                remote="prgs",
-                worktree_path=scratch,
-            )
-        self.assertTrue(res["success"])
-        self.assertEqual(res["worktree_path"], os.path.realpath(scratch))
-        mock_git.assert_called_once_with(os.path.realpath(scratch))
-
-    @patch("mcp_server.api_get_all", return_value=[])
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_fails_when_declared_worktree_dirty(self, _auth, _api):
-        with patch(
-            "mcp_server.issue_lock_worktree.read_worktree_git_state",
-            return_value={
-                "current_branch": "master",
-                "porcelain_status": " M gitea_mcp_server.py\n",
-            },
-        ):
-            with self.assertRaises(RuntimeError) as ctx:
-                gitea_lock_issue(
-                    issue_number=249,
-                    branch_name="feat/issue-249-issue-lock-scratch-worktree",
-                    remote="prgs",
-                    worktree_path="/tmp/scratch/wt",
-                )
-        self.assertIn("tracked file edits exist before issue lock", str(ctx.exception))
-
-    @patch("mcp_server.api_get_all", return_value=[])
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_fails_when_declared_worktree_not_on_base(self, _auth, _api):
-        with patch(
-            "mcp_server.issue_lock_worktree.read_worktree_git_state",
-            return_value={
-                "current_branch": "feat/issue-243-forbidden-git-gaps",
-                "porcelain_status": "",
-            },
-        ):
-            with self.assertRaises(RuntimeError) as ctx:
-                gitea_lock_issue(
-                    issue_number=249,
-                    branch_name="feat/issue-249-issue-lock-scratch-worktree",
-                    remote="prgs",
-                    worktree_path="/tmp/scratch/wt",
-                )
-        self.assertIn("issue lock must be taken from base branch", str(ctx.exception))
-
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_missing_lock_fails(self, _auth, _role):
-        if os.path.exists(ISSUE_LOCK_FILE):
-            os.remove(ISSUE_LOCK_FILE)
+        if os.path.exists("/tmp/gitea_issue_lock.json"):
+            os.remove("/tmp/gitea_issue_lock.json")
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(RuntimeError) as ctx:
                 gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
@@ -3067,9 +2827,8 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_branch_mismatch_fails(self, _auth, _role):
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump(_sample_issue_lock(
-                issue_number=196, branch_name="feat/issue-196-mutations"), f)
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(ValueError) as ctx:
                 gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
@@ -3079,9 +2838,8 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_forbidden_terms_fails(self, _auth, _role):
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump(_sample_issue_lock(
-                issue_number=196, branch_name="feat/issue-196-mutations"), f)
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             for term in ("equivalent to #196", "related to #196", "same as #196"):
                 with self.assertRaises(ValueError) as ctx:
@@ -3092,57 +2850,13 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump(_sample_issue_lock(
-                issue_number=196, branch_name="feat/issue-196-mutations"), f)
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(ValueError) as ctx:
                 gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
         self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
 
-    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
-           return_value=(True, []))
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_worktree_mismatch_fails(self, _auth, _role):
-        scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-pr")
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump(_sample_issue_lock(
-                issue_number=249,
-                branch_name="feat/issue-249-issue-lock-scratch-worktree",
-                worktree_path=scratch,
-            ), f)
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
-            with self.assertRaises(ValueError) as ctx:
-                gitea_create_pr(
-                    title="feat: lock scratch worktree Closes #249",
-                    head="feat/issue-249-issue-lock-scratch-worktree",
-                    remote="prgs",
-                    worktree_path="/tmp/other-scratch",
-                )
-        self.assertIn("does not match locked worktree", str(ctx.exception))
-
-    @patch("mcp_server.api_request")
-    @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_honors_scratch_worktree_lock(self, _auth, _role, mock_api):
-        scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-e2e")
-        mock_api.return_value = {"number": 250, "html_url": "https://example/pr/250"}
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump(_sample_issue_lock(
-                issue_number=249,
-                branch_name="feat/issue-249-issue-lock-scratch-worktree",
-                worktree_path=scratch,
-            ), f)
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
-            res = gitea_create_pr(
-                title="feat: issue-lock scratch worktree Closes #249",
-                head="feat/issue-249-issue-lock-scratch-worktree",
-                remote="prgs",
-                worktree_path=scratch,
-            )
-        self.assertEqual(res["number"], 250)
-
 
 # ---------------------------------------------------------------------------
 # Pre-flight ordering and workspace edit block (#210)

tests/test_pr_queue_inventory.py

@@ -119,28 +119,12 @@ class TestPRQueueInventory(unittest.TestCase):
         mock_get_all.return_value = [
             {"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other_user"}}
         ]
-        # mock_api: inventory whoami, eligibility whoami, eligibility PR,
-        # POST review (#244: state + visible-verdict GET reviews).
+        # mock_api: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility), 4) /pulls/1/reviews (POST review)
         mock_api.side_effect = [
-            {"login": "reviewer1"},
-            {"login": "reviewer1"},
-            {
-                "user": {"login": "other_user"},
-                "state": "open",
-                "head": {"sha": "abc1"},
-                "mergeable": True,
-            },
-            {"id": 100, "state": "APPROVED"},
-            [
-                {
-                    "id": 100,
-                    "user": {"login": "reviewer1"},
-                    "state": "APPROVED",
-                    "commit_id": "abc1",
-                    "submitted_at": "2026-07-06T10:00:00Z",
-                    "dismissed": False,
-                }
-            ],
+            {"login": "reviewer1"}, # inventory whoami
+            {"login": "reviewer1"}, # submit eligibility whoami
+            {"user": {"login": "other_user"}, "state": "open", "head": {"sha": "abc1"}, "mergeable": True}, # submit eligibility PR
+            {"id": 100}, # POST review
         ]
 
         from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
@@ -277,79 +261,12 @@ class TestPRQueueInventory(unittest.TestCase):
         for call in mock_api.call_args_list:
             self.assertEqual(call.args[0], "GET")
 
-    @patch("mcp_server._local_git_remote_url")
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("mcp_server.get_profile")
-    def test_empty_inventory_runs_trust_gate_and_blocks_without_trusted_empty(
-        self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
-    ):
-        mock_get_profile.return_value = {
-            "profile_name": "gitea-author",
-            "allowed_operations": ["read"],
-            "forbidden_operations": [],
-            "base_url": None,
-        }
-        mock_get_all.return_value = []
-        mock_api.return_value = {"login": "jcwalker3"}
-        mock_local_url.return_value = None
-
-        result = gitea_review_pr(
-            pr_number=1,
-            event="APPROVE",
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-        )
-        self.assertFalse(result["success"])
-        msg = result["message"]
-        self.assertIn("pr_inventory_trust_gate.status: untrusted_empty", msg)
-        self.assertIn("Empty-queue claim blocked", msg)
-        self.assertNotIn("Open PRs found: 0 (trusted_empty)", msg)
-
-    @patch("mcp_server._local_git_remote_url")
-    @patch("mcp_server.api_get_all")
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    @patch("mcp_server.get_profile")
-    def test_empty_inventory_trusted_empty_when_gate_passes(
-        self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
-    ):
-        mock_get_profile.return_value = {
-            "profile_name": "gitea-author",
-            "allowed_operations": ["read"],
-            "forbidden_operations": [],
-            "base_url": None,
-        }
-        mock_get_all.return_value = []
-        mock_api.return_value = {"login": "jcwalker3"}
-        mock_local_url.return_value = (
-            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
-        )
-
-        result = gitea_review_pr(
-            pr_number=1,
-            event="APPROVE",
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-        )
-        self.assertFalse(result["success"])
-        msg = result["message"]
-        self.assertIn("pr_inventory_trust_gate.status: trusted_empty", msg)
-        self.assertIn("Open PRs found: 0 (trusted_empty)", msg)
-
-    @patch("mcp_server._local_git_remote_url")
-    @patch("mcp_server.api_get_all")
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    @patch("mcp_server.get_profile")
-    def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url):
+    def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
         """Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory."""
-        mock_local_url.return_value = (
-            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
-        )
         for event in ["APPROVE", "REQUEST_CHANGES"]:
             mock_get_profile.return_value = {
                 "profile_name": "gitea-author",

tests/test_review_proofs.py

@@ -21,21 +21,11 @@ import unittest
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
 from review_proofs import (  # noqa: E402
-    ISSUE_SELECTION_CONTINUATION_EXPLICIT,
-    ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
     assess_author_pr_report,
     assess_capability_evidence,
     assess_capability_proof,
-    assess_contradictory_no_pr_claim,
-    assess_continuation_mode_report,
     assess_controller_handoff,
-    assess_edited_pr_inventory_coverage,
-    assess_empty_queue_report,
-    assess_fresh_issue_selection,
     assess_inventory_completeness,
-    assess_issue_selection_final_report,
-    assess_queue_target_final_report,
-    assess_reviewer_queue_inventory,
     assess_live_state_recheck,
     assess_review_mutation_final_report,
     assess_role_boundary,
@@ -44,9 +34,7 @@ from review_proofs import (  # noqa: E402
     assess_sweep_evidence,
     assess_validation_report,
     build_final_report,
-    classify_issue_for_selection,
     pr_inventory_trust_gate,
-    reconcile_queue_target,
     resolve_repos_from_user_reference,
     verify_pinned_head_checkout,
 )
@@ -195,24 +183,6 @@ def _good_review_mutation():
     return assess_review_mutation_final_report(report, lock)
 
 
-def _good_worktree(**overrides):
-    proof = {
-        "worktree_path": "/repo/branches/review-feat-issue-224",
-        "porcelain_status": "",
-        "pr_scope_files": ["docs/wiki/Repositories.md"],
-        "scratch_used": True,
-        "scratch_path": "/repo/branches/review-feat-issue-224",
-        "git_commands": [
-            "git fetch prgs master feat/issue-224-wiki-proof-refresh",
-            "git diff prgs/master...prgs/feat/issue-224-wiki-proof-refresh",
-        ],
-    }
-    proof.update(overrides)
-    from reviewer_worktree import assess_reviewer_worktree_proof  # noqa: E402
-
-    return assess_reviewer_worktree_proof(proof)
-
-
 def _good_role_boundary_179(**overrides):
     kwargs = {
         "task_role": "reviewer",
@@ -625,13 +595,6 @@ class TestFinalReport(unittest.TestCase):
             "controller_handoff": _good_handoff(),
             "capability_proof": _good_capability_proof(),
             "sweep_proof": _good_secret_sweep(),
-            "worktree_proof": {
-                "worktree_path": "/repo/branches/review-feat-issue-224",
-                "porcelain_status": "",
-                "pr_scope_files": ["docs/wiki/Repositories.md"],
-                "scratch_used": True,
-                "scratch_path": "/repo/branches/review-feat-issue-224",
-            },
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -936,17 +899,12 @@ class TestControllerHandoff(unittest.TestCase):
         result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
         self.assertEqual(result["verdict"], "incomplete")
         self.assertIn("Pinned reviewed head", result["missing_fields"])
-        self.assertIn("Worktree path", result["missing_fields"])
         self.assertIn("Merge result", result["missing_fields"])
 
         complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected PR: #999",
             "- Reviewer eligibility: passed",
             "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Worktree path: /repo/branches/review-pr-999",
-            "- Worktree dirty: no",
-            "- Scratch worktree used: yes (/repo/branches/review-pr-999)",
-            "- Unrelated local mutations: none",
             "- Review decision: approve",
             "- Merge result: merged",
             "- Linked issue status: closed",
@@ -955,46 +913,24 @@ class TestControllerHandoff(unittest.TestCase):
         result = assess_controller_handoff(complete, role="review")
         self.assertEqual(result["verdict"], "complete")
 
-    def _author_role_fields(self, issue_number=182, pr_number=999):
-        return [
-            f"- Selected issue: #{issue_number}",
-            "- Issue lock proof: lock before diff on feat/x @ master",
-            "- Claim/comment status: comment-claimed",
-            f"- PR number opened: #{pr_number}",
-            "- No review/merge: confirmed",
-        ]
-
-    def test_handoff_role_fields_author_includes_issue_lock_proof(self):
-        from review_proofs import HANDOFF_ROLE_FIELDS
-        names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]]
-        self.assertIn("Issue lock proof", names)
-
     def test_author_role_requires_author_fields(self):
-        complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields())
-        result = assess_controller_handoff(complete, role="author")
-        self.assertEqual(result["verdict"], "complete")
-
-        result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
-        self.assertEqual(result["verdict"], "incomplete")
-        self.assertIn("Issue lock proof", result["missing_fields"])
-        self.assertIn("No review/merge confirmation", result["missing_fields"])
-
-    def test_author_role_requires_issue_lock_proof(self):
-        without_lock = self.BASE_HANDOFF + "\n" + "\n".join([
+        complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected issue: #182",
             "- Claim/comment status: comment-claimed",
             "- PR number opened: #999",
             "- No review/merge: confirmed",
         ])
-        result = assess_controller_handoff(without_lock, role="author")
+        result = assess_controller_handoff(complete, role="author")
+        self.assertEqual(result["verdict"], "complete")
+
+        result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
         self.assertEqual(result["verdict"], "incomplete")
-        self.assertIn("Issue lock proof", result["missing_fields"])
+        self.assertIn("No review/merge confirmation", result["missing_fields"])
 
     def test_author_role_rejects_equivalent_or_multiple_issues(self):
         # 1. equivalent reference blocked
         incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected issue: Issue #194 / #196 equivalent",
-            "- Issue lock proof: lock before diff on feat/x @ master",
             "- Claim/comment status: comment-claimed",
             "- PR number opened: #999",
             "- No review/merge: confirmed",
@@ -1006,7 +942,6 @@ class TestControllerHandoff(unittest.TestCase):
         # 2. multiple issues blocked
         incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected issue: #194, #196",
-            "- Issue lock proof: lock before diff on feat/x @ master",
             "- Claim/comment status: comment-claimed",
             "- PR number opened: #999",
             "- No review/merge: confirmed",
@@ -1018,7 +953,6 @@ class TestControllerHandoff(unittest.TestCase):
     def test_author_role_rejects_fuzzy_pr_number(self):
         incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected issue: #196",
-            "- Issue lock proof: lock before diff on feat/issue-196 @ master",
             "- Claim/comment status: comment-claimed",
             "- PR number opened: PR #203 / #204 equivalent",
             "- No review/merge: confirmed",
@@ -1031,10 +965,6 @@ class TestControllerHandoff(unittest.TestCase):
         complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Repositories checked: Gitea-Tools, mcp-control-plane",
             "- Open PR counts: 2 / 0",
-            "- PR inventory trust gate: trusted_nonempty / trusted_empty",
-            "- Trust gate reasons: none",
-            "- Trust gate corroborated: true",
-            "- Inventory profile: prgs-reviewer",
             "- Selected PR or reason: none eligible (self-authored)",
             "- Inventory completeness: complete, no pagination needed",
         ])
@@ -1054,17 +984,23 @@ class TestControllerHandoff(unittest.TestCase):
 
     def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
         # 1. Workspace mutations: none is rejected when local_edits is True
-        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
-            self._author_role_fields(issue_number=196, pr_number=203))
+        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #203",
+            "- No review/merge: confirmed",
+        ])
         res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
         self.assertEqual(res["verdict"], "incomplete")
         self.assertIn("Workspace mutations", res["missing_fields"])
 
         # 2. Workspace mutations: edited files is allowed when local_edits is True
-        complete_eq = self.BASE_HANDOFF.replace(
-            "- Workspace mutations: none",
-            "- Workspace mutations: edited review_proofs.py",
-        ) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
+        complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
+            "- Selected issue: #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #203",
+            "- No review/merge: confirmed",
+        ])
         res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
         self.assertEqual(res2["verdict"], "complete")
 
@@ -1145,144 +1081,6 @@ class TestReviewMutationFinalReport(unittest.TestCase):
         self.assertTrue(final["review_mutation_complete"])
 
 
-class TestQueueTargetReconciliation(unittest.TestCase):
-    """Queue target lock: reconcile operator-supplied backlog before inventory (#200)."""
-
-    CONFIGURED = [
-        "Scaled-Tech-Consulting/Gitea-Tools",
-        "Scaled-Tech-Consulting/mcp-control-plane",
-    ]
-    GITEA_TOOLS = "Scaled-Tech-Consulting/Gitea-Tools"
-    MCP = "Scaled-Tech-Consulting/mcp-control-plane"
-    OPERATOR_CONTEXT = (
-        "six open PRs in Scaled-Tech-Consulting/Gitea-Tools including "
-        "#195, #193, #192, #190, #187, and #181"
-    )
-    PROFILE = {
-        "profile_name": "prgs-reviewer",
-        "allowed_operations": ["read", "gitea.read"],
-    }
-
-    def test_supplied_gitea_tools_prs_but_inventoried_mcp_is_mismatch(self):
-        lock = reconcile_queue_target(
-            operator_context=self.OPERATOR_CONTEXT,
-            inventoried_repo=self.MCP,
-            configured_repos=self.CONFIGURED,
-        )
-        self.assertEqual(lock["status"], "target_repo_mismatch")
-        self.assertEqual(lock["resolved_repo"], self.GITEA_TOOLS)
-        self.assertEqual(lock["resolution_source"], "operator_context")
-        self.assertIn(195, lock["supplied_pr_numbers"])
-        self.assertFalse(lock["allow_clean_stop"])
-        self.assertFalse(lock["allow_trusted_empty"])
-
-    def test_wrong_repo_zero_open_cannot_stop_cleanly(self):
-        lock = reconcile_queue_target(
-            operator_context=self.OPERATOR_CONTEXT,
-            inventoried_repo=self.MCP,
-            configured_repos=self.CONFIGURED,
-        )
-        gate = pr_inventory_trust_gate(
-            [],
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="mcp-control-plane",
-            state="open",
-            authenticated_profile=self.PROFILE,
-            local_remote_url=(
-                "https://gitea.prgs.cc/Scaled-Tech-Consulting/"
-                "mcp-control-plane.git"
-            ),
-            corroboration_open_pr_counter=0,
-            queue_target_lock=lock,
-        )
-        self.assertEqual(gate["status"], "target_repo_mismatch")
-        self.assertNotEqual(gate["status"], "trusted_empty")
-
-    def test_supplied_pr_numbers_reconciled_before_empty_stop(self):
-        lock = reconcile_queue_target(
-            operator_context=(
-                "currently we have 6 open PRs: #195, #193, #192, "
-                "#190, #187, #181 in Gitea-Tools"
-            ),
-            inventoried_repo=self.GITEA_TOOLS,
-            configured_repos=self.CONFIGURED,
-        )
-        self.assertEqual(lock["status"], "resolved")
-        self.assertEqual(len(lock["supplied_pr_numbers"]), 6)
-        self.assertTrue(all(item["matches_inventoried_repo"]
-                            for item in lock["reconciliation"]))
-
-    def test_ambiguous_repo_context_fails_closed(self):
-        lock = reconcile_queue_target(
-            operator_context=(
-                "open PRs in mcp-control-plane and gitea-tools including PR #195"
-            ),
-            inventoried_repo=self.MCP,
-            configured_repos=self.CONFIGURED,
-        )
-        self.assertEqual(lock["status"], "unresolved")
-        self.assertFalse(lock["allow_trusted_empty"])
-        self.assertTrue(
-            any("could not be resolved" in r for r in lock["reasons"])
-        )
-
-    def test_conflicting_directive_vs_backlog_fails_closed(self):
-        lock = reconcile_queue_target(
-            operator_context="Repository: Scaled-Tech-Consulting/mcp-control-plane",
-            supplied_pr_backlog=[
-                {"number": 195, "repo": self.GITEA_TOOLS},
-            ],
-            inventoried_repo=self.MCP,
-            configured_repos=self.CONFIGURED,
-        )
-        self.assertEqual(lock["status"], "unresolved")
-        self.assertIn("conflicts", " ".join(lock["reasons"]).lower())
-
-    def test_final_report_must_document_reconciliation(self):
-        lock = reconcile_queue_target(
-            operator_context=self.OPERATOR_CONTEXT,
-            inventoried_repo=self.GITEA_TOOLS,
-            configured_repos=self.CONFIGURED,
-        )
-        incomplete = assess_queue_target_final_report(
-            "Open PRs: 0. Stopping.", lock
-        )
-        self.assertFalse(incomplete["complete"])
-        self.assertTrue(incomplete["downgraded"])
-
-        complete_report = "\n".join([
-            "Queue inventory complete.",
-            f"Resolved repo: {self.GITEA_TOOLS}",
-            "Resolution source: operator_context",
-            "queue_target_lock.status: resolved",
-            "Supplied PR reconciliation: #195, #193, #192, #190, #187, #181",
-        ])
-        complete = assess_queue_target_final_report(complete_report, lock)
-        self.assertTrue(complete["complete"])
-        self.assertFalse(complete["downgraded"])
-
-    def test_assess_reviewer_queue_inventory_blocks_mismatch(self):
-        result = assess_reviewer_queue_inventory(
-            [{
-                "repo": self.MCP,
-                "state_filter": "open",
-                "pagination_complete": True,
-                "open_pr_count": 0,
-                "list_prs_response": [],
-                "remote": "prgs",
-                "authenticated_profile": self.PROFILE,
-                "local_remote_url": (
-                    "https://gitea.prgs.cc/Scaled-Tech-Consulting/"
-                    "mcp-control-plane.git"
-                ),
-            }],
-            operator_context=self.OPERATOR_CONTEXT,
-        )
-        self.assertFalse(result["can_claim_empty_queue"])
-        self.assertIn("target_repo_mismatch", str(result["trust_gates"]))
-
-
 class TestPRInventoryTrustGate(unittest.TestCase):
     """Issue #194: unit tests for the PR inventory trust gate."""
 
@@ -1365,157 +1163,6 @@ class TestPRInventoryTrustGate(unittest.TestCase):
         self.assertTrue(res["corroborated"])
 
 
-class TestAssessReviewerQueueInventory(unittest.TestCase):
-    """Issue #196: trust gate wired into canonical queue inventory."""
-
-    def _repo_report(self, **overrides):
-        report = {
-            "repo": "Scaled-Tech-Consulting/Gitea-Tools",
-            "state_filter": "open",
-            "pagination_complete": True,
-            "open_pr_count": 0,
-            "list_prs_response": [],
-            "remote": "prgs",
-            "authenticated_profile": {
-                "profile_name": "prgs-reviewer",
-                "allowed_operations": ["read", "gitea.read"],
-            },
-            "local_remote_url": (
-                "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
-            ),
-        }
-        report.update(overrides)
-        return report
-
-    def test_empty_without_corroboration_blocks_empty_queue_claim(self):
-        result = assess_reviewer_queue_inventory([
-            self._repo_report(pagination_complete=False),
-            self._repo_report(
-                repo="Scaled-Tech-Consulting/mcp-control-plane",
-                open_pr_count=0,
-                pagination_complete=False,
-            ),
-        ])
-        self.assertFalse(result["can_claim_empty_queue"])
-        self.assertIn("untrusted_empty", str(result["trust_gates"]))
-
-    def test_trusted_empty_with_finality_allows_empty_queue_claim(self):
-        result = assess_reviewer_queue_inventory([
-            self._repo_report(),
-            self._repo_report(
-                repo="Scaled-Tech-Consulting/mcp-control-plane",
-                local_remote_url=(
-                    "https://gitea.prgs.cc/Scaled-Tech-Consulting/"
-                    "mcp-control-plane.git"
-                ),
-            ),
-        ])
-        self.assertTrue(result["can_claim_empty_queue"])
-        self.assertTrue(result["can_claim_exhaustive"])
-
-    def test_user_context_indicating_open_prs_blocks_empty_claim(self):
-        result = assess_reviewer_queue_inventory(
-            [self._repo_report()],
-            user_context="please review open PR #195 in the queue",
-        )
-        self.assertFalse(result["can_claim_empty_queue"])
-        self.assertTrue(result["blockers"])
-
-    def test_nonempty_inventory_skips_empty_trust_gate_block(self):
-        result = assess_reviewer_queue_inventory([
-            self._repo_report(open_pr_count=2),
-            self._repo_report(
-                repo="Scaled-Tech-Consulting/mcp-control-plane",
-                open_pr_count=1,
-            ),
-        ])
-        self.assertTrue(result["complete"])
-        self.assertEqual(result["trust_gates"], {})
-
-
-class TestAssessEmptyQueueReport(unittest.TestCase):
-    """Issue #198: empty-queue reports require formal trust-gate proof."""
-
-    def _trusted_report(self, **extra):
-        lines = [
-            "Queue inventory complete.",
-            "Repository: Scaled-Tech-Consulting/Gitea-Tools",
-            "Open PR count: 0",
-            "pr_inventory_trust_gate.status: trusted_empty",
-            "pr_inventory_trust_gate.corroborated: true",
-            "Inventory profile: prgs-reviewer",
-            "Workflow correctly stops with nothing to review.",
-        ]
-        lines.extend(extra)
-        return "\n".join(lines)
-
-    def test_non_empty_report_not_claimed(self):
-        result = assess_empty_queue_report("Reviewed PR #236 and merged.")
-        self.assertFalse(result["claimed"])
-        self.assertTrue(result["proven"])
-
-    def test_empty_claim_without_trust_gate_blocked(self):
-        report = (
-            "Open PR count: 0\n"
-            "Pagination complete: yes\n"
-            "Queue cleared."
-        )
-        result = assess_empty_queue_report(report)
-        self.assertTrue(result["claimed"])
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["block"])
-
-    def test_trusted_empty_report_with_required_fields_passes(self):
-        result = assess_empty_queue_report(self._trusted_report())
-        self.assertTrue(result["proven"])
-
-    def test_weak_merge_commit_corroboration_blocked(self):
-        report = (
-            "Open PR count: 0\n"
-            "Master latest commit is merge of PR #79 so queue is empty."
-        )
-        result = assess_empty_queue_report(report)
-        self.assertFalse(result["proven"])
-        self.assertTrue(
-            any("weak corroboration" in r for r in result["reasons"])
-        )
-
-    def test_author_session_reviewer_queue_wording_blocked(self):
-        result = assess_empty_queue_report(
-            self._trusted_report(),
-            task_role="author",
-        )
-        self.assertFalse(result["proven"])
-
-    def test_build_final_report_downgrades_weak_empty_queue(self):
-        final = build_final_report(
-            checkout_proof=_good_checkout(),
-            inventory=_good_inventory(),
-            validation=_good_validation(),
-            contamination=_good_contamination(),
-            identity_eligible=True,
-            merge_performed=False,
-            issue_status_verified=True,
-            capability_evidence=_good_capability_evidence(),
-            sweep=_good_sweep(),
-            live_state=_good_live_state(),
-            role_boundary=_good_role_boundary(),
-            review_mutation=_good_review_mutation(),
-            controller_handoff=_good_handoff(),
-            capability_proof=_good_capability_proof(),
-            sweep_proof=_good_secret_sweep(),
-            worktree_proof={
-                "worktree_path": "/repo/branches/review-pr-1",
-                "porcelain_status": "",
-                "scratch_used": True,
-                "scratch_path": "/repo/branches/review-pr-1",
-            },
-            report_text="Open PR count: 0. Queue cleared.",
-        )
-        self.assertNotEqual(final["grade"], "A")
-        self.assertFalse(final["empty_queue_trust_gate_proven"])
-
-
 class TestCapabilityEvidence(unittest.TestCase):
     """#179 gap 1: capability claims need exact evidence."""
 
@@ -1659,13 +1306,6 @@ class TestFinalReport179Bar(unittest.TestCase):
             "controller_handoff": _good_handoff(),
             "capability_proof": _good_capability_proof(),
             "sweep_proof": _good_secret_sweep(),
-            "worktree_proof": {
-                "worktree_path": "/repo/branches/review-feat-issue-224",
-                "porcelain_status": "",
-                "pr_scope_files": ["docs/wiki/Repositories.md"],
-                "scratch_used": True,
-                "scratch_path": "/repo/branches/review-feat-issue-224",
-            },
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -1810,120 +1450,5 @@ class TestAuthorReporting(unittest.TestCase):
         self.assertFalse(result["complete"])
 
 
-class TestIssueSelectionContinuation(unittest.TestCase):
-    """Issue #188: continuation mode wall for issues with open PRs."""
-
-    OLD_SHA = PINNED
-    NEW_SHA = OTHER
-    OPEN_PR = [{"number": 187, "head": {"ref": "feat/issue-183-harden-author-run-reporting"}}]
-
-    def test_open_pr_issue_excluded_from_fresh_selection(self):
-        classified = classify_issue_for_selection(
-            183, open_prs=self.OPEN_PR,
-        )
-        self.assertEqual(classified["status"], ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR)
-        self.assertFalse(classified["selectable_for_fresh_work"])
-        blocked = assess_fresh_issue_selection([classified])
-        self.assertTrue(blocked["downgraded"])
-
-    def test_explicit_continuation_allows_represented_issue(self):
-        classified = classify_issue_for_selection(
-            183,
-            open_prs=self.OPEN_PR,
-            operator_continuation_requested=True,
-        )
-        self.assertEqual(classified["status"], ISSUE_SELECTION_CONTINUATION_EXPLICIT)
-        blocked = assess_fresh_issue_selection([classified])
-        self.assertFalse(blocked["downgraded"])
-
-    def test_contradictory_no_pr_claim_downgrades(self):
-        report = (
-            "Selected issue #183; no duplicate PR open. "
-            "Updated PR #187 on branch feat/issue-183-harden-author-run-reporting."
-        )
-        result = assess_contradictory_no_pr_claim(
-            report, edited_pr_numbers=[187], issue_open_pr_map={183: 187},
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_edited_pr_must_appear_in_inventory(self):
-        report = "Open PR inventory: PR #195 only."
-        result = assess_edited_pr_inventory_coverage(
-            report,
-            edited_pr_numbers=[187],
-            inventoried_pr_numbers=[195],
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_continuation_report_requires_old_and_new_head(self):
-        report = (
-            "Issue #182 continuation mode. PR #186. "
-            f"old head {self.OLD_SHA} -> new head {self.NEW_SHA}. "
-            "PR author: jcwalker3. Branch: feat/issue-182-controller-handoff. "
-            "Session authored PR: yes. Continuation allowed: operator requested."
-        )
-        result = assess_continuation_mode_report(
-            report,
-            pr_number=186,
-            pr_author="jcwalker3",
-            branch="feat/issue-182-controller-handoff",
-            old_head_sha=self.OLD_SHA,
-            new_head_sha=self.NEW_SHA,
-            session_authored_pr=True,
-            continuation_allowed_reason="operator requested continuation",
-        )
-        self.assertTrue(result["complete"])
-
-    def test_issue_selection_final_report_continuation_earns_a(self):
-        report = "\n".join([
-            "Issue #182 continuation mode — no new issue claimed.",
-            f"PR #186 updated: old head {self.OLD_SHA}, "
-            f"new head {self.NEW_SHA}.",
-            "PR author: jcwalker3. Branch: feat/issue-182-controller-handoff.",
-            "Session authored PR: yes.",
-            "Continuation allowed: operator requested rebase.",
-            "Open PR inventory included PR #186.",
-            "## Controller Handoff",
-            "- Task: continuation",
-            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
-            "- Role: author",
-            "- Identity: prgs-author",
-            "- Issue/PR: #182 / PR #186",
-            "- Branch/SHA: feat/issue-182-controller-handoff",
-            "- Files changed: review_proofs.py",
-            "- Validation: tests passed",
-            "- Mutations: push_branch",
-            "- Workspace mutations: none",
-            "- Current status: PR mergeable",
-            "- Blockers: none",
-            "- Next: review",
-            "- Safety: no review/merge",
-            "- Continuation mode: issue #182 continuation",
-            "- Existing PR: #186",
-            "- PR author: jcwalker3",
-            "- Branch: feat/issue-182-controller-handoff",
-            f"- Old PR head: {self.OLD_SHA}",
-            f"- New PR head: {self.NEW_SHA}",
-            "- Session authored PR: yes",
-            "- Why continuation allowed: operator requested rebase",
-        ])
-        result = assess_issue_selection_final_report(
-            report,
-            mode="continuation",
-            continuation_proof={
-                "pr_number": 186,
-                "pr_author": "jcwalker3",
-                "branch": "feat/issue-182-controller-handoff",
-                "old_head_sha": self.OLD_SHA,
-                "new_head_sha": self.NEW_SHA,
-                "session_authored_pr": True,
-                "continuation_allowed_reason": "operator requested rebase",
-            },
-            edited_pr_numbers=[186],
-            inventoried_pr_numbers=[186],
-        )
-        self.assertEqual(result["grade"], "A")
-
-
 if __name__ == "__main__":
     unittest.main()

tests/test_reviewer_worktree.py

@@ -1,183 +0,0 @@
-"""Tests for reviewer worktree safety proofs (Issue #233)."""
-import sys
-import unittest
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-from reviewer_worktree import (  # noqa: E402
-    assess_author_worktree_continuity,
-    assess_reviewer_git_command_log,
-    assess_reviewer_worktree_proof,
-    files_outside_pr_scope,
-    is_forbidden_reviewer_git_command,
-    is_readonly_reviewer_git_command,
-    parse_dirty_tracked_files,
-)
-
-
-class TestParseDirtyTrackedFiles(unittest.TestCase):
-    def test_ignores_untracked_files(self):
-        porcelain = "?? untracked.txt\n M tracked.py\n"
-        self.assertEqual(parse_dirty_tracked_files(porcelain), ["tracked.py"])
-
-    def test_parses_renamed_paths(self):
-        porcelain = "R  old.py -> new.py\n"
-        self.assertEqual(parse_dirty_tracked_files(porcelain), ["new.py"])
-
-
-class TestFilesOutsidePrScope(unittest.TestCase):
-    def test_all_dirty_in_scope_is_clean(self):
-        self.assertEqual(
-            files_outside_pr_scope(
-                ["docs/wiki/Repositories.md"],
-                ["docs/wiki/Repositories.md"],
-            ),
-            [],
-        )
-
-    def test_unrelated_dirty_files_detected(self):
-        self.assertEqual(
-            files_outside_pr_scope(
-                ["review_proofs.py", "docs/wiki/Repositories.md"],
-                ["docs/wiki/Repositories.md"],
-            ),
-            ["review_proofs.py"],
-        )
-
-
-class TestForbiddenGitCommands(unittest.TestCase):
-    def test_blocks_stash_operations(self):
-        self.assertTrue(is_forbidden_reviewer_git_command("git stash"))
-        self.assertTrue(
-            is_forbidden_reviewer_git_command(
-                'git stash push -m "reviewer-temp-stash" -- tests/test_mcp_server.py'
-            )
-        )
-        self.assertTrue(is_forbidden_reviewer_git_command("git stash pop"))
-        self.assertTrue(is_forbidden_reviewer_git_command("git stash drop"))
-
-    def test_blocks_checkout_reset_and_clean(self):
-        self.assertTrue(
-            is_forbidden_reviewer_git_command(
-                "git checkout -- review_proofs.py tests/test_mcp_server.py"
-            )
-        )
-        self.assertTrue(is_forbidden_reviewer_git_command("git reset --hard"))
-        self.assertTrue(is_forbidden_reviewer_git_command("git clean -fd"))
-
-    def test_blocks_checkout_restore_and_switch_bypasses(self):
-        """Issue #243: blocklist gaps closed via readonly allowlist model."""
-        blocked = (
-            "git checkout HEAD -- review_proofs.py",
-            "git checkout prgs/master -- review_proofs.py",
-            "git checkout .",
-            "git switch -",
-            "git switch -C feat/other-branch",
-            "git switch master",
-            "git stash store",
-            "git stash branch wip-stash",
-        )
-        for cmd in blocked:
-            with self.subTest(cmd=cmd):
-                self.assertTrue(is_forbidden_reviewer_git_command(cmd))
-                self.assertFalse(is_readonly_reviewer_git_command(cmd))
-
-    def test_allows_readonly_commands(self):
-        for cmd in (
-            "git fetch prgs master",
-            "git status --porcelain",
-            "git diff prgs/master...HEAD",
-            "git rev-parse HEAD",
-            "git -C /repo log -1",
-        ):
-            with self.subTest(cmd=cmd):
-                self.assertFalse(is_forbidden_reviewer_git_command(cmd))
-                self.assertTrue(is_readonly_reviewer_git_command(cmd))
-
-
-class TestAssessReviewerWorktreeProof(unittest.TestCase):
-    def test_clean_worktree_proceeds(self):
-        result = assess_reviewer_worktree_proof({
-            "worktree_path": "/repo/branches/review-pr-231",
-            "porcelain_status": "",
-            "pr_scope_files": ["docs/wiki/Repositories.md"],
-            "scratch_used": False,
-            "git_commands": ["git fetch prgs master", "git diff prgs/master...HEAD"],
-        })
-        self.assertTrue(result["proven"])
-        self.assertFalse(result["block"])
-
-    def test_dirty_unrelated_without_scratch_blocks(self):
-        result = assess_reviewer_worktree_proof({
-            "worktree_path": "/repo",
-            "dirty_files": ["review_proofs.py", "tests/test_review_proofs.py"],
-            "pr_scope_files": ["docs/wiki/Repositories.md"],
-            "scratch_used": False,
-        })
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["block"])
-        self.assertIn("review_proofs.py", result["unrelated_dirty_files"][0])
-
-    def test_scratch_worktree_allows_dirty_main_repo(self):
-        result = assess_reviewer_worktree_proof({
-            "worktree_path": "/repo",
-            "dirty_files": ["review_proofs.py"],
-            "pr_scope_files": ["docs/wiki/Repositories.md"],
-            "scratch_used": True,
-            "scratch_path": "/repo/branches/review-feat-issue-224",
-        })
-        self.assertTrue(result["proven"])
-
-    def test_forbidden_command_history_blocks(self):
-        result = assess_reviewer_worktree_proof({
-            "worktree_path": "/repo/branches/review-pr-231",
-            "porcelain_status": "",
-            "git_commands": ["git stash push -m temp -- review_proofs.py"],
-        })
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["forbidden_commands"])
-
-    def test_unrelated_mutations_claimed_blocks(self):
-        result = assess_reviewer_worktree_proof({
-            "worktree_path": "/repo",
-            "porcelain_status": "",
-            "unrelated_mutations_claimed": True,
-        })
-        self.assertFalse(result["proven"])
-
-
-class TestAuthorContinuity(unittest.TestCase):
-    def test_author_may_keep_dirty_worktree(self):
-        result = assess_author_worktree_continuity({
-            "task_role": "author",
-            "dirty_files": ["feat.py"],
-        })
-        self.assertTrue(result["allowed"])
-
-    def test_reviewer_dirty_worktree_uses_reviewer_gate(self):
-        result = assess_author_worktree_continuity({
-            "task_role": "reviewer",
-            "worktree_path": "/repo",
-            "dirty_files": ["other.py"],
-            "pr_scope_files": ["docs/a.md"],
-            "scratch_used": False,
-        })
-        self.assertFalse(result["proven"])
-
-
-class TestAssessReviewerGitCommandLog(unittest.TestCase):
-    def test_empty_log_is_clean(self):
-        result = assess_reviewer_git_command_log([])
-        self.assertTrue(result["proven"])
-
-    def test_mixed_log_blocks_on_forbidden(self):
-        result = assess_reviewer_git_command_log([
-            "git fetch prgs",
-            "git stash",
-        ])
-        self.assertFalse(result["proven"])
-        self.assertEqual(len(result["forbidden_commands"]), 1)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_role_session_router.py

@@ -168,74 +168,6 @@ class TestRoleSessionRouter(unittest.TestCase):
         self.assertFalse(result["success"])
         self.assertIn("switching is disabled", result["message"])
 
-    def _switching_env(self, profile):
-        switching_config = dict(CONFIG)
-        switching_config["rules"] = {"allow_runtime_switching": True}
-        switching_config_path = os.path.join(self._dir.name, "profiles_switching.json")
-        with open(switching_config_path, "w", encoding="utf-8") as fh:
-            fh.write(json.dumps(switching_config))
-        return {
-            "GITEA_MCP_CONFIG": switching_config_path,
-            "GITEA_MCP_PROFILE": profile,
-            "GITEA_TOKEN_AUTHOR": "author-pass",
-            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
-        }
-
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
-    def test_dynamic_switch_to_reviewer_profile(self, _auth, mock_api):
-        mock_api.side_effect = lambda method, url, header: (
-            {"login": "sysadmin"} if "reviewer-pass" in str(header) else {"login": "jcwalker3"}
-        )
-        with patch.dict(os.environ, self._switching_env("prgs-author")):
-            self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
-            route = mcp_server.gitea_route_task_session(
-                task_type="review_pr", remote="prgs"
-            )
-            self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
-            self.assertTrue(route["downstream_allowed"])
-            self.assertEqual(gitea_config.selected_profile_name(), "prgs-reviewer")
-
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value="token author-pass")
-    def test_dynamic_switch_to_author_profile(self, _auth, mock_api):
-        mock_api.side_effect = lambda method, url, header: (
-            {"login": "jcwalker3"} if "author-pass" in str(header) else {"login": "sysadmin"}
-        )
-        with patch.dict(os.environ, self._switching_env("prgs-reviewer")):
-            self.assertEqual(gitea_config.selected_profile_name(), "prgs-reviewer")
-            route = mcp_server.gitea_route_task_session(
-                task_type="create_issue", remote="prgs"
-            )
-            self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
-            self.assertTrue(route["downstream_allowed"])
-            self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
-
-    def test_dynamic_switch_blocked_for_wrong_role(self):
-        with patch.dict(os.environ, self._switching_env("prgs-author")):
-            config_data = dict(CONFIG)
-            config_data["rules"] = {"allow_runtime_switching": True}
-            config_data["profiles"]["prgs-author"]["allowed_operations"] = [
-                "gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push"
-            ]
-            custom_path = os.path.join(self._dir.name, "profiles_wrong_role.json")
-            with open(custom_path, "w", encoding="utf-8") as fh:
-                fh.write(json.dumps(config_data))
-
-            env = {
-                "GITEA_MCP_CONFIG": custom_path,
-                "GITEA_MCP_PROFILE": "prgs-author",
-                "GITEA_TOKEN_AUTHOR": "author-pass",
-                "GITEA_TOKEN_REVIEWER": "reviewer-pass",
-            }
-            with patch.dict(os.environ, env):
-                route = mcp_server.gitea_route_task_session(
-                    task_type="comment_issue", remote="prgs"
-                )
-                self.assertEqual(route["route_result"], role_session_router.ROUTE_TO_AUTHOR)
-                self.assertFalse(route["downstream_allowed"])
-                self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
-
     def test_handoff_requires_route_fields(self):
         incomplete = assess_role_route_handoff("Task done.")
         self.assertFalse(incomplete["complete"])