feat(reviewer): add queue-target lock before PR inventory trust gate (#200)

Operators can supply an active PR backlog for one repo while agents
inventory another and falsely report trusted_empty. reconcile_queue_target
locks the target repo from operator context/backlog before list_prs runs;
pr_inventory_trust_gate blocks target_repo_mismatch and unresolved locks.
assess_reviewer_queue_inventory wires the lock into the canonical path.

Tests cover Gitea-Tools backlog vs mcp-control-plane inventory mismatch,
ambiguous dual-repo context, conflicting directive vs backlog, and final-
report field requirements.

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

capability_stop_terminal.py

@@ -120,6 +120,11 @@ def assess_capability_stop_report(
     capability_denied: bool = True,
 ) -> dict:
     """Validate final report purity after reviewer capability denial."""
+    from review_proofs import (
+        assess_empty_queue_report,
+        parse_trust_gate_status_from_report,
+    )
+
     text = report_text or ""
     lower = text.lower()
     violations = []
@@ -148,13 +153,19 @@ def assess_capability_stop_report(
         r"inventory empty",
         re.I,
     )
+    parsed_status = parse_trust_gate_status_from_report(text)
+    effective_status = trust_gate_status or parsed_status
     if empty_queue_patterns.search(text):
-        if trust_gate_status != "trusted_empty":
+        if effective_status != "trusted_empty":
             violations.append(
                 "empty-queue claim after capability stop without "
                 "pr_inventory_trust_gate.status == trusted_empty"
             )
 
+    empty_queue = assess_empty_queue_report(text)
+    if empty_queue.get("claimed") and not empty_queue.get("proven"):
+        violations.extend(empty_queue.get("reasons") or [])
+
     ok, elig_violations = validate_eligibility_wording(text)
     violations.extend(elig_violations)
 

docs/wiki/Repositories.md

@@ -31,7 +31,8 @@ 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 `11549ee` |
-| `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) |
+| `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` |
+
 
 Update this table whenever a wiki is published, re-synced, or found stale.

gitea_mcp_server.py

@@ -259,6 +259,7 @@ 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
 
 
 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
@@ -2274,6 +2275,23 @@ 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,
@@ -2341,6 +2359,8 @@ 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
@@ -2425,7 +2445,28 @@ def gitea_review_pr(
             if inventory_msg:
                 report_lines.append(inventory_msg)
             else:
-                report_lines.append("Open PRs found: 0")
+                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"
+                    )
     else:
         report_lines.append(inventory_msg)
 

review_proofs.py

@@ -17,6 +17,7 @@ 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}$")
 
@@ -68,6 +69,251 @@ 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"
@@ -637,7 +883,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):
+                       sweep_proof=None, worktree_proof=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
 
     Combines the individual proof verdicts into the final-report fields the
@@ -707,6 +953,14 @@ 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"))
@@ -719,6 +973,7 @@ 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:
@@ -772,6 +1027,11 @@ 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", []))
 
     merge_allowed = (
         identity_eligible
@@ -782,6 +1042,7 @@ 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 = []
@@ -825,6 +1086,11 @@ 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")
+        ),
     }
 
 
@@ -971,6 +1237,12 @@ 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")),
@@ -1207,11 +1479,31 @@ 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",
@@ -1281,9 +1573,120 @@ 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
+
+
 def assess_duplicate_search_proof(report_text, matches):
     """#207: reject LLM duplicate summaries that omit known title matches."""
     return issue_duplicate_gate.assess_duplicate_search_proof(

reviewer_worktree.py

@@ -0,0 +1,219 @@
+"""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``).
+"""
+
+from __future__ import annotations
+
+import re
+import shlex
+
+# Subcommands that mutate unrelated local state — forbidden for reviewers.
+_FORBIDDEN_REVIEWER_GIT = re.compile(
+    r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
+    r"(?:stash(?:\s+(?:push|pop|drop|apply|clear|list))?|"
+    r"checkout\s+--|"
+    r"restore\s+|"
+    r"reset(?:\s+(?:--hard|--soft|--mixed|--merge))?|"
+    r"clean(?:\s+(?:-f|-fd|-fdx|-x|-d|-n))*|"
+    r"cherry-pick|"
+    r"rebase|"
+    r"merge|"
+    r"commit(?:\s+(?:--amend|-a|-am))?|"
+    r"push(?:\s+(?:--force|--force-with-lease))?|"
+    r"branch\s+-[dD]|"
+    r"worktree\s+remove)",
+    re.IGNORECASE,
+)
+
+# Read-only git operations reviewers may use for validation.
+_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,
+)
+
+
+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_forbidden_reviewer_git_command(command: str) -> bool:
+    """True when a shell command would mutate unrelated local/remote git state."""
+    text = (command or "").strip()
+    if not text:
+        return False
+    return bool(_FORBIDDEN_REVIEWER_GIT.search(text))
+
+
+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:
+        return False
+    if is_forbidden_reviewer_git_command(text):
+        return False
+    return bool(_READONLY_REVIEWER_GIT.search(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

@@ -216,7 +216,10 @@ 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`).
+   `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.
 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

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

@@ -17,9 +17,30 @@ 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.
+  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_capability_stop_terminal.py

@@ -148,6 +148,19 @@ 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_pr_queue_inventory.py

@@ -261,12 +261,79 @@ 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_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
+    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):
         """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

@@ -26,6 +26,8 @@ from review_proofs import (  # noqa: E402
     assess_capability_proof,
     assess_controller_handoff,
     assess_inventory_completeness,
+    assess_queue_target_final_report,
+    assess_reviewer_queue_inventory,
     assess_live_state_recheck,
     assess_review_mutation_final_report,
     assess_role_boundary,
@@ -35,6 +37,7 @@ from review_proofs import (  # noqa: E402
     assess_validation_report,
     build_final_report,
     pr_inventory_trust_gate,
+    reconcile_queue_target,
     resolve_repos_from_user_reference,
     verify_pinned_head_checkout,
 )
@@ -183,6 +186,24 @@ 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",
@@ -595,6 +616,13 @@ 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)
@@ -899,12 +927,17 @@ 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",
@@ -1081,6 +1114,144 @@ 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."""
 
@@ -1163,6 +1334,74 @@ 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 TestCapabilityEvidence(unittest.TestCase):
     """#179 gap 1: capability claims need exact evidence."""
 
@@ -1306,6 +1545,13 @@ 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)

tests/test_reviewer_worktree.py

@@ -0,0 +1,166 @@
+"""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_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()