feat: enforce capability stop terminal mode after reviewer denial (Issue #197)

Enter terminal mode when review_pr/merge_pr capability is denied. Block
reviewer queue tools (list_prs, eligibility checks) and add report-purity
validators for forbidden PR selection, fallback, and empty-queue claims.

Closes #197

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

author_proofs.py

@@ -0,0 +1,217 @@
+"""Fail-closed branch-identity proofs for author workflows (#177).
+
+Author-side counterpart of the reviewer proofs in ``review_proofs.py``
+(#173). During the #173 implementation itself, a commit landed on local
+``master`` because the shared checkout's branch moved mid-session (origin
+incident of #177). These helpers turn that from an after-the-fact repair
+into a fail-closed gate: an author workflow must prove its local git state
+before staging, committing, or pushing.
+
+The helpers are pure (no git calls): the workflow gathers the raw facts
+(``git branch --show-current``, ``git rev-parse HEAD``, the push refspec,
+the branch named in the issue claim) and passes them in, so the same logic
+works from prompts, harness assertions, and tests. Shared-worktree branch
+switches by other sessions are treated as expected events to detect, not
+exceptional ones. Nothing here weakens the review/merge/permission gates.
+"""
+
+PROTECTED_BRANCHES = frozenset(
+    {"master", "main", "develop", "development", "dev"}
+)
+
+
+def _clean(name):
+    return (name or "").strip()
+
+
+def verify_branch_for_commit(current_branch, intended_branch):
+    """Required behavior 1: prove the branch before staging/committing.
+
+    Proven only when both names are present, the intended branch is not a
+    protected branch, and the current branch equals the intended one (which
+    also rules out being on any protected branch). Returns {'proven',
+    'block', 'reasons', 'current_branch', 'intended_branch'}.
+    """
+    reasons = []
+    current = _clean(current_branch)
+    intended = _clean(intended_branch)
+
+    if not current:
+        reasons.append(
+            "current branch unknown (detached HEAD or state not read); "
+            "fail closed"
+        )
+    if not intended:
+        reasons.append("intended feature branch not stated; fail closed")
+    if intended and intended in PROTECTED_BRANCHES:
+        reasons.append(
+            f"intended branch '{intended}' is a protected branch; author "
+            "work must target a feature branch"
+        )
+    if current and current in PROTECTED_BRANCHES:
+        reasons.append(
+            f"current branch '{current}' is a protected branch; committing "
+            "here is blocked"
+        )
+    if current and intended and current != intended:
+        reasons.append(
+            f"current branch '{current}' is not the intended feature branch "
+            f"'{intended}'; stop before staging/committing"
+        )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "current_branch": current or None,
+        "intended_branch": intended or None,
+    }
+
+
+def detect_branch_drift(branch_at_validation, head_at_validation,
+                        current_branch, current_head):
+    """Required behaviors 2–3: stop when branch or HEAD moved mid-session.
+
+    Compares the branch name and HEAD SHA captured at validation time with
+    the state observed immediately before commit/push. Any difference —
+    including an external branch switch in a shared worktree — is drift and
+    blocks until reconciled. Missing state fails closed.
+    """
+    reasons = []
+    branch_then = _clean(branch_at_validation)
+    branch_now = _clean(current_branch)
+    head_then = _clean(head_at_validation).lower()
+    head_now = _clean(current_head).lower()
+
+    if not branch_then or not head_then:
+        reasons.append("validation-time branch/HEAD not recorded; fail closed")
+    if not branch_now or not head_now:
+        reasons.append("current branch/HEAD not read; fail closed")
+
+    if branch_then and branch_now and branch_then != branch_now:
+        reasons.append(
+            f"branch changed from '{branch_then}' to '{branch_now}' since "
+            "validation — possible external branch switch in a shared "
+            "worktree; stop and reconcile before committing"
+        )
+    if head_then and head_now and head_then != head_now:
+        reasons.append(
+            "HEAD moved since validation; re-validate on the current HEAD "
+            "before committing"
+        )
+
+    drifted = bool(reasons)
+    return {"drifted": drifted, "block": drifted, "reasons": reasons}
+
+
+def verify_push_target(current_branch, remote_target_branch, intended_branch):
+    """Acceptance: a push needs local, remote, and intended branches to match.
+
+    Proven only when all three names are present, equal, and not a
+    protected branch — a feature-branch workflow never pushes a protected
+    branch, and never pushes to a refspec other than its own branch.
+    """
+    reasons = []
+    current = _clean(current_branch)
+    remote_target = _clean(remote_target_branch)
+    intended = _clean(intended_branch)
+
+    if not current:
+        reasons.append("current branch unknown; fail closed")
+    if not remote_target:
+        reasons.append("remote target branch not stated; fail closed")
+    if not intended:
+        reasons.append("intended feature branch not stated; fail closed")
+
+    for label, name in (("current", current), ("remote target", remote_target),
+                        ("intended", intended)):
+        if name and name in PROTECTED_BRANCHES:
+            reasons.append(
+                f"{label} branch '{name}' is a protected branch; author "
+                "pushes to protected branches are blocked"
+            )
+
+    if current and remote_target and current != remote_target:
+        reasons.append(
+            f"push target '{remote_target}' does not match the local branch "
+            f"'{current}'"
+        )
+    if current and intended and current != intended:
+        reasons.append(
+            f"local branch '{current}' does not match the intended feature "
+            f"branch '{intended}'"
+        )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+    }
+
+
+def assess_protected_branch_commit(commit_branch, pushed=False,
+                                   repair_reported=True):
+    """Required behavior 4: handle an accidental protected-branch commit.
+
+    If a commit landed on a protected branch: it must never be pushed, a
+    repair is required, and the repair must be *reported* — silently
+    continuing after (or without) repair is a violation, as is having
+    pushed the accident.
+    """
+    branch = _clean(commit_branch)
+    accident = branch in PROTECTED_BRANCHES
+
+    violations = []
+    if accident:
+        if pushed:
+            violations.append(
+                f"accidental commit on protected branch '{branch}' was "
+                "pushed; protected-branch pushes are forbidden"
+            )
+        if not repair_reported:
+            violations.append(
+                "protected-branch commit repair was not reported; the "
+                "workflow must surface the accident and the repair steps, "
+                "never silently continue"
+            )
+
+    return {
+        "accident": accident,
+        "must_not_push": accident,
+        "repair_required": accident,
+        "violations": violations,
+    }
+
+
+def build_commit_push_report(commit_proof, drift, push_proof, accident=None):
+    """Acceptance: final report carries branch proof before commit and push.
+
+    Combines the individual proofs; any failed proof, detected drift, or
+    accident violation makes the status 'blocked' — the workflow stops and
+    reports instead of continuing.
+    """
+    accident = accident or {"accident": False, "violations": []}
+    violations = list(accident.get("violations", []))
+
+    blocked = (
+        not commit_proof.get("proven")
+        or drift.get("drifted")
+        or not push_proof.get("proven")
+        or bool(violations)
+    )
+
+    return {
+        "status": "blocked" if blocked else "ok",
+        "branch_proof_before_commit": bool(commit_proof.get("proven")),
+        "branch_proof_before_push": bool(push_proof.get("proven")),
+        "drift_detected": bool(drift.get("drifted")),
+        "protected_branch_accident": bool(accident.get("accident")),
+        "violations": violations,
+        "reasons": (
+            list(commit_proof.get("reasons", []))
+            + list(drift.get("reasons", []))
+            + list(push_proof.get("reasons", []))
+        ),
+    }

capability_stop_terminal.py

@@ -0,0 +1,202 @@
+"""Hard-stop terminal mode after reviewer capability denial (#197)."""
+
+from __future__ import annotations
+
+import re
+
+TERMINAL_REPORT_HEADING = (
+    "Cannot perform reviewer task under current profile. "
+    "No reviewer mutations performed."
+)
+
+REVIEWER_CAPABILITY_TASKS = frozenset({
+    "review_pr",
+    "merge_pr",
+    "blind_pr_queue_review",
+    "request_changes_pr",
+    "approve_pr",
+})
+
+BLOCKED_QUEUE_TOOLS = frozenset({
+    "list_prs",
+    "check_pr_eligibility",
+    "view_pr",
+    "submit_pr_review",
+    "dry_run_pr_review",
+    "merge_pr",
+    "review_pr",
+})
+
+_session_terminal: dict | None = None
+
+
+def enter_from_capability_result(capability: dict) -> dict | None:
+    """Enter terminal mode when a reviewer/merge task is denied."""
+    global _session_terminal
+    task = (capability or {}).get("requested_task", "")
+    required_role = (capability or {}).get("required_role_kind")
+    if not capability.get("stop_required"):
+        return None
+    if required_role != "reviewer" and task not in REVIEWER_CAPABILITY_TASKS:
+        return None
+    record = {
+        "active": True,
+        "requested_task": task,
+        "required_role_kind": required_role,
+        "active_profile": capability.get("active_profile"),
+        "active_identity": capability.get("active_identity"),
+        "stop_required": True,
+        "exact_safe_next_action": capability.get("exact_safe_next_action"),
+        "terminal_message": TERMINAL_REPORT_HEADING,
+    }
+    _session_terminal = record
+    return dict(record)
+
+
+def enter_from_route_result(route: dict) -> dict | None:
+    """Enter terminal mode from a role router wrong_role_stop (#206 compat)."""
+    if (route or {}).get("route_result") != "wrong_role_stop":
+        return None
+    if route.get("required_role") != "reviewer":
+        return None
+    return enter_from_capability_result({
+        "requested_task": route.get("task_type"),
+        "required_role_kind": "reviewer",
+        "stop_required": True,
+        "active_profile": route.get("active_profile"),
+        "active_identity": None,
+        "exact_safe_next_action": route.get("message"),
+    })
+
+
+def is_active() -> bool:
+    return bool(_session_terminal and _session_terminal.get("active"))
+
+
+def active_record() -> dict | None:
+    if not is_active():
+        return None
+    return dict(_session_terminal)
+
+
+def clear():
+    global _session_terminal
+    _session_terminal = None
+
+
+def check_reviewer_queue_tool(tool_name: str) -> tuple[bool, list[str]]:
+    """Return (allowed, reasons). False when terminal mode blocks queue work."""
+    if not is_active():
+        return True, []
+    name = (tool_name or "").strip().lower().removeprefix("gitea_")
+    if name in BLOCKED_QUEUE_TOOLS:
+        return False, [
+            TERMINAL_REPORT_HEADING,
+            f"Reviewer queue tool '{tool_name}' is blocked after "
+            "capability denial (fail closed).",
+            "Relaunch a reviewer MCP namespace to perform reviewer work.",
+        ]
+    return True, []
+
+
+def validate_eligibility_wording(text: str) -> tuple[bool, list[str]]:
+    """Reject session-based eligibility reasoning (#197)."""
+    lower = (text or "").lower()
+    violations = []
+    if "not authored by this session" in lower:
+        violations.append(
+            "eligibility must use authenticated account identity, not "
+            "'this session' wording"
+        )
+    if re.search(r"not (?:self-)?authored by (?:the )?session", lower):
+        violations.append("session-based eligibility reasoning is invalid")
+    return (len(violations) == 0), violations
+
+
+def assess_capability_stop_report(
+    report_text: str,
+    *,
+    trust_gate_status: str | None = None,
+    capability_denied: bool = True,
+) -> dict:
+    """Validate final report purity after reviewer capability denial."""
+    text = report_text or ""
+    lower = text.lower()
+    violations = []
+
+    if capability_denied and TERMINAL_REPORT_HEADING.lower() not in lower:
+        violations.append("missing required terminal report heading")
+
+    forbidden_patterns = [
+        ("pr selection", re.compile(
+            r"selected pr|pr #\d+ (?:to review|selected)|eligible pr|"
+            r"next pr to review", re.I)),
+        ("sibling repo inventory", re.compile(
+            r"sibling repo|other repo|mcp-control-plane|gitea-tools and", re.I)),
+        ("author fallback", re.compile(
+            r"rebase conflicted|author-side fallback|have me rebase|"
+            r"implement the fix|push a branch|open a pr for", re.I)),
+        ("invalid session eligibility", re.compile(
+            r"not authored by this session", re.I)),
+    ]
+    for label, pattern in forbidden_patterns:
+        if pattern.search(text):
+            violations.append(f"forbidden after hard stop: {label}")
+
+    empty_queue_patterns = re.compile(
+        r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
+        r"inventory empty",
+        re.I,
+    )
+    if empty_queue_patterns.search(text):
+        if trust_gate_status != "trusted_empty":
+            violations.append(
+                "empty-queue claim after capability stop without "
+                "pr_inventory_trust_gate.status == trusted_empty"
+            )
+
+    ok, elig_violations = validate_eligibility_wording(text)
+    violations.extend(elig_violations)
+
+    if violations:
+        return {
+            "pure": False,
+            "downgraded": True,
+            "violations": violations,
+            "reasons": violations,
+        }
+    return {
+        "pure": True,
+        "downgraded": False,
+        "violations": [],
+        "reasons": [],
+    }
+
+
+def build_terminal_report(capability: dict) -> dict:
+    """Minimal allowed report fields after hard stop."""
+    return {
+        "terminal_mode": True,
+        "heading": TERMINAL_REPORT_HEADING,
+        "authenticated_profile": capability.get("active_profile"),
+        "authenticated_identity": capability.get("active_identity"),
+        "denied_task": capability.get("requested_task"),
+        "required_role_kind": capability.get("required_role_kind"),
+        "stop_required": capability.get("stop_required"),
+        "required_action": capability.get("exact_safe_next_action"),
+        "mutations_performed": False,
+        "allowed_sections": [
+            "authenticated identity/profile",
+            "denied capability result",
+            "reason task cannot proceed",
+            "required reviewer profile/identity",
+            "mutation confirmation (none)",
+        ],
+        "forbidden_sections": [
+            "PR selection",
+            "sibling-repo queue recommendations",
+            "author-side fallback suggestions",
+            "empty-queue claims without trusted_empty",
+            "session-based eligibility wording",
+        ],
+    }

mcp_server.py

@@ -45,6 +45,7 @@ from gitea_auth import (  # noqa: E402
 )
 import gitea_audit  # noqa: E402
 import gitea_config  # noqa: E402
+import capability_stop_terminal  # noqa: E402
 
 
 def _reveal_endpoints() -> bool:
@@ -406,6 +407,11 @@ def gitea_list_prs(
         'mergeable', 'updated_at' ('url' only with the reveal opt-in).
         The additional 'updated_at' aids stale/conflicting queue detection.
     """
+    allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool(
+        "list_prs"
+    )
+    if not allowed:
+        raise RuntimeError("; ".join(block_reasons))
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls?state={state}"
@@ -517,6 +523,17 @@ def gitea_check_pr_eligibility(
         'permission_report' (#142).
     """
     action = (action or "").strip().lower()
+    if action in ("review", "approve", "request_changes", "merge"):
+        allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool(
+            "check_pr_eligibility"
+        )
+        if not allowed:
+            return {
+                "eligible": False,
+                "requested_action": action,
+                "reasons": block_reasons,
+                "terminal_mode": True,
+            }
     profile = get_profile()
     result = {
         "eligible": False,
@@ -2307,26 +2324,26 @@ _PROJECT_SKILLS = {
             "committed.",
         ],
     },
-    "jenkins-readonly": {
+    "jenkins-mcp": {
         "description": "Read-only Jenkins CI inspection (jobs, builds, "
-                       "logs). Actual server name: jenkins-mcp (see mcp-control-plane).",
+                       "logs).",
         "when_to_use": "Checking CI state once Jenkins MCP tools exist.",
         "required_operations": ["jenkins.read"],
         "status": "designed-not-implemented",
-        "notes": "Server code exists in mcp-control-plane as jenkins-mcp (read tools + gated trigger); registration pending (#55); docs in Gitea-Tools use historical name. Report SKIPPED if not connected. Do not substitute shell/API. Trigger requires dedicated profile (see #56).",
+        "notes": "Server code exists in mcp-control-plane as jenkins-mcp (read tools + gated trigger); registration pending. To register for discoverability in clients (Codex/Gemini/Grok/etc.): add to client MCP config under the jenkins-mcp name, reconnect/reload the client session after registration. Report SKIPPED if not connected. Do not substitute shell/API. Trigger requires dedicated profile (see #56).",
         "steps": [
             "Confirm a Jenkins MCP server is connected (jenkins-mcp); if not, report "
             "SKIPPED.",
             "Use read-only operations only; never trigger unless using dedicated profile + confirmation.",
         ],
     },
-    "glitchtip-readonly": {
+    "glitchtip-mcp": {
-        "description": "Read-only GlitchTip error/event inspection. Actual server name: glitchtip-mcp (see mcp-control-plane).",
+        "description": "Read-only GlitchTip error/event inspection.",
         "when_to_use": "Investigating reported errors once GlitchTip MCP "
                        "tools exist.",
         "required_operations": ["glitchtip.read"],
         "status": "designed-not-implemented",
-        "notes": "Server code exists in mcp-control-plane as glitchtip-mcp (read-only tools); registration pending (#55); filing orchestrator is partial in mcp-control-plane (see #57). Report SKIPPED if not connected. Filing to Gitea is separate orchestrator, not in this server.",
+        "notes": "Server code exists in mcp-control-plane as glitchtip-mcp (read-only tools); registration pending. To register for discoverability in clients (Codex/Gemini/Grok/etc.): add to client MCP config under the glitchtip-mcp name, reconnect/reload the client session after registration. Filing orchestrator is partial in mcp-control-plane (see #57). Report SKIPPED if not connected. Filing to Gitea is separate orchestrator, not in this server.",
         "steps": [
             "Confirm a GlitchTip MCP server is connected (glitchtip-mcp); if not, report "
             "SKIPPED.",
@@ -3476,7 +3493,7 @@ def gitea_resolve_task_capability(
             "STOP: the active profile cannot perform the requested task; "
             "follow exact_safe_next_action instead of improvising.")
 
-    return {
+    result = {
         "requested_task": task,
         "required_operation_permission": required_permission,
         "required_role_kind": required_role,
@@ -3491,6 +3508,33 @@ def gitea_resolve_task_capability(
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
     }
+    if stop_required:
+        terminal = capability_stop_terminal.enter_from_capability_result(result)
+        if terminal:
+            result["terminal_mode"] = True
+            result["terminal_report"] = (
+                capability_stop_terminal.build_terminal_report(result)
+            )
+    return result
+
+
+@mcp.tool()
+def gitea_capability_stop_terminal_report() -> dict:
+    """Read-only: terminal report template after reviewer capability denial (#197)."""
+    record = capability_stop_terminal.active_record()
+    if not record:
+        return {
+            "terminal_mode": False,
+            "reasons": ["capability stop terminal mode is not active"],
+        }
+    return capability_stop_terminal.build_terminal_report({
+        "requested_task": record.get("requested_task"),
+        "required_role_kind": record.get("required_role_kind"),
+        "active_profile": record.get("active_profile"),
+        "active_identity": record.get("active_identity"),
+        "stop_required": record.get("stop_required"),
+        "exact_safe_next_action": record.get("exact_safe_next_action"),
+    })
 
 
 # ── Entry point ───────────────────────────────────────────────────────────────

review_proofs.py

@@ -330,9 +330,261 @@ def assess_self_review_contamination(reviewer_identity, pr_author,
     }
 
 
+def assess_capability_evidence(capability_claims):
+    """#179 gap 1: a capability claim needs exact evidence, not assertion.
+
+    *capability_claims* is a list of ``{'task', 'allowed',
+    'evidence_source'}`` dicts, one per capability the report claims (e.g.
+    review_pr, merge_pr). Each claim must name its task, be allowed, and
+    cite an exact evidence source (``gitea_resolve_task_capability`` output
+    or equivalent runtime-context evidence). No claims at all fails closed.
+    """
+    reasons = []
+    claims = capability_claims or []
+    if not claims:
+        reasons.append(
+            "no capability evidence provided; capability checks may not be "
+            "claimed as passed"
+        )
+    for claim in claims:
+        task = (claim.get("task") or "").strip() or "<unnamed task>"
+        if claim.get("allowed") is not True:
+            reasons.append(
+                f"capability '{task}' is not proven allowed; fail closed"
+            )
+        if not (claim.get("evidence_source") or "").strip():
+            reasons.append(
+                f"capability '{task}' claimed without exact evidence "
+                "(cite gitea_resolve_task_capability output or equivalent)"
+            )
+    proven = not reasons
+    return {"proven": proven, "reasons": reasons, "claims": len(claims)}
+
+
+def assess_sweep_evidence(sweep):
+    """#179 gap 2: secret/provenance sweeps must state exact method + scope.
+
+    *sweep* keys: ``command`` (the exact command, script, grep pattern, or
+    named sweep method), ``scope`` (what was scanned, e.g. 'full PR diff
+    against prgs/master'), ``clean`` (bool result). A vague summary without
+    the exact method is downgraded; a missing sweep fails closed.
+    """
+    if not sweep:
+        return {
+            "verdict": "missing",
+            "proven": False,
+            "reasons": ["no secret/provenance sweep reported; fail closed"],
+        }
+    reasons = []
+    if not (sweep.get("command") or "").strip():
+        reasons.append(
+            "sweep method/command not stated exactly (command, script, "
+            "pattern, or named sweep method required)"
+        )
+    if not (sweep.get("scope") or "").strip():
+        reasons.append("sweep scope not stated (what diff/files were scanned)")
+    if not isinstance(sweep.get("clean"), bool):
+        reasons.append("sweep result not stated as clean/not-clean")
+    verdict = "exact" if not reasons else "vague"
+    return {
+        "verdict": verdict,
+        "proven": verdict == "exact",
+        "reasons": reasons,
+        "clean": sweep.get("clean") if isinstance(sweep.get("clean"), bool)
+        else None,
+    }
+
+
+def assess_live_state_recheck(recheck):
+    """#179 gap 3: explicit live-state recheck before review/merge mutation.
+
+    *recheck* keys: ``pr_state``, ``pinned_head_sha``, ``live_head_sha``,
+    ``pinned_base_ref``, ``live_base_ref``, ``blocking_change_requests``.
+    Proven only when the PR is still open, the live head equals the pinned
+    head (full 40-hex), the base branch is unchanged, and blocking review
+    state was checked and is absent. Not performing the recheck fails
+    closed and blocks mutation.
+    """
+    if not recheck:
+        return {
+            "proven": False,
+            "block": True,
+            "reasons": [
+                "final live-state recheck not performed before mutation; "
+                "fail closed"
+            ],
+        }
+    reasons = []
+    if (recheck.get("pr_state") or "").strip().lower() != "open":
+        reasons.append(
+            f"PR state is '{recheck.get('pr_state')}', not open; stop"
+        )
+
+    pinned = (recheck.get("pinned_head_sha") or "").strip().lower()
+    live = (recheck.get("live_head_sha") or "").strip().lower()
+    if not (_FULL_SHA.match(pinned) and _FULL_SHA.match(live)):
+        reasons.append(
+            "pinned/live head SHAs missing or not full 40-hex; fail closed"
+        )
+    elif pinned != live:
+        reasons.append(
+            "live head SHA no longer equals the pinned head; re-pin and "
+            "re-validate before mutation"
+        )
+
+    base_pinned = _normalize_ref(recheck.get("pinned_base_ref"))
+    base_live = _normalize_ref(recheck.get("live_base_ref"))
+    if not base_pinned or not base_live:
+        reasons.append("base refs missing from live-state recheck; fail closed")
+    elif base_pinned != base_live:
+        reasons.append(
+            f"base branch changed from '{base_pinned}' to '{base_live}'"
+        )
+
+    blocking = recheck.get("blocking_change_requests")
+    if blocking is None:
+        reasons.append(
+            "blocking review state not checked; fail closed"
+        )
+    elif blocking:
+        reasons.append(
+            "an undismissed REQUEST_CHANGES / blocking review state remains "
+            "unresolved"
+        )
+
+    proven = not reasons
+    return {"proven": proven, "block": not proven, "reasons": reasons}
+
+
+def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
+                         justification=None):
+    """Assess reviewer/author role separation for blind queue workflows.
+
+    Issue #175 blocks a reviewer queue task from silently becoming author
+    implementation work. Issue #179 also requires reviewer workflows to
+    report namespace use and justify any foreign namespace calls. This helper
+    accepts both forms:
+
+    - the #175 dict proof with mutation details, or
+    - the #179 keyword form: ``task_role``, ``namespaces_used``,
+      ``justification``.
+    """
+    if proof is None:
+        namespaces_reported = namespaces_used is not None
+        namespaces = list(namespaces_used or [])
+        proof = {
+            "task_role": task_role,
+            "reviewer_namespace_used": any(
+                "reviewer" in (namespace or "").lower()
+                for namespace in namespaces
+            ),
+            "author_namespace_used": any(
+                "author" in (namespace or "").lower()
+                for namespace in namespaces
+            ),
+            "mixed_namespace_justification": justification,
+            "author_mutations": [],
+            "review_mutations": [],
+            "_namespaces_used": namespaces,
+            "_namespaces_reported": namespaces_reported,
+        }
+    else:
+        proof = dict(proof or {})
+
+    task_role = (proof.get("task_role") or "").strip().lower()
+    task_kind = (proof.get("task_kind") or "").strip().lower()
+    author_mutations = list(proof.get("author_mutations") or [])
+    review_mutations = list(proof.get("review_mutations") or [])
+    reviewer_used = bool(proof.get("reviewer_namespace_used"))
+    author_used = bool(proof.get("author_namespace_used"))
+    authorized = bool(proof.get("operator_authorized_author_work"))
+    mixed_justification = (
+        proof.get("mixed_namespace_justification") or ""
+    ).strip()
+    scratch_claimed = bool(proof.get("scratch_evidence_claimed"))
+    scratch_durable = bool(proof.get("scratch_evidence_durable"))
+
+    reasons = []
+    violations = []
+
+    if task_role not in {"reviewer", "author"}:
+        reasons.append("task role missing or unknown; role boundary unproven")
+    if proof.get("_namespaces_reported") is False:
+        reasons.append("namespaces used were not reported; fail closed")
+
+    if task_role == "reviewer":
+        if author_mutations and not authorized:
+            violations.append(
+                "reviewer task performed author mutations without explicit "
+                "operator authorization"
+            )
+        if author_used and not mixed_justification:
+            reasons.append(
+                "reviewer task used author namespace without an explicit "
+                "justification"
+            )
+        if task_kind == "blind_pr_queue_review" and author_mutations:
+            if not authorized:
+                violations.append(
+                    "blind PR queue review silently pivoted into author "
+                    "implementation"
+                )
+    elif task_role == "author":
+        if review_mutations:
+            violations.append(
+                "author task performed reviewer-only mutations"
+            )
+
+    if reviewer_used and author_used and not mixed_justification:
+        reasons.append(
+            "mixed reviewer+author namespace use was not reported as a "
+            "role-boundary event"
+        )
+
+    if scratch_claimed and not scratch_durable:
+        reasons.append(
+            "scratch-only notes were claimed as durable evidence"
+        )
+
+    if violations:
+        status = "violation"
+        safe_next_action = "stop; report role-boundary violation"
+    elif reasons:
+        status = "warning"
+        safe_next_action = "downgrade final report; do not claim A-level proof"
+    else:
+        status = "clean"
+        safe_next_action = "proceed"
+
+    namespaces = proof.get("_namespaces_used")
+    if namespaces is None:
+        namespaces = []
+        if reviewer_used:
+            namespaces.append("gitea-reviewer")
+        if author_used:
+            namespaces.append("gitea-author")
+    foreign = [
+        namespace for namespace in namespaces
+        if task_role and task_role not in (namespace or "").lower()
+    ]
+
+    return {
+        "status": status,
+        "clean": status == "clean",
+        "proven": status == "clean",
+        "reasons": reasons,
+        "violations": violations,
+        "safe_next_action": safe_next_action,
+        "foreign_namespaces": foreign,
+        "justified": bool(mixed_justification),
+    }
+
+
 def build_final_report(checkout_proof, inventory, validation, contamination,
                        identity_eligible, merge_performed,
-                       issue_status_verified):
+                       issue_status_verified,
+                       capability_evidence=None, sweep=None, live_state=None,
+                       role_boundary=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
 
     Combines the individual proof verdicts into the final-report fields the
@@ -343,11 +595,47 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
     - 'downgraded' — one or more proofs missing/weak; do not merge.
     - 'blocked'    — a *violation*: a merge was claimed although the proofs
       did not allow one.
+
+    #179 raises the A bar: the report must also carry exact capability
+    evidence (``assess_capability_evidence``), an exact secret/provenance
+    sweep (``assess_sweep_evidence``), a pre-mutation live-state recheck
+    (``assess_live_state_recheck`` — also required for ``merge_allowed``),
+    and a clean role boundary (``assess_role_boundary``). Omitting any of
+    them downgrades; a merge without the live recheck is a violation.
     """
     contamination_status = contamination.get("status", "unknown")
     checkout_proven = bool(checkout_proof.get("proven"))
     validation_claimable = bool(validation.get("claimable"))
     validation_strong = validation.get("verdict") == "strong"
+    role_boundary = role_boundary or {
+        "status": "warning",
+        "reasons": ["role-boundary proof missing"],
+        "violations": [],
+    }
+    role_status = role_boundary.get("status", "warning")
+
+    capability_evidence = capability_evidence or {
+        "proven": False,
+        "reasons": ["capability evidence not provided (#179)"],
+    }
+    sweep = sweep or {
+        "verdict": "missing",
+        "proven": False,
+        "reasons": ["secret/provenance sweep evidence not provided (#179)"],
+    }
+    live_state = live_state or {
+        "proven": False,
+        "block": True,
+        "reasons": ["pre-mutation live-state recheck not provided (#179)"],
+    }
+    role_boundary = role_boundary or {
+        "proven": False,
+        "reasons": ["role-boundary/namespace usage not reported (#179)"],
+    }
+    capability_proven = bool(capability_evidence.get("proven"))
+    sweep_proven = bool(sweep.get("proven"))
+    live_state_proven = bool(live_state.get("proven"))
+    role_boundary_clean = bool(role_boundary.get("proven"))
 
     downgrade_reasons = []
     if not identity_eligible:
@@ -367,15 +655,38 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         downgrade_reasons.append(
             f"session contamination status is '{contamination_status}'"
         )
+    if role_status != "clean":
+        downgrade_reasons.append(f"role boundary status is '{role_status}'")
+        downgrade_reasons.extend(role_boundary.get("reasons", []))
     if not issue_status_verified:
         downgrade_reasons.append("linked issue status not verified")
+    if not capability_proven:
+        downgrade_reasons.append("exact capability evidence missing (#179)")
+        downgrade_reasons.extend(capability_evidence.get("reasons", []))
+    if not sweep_proven:
+        downgrade_reasons.append(
+            f"secret/provenance sweep evidence is "
+            f"{sweep.get('verdict', 'missing')} (#179)"
+        )
+        downgrade_reasons.extend(sweep.get("reasons", []))
+    if not live_state_proven:
+        downgrade_reasons.append(
+            "pre-mutation live-state recheck missing or failed (#179)"
+        )
+        downgrade_reasons.extend(live_state.get("reasons", []))
+    if not role_boundary_clean:
+        downgrade_reasons.append("role/namespace boundary not clean (#179)")
+        downgrade_reasons.extend(role_boundary.get("reasons", []))
 
     merge_allowed = (
         identity_eligible
         and checkout_proven
         and contamination_status == "clean"
+        and role_status == "clean"
         and validation_claimable
         and validation.get("verdict") != "invalid"
+        # #179: no merge without a proven final live-state recheck.
+        and live_state_proven
     )
 
     violations = []
@@ -384,6 +695,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "merge was performed/claimed although the proofs did not allow "
             "one; this run is blocked, not graded"
         )
+    violations.extend(role_boundary.get("violations", []))
 
     if violations:
         grade = "blocked"
@@ -400,6 +712,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "pr_author_distinct_from_reviewer":
             contamination_status in ("clean",),
         "session_contamination": contamination_status,
+        "role_boundary": role_status,
         "inventory_complete": bool(inventory.get("complete")),
         "validated_on_pinned_head": checkout_proven and validation_claimable,
         "validation_passed":
@@ -408,6 +721,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "merge_allowed": merge_allowed,
         "merge_performed": bool(merge_performed),
         "issue_status_verified": bool(issue_status_verified),
+        "capability_evidence_proven": capability_proven,
+        "sweep_verdict": sweep.get("verdict"),
+        "live_state_recheck_proven": live_state_proven,
+        "role_boundary_clean": role_boundary_clean,
     }
 
 
@@ -533,3 +850,113 @@ def assess_controller_handoff(report_text, role=None):
         "missing_fields": [],
         "reasons": [],
     }
+
+
+def assess_capability_stop_terminal_report(
+    report_text,
+    *,
+    trust_gate_status=None,
+    capability_denied=True,
+):
+    """Issue #197: reports after reviewer capability denial must stay pure."""
+    from capability_stop_terminal import assess_capability_stop_report
+
+    return assess_capability_stop_report(
+        report_text,
+        trust_gate_status=trust_gate_status,
+        capability_denied=capability_denied,
+    )
+
+
+# ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
+#
+# A reviewer agent may not convert an empty PR list response into a definitive
+# "no open PRs" conclusion unless the inventory result is independently proven
+# trustworthy.
+
+def pr_inventory_trust_gate(
+    list_prs_response: list | None,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    state: str | None = None,
+    authenticated_profile: dict | None = None,
+    local_remote_url: str | None = None,
+    user_context: str | None = None,
+    corroboration_open_pr_counter: int | None = None,
+    has_finality_metadata: bool = False,
+) -> dict:
+    """Evaluate whether an empty PR list is trusted or untrusted.
+
+    Returns a dict with 'status', 'reasons', and 'corroborated'.
+    """
+    if list_prs_response is None or not isinstance(list_prs_response, list):
+        return {
+            "status": "inventory_error",
+            "reasons": ["PR list response is invalid (not a list or None)"],
+            "corroborated": False,
+        }
+
+    if len(list_prs_response) > 0:
+        return {
+            "status": "trusted_nonempty",
+            "reasons": [],
+            "corroborated": False,
+        }
+
+    reasons = []
+
+    # 1. Exact remote, owner, repo, and state filter resolved correctly
+    if not remote or remote not in ("dadeschools", "prgs"):
+        reasons.append("remote instance is invalid or unresolved")
+    if not org or not org.strip():
+        reasons.append("owner/org is invalid or unresolved")
+    if not repo or not repo.strip():
+        reasons.append("repository name is invalid or unresolved")
+    if state != "open":
+        reasons.append("state filter is not 'open'")
+
+    # 2. Authenticated profile permission check
+    if not authenticated_profile or not isinstance(authenticated_profile, dict):
+        reasons.append("authenticated profile is missing or invalid")
+    else:
+        allowed = authenticated_profile.get("allowed_operations") or []
+        if "gitea.read" not in allowed and "read" not in allowed:
+            reasons.append("authenticated profile lacks read permissions")
+
+    # 3. Pagination/finality metadata or independent read path corroboration
+    corroborated = False
+    if has_finality_metadata:
+        corroborated = True
+    elif corroboration_open_pr_counter == 0:
+        corroborated = True
+    else:
+        reasons.append("pagination finality not proven and open_pr_counter corroboration is missing or non-zero")
+
+    # 4. Local checkout remote URL matching the target repo
+    if not local_remote_url or not isinstance(local_remote_url, str):
+        reasons.append("local checkout remote URL is missing or invalid")
+    else:
+        expected = f"{org}/{repo}".lower()
+        if expected not in local_remote_url.lower():
+            reasons.append(f"local remote URL does not match target repository '{org}/{repo}'")
+
+    # 5. User context check (indicators that PRs should exist)
+    if user_context and isinstance(user_context, str):
+        indicators = ["pr #", "pull request #", "open pr", "pr queue"]
+        found = [ind for ind in indicators if ind in user_context.lower()]
+        if found:
+            reasons.append(f"user context indicates open PRs should exist (matched: {', '.join(found)})")
+
+    if reasons:
+        return {
+            "status": "untrusted_empty",
+            "reasons": reasons,
+            "corroborated": corroborated,
+        }
+
+    return {
+        "status": "trusted_empty",
+        "reasons": [],
+        "corroborated": corroborated,
+    }

skills/llm-project-workflow/SKILL.md

@@ -150,12 +150,33 @@ Worktree folder = branch with `/` replaced by `-`
 6. Implement the narrow scope only — no unrelated refactors or formatting churn.
 7. Add/update focused tests when behavior changes.
 8. Run the checks (tests, compile/lint, `git diff --check`, secret scan).
-9. Commit with an issue-linked message.
+   Record the branch name and `HEAD` SHA at validation time — the drift
-10. Push the branch.
+   check in step 9 compares against exactly this state.
-11. Open a PR to `master`.
+9. **Branch proof before commit (#177):** prove and state, immediately
-12. **If you are the author, stop before review/merge.**
+   before staging/committing (`author_proofs.verify_branch_for_commit`,
-13. **Normal issue work must not directly push to `master`.** PR content should be merged through the forge PR merge mechanism.
+   `author_proofs.detect_branch_drift`):
-14. Direct push to `master` is allowed only as a documented recovery exception. If used, the final report must include:
+   - current branch (`git branch --show-current`) equals the intended
+     feature branch from the issue claim
+   - current branch is not `master`, `main`, `develop`, `development`, or
+     `dev`
+   - branch and `HEAD` have not changed since validation (step 8) — in a
+     shared checkout another session may switch branches mid-session;
+     treat that as expected and **stop before committing** when detected
+   If any check fails, stop and reconcile; do not commit.
+10. Commit with an issue-linked message.
+11. **Branch proof before push (#177):** prove that the local branch, the
+    push target branch, and the intended issue branch all match, and that
+    none of them is a protected branch
+    (`author_proofs.verify_push_target`). If a commit accidentally landed
+    on a protected branch, do **not** push: report the accident and the
+    exact repair steps (`author_proofs.assess_protected_branch_commit`) —
+    never silently continue after a repair.
+12. Push the branch.
+13. Open a PR to `master`. The final report must include the branch proofs
+    from steps 9 and 11 (`author_proofs.build_commit_push_report`).
+14. **If you are the author, stop before review/merge.**
+15. **Normal issue work must not directly push to `master`.** PR content should be merged through the forge PR merge mechanism.
+16. Direct push to `master` is allowed only as a documented recovery exception. If used, the final report must include:
     - why the PR merge path could not be used
     - exact commits pushed
     - PR metadata state
@@ -196,26 +217,50 @@ Worktree folder = branch with `/` replaced by `-`
    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`).
-7. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
+7. **Role-boundary proof (#175):** a reviewer queue task must not silently
-8. Run the tests. Validation reporting must include the exact command and
+   become author implementation. If no eligible PR exists, stop with the
+   queue report. Do not claim issues, create branches, commit, push, or open
+   PRs unless the operator explicitly retasks the run as author work. Mixed
+   reviewer+author namespace use must be reported with a justification, and
+   scratch-only notes are not durable evidence unless posted or committed
+   intentionally (`review_proofs.assess_role_boundary`).
+8. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
+9. Run the tests. Validation reporting must include the exact command and
    exact results: pass/fail, counts of tests passed/skipped/failed, any
    ignored paths and why they are safe to ignore, and whether the command
    differs from the repository's canonical validation command. Only claim a
    validation result after the command has completed and its output has
    been read (`review_proofs.assess_validation_report`).
-9. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
+10. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
-10. The final report must distinguish (`review_proofs.build_final_report`):
+11. **#179 A-bar proofs** (all fail closed when missing —
+    `review_proofs.assess_capability_evidence`, `assess_sweep_evidence`,
+    `assess_live_state_recheck`, `assess_role_boundary`):
+    - Capability claims must cite exact `gitea_resolve_task_capability`
+      output (or runtime context); a bare "capability checks passed" is
+      downgraded.
+    - The secret/provenance sweep must state the exact command/script/
+      pattern/named method and the scope scanned.
+    - Immediately before submitting a review verdict (and again before any
+      merge), re-read live PR state and prove: still open, live head ==
+      pinned head, base unchanged, no unresolved blocking review state.
+    - Reviewer runs stay in the reviewer namespace; any author-namespace
+      call requires an explicit justification in the report.
+12. The final report must distinguish (`review_proofs.build_final_report`):
     identity eligible; PR author different from reviewer; session
     contamination absent (with evidence); validation performed on the pinned
-    head; merge performed; issue status verified. If any proof is missing,
+    head; capability evidence; sweep verdict; live-state recheck; role
-    stop or downgrade the result instead of merging confidently.
+    boundary; merge performed; issue status verified. If any proof is
+    missing, stop or downgrade the result instead of merging confidently.
 
 ## G. Merge / cleanup workflow
 
 Only an eligible (non-author) reviewer merges. Before merging: always verify
-the authenticated identity **and** the PR author; respect runtime profile
+the authenticated identity **and** the PR author; cite exact capability
-gates; run independent validation (do not trust the author's reported
+evidence for merge_pr (#179); respect runtime profile gates; run independent
-results); and merge with a **pinned head SHA** and, where supported, the
+validation (do not trust the author's reported results); perform the **final
+live-state recheck** (#179 — PR still open, live head == pinned head, base
+unchanged, no unresolved blocking review state) immediately before the merge
+mutation; and merge with a **pinned head SHA** and, where supported, the
 **expected changed-file set**, so a moved head or widened diff refuses the
 merge. After a real merge:
 

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

@@ -20,10 +20,21 @@ Steps:
    *If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
 2. Verify authenticated identity + active profile.
 3. Confirm PR #<pr>: author (not you), state open, mergeable, review approved. Check if PR body uses `Closes #N` or `Fixes #N`; if it uses `Implements #N` or `Refs #N`, manual closing will be needed in step 29.
-4. If any gate fails → STOP and report.
+4. Capability evidence (#179): cite the exact gitea_resolve_task_capability
-4. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
+   output (or runtime context) proving merge_pr is allowed — a bare
-   optionally pinning the reviewed head SHA / changed-file set.
+   "capability checks passed" claim is downgraded.
-5. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
+5. Final live-state recheck (#179), immediately before the merge mutation —
+   re-read the live PR and prove:
+   - PR still open
+   - live head SHA still equals the pinned/reviewed head SHA
+   - base branch unchanged
+   - no undismissed REQUEST_CHANGES / blocking review state remains
+   If any recheck fails → STOP, re-pin, re-validate.
+6. If any gate fails → STOP and report.
+7. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
+   pinning the reviewed head SHA (expected_head_sha) and, where supported,
+   the changed-file set.
+8. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
    *Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.*
 
 Then run the cleanup template (worktree-cleanup.md):

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

@@ -23,6 +23,10 @@ Rules (llm-project-workflow):
 - 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).
+- Do not pivot from a reviewer queue task into author implementation unless
+  the operator explicitly retasks the run. If author namespace was used, the
+  final report must justify why; author mutations after reviewer queue work
+  without explicit authorization are a role-boundary violation.
 - Do not merge if any check fails.
 
 Steps:
@@ -32,6 +36,11 @@ Steps:
    - Target task role: reviewer identity (must NOT be the PR author)
    *If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
 2. Verify your authenticated identity (whoami) and the active profile.
+   Capability evidence (#179): cite the exact gitea_resolve_task_capability
+   output (or runtime context) for review_pr (and merge_pr if merging later);
+   a bare "capability checks passed" claim is downgraded. Stay in the
+   reviewer namespace: any author-namespace call must be justified in the
+   report (#179).
 3. Fetch the PR facts: PR author, head SHA, state (must be open), base branch.
    Pin the head SHA in your notes; every later step validates THAT SHA.
 4. If authenticated user == PR author → STOP (no self-review).
@@ -39,6 +48,10 @@ Steps:
    cannot evidence whether this session authored/touched the PR branch,
    report contamination as UNKNOWN (not contaminated, not clean) and choose
    another PR or stop.
+   Role-boundary claims must also be evidence-backed (#175): report whether
+   reviewer namespace, author namespace, author mutations, or review mutations
+   occurred. Use `review_proofs.assess_role_boundary`; if it is not clean,
+   downgrade or stop instead of claiming an A-level run.
 5. scripts/worktree-review <pr-head-branch>          # detached, branches/review-*
    cd branches/review-<pr-head-branch-slug>
 6. Checkout proof (#173) — prove and state, before any diff review or
@@ -50,12 +63,23 @@ Steps:
    If HEAD does not match the pinned head → STOP before review/merge.
 7. Confirm the worktree is clean. Inspect the FULL diff; confirm scope matches
    issue #<n>; flag any unrelated files, secrets, or formatting churn. Check that the PR body correctly uses Gitea-closing keywords (`Closes #N` or `Fixes #N`) instead of non-closing ones (`Implements #N`, `Refs #N`).
+   Secret/provenance sweep must be exact (#179): state the exact command,
+   script, grep pattern, or named sweep method AND the scope scanned (e.g.
+   `git diff prgs/master...HEAD | grep -inE '<pattern>'`); "checked the diff
+   for secrets" alone is downgraded.
 8. Run the test suite; report the exact command and exact results — pass/fail
    plus passed/skipped/failed counts, any ignored paths and why they are safe
    to ignore, and whether the command differs from the repository's canonical
    validation command. Only claim a result after the output has been read.
-9. Post the review verdict: approve only if scope is clean and checks pass;
+9. Final live-state recheck (#179), immediately before submitting the review
-   otherwise request changes with specifics. Never merge from this review step.
+   verdict — re-read the live PR and prove:
+   - PR still open
+   - live head SHA still equals the pinned head SHA from step 3
+   - base branch unchanged
+   - no undismissed REQUEST_CHANGES / blocking review state left unaccounted
+   If anything moved → STOP, re-pin, re-validate before any verdict.
+10. Post the review verdict: approve only if scope is clean and checks pass;
+    otherwise request changes with specifics. Never merge from this review step.
    Include a "Review Metadata" block (attribution only — docs/llm-agent-sha.md):
 
    Review Metadata:

skills/llm-project-workflow/templates/start-issue.md

@@ -26,8 +26,19 @@ Steps:
    cd branches/<type>-issue-<n>-<slug>
 6. Implement the narrow scope only; add/update focused tests if behavior changes.
 7. Checks: run the test suite, compile/lint changed files, git diff --check,
-   and scan the diff for secrets.
+   and scan the diff for secrets. Record the branch name and HEAD SHA at
-8. Commit (issue-linked message), push the branch, open a PR to master.
+   validation time.
+8. Branch proof before commit (#177) — prove and state:
+   - git branch --show-current == the intended issue branch from step 5
+   - the branch is NOT master/main/develop/development/dev
+   - branch and HEAD unchanged since step 7 (another session can switch a
+     shared checkout mid-session; if drift is detected, STOP and reconcile
+     before committing)
+   If a commit accidentally lands on a protected branch: do NOT push;
+   report the accident and the exact repair steps — never silently continue.
+9. Commit (issue-linked message). Branch proof before push (#177): local
+   branch == push target branch == intended issue branch, none protected.
+   Then push the branch and open a PR to master.
    *The PR body MUST use closing keywords like `Closes #N` or `Fixes #N` to close the issue; do NOT use `Implements #N` or `Refs #N` for closing, as Gitea will not auto-close it.*
    Include an "LLM Handoff Metadata" block in the PR body (attribution only;
    never an eligibility input — docs/llm-agent-sha.md):
@@ -40,7 +51,7 @@ Steps:
    - Branch: <branch>
    - Worktree: <worktree path>
    - Self-review allowed: no
-9. Stop before review/merge — you are the author.
+10. Stop before review/merge — you are the author.
 
 Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
 §K (compact; long form only on the high-risk triggers), including the author

tests/test_author_proofs.py

@@ -0,0 +1,234 @@
+"""Tests for author-side branch-identity proofs (Issue #177).
+
+Issue #177 (author-side counterpart of the #173 reviewer proofs) requires
+author workflows to *prove* local git state before staging, committing, or
+pushing, instead of discovering drift after the fact:
+
+1. The current branch equals the intended feature branch and is never a
+   protected branch (master/main/develop/development/dev).
+2. Branch or HEAD drift between validation and commit — including external
+   branch switches in a shared worktree — stops the workflow.
+3. A push requires local branch, remote target branch, and intended issue
+   branch to all match.
+4. An accidental commit on a protected branch must not be pushed and its
+   repair must be reported, never silently continued.
+
+These are the harness assertions from the issue's Required behavior 5.
+"""
+import sys
+import unittest
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+from author_proofs import (  # noqa: E402
+    PROTECTED_BRANCHES,
+    assess_protected_branch_commit,
+    build_commit_push_report,
+    detect_branch_drift,
+    verify_branch_for_commit,
+    verify_push_target,
+)
+
+FEATURE = "feat/issue-177-branch-drift-proofs"
+HEAD_1 = "64dc334a92685b7b6a1fdb7ffe363f02a69f5dbd"
+HEAD_2 = "ccc5ef79dfe629853e144763238593bd808d57e0"
+
+
+class TestProtectedBranches(unittest.TestCase):
+    def test_known_protected_names(self):
+        for name in ("master", "main", "develop", "development", "dev"):
+            self.assertIn(name, PROTECTED_BRANCHES)
+
+
+class TestVerifyBranchForCommit(unittest.TestCase):
+    """Required behavior 1: prove the branch before staging/committing."""
+
+    def test_on_intended_feature_branch_is_proven(self):
+        proof = verify_branch_for_commit(FEATURE, FEATURE)
+        self.assertTrue(proof["proven"])
+        self.assertFalse(proof["block"])
+
+    def test_commit_attempted_while_on_master_is_blocked(self):
+        # Harness assertion (behavior 5, bullet 1).
+        proof = verify_branch_for_commit("master", FEATURE)
+        self.assertFalse(proof["proven"])
+        self.assertTrue(proof["block"])
+        self.assertTrue(any("master" in r for r in proof["reasons"]))
+
+    def test_every_protected_branch_is_blocked_as_current(self):
+        for name in PROTECTED_BRANCHES:
+            proof = verify_branch_for_commit(name, FEATURE)
+            self.assertTrue(proof["block"], name)
+
+    def test_intended_branch_may_not_be_protected(self):
+        proof = verify_branch_for_commit("master", "master")
+        self.assertFalse(proof["proven"])
+        self.assertTrue(proof["block"])
+
+    def test_wrong_feature_branch_is_blocked(self):
+        proof = verify_branch_for_commit("feat/issue-178-other-work", FEATURE)
+        self.assertFalse(proof["proven"])
+        self.assertTrue(proof["block"])
+
+    def test_missing_current_branch_fails_closed(self):
+        proof = verify_branch_for_commit("", FEATURE)
+        self.assertTrue(proof["block"])
+
+    def test_missing_intended_branch_fails_closed(self):
+        proof = verify_branch_for_commit(FEATURE, None)
+        self.assertTrue(proof["block"])
+
+
+class TestBranchDrift(unittest.TestCase):
+    """Required behaviors 2 + 3: drift between validation and commit stops
+    the workflow."""
+
+    def test_no_drift_when_branch_and_head_unchanged(self):
+        drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1)
+        self.assertFalse(drift["drifted"])
+        self.assertFalse(drift["block"])
+
+    def test_branch_drift_between_validation_and_commit_is_blocked(self):
+        # Harness assertion (behavior 5, bullet 2).
+        drift = detect_branch_drift(FEATURE, HEAD_1, "feat/other", HEAD_1)
+        self.assertTrue(drift["drifted"])
+        self.assertTrue(drift["block"])
+
+    def test_shared_worktree_branch_switch_is_detected(self):
+        # Harness assertion (behavior 5, bullet 4): an external session
+        # switching the shared checkout to another branch (e.g. master)
+        # must be detected as drift, not treated as exceptional noise.
+        drift = detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1)
+        self.assertTrue(drift["drifted"])
+        self.assertTrue(drift["block"])
+        self.assertTrue(any("switch" in r.lower() for r in drift["reasons"]))
+
+    def test_head_moved_since_validation_is_blocked(self):
+        drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_2)
+        self.assertTrue(drift["drifted"])
+        self.assertTrue(drift["block"])
+        self.assertTrue(any("HEAD" in r for r in drift["reasons"]))
+
+    def test_missing_state_fails_closed(self):
+        drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, None)
+        self.assertTrue(drift["drifted"])
+        self.assertTrue(drift["block"])
+
+
+class TestVerifyPushTarget(unittest.TestCase):
+    """Required behavior 1 (push leg) + acceptance: push needs proof that
+    local, remote, and intended branches all match."""
+
+    def test_matching_local_remote_and_intended_is_proven(self):
+        proof = verify_push_target(FEATURE, FEATURE, FEATURE)
+        self.assertTrue(proof["proven"])
+        self.assertFalse(proof["block"])
+
+    def test_push_target_mismatch_is_blocked(self):
+        # Harness assertion (behavior 5, bullet 3).
+        proof = verify_push_target(FEATURE, "feat/issue-178-other-work", FEATURE)
+        self.assertFalse(proof["proven"])
+        self.assertTrue(proof["block"])
+
+    def test_local_branch_differs_from_intended_is_blocked(self):
+        proof = verify_push_target("feat/other", FEATURE, FEATURE)
+        self.assertTrue(proof["block"])
+
+    def test_pushing_a_protected_branch_is_blocked(self):
+        proof = verify_push_target("master", "master", "master")
+        self.assertFalse(proof["proven"])
+        self.assertTrue(proof["block"])
+
+    def test_missing_remote_target_fails_closed(self):
+        proof = verify_push_target(FEATURE, "", FEATURE)
+        self.assertTrue(proof["block"])
+
+
+class TestProtectedBranchAccident(unittest.TestCase):
+    """Required behavior 4: accidental protected-branch commits must not be
+    pushed and their repair must be reported."""
+
+    def test_feature_branch_commit_is_not_an_accident(self):
+        result = assess_protected_branch_commit(FEATURE)
+        self.assertFalse(result["accident"])
+        self.assertEqual(result["violations"], [])
+
+    def test_commit_on_master_is_an_accident_and_must_not_push(self):
+        result = assess_protected_branch_commit(
+            "master", pushed=False, repair_reported=True
+        )
+        self.assertTrue(result["accident"])
+        self.assertTrue(result["must_not_push"])
+        self.assertEqual(result["violations"], [])
+        self.assertTrue(result["repair_required"])
+
+    def test_pushing_the_accident_is_a_violation(self):
+        result = assess_protected_branch_commit(
+            "master", pushed=True, repair_reported=True
+        )
+        self.assertTrue(any("push" in v.lower() for v in result["violations"]))
+
+    def test_silent_repair_is_a_violation(self):
+        # Harness assertion (behavior 5, bullet 5): the repair path must not
+        # silently continue without reporting.
+        result = assess_protected_branch_commit(
+            "master", pushed=False, repair_reported=False
+        )
+        self.assertTrue(any("report" in v.lower() for v in result["violations"]))
+
+
+class TestCommitPushReport(unittest.TestCase):
+    """Acceptance criteria: the final report includes branch proof before
+    commit and before push, and blocks instead of continuing."""
+
+    def _report(self, **overrides):
+        kwargs = {
+            "commit_proof": verify_branch_for_commit(FEATURE, FEATURE),
+            "drift": detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1),
+            "push_proof": verify_push_target(FEATURE, FEATURE, FEATURE),
+            "accident": assess_protected_branch_commit(FEATURE),
+        }
+        kwargs.update(overrides)
+        return build_commit_push_report(**kwargs)
+
+    def test_fully_proven_report_is_ok(self):
+        report = self._report()
+        self.assertEqual(report["status"], "ok")
+        self.assertTrue(report["branch_proof_before_commit"])
+        self.assertTrue(report["branch_proof_before_push"])
+        self.assertFalse(report["drift_detected"])
+        self.assertEqual(report["violations"], [])
+
+    def test_commit_proof_failure_blocks(self):
+        report = self._report(
+            commit_proof=verify_branch_for_commit("master", FEATURE)
+        )
+        self.assertEqual(report["status"], "blocked")
+        self.assertFalse(report["branch_proof_before_commit"])
+
+    def test_drift_blocks(self):
+        report = self._report(
+            drift=detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1)
+        )
+        self.assertEqual(report["status"], "blocked")
+        self.assertTrue(report["drift_detected"])
+
+    def test_push_proof_failure_blocks(self):
+        report = self._report(
+            push_proof=verify_push_target(FEATURE, "feat/other", FEATURE)
+        )
+        self.assertEqual(report["status"], "blocked")
+        self.assertFalse(report["branch_proof_before_push"])
+
+    def test_accident_violations_block(self):
+        report = self._report(
+            accident=assess_protected_branch_commit(
+                "master", pushed=False, repair_reported=False
+            )
+        )
+        self.assertEqual(report["status"], "blocked")
+        self.assertTrue(report["violations"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_capability_stop_terminal.py

@@ -0,0 +1,164 @@
+"""Tests for capability stop terminal mode (#197)."""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import capability_stop_terminal
+import gitea_config
+import mcp_server
+from review_proofs import assess_capability_stop_terminal_report
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "prgs-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "jcwalker3",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": [
+                "gitea.read", "gitea.issue.create", "gitea.pr.create",
+                "gitea.branch.push",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
+            ],
+            "execution_profile": "prgs-author",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+
+class TestCapabilityStopTerminal(unittest.TestCase):
+    def setUp(self):
+        capability_stop_terminal.clear()
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.example.com",
+                "org": "Scaled-Tech-Consulting",
+                "repo": "Gitea-Tools",
+            },
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir = tempfile.TemporaryDirectory()
+        self.config_path = os.path.join(self._dir.name, "profiles.json")
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(CONFIG))
+
+    def tearDown(self):
+        self._remotes.stop()
+        capability_stop_terminal.clear()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir.cleanup()
+
+    def _env(self):
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": "prgs-author",
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+        }
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_review_pr_stop_enters_terminal_mode(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="review_pr", remote="prgs"
+            )
+        self.assertTrue(res["stop_required"])
+        self.assertTrue(res.get("terminal_mode"))
+        self.assertTrue(capability_stop_terminal.is_active())
+        self.assertIn(
+            capability_stop_terminal.TERMINAL_REPORT_HEADING,
+            res["terminal_report"]["heading"],
+        )
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_list_prs_blocked_after_capability_stop(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.gitea_list_prs(remote="prgs")
+        self.assertIn("Cannot perform reviewer task", str(ctx.exception))
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_eligibility_check_blocked_after_stop(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            res = mcp_server.gitea_check_pr_eligibility(
+                pr_number=193, action="review", remote="prgs"
+            )
+        self.assertFalse(res["eligible"])
+        self.assertTrue(res.get("terminal_mode"))
+
+    def test_report_with_pr_selection_impure(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Selected PR #193 for review anyway."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertFalse(result["pure"])
+
+    def test_session_eligibility_wording_blocked(self):
+        ok, violations = capability_stop_terminal.validate_eligibility_wording(
+            "PR 193 is not authored by this session so it is eligible."
+        )
+        self.assertFalse(ok)
+        self.assertTrue(violations)
+
+    def test_rebase_fallback_blocked_in_report(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Or have me rebase conflicted PR 193."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertFalse(result["pure"])
+        self.assertTrue(
+            any("author fallback" in v for v in result["violations"])
+        )
+
+    def test_empty_queue_without_trusted_empty_blocked(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "The repo has 0 open PRs."
+        )
+        result = assess_capability_stop_terminal_report(
+            report, trust_gate_status="untrusted_empty"
+        )
+        self.assertFalse(result["pure"])
+
+    def test_pure_terminal_report_passes(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Identity: jcwalker3 / prgs-author.\n"
+            "Required: prgs-reviewer.\n"
+            "No mutations performed."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertTrue(result["pure"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_operator_guide.py

@@ -46,8 +46,8 @@ EXPECTED_SKILLS = [
     "gitea-resolve-task-capability",
     "profile-switching",
     "redaction-security-review",
-    "jenkins-readonly",
+    "jenkins-mcp",
-    "glitchtip-readonly",
+    "glitchtip-mcp",
     "release-operator",
 ]
 
@@ -234,8 +234,8 @@ class TestProjectSkills(GuideTestBase):
         with patch.dict(os.environ, AUTHOR_ENV, clear=True):
             r = mcp_list_project_skills()
         by_name = {s["name"]: s for s in r["skills"]}
-        self.assertNotEqual(by_name["jenkins-readonly"]["status"], "available")
+        self.assertNotEqual(by_name["jenkins-mcp"]["status"], "available")
-        self.assertNotEqual(by_name["glitchtip-readonly"]["status"], "available")
+        self.assertNotEqual(by_name["glitchtip-mcp"]["status"], "available")
 
     def test_no_urls_in_registry(self):
         with patch.dict(os.environ, AUTHOR_ENV, clear=True):
@@ -245,6 +245,17 @@ class TestProjectSkills(GuideTestBase):
         self.assertNotIn("http://", blob)
         self.assertNotIn("keychain:", blob)
 
+    def test_enabled_but_no_usable_tools_negative_assertion(self):
+        """Negative assertion for 'enabled but no usable tools' (per issue #146)."""
+        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
+            r = mcp_list_project_skills()
+        by_name = {s["name"]: s for s in r["skills"]}
+        # jenkins-mcp is designed-not-implemented; even if "enabled" in config,
+        # it should not be usable/available to current profile without tools.
+        self.assertIn("jenkins-mcp", by_name)
+        self.assertEqual(by_name["jenkins-mcp"]["status"], "designed-not-implemented")
+        self.assertFalse(by_name["jenkins-mcp"].get("available_to_current_profile", False))
+
 
 # ---------------------------------------------------------------------------
 # mcp_get_skill_guide

tests/test_review_proofs.py

@@ -21,11 +21,16 @@ import unittest
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
 from review_proofs import (  # noqa: E402
+    assess_capability_evidence,
     assess_controller_handoff,
     assess_inventory_completeness,
+    assess_live_state_recheck,
+    assess_role_boundary,
     assess_self_review_contamination,
+    assess_sweep_evidence,
     assess_validation_report,
     build_final_report,
+    pr_inventory_trust_gate,
     resolve_repos_from_user_reference,
     verify_pinned_head_checkout,
 )
@@ -99,6 +104,78 @@ def _good_contamination():
     )
 
 
+def _good_role_boundary():
+    return assess_role_boundary(
+        {
+            "task_role": "reviewer",
+            "task_kind": "blind_pr_queue_review",
+            "reviewer_namespace_used": True,
+            "author_namespace_used": False,
+            "author_mutations": [],
+            "review_mutations": [],
+            "operator_authorized_author_work": False,
+            "scratch_evidence_claimed": False,
+        }
+    )
+
+
+def _good_capability_evidence():
+    return assess_capability_evidence([
+        {
+            "task": "review_pr",
+            "allowed": True,
+            "evidence_source": (
+                "gitea_resolve_task_capability(review_pr) output: "
+                "allowed_in_current_session=true, profile prgs-reviewer"
+            ),
+        },
+        {
+            "task": "merge_pr",
+            "allowed": True,
+            "evidence_source": (
+                "gitea_resolve_task_capability(merge_pr) output: "
+                "allowed_in_current_session=true, profile prgs-reviewer"
+            ),
+        },
+    ])
+
+
+def _good_sweep(**overrides):
+    sweep = {
+        "command": (
+            "git diff prgs/master...HEAD | grep -inE "
+            "'password|token|secret|api[_-]?key|authorization|bearer|https?://'"
+        ),
+        "scope": "full PR diff against prgs/master",
+        "clean": True,
+    }
+    sweep.update(overrides)
+    return assess_sweep_evidence(sweep)
+
+
+def _good_live_state(**overrides):
+    recheck = {
+        "pr_state": "open",
+        "pinned_head_sha": PINNED,
+        "live_head_sha": PINNED,
+        "pinned_base_ref": "master",
+        "live_base_ref": "master",
+        "blocking_change_requests": False,
+    }
+    recheck.update(overrides)
+    return assess_live_state_recheck(recheck)
+
+
+def _good_role_boundary_179(**overrides):
+    kwargs = {
+        "task_role": "reviewer",
+        "namespaces_used": ["gitea-reviewer"],
+        "justification": None,
+    }
+    kwargs.update(overrides)
+    return assess_role_boundary(**kwargs)
+
+
 class TestCheckoutProof(unittest.TestCase):
     """Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
 
@@ -368,6 +445,74 @@ class TestSelfReviewContamination(unittest.TestCase):
         self.assertEqual(result["status"], "unknown")
 
 
+class TestRoleBoundary(unittest.TestCase):
+    """Issue #175: reviewer queue tasks must not pivot into author work."""
+
+    def test_reviewer_queue_without_author_mutations_is_clean(self):
+        result = _good_role_boundary()
+        self.assertEqual(result["status"], "clean")
+        self.assertEqual(result["violations"], [])
+
+    def test_reviewer_queue_author_mutation_without_authorization_violates(self):
+        result = assess_role_boundary(
+            {
+                "task_role": "reviewer",
+                "task_kind": "blind_pr_queue_review",
+                "reviewer_namespace_used": True,
+                "author_namespace_used": True,
+                "author_mutations": ["claim issue #171", "push branch"],
+                "operator_authorized_author_work": False,
+                "mixed_namespace_justification": (
+                    "author namespace was used for implementation"
+                ),
+            }
+        )
+        self.assertEqual(result["status"], "violation")
+        self.assertTrue(any("pivot" in r for r in result["violations"]))
+
+    def test_mixed_namespace_use_without_justification_is_warning(self):
+        result = assess_role_boundary(
+            {
+                "task_role": "reviewer",
+                "task_kind": "blind_pr_queue_review",
+                "reviewer_namespace_used": True,
+                "author_namespace_used": True,
+                "author_mutations": [],
+            }
+        )
+        self.assertEqual(result["status"], "warning")
+        self.assertTrue(
+            any("mixed" in r.lower() for r in result["reasons"])
+        )
+
+    def test_author_task_cannot_perform_review_mutations(self):
+        result = assess_role_boundary(
+            {
+                "task_role": "author",
+                "reviewer_namespace_used": False,
+                "author_namespace_used": True,
+                "review_mutations": ["approve PR"],
+            }
+        )
+        self.assertEqual(result["status"], "violation")
+        self.assertTrue(
+            any("reviewer-only" in r for r in result["violations"])
+        )
+
+    def test_scratch_only_notes_are_not_durable_evidence(self):
+        result = assess_role_boundary(
+            {
+                "task_role": "reviewer",
+                "task_kind": "blind_pr_queue_review",
+                "reviewer_namespace_used": True,
+                "scratch_evidence_claimed": True,
+                "scratch_evidence_durable": False,
+            }
+        )
+        self.assertEqual(result["status"], "warning")
+        self.assertTrue(any("scratch-only" in r for r in result["reasons"]))
+
+
 class TestFinalReport(unittest.TestCase):
     """Required behavior 6 + acceptance criteria: the report must
     distinguish each proof, and only a fully proven run earns an "A"."""
@@ -381,6 +526,10 @@ class TestFinalReport(unittest.TestCase):
             "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(),
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -393,6 +542,7 @@ class TestFinalReport(unittest.TestCase):
         self.assertTrue(report["identity_eligible"])
         self.assertTrue(report["pr_author_distinct_from_reviewer"])
         self.assertEqual(report["session_contamination"], "clean")
+        self.assertEqual(report["role_boundary"], "clean")
         self.assertTrue(report["validated_on_pinned_head"])
         self.assertFalse(report["merge_performed"])
         self.assertTrue(report["issue_status_verified"])
@@ -465,6 +615,37 @@ class TestFinalReport(unittest.TestCase):
         self.assertNotEqual(report["grade"], "A")
         self.assertFalse(report["merge_allowed"])
 
+    def test_missing_role_boundary_downgrades_and_blocks_merge(self):
+        kwargs = {
+            "checkout_proof": _good_checkout(),
+            "inventory": _good_inventory(),
+            "validation": _good_validation(),
+            "contamination": _good_contamination(),
+            "identity_eligible": True,
+            "merge_performed": False,
+            "issue_status_verified": True,
+        }
+        report = build_final_report(**kwargs)
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["merge_allowed"])
+        self.assertEqual(report["role_boundary"], "warning")
+
+    def test_role_boundary_violation_blocks_report(self):
+        boundary = assess_role_boundary(
+            {
+                "task_role": "reviewer",
+                "task_kind": "blind_pr_queue_review",
+                "reviewer_namespace_used": True,
+                "author_namespace_used": True,
+                "author_mutations": ["create PR"],
+                "operator_authorized_author_work": False,
+                "mixed_namespace_justification": "implementation pivot",
+            }
+        )
+        report = self._report(role_boundary=boundary)
+        self.assertEqual(report["grade"], "blocked")
+        self.assertFalse(report["merge_allowed"])
+
 
 class TestStdoutIsolation(unittest.TestCase):
     """Regression test for #178: tests must not close or corrupt stdout/stderr
@@ -679,5 +860,291 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertIn("issue #182", skill)
 
 
+class TestPRInventoryTrustGate(unittest.TestCase):
+    """Issue #194: unit tests for the PR inventory trust gate."""
+
+    def setUp(self):
+        self.profile = {
+            "profile_name": "prgs-reviewer",
+            "allowed_operations": ["read", "gitea.read", "gitea.pr.approve"],
+        }
+        self.local_url = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+
+    def test_trusted_nonempty(self):
+        res = pr_inventory_trust_gate([{"number": 1}])
+        self.assertEqual(res["status"], "trusted_nonempty")
+        self.assertFalse(res["corroborated"])
+
+    def test_inventory_error_none_or_not_list(self):
+        self.assertEqual(pr_inventory_trust_gate(None)["status"], "inventory_error")
+        self.assertEqual(pr_inventory_trust_gate("not a list")["status"], "inventory_error")
+
+    def test_untrusted_empty_no_pagination_or_corroboration(self):
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=self.profile,
+            local_remote_url=self.local_url, user_context=None,
+            corroboration_open_pr_counter=None, has_finality_metadata=False
+        )
+        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertIn("pagination finality not proven and open_pr_counter corroboration is missing or non-zero", res["reasons"])
+
+    def test_untrusted_empty_profile_permission_mismatch(self):
+        bad_profile = {"profile_name": "prgs-bad", "allowed_operations": ["write"]}
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=bad_profile,
+            local_remote_url=self.local_url, user_context=None,
+            corroboration_open_pr_counter=0, has_finality_metadata=False
+        )
+        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertIn("authenticated profile lacks read permissions", res["reasons"])
+
+    def test_untrusted_empty_remote_url_mismatch(self):
+        bad_url = "https://gitea.prgs.cc/other-org/other-repo.git"
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=self.profile,
+            local_remote_url=bad_url, user_context=None,
+            corroboration_open_pr_counter=0, has_finality_metadata=False
+        )
+        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertIn("local remote URL does not match target repository 'Scaled-Tech-Consulting/Gitea-Tools'", res["reasons"])
+
+    def test_untrusted_empty_user_context_indicates_prs(self):
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=self.profile,
+            local_remote_url=self.local_url, user_context="please check open PR #181",
+            corroboration_open_pr_counter=0, has_finality_metadata=False
+        )
+        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertTrue(any("user context indicates open PRs should exist" in r for r in res["reasons"]))
+
+    def test_trusted_empty_with_corroboration(self):
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=self.profile,
+            local_remote_url=self.local_url, user_context=None,
+            corroboration_open_pr_counter=0, has_finality_metadata=False
+        )
+        self.assertEqual(res["status"], "trusted_empty")
+        self.assertTrue(res["corroborated"])
+
+    def test_trusted_empty_with_finality_metadata(self):
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=self.profile,
+            local_remote_url=self.local_url, user_context=None,
+            corroboration_open_pr_counter=None, has_finality_metadata=True
+        )
+        self.assertEqual(res["status"], "trusted_empty")
+        self.assertTrue(res["corroborated"])
+
+
+class TestCapabilityEvidence(unittest.TestCase):
+    """#179 gap 1: capability claims need exact evidence."""
+
+    def test_evidence_backed_claims_are_proven(self):
+        result = _good_capability_evidence()
+        self.assertTrue(result["proven"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_claim_without_evidence_source_is_not_proven(self):
+        result = assess_capability_evidence([
+            {"task": "review_pr", "allowed": True, "evidence_source": ""},
+        ])
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("evidence" in r.lower() for r in result["reasons"]))
+
+    def test_no_claims_at_all_fails_closed(self):
+        result = assess_capability_evidence([])
+        self.assertFalse(result["proven"])
+
+    def test_disallowed_task_is_not_proven(self):
+        result = assess_capability_evidence([
+            {
+                "task": "merge_pr",
+                "allowed": False,
+                "evidence_source": "gitea_resolve_task_capability output",
+            },
+        ])
+        self.assertFalse(result["proven"])
+
+
+class TestSweepEvidence(unittest.TestCase):
+    """#179 gap 2: secret/provenance sweep must be exact."""
+
+    def test_exact_sweep_is_proven(self):
+        result = _good_sweep()
+        self.assertEqual(result["verdict"], "exact")
+        self.assertTrue(result["proven"])
+
+    def test_vague_sweep_without_command_is_downgraded(self):
+        result = _good_sweep(command="")
+        self.assertEqual(result["verdict"], "vague")
+        self.assertFalse(result["proven"])
+
+    def test_sweep_without_scope_is_downgraded(self):
+        result = _good_sweep(scope="")
+        self.assertEqual(result["verdict"], "vague")
+        self.assertFalse(result["proven"])
+
+    def test_missing_sweep_fails_closed(self):
+        result = assess_sweep_evidence(None)
+        self.assertEqual(result["verdict"], "missing")
+        self.assertFalse(result["proven"])
+
+    def test_unstated_result_is_downgraded(self):
+        result = _good_sweep(clean=None)
+        self.assertFalse(result["proven"])
+
+
+class TestLiveStateRecheck(unittest.TestCase):
+    """#179 gap 3: explicit pre-mutation live-state recheck."""
+
+    def test_clean_recheck_is_proven(self):
+        result = _good_live_state()
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_missing_recheck_fails_closed(self):
+        result = assess_live_state_recheck(None)
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_closed_pr_blocks(self):
+        result = _good_live_state(pr_state="closed")
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_moved_head_blocks(self):
+        result = _good_live_state(live_head_sha=OTHER)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("head" in r.lower() for r in result["reasons"]))
+
+    def test_changed_base_blocks(self):
+        result = _good_live_state(live_base_ref="develop")
+        self.assertFalse(result["proven"])
+
+    def test_unresolved_blocking_reviews_block(self):
+        result = _good_live_state(blocking_change_requests=True)
+        self.assertFalse(result["proven"])
+
+    def test_unchecked_blocking_state_fails_closed(self):
+        result = _good_live_state(blocking_change_requests=None)
+        self.assertFalse(result["proven"])
+
+
+class TestRoleBoundary179(unittest.TestCase):
+    """#179 gap 4: reviewer flows avoid unjustified author-namespace use."""
+
+    def test_native_namespace_only_is_clean(self):
+        result = _good_role_boundary_179()
+        self.assertTrue(result["proven"])
+
+    def test_foreign_namespace_without_justification_is_downgraded(self):
+        result = _good_role_boundary_179(
+            namespaces_used=["gitea-reviewer", "gitea-author"]
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("justif" in r.lower() for r in result["reasons"]))
+
+    def test_foreign_namespace_with_justification_is_clean(self):
+        result = _good_role_boundary_179(
+            namespaces_used=["gitea-reviewer", "gitea-author"],
+            justification=(
+                "author namespace read-only whoami used to evidence "
+                "self-review contamination status"
+            ),
+        )
+        self.assertTrue(result["proven"])
+
+    def test_unreported_namespaces_fail_closed(self):
+        result = _good_role_boundary_179(namespaces_used=None)
+        self.assertFalse(result["proven"])
+
+
+class TestFinalReport179Bar(unittest.TestCase):
+    """#179 acceptance adds capability, sweep, live-state, and role proofs."""
+
+    def _report(self, **overrides):
+        kwargs = {
+            "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(),
+        }
+        kwargs.update(overrides)
+        return build_final_report(**kwargs)
+
+    def test_all_179_proofs_present_is_grade_a(self):
+        report = self._report()
+        self.assertEqual(report["grade"], "A")
+        self.assertTrue(report["capability_evidence_proven"])
+        self.assertEqual(report["sweep_verdict"], "exact")
+        self.assertTrue(report["live_state_recheck_proven"])
+        self.assertTrue(report["role_boundary_clean"])
+
+    def test_missing_capability_evidence_downgrades(self):
+        report = self._report(capability_evidence=None)
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["capability_evidence_proven"])
+
+    def test_unevidenced_capability_claim_downgrades(self):
+        report = self._report(
+            capability_evidence=assess_capability_evidence([
+                {"task": "review_pr", "allowed": True, "evidence_source": ""},
+            ])
+        )
+        self.assertNotEqual(report["grade"], "A")
+
+    def test_vague_sweep_downgrades(self):
+        report = self._report(sweep=_good_sweep(command=""))
+        self.assertNotEqual(report["grade"], "A")
+        self.assertEqual(report["sweep_verdict"], "vague")
+
+    def test_missing_sweep_downgrades(self):
+        report = self._report(sweep=None)
+        self.assertNotEqual(report["grade"], "A")
+
+    def test_missing_live_state_recheck_downgrades_and_blocks_merge(self):
+        report = self._report(live_state=None)
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["merge_allowed"])
+        self.assertFalse(report["live_state_recheck_proven"])
+
+    def test_stale_live_state_blocks_merge(self):
+        report = self._report(live_state=_good_live_state(live_head_sha=OTHER))
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["merge_allowed"])
+
+    def test_merge_claim_without_live_recheck_is_a_violation(self):
+        report = self._report(live_state=None, merge_performed=True)
+        self.assertEqual(report["grade"], "blocked")
+        self.assertTrue(report["violations"])
+
+    def test_unjustified_author_namespace_downgrades(self):
+        report = self._report(
+            role_boundary=_good_role_boundary_179(
+                namespaces_used=["gitea-reviewer", "gitea-author"]
+            )
+        )
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["role_boundary_clean"])
+
+    def test_positive_baseline_from_173_still_holds(self):
+        report = self._report()
+        self.assertTrue(report["inventory_complete"])
+        self.assertTrue(report["validated_on_pinned_head"])
+
+
 if __name__ == "__main__":
     unittest.main()