Merge pull request 'feat(continuation): formalize continuation proofs for claim and push (closes #189)' (#253) from feat/issue-189-continuation-mode-proofs into feat/issue-188-continuation-selection-wall

Extend #188 continuation mode with issue claim status evidence,
force-with-lease push reporting, canonical secret sweep helper, and
composite final-report checks wired into assess_issue_selection_final_report.

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

@@ -32,6 +32,7 @@ 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) |
+| `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}$")
 
@@ -637,7 +638,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
@@ -663,6 +664,8 @@ 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"))
@@ -707,6 +710,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 +730,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 +784,16 @@ 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
@@ -782,6 +804,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 +848,17 @@ 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"),
     }
 
 
@@ -971,6 +1005,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")),
@@ -978,6 +1018,7 @@ 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")),
@@ -988,13 +1029,40 @@ 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")),
+        ("Issue claim status", ("issue claim", "claim status",
+                                "status:in-progress")),
+        ("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")),
+    ),
 }
 
+# Canonical secret/provenance sweep for comparable continuation runs (#189).
+CANONICAL_SECRET_SWEEP_COMMAND = (
+    "git diff prgs/master...HEAD | rg -i "
+    "'(token|password|secret|api[_-]?key|authorization:)'"
+)
+CANONICAL_SECRET_SWEEP_SCOPE = "full feature-branch diff against prgs/master"
+
 
 def _handoff_section_lines(report_text):
     """Return the lines of the Controller Handoff section, or None."""
@@ -1284,6 +1352,616 @@ def pr_inventory_trust_gate(
     }
 
 
+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,
+) -> 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] = {}
+    blockers: list[str] = []
+    can_claim_empty = bool(completeness.get("complete"))
+
+    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 = []
+
+        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,
+        )
+        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,
+        "blockers": blockers,
+        "reasons": list(completeness.get("reasons") or []) + blockers,
+    }
+
+
+def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]:
+    """Render trust-gate lines for MCP inventory output."""
+    lines = [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 canonical_secret_sweep_report(*, clean: bool) -> dict:
+    """Return a sweep report using the canonical command/scope (#189)."""
+    return {
+        "method": CANONICAL_SECRET_SWEEP_COMMAND,
+        "scope": CANONICAL_SECRET_SWEEP_SCOPE,
+        "clean": clean,
+    }
+
+
+def assess_force_with_lease_push_report(
+    report_text: str,
+    *,
+    old_remote_head: str | None = None,
+    expected_lease_head: str | None = None,
+    new_pushed_head: str | None = None,
+    branch_pushed: str | None = None,
+    used_force_with_lease: bool | None = None,
+) -> dict:
+    """Issue #189: force-with-lease pushes must disclose lease evidence."""
+    if not used_force_with_lease:
+        return {"complete": True, "downgraded": False, "reasons": []}
+
+    text = report_text or ""
+    lower = text.lower()
+    reasons = []
+
+    if "force-with-lease" not in lower and "force with lease" not in lower:
+        reasons.append(
+            "continuation push used force-with-lease but report does not say so"
+        )
+
+    for label, sha in (
+        ("old remote", old_remote_head),
+        ("lease", expected_lease_head),
+        ("pushed", new_pushed_head),
+    ):
+        if not sha:
+            reasons.append(f"force-with-lease proof missing {label} head SHA")
+        elif not _FULL_SHA.match(sha.lower()):
+            reasons.append(
+                f"force-with-lease {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 branch_pushed:
+        if branch_pushed.lower() not in lower:
+            reasons.append("continuation report missing pushed branch name")
+        only_branch_tokens = (
+            "only feature branch",
+            "push branch only",
+            "only the feature branch",
+            "only pushed feature branch",
+        )
+        if not any(t in lower for t in only_branch_tokens):
+            reasons.append(
+                "continuation report missing confirmation that only the "
+                "feature branch was pushed"
+            )
+
+    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,
+    issue_number: int | None = None,
+    issue_claim_status: 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/#189: 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 issue_number is not None:
+        if (
+            f"#{issue_number}" not in lower
+            and f"issue #{issue_number}" not in lower
+            and f"issue {issue_number}" not in lower
+        ):
+            reasons.append(f"continuation report missing issue #{issue_number}")
+    if issue_claim_status:
+        claim_lower = issue_claim_status.lower()
+        claim_tokens = (
+            claim_lower,
+            "claim status",
+            "issue claim",
+            "status:in-progress",
+        )
+        if not any(t in lower for t in claim_tokens):
+            reasons.append("continuation report missing issue claim status")
+    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"),
+            issue_number=proof.get("issue_number"),
+            issue_claim_status=proof.get("issue_claim_status"),
+            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"),
+        )
+        push_proof = proof.get("push_proof") or {}
+        checks["force_with_lease_push"] = assess_force_with_lease_push_report(
+            report_text,
+            old_remote_head=push_proof.get("old_remote_head"),
+            expected_lease_head=push_proof.get("expected_lease_head"),
+            new_pushed_head=push_proof.get("new_pushed_head"),
+            branch_pushed=push_proof.get("branch_pushed"),
+            used_force_with_lease=push_proof.get("used_force_with_lease"),
+        )
+        sweep = proof.get("secret_sweep")
+        if sweep is not None:
+            checks["secret_sweep"] = assess_secret_sweep(sweep)
+
+    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("proven") is False:
+            downgraded = True
+            reasons.extend(
+                f"{name}: {r}" for r in (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,
+    }
+
+
 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

@@ -379,6 +379,17 @@ 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/#189): `Continuation mode:`, `Existing PR:`,
+  `PR author:`, `Issue claim status:`, `Branch:`, `Old PR head:`,
+  `New PR head:`, `Session authored PR:`, `Why continuation allowed:` —
+  when rebasing with `git push --force-with-lease`, also record
+  `Old remote head:`, `Lease head:`, `Pushed head:`, and confirm
+  `Push branch only:` (feature branch only). Use the canonical secret sweep
+  from `review_proofs.CANONICAL_SECRET_SWEEP_COMMAND` so runs are
+  comparable. Issues with open PRs are excluded from fresh selection unless
+  the 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,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_mcp_server.py

@@ -65,6 +65,20 @@ 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
@@ -127,15 +141,19 @@ 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):
-        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
+        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
+        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         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, []))
@@ -144,13 +162,28 @@ 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):
-        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
+        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
+        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         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
@@ -2770,8 +2803,8 @@ class TestIssueLocking(unittest.TestCase):
     """Test issue locking and PR gating constraints."""
 
     def tearDown(self):
-        if os.path.exists("/tmp/gitea_issue_lock.json"):
-            os.remove("/tmp/gitea_issue_lock.json")
+        if os.path.exists(ISSUE_LOCK_FILE):
+            os.remove(ISSUE_LOCK_FILE)
 
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -2779,7 +2812,7 @@ class TestIssueLocking(unittest.TestCase):
         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("/tmp/gitea_issue_lock.json"))
+        self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
 
     def test_lock_issue_mismatch_branch_fails(self):
         with self.assertRaises(ValueError) as ctx:
@@ -2816,8 +2849,8 @@ class TestIssueLocking(unittest.TestCase):
            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("/tmp/gitea_issue_lock.json"):
-            os.remove("/tmp/gitea_issue_lock.json")
+        if os.path.exists(ISSUE_LOCK_FILE):
+            os.remove(ISSUE_LOCK_FILE)
         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")
@@ -2827,8 +2860,9 @@ 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("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        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 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")
@@ -2838,8 +2872,9 @@ 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("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        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 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:
@@ -2850,8 +2885,9 @@ 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("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        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 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")

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

@@ -21,11 +21,23 @@ 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_force_with_lease_push_report,
+    canonical_secret_sweep_report,
+    CANONICAL_SECRET_SWEEP_COMMAND,
     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_reviewer_queue_inventory,
     assess_live_state_recheck,
     assess_review_mutation_final_report,
     assess_role_boundary,
@@ -34,6 +46,7 @@ from review_proofs import (  # noqa: E402
     assess_sweep_evidence,
     assess_validation_report,
     build_final_report,
+    classify_issue_for_selection,
     pr_inventory_trust_gate,
     resolve_repos_from_user_reference,
     verify_pinned_head_checkout,
@@ -183,6 +196,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 +626,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 +937,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",
@@ -913,24 +956,46 @@ class TestControllerHandoff(unittest.TestCase):
         result = assess_controller_handoff(complete, role="review")
         self.assertEqual(result["verdict"], "complete")
 
-    def test_author_role_requires_author_fields(self):
-        complete = self.BASE_HANDOFF + "\n" + "\n".join([
-            "- Selected issue: #182",
+    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",
-            "- PR number opened: #999",
+            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([
+            "- Selected issue: #182",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #999",
+            "- No review/merge: confirmed",
+        ])
+        result = assess_controller_handoff(without_lock, role="author")
+        self.assertEqual(result["verdict"], "incomplete")
+        self.assertIn("Issue lock proof", 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",
@@ -942,6 +1007,7 @@ 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",
@@ -953,6 +1019,7 @@ 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",
@@ -965,6 +1032,10 @@ 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",
         ])
@@ -984,23 +1055,17 @@ 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([
-            "- Selected issue: #196",
-            "- Claim/comment status: comment-claimed",
-            "- PR number opened: #203",
-            "- No review/merge: confirmed",
-        ])
+        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
+            self._author_role_fields(issue_number=196, pr_number=203))
         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([
-            "- Selected issue: #196",
-            "- Claim/comment status: comment-claimed",
-            "- PR number opened: #203",
-            "- No review/merge: confirmed",
-        ])
+        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))
         res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
         self.assertEqual(res2["verdict"], "complete")
 
@@ -1163,6 +1228,157 @@ 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."""
 
@@ -1306,6 +1522,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)
@@ -1450,5 +1673,254 @@ 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",
+            "- Issue claim status: status:in-progress",
+            "- 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",
+                "issue_number": 182,
+                "issue_claim_status": "status:in-progress",
+                "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")
+
+
+class TestContinuationModeProofs(unittest.TestCase):
+    """Issue #189: formal continuation proofs beyond #188 selection wall."""
+
+    OLD_SHA = PINNED
+    NEW_SHA = OTHER
+    REMOTE_OLD = "601c608c00000000000000000000000000000000"
+    LEASE_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
+    PUSHED_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
+
+    def test_force_with_lease_requires_lease_evidence(self):
+        report = (
+            "Issue #182 continuation. force-with-lease push to "
+            "feat/issue-182-controller-handoff-enforcement."
+        )
+        result = assess_force_with_lease_push_report(
+            report,
+            old_remote_head=self.REMOTE_OLD,
+            expected_lease_head=self.LEASE_SHA,
+            new_pushed_head=self.PUSHED_SHA,
+            branch_pushed="feat/issue-182-controller-handoff-enforcement",
+            used_force_with_lease=True,
+        )
+        self.assertTrue(result["downgraded"])
+
+    def test_force_with_lease_complete_with_full_evidence(self):
+        report = (
+            "Issue #182 continuation with git push --force-with-lease. "
+            f"Old remote head {self.REMOTE_OLD}. "
+            f"Lease head {self.LEASE_SHA}. "
+            f"Pushed head {self.PUSHED_SHA}. "
+            "Branch feat/issue-182-controller-handoff-enforcement. "
+            "Push branch only: only the feature branch was pushed."
+        )
+        result = assess_force_with_lease_push_report(
+            report,
+            old_remote_head=self.REMOTE_OLD,
+            expected_lease_head=self.LEASE_SHA,
+            new_pushed_head=self.PUSHED_SHA,
+            branch_pushed="feat/issue-182-controller-handoff-enforcement",
+            used_force_with_lease=True,
+        )
+        self.assertTrue(result["complete"])
+
+    def test_continuation_requires_issue_claim_status(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-test."
+        )
+        result = assess_continuation_mode_report(
+            report,
+            pr_number=186,
+            pr_author="jcwalker3",
+            issue_number=182,
+            issue_claim_status="status:in-progress",
+            branch="feat/issue-182-test",
+            old_head_sha=self.OLD_SHA,
+            new_head_sha=self.NEW_SHA,
+        )
+        self.assertTrue(result["downgraded"])
+
+    def test_canonical_secret_sweep_helper(self):
+        sweep = canonical_secret_sweep_report(clean=True)
+        self.assertEqual(sweep["method"], CANONICAL_SECRET_SWEEP_COMMAND)
+        self.assertTrue(sweep["clean"])
+
+    def test_continuation_final_report_with_push_and_sweep_earns_a(self):
+        branch = "feat/issue-182-controller-handoff-enforcement"
+        report = "\n".join([
+            "Issue #182 continuation mode — status:in-progress already set.",
+            f"PR #186 updated via git push --force-with-lease.",
+            f"Old PR head {self.OLD_SHA}; new PR head {self.NEW_SHA}.",
+            f"Old remote head {self.REMOTE_OLD}.",
+            f"Lease head {self.LEASE_SHA}. Pushed head {self.PUSHED_SHA}.",
+            f"Branch {branch}. Push branch only: only the feature branch.",
+            "PR author: jcwalker3. 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",
+            f"- Branch/SHA: {branch}",
+            "- 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",
+            "- Issue claim status: status:in-progress",
+            f"- Branch: {branch}",
+            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",
+                "issue_number": 182,
+                "issue_claim_status": "status:in-progress",
+                "branch": branch,
+                "old_head_sha": self.OLD_SHA,
+                "new_head_sha": self.NEW_SHA,
+                "session_authored_pr": True,
+                "continuation_allowed_reason": "operator requested rebase",
+                "push_proof": {
+                    "old_remote_head": self.REMOTE_OLD,
+                    "expected_lease_head": self.LEASE_SHA,
+                    "new_pushed_head": self.PUSHED_SHA,
+                    "branch_pushed": branch,
+                    "used_force_with_lease": True,
+                },
+                "secret_sweep": canonical_secret_sweep_report(clean=True),
+            },
+            edited_pr_numbers=[186],
+            inventoried_pr_numbers=[186],
+        )
+        self.assertEqual(result["grade"], "A")
+
+
 if __name__ == "__main__":
     unittest.main()

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