feat: block reviewer local Gitea fallbacks during normal review (Closes #324)

Merge prgs/master into feat/issue-330-proof-wording-enforcement and keep
both assess_proof_wording (#330) and git ref mutation reporting (#297).

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

review_proofs.py

@@ -1052,6 +1052,125 @@ def _prs_from_list_response(list_prs_response: list | dict | None) -> list | Non
     return None
 
 
+# ── Workspace-vs-worktree mutation consistency (Issue #313) ──────────────────
+#
+# A worktree reset/checkout/clean is a workspace mutation even when the
+# reviewer edited no files by hand. Final reports must therefore never say
+# "Workspace mutations: none" once a worktree mutation occurred, and every
+# observed worktree / git ref mutation must appear under its precise
+# category field.
+
+WORKTREE_MUTATION_MARKERS = (
+    "reset",
+    "checkout",
+    "switch",
+    "restore",
+    "clean",
+    "stash",
+    "merge",
+    "rebase",
+    "cherry-pick",
+    "worktree add",
+    "worktree remove",
+    "worktree prune",
+)
+
+GIT_REF_MUTATION_MARKERS = (
+    "fetch",
+    "pull",
+)
+
+
+def _report_labeled_fields(report_text):
+    """Return {lowercase label: value} for every 'Label: value' line."""
+    fields = {}
+    for line in (report_text or "").splitlines():
+        stripped = line.strip().lstrip("-*").strip()
+        if ":" in stripped:
+            k, v = stripped.split(":", 1)
+            fields[k.strip().lower()] = v.strip()
+    return fields
+
+
+def _field_claims_none(fields, label):
+    """True when *label* is present and its value is empty or 'none...'."""
+    value = fields.get(label)
+    if value is None:
+        return False
+    value = value.strip().lower()
+    return not value or value.startswith("none")
+
+
+def _classify_git_commands(observed_commands):
+    """Split observed git commands into worktree and ref mutations."""
+    worktree_cmds = []
+    ref_cmds = []
+    for cmd in observed_commands or []:
+        lowered = " ".join((cmd or "").lower().split())
+        if "git" not in lowered:
+            continue
+        if any(marker in lowered for marker in WORKTREE_MUTATION_MARKERS):
+            worktree_cmds.append(cmd)
+        elif any(marker in lowered for marker in GIT_REF_MUTATION_MARKERS):
+            ref_cmds.append(cmd)
+    return worktree_cmds, ref_cmds
+
+
+def assess_workspace_mutation_consistency(report_text, observed_commands=None):
+    """#313: reject 'Workspace mutations: none' after worktree mutations.
+
+    *observed_commands* is the optional list of git commands the workflow
+    actually ran. Even without it, a report that itself lists a non-none
+    ``Worktree mutations`` value while claiming ``Workspace mutations:
+    none`` is internally contradictory and fails.
+
+    Returns {'complete', 'downgraded', 'reasons'}; fails closed on any
+    contradiction or unreported observed mutation.
+    """
+    fields = _report_labeled_fields(report_text)
+    worktree_cmds, ref_cmds = _classify_git_commands(observed_commands)
+
+    reported_worktree = fields.get("worktree mutations")
+    reported_worktree_nonnone = bool(
+        reported_worktree and not reported_worktree.strip().lower().startswith("none")
+    )
+
+    reasons = []
+
+    workspace_none = _field_claims_none(fields, "workspace mutations")
+    if workspace_none and (worktree_cmds or reported_worktree_nonnone):
+        detail = worktree_cmds[0] if worktree_cmds else reported_worktree
+        reasons.append(
+            "report claims 'Workspace mutations: none' but worktree "
+            f"mutations occurred ({detail}); use precise categories such as "
+            "'File edits by reviewer: none' plus 'Worktree mutations: <cmd>'"
+        )
+
+    if worktree_cmds and (
+        "worktree mutations" not in fields
+        or _field_claims_none(fields, "worktree mutations")
+    ):
+        reasons.append(
+            "observed worktree mutation not reported under 'Worktree "
+            f"mutations': {worktree_cmds[0]}"
+        )
+
+    if ref_cmds and (
+        "git ref mutations" not in fields
+        or _field_claims_none(fields, "git ref mutations")
+    ):
+        reasons.append(
+            "observed git ref mutation not reported under 'Git ref "
+            f"mutations': {ref_cmds[0]}"
+        )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "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.
@@ -3325,6 +3444,13 @@ def assess_email_disclosure(
     }
 
 
+def assess_reviewer_fallback_report(report_text, **kwargs):
+    """#324: block local Gitea fallbacks during normal reviewer workflows."""
+    from reviewer_fallback import assess_reviewer_fallback_report as _assess
+
+    return _assess(report_text, **kwargs)
+
+
 _QUEUE_STATUS_REPORT_HINT = re.compile(
     r"queue[- ]status|selected pr:\s*none|no pr selected|"
     r"queue[- ]status[- ]only",
@@ -3384,6 +3510,17 @@ _PROOF_WORDING_RULES: tuple[dict, ...] = (
         "text_evidence": _PAGINATION_FINALITY_EVIDENCE,
         "allow_prior_label": False,
     },
+    {
+        "id": "same_as_master",
+        "pattern": re.compile(r"\bsame as master\b", re.I),
+        "session_keys": ("baseline_worktree_proof", "clean_baseline_proof"),
+        "text_evidence": re.compile(
+            r"baseline worktree|clean baseline|diff base.*master|"
+            r"base[- ]equivalent.*master|worktree proof",
+            re.I,
+        ),
+        "allow_prior_label": False,
+    },
     {
         "id": "no_file_edits",
         "pattern": re.compile(
@@ -3449,7 +3586,12 @@ def assess_proof_wording(
     *,
     session_evidence: dict | None = None,
 ) -> dict:
-    """Issue #330: reject unsupported proof phrases in final reports."""
+    """Issue #330: reject unsupported proof phrases in final reports.
+
+    Forbidden wording such as ``inventory complete``, ``live proof``, and
+    ``same as master`` is allowed only when matching current-session evidence
+    appears in the report text or in *session_evidence*.
+    """
     text = report_text or ""
     evidence = dict(session_evidence or {})
     violations: list[str] = []

reviewer_fallback.py

@@ -0,0 +1,213 @@
+"""Reviewer local-fallback detection for normal PR review workflows (#324).
+
+Normal reviewer runs must use MCP tools. Reading profile secret files or
+running local Gitea helper scripts while MCP is available is a fail-closed
+violation. Explicit recovery mode may use local fallback only with full proof.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+LOCAL_GITEA_SCRIPT_NAMES = (
+    "list_prs.py",
+    "view_pr.py",
+    "create_pr.py",
+    "merge_pr.py",
+    "edit_pr.py",
+    "list_issues.py",
+    "delete_branch.py",
+    "create_issue.py",
+    "close_issue.py",
+    "review_pr.py",
+    "mark_issue.py",
+    "mirror_refs.sh",
+)
+
+PROFILE_SECRET_MARKERS = (
+    "profiles.json",
+    ".config/gitea-tools/profiles",
+    "gitea_auth.py",
+    "gitea_config.py",
+    "keychain",
+    "credential fill",
+    "token store",
+)
+
+_RECOVERY_MODE_RE = re.compile(
+    r"\b(?:recovery mode|explicit recovery|mcp unavailable|mcp not available|"
+    r"mcp tools unavailable|no mcp path)\b",
+    re.IGNORECASE,
+)
+_FALLBACK_CLASSIFICATION_RE = re.compile(
+    r"\b(?:local fallback|fallback mode|used local gitea|ran local script)\b",
+    re.IGNORECASE,
+)
+_MCP_TOOL_USE_RE = re.compile(
+    r"\b(?:gitea_list_prs|gitea_view_pr|gitea_review_pr|gitea_merge_pr|"
+    r"gitea_resolve_task_capability|gitea_whoami|mcp tool)\b",
+    re.IGNORECASE,
+)
+_LOCAL_SCRIPT_RE = re.compile(
+    r"(?:^|[\s\"'`/])(?:" + "|".join(re.escape(name) for name in LOCAL_GITEA_SCRIPT_NAMES) + r")",
+    re.IGNORECASE,
+)
+_PROFILE_ACCESS_RE = re.compile(
+    r"(?:read|open|inspect|cat|view|access(?:ed)?|loaded?)\s+(?:file\s+)?[`'\"]?"
+    r"[^`'\"]*profiles\.json|profiles\.json[`'\"]?\s+(?:read|opened|inspected|accessed)|"
+    r"~/?\.config/gitea-tools/profiles",
+    re.IGNORECASE,
+)
+_LOCAL_ENV_SCRIPT_RE = re.compile(
+    r"GITEA_MCP_(?:CONFIG|PROFILE)=.*\b(?:python\s+)?(?:list_prs|view_pr|create_pr|merge_pr)\.py",
+    re.IGNORECASE,
+)
+_RECOVERY_PROOF_FIELDS = (
+    ("identity proof", ("identity proof", "exact identity proof")),
+    ("profile proof", ("profile proof", "exact profile proof")),
+    ("repo proof", ("repo proof", "exact repo proof")),
+    ("capability proof", ("capability proof", "exact capability proof")),
+    ("mcp unavailable reason", (
+        "why mcp was unavailable",
+        "mcp unavailable",
+        "mcp unavailability",
+    )),
+)
+
+
+def _collect_observed_text(
+    report_text: str,
+    action_log: list[dict] | None,
+) -> str:
+    chunks = [report_text or ""]
+    for entry in action_log or []:
+        for key in ("command", "action", "detail", "path", "script"):
+            value = entry.get(key)
+            if value:
+                chunks.append(str(value))
+    return "\n".join(chunks)
+
+
+def _detect_profile_secret_access(text: str) -> list[str]:
+    reasons = []
+    lower = text.lower()
+    if _PROFILE_ACCESS_RE.search(text):
+        reasons.append("report or action log shows profiles.json/profile secret access")
+    for marker in PROFILE_SECRET_MARKERS:
+        if marker == "profiles.json":
+            continue
+        if marker in lower and "do not" not in lower and "must not" not in lower:
+            if any(verb in lower for verb in ("read ", "open ", "inspect ", "cat ", "loaded ")):
+                reasons.append(f"profile secret surface '{marker}' accessed during review")
+    return reasons
+
+
+def _detect_local_script_use(text: str) -> list[str]:
+    reasons = []
+    match = _LOCAL_SCRIPT_RE.search(text)
+    if match:
+        reasons.append(
+            f"local Gitea helper script invoked ({match.group(0).strip()})"
+        )
+    if _LOCAL_ENV_SCRIPT_RE.search(text):
+        reasons.append(
+            "local Gitea script run with GITEA_MCP_CONFIG/GITEA_MCP_PROFILE env overrides"
+        )
+    return reasons
+
+
+def _recovery_proof_missing(text: str) -> list[str]:
+    lower = text.lower()
+    missing = []
+    for label, aliases in _RECOVERY_PROOF_FIELDS:
+        if not any(alias in lower for alias in aliases):
+            missing.append(label)
+    if not _FALLBACK_CLASSIFICATION_RE.search(text):
+        missing.append("fallback classification")
+    if not _RECOVERY_MODE_RE.search(text):
+        missing.append("recovery mode declaration")
+    return missing
+
+
+def assess_reviewer_fallback_report(
+    report_text: str,
+    *,
+    action_log: list[dict] | None = None,
+    recovery_mode: bool = False,
+    mcp_available: bool = True,
+    mcp_tools_used: bool | None = None,
+) -> dict[str, Any]:
+    """Fail closed when normal reviewer workflows use local Gitea fallbacks (#324)."""
+    text = _collect_observed_text(report_text, action_log)
+    lower = (report_text or "").lower()
+
+    profile_violations = _detect_profile_secret_access(text)
+    script_violations = _detect_local_script_use(text)
+    violations = profile_violations + script_violations
+
+    if mcp_tools_used is None:
+        mcp_tools_used = bool(_MCP_TOOL_USE_RE.search(report_text or ""))
+    elif mcp_tools_used is False and _MCP_TOOL_USE_RE.search(report_text or ""):
+        mcp_tools_used = True
+
+    reasons: list[str] = []
+    recovery_declared = recovery_mode or bool(_RECOVERY_MODE_RE.search(report_text or ""))
+
+    if violations and mcp_available and not recovery_declared:
+        reasons.extend(violations)
+        if mcp_tools_used:
+            reasons.append(
+                "MCP tools were available and used; local fallback/profile access is forbidden"
+            )
+        else:
+            reasons.append(
+                "MCP path is available; normal review must not use local Gitea fallbacks"
+            )
+
+    if violations and recovery_declared:
+        missing = _recovery_proof_missing(report_text or "")
+        if missing:
+            reasons.append(
+                "recovery-mode fallback missing proof fields: "
+                + ", ".join(missing)
+            )
+
+    if (
+        not violations
+        and recovery_declared
+        and _FALLBACK_CLASSIFICATION_RE.search(report_text or "")
+        and not recovery_mode
+    ):
+        missing = _recovery_proof_missing(report_text or "")
+        if missing:
+            reasons.append(
+                "report claims local fallback but recovery proof is incomplete: "
+                + ", ".join(missing)
+            )
+
+    proven = not reasons
+    blocked = bool(reasons) and not recovery_declared or any(
+        "recovery-mode fallback missing" in r or "recovery proof is incomplete" in r
+        for r in reasons
+    )
+    return {
+        "proven": proven,
+        "block": blocked or (bool(reasons) and mcp_available and not recovery_declared),
+        "downgraded": bool(reasons) and not blocked,
+        "recovery_mode": recovery_declared,
+        "mcp_available": mcp_available,
+        "mcp_tools_used": mcp_tools_used,
+        "profile_violations": profile_violations,
+        "script_violations": script_violations,
+        "reasons": reasons,
+        "safe_next_action": (
+            "use MCP reviewer tools only; do not read profiles.json or run local Gitea scripts"
+            if reasons and not recovery_declared
+            else (
+                "complete recovery-mode fallback proof before claiming local fallback"
+                if reasons
+                else "proceed with MCP tools"
+            )
+        ),
+    }

tests/test_llm_workflow_split.py

@@ -118,6 +118,12 @@ def test_start_issue_template_references_work_issue_workflow():
     assert "workflows/work-issue.md" in text
 
 
+def test_reviewer_fallback_verifier_exported():
+    from review_proofs import assess_reviewer_fallback_report
+
+    assert callable(assess_reviewer_fallback_report)
+
+
 def test_create_issue_final_report_verifier_exported():
     from review_proofs import assess_create_issue_final_report
 

tests/test_proof_wording.py

@@ -0,0 +1,93 @@
+"""Doc-contract checks for proof wording enforcement (#330)."""
+import unittest
+
+from review_proofs import assess_proof_wording, build_final_report
+
+
+class TestProofWording(unittest.TestCase):
+    def test_rejects_inventory_complete_without_pagination_proof(self):
+        result = assess_proof_wording(
+            "Open PR inventory is complete because only 3 PRs were returned."
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_rejects_default_page_size_assumption(self):
+        result = assess_proof_wording(
+            "Inventory exhaustive: fewer than the default Gitea page size returned."
+        )
+        self.assertFalse(result["proven"])
+        self.assertIn("default_page_size_assumption", result["flagged_rules"])
+
+    def test_allows_inventory_complete_with_finality_metadata(self):
+        result = assess_proof_wording(
+            "Inventory complete. pagination_complete: true, is_final_page: true"
+        )
+        self.assertTrue(result["proven"])
+
+    def test_allows_inventory_complete_with_session_evidence(self):
+        result = assess_proof_wording(
+            "Queue inventory complete.",
+            session_evidence={"pagination_complete": True},
+        )
+        self.assertTrue(result["proven"])
+
+    def test_rejects_live_proof_without_session_evidence(self):
+        result = assess_proof_wording("Live proof confirms mergeable state.")
+        self.assertFalse(result["proven"])
+        self.assertIn("live_proof", result["flagged_rules"])
+
+    def test_allows_prior_blocker_reuse_wording(self):
+        result = assess_proof_wording(
+            "Prior blocker reused; head SHA unchanged since blocking review. "
+            "Live proof not claimed for this skip."
+        )
+        self.assertTrue(result["proven"])
+
+    def test_allows_live_proof_with_current_session_evidence(self):
+        result = assess_proof_wording(
+            "Live proof: gitea_view_pr revalidated in the current session."
+        )
+        self.assertTrue(result["proven"])
+
+    def test_rejects_same_as_master_without_baseline_proof(self):
+        result = assess_proof_wording("Diff is same as master.")
+        self.assertFalse(result["proven"])
+        self.assertIn("same_as_master", result["flagged_rules"])
+
+    def test_allows_same_as_master_with_baseline_wording(self):
+        result = assess_proof_wording(
+            "Clean baseline worktree proof: diff base matches master."
+        )
+        self.assertTrue(result["proven"])
+
+    def test_rejects_file_edits_none_without_ledger(self):
+        result = assess_proof_wording("Workspace mutations: file edits none.")
+        self.assertFalse(result["proven"])
+        self.assertIn("no_file_edits", result["flagged_rules"])
+
+    def test_allows_file_edits_none_with_mutation_ledger(self):
+        result = assess_proof_wording(
+            "Mutation ledger: tracked file edits: none",
+            session_evidence={"mutation_ledger_clean": True},
+        )
+        self.assertTrue(result["proven"])
+
+    def test_build_final_report_downgrades_on_proof_wording_violation(self):
+        report = build_final_report(
+            {"proven": True},
+            {"complete": True},
+            {"claimable": True, "verdict": "strong"},
+            {"status": "clean"},
+            True,
+            False,
+            True,
+            report_text="Inventory complete after listing 4 open PRs.",
+        )
+        self.assertEqual(report["grade"], "downgraded")
+        self.assertFalse(report["proof_wording_proven"])
+        self.assertTrue(report["proof_wording_violations"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_fallback.py

@@ -0,0 +1,112 @@
+"""Tests for reviewer local-fallback verifier (#324)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_fallback import assess_reviewer_fallback_report  # noqa: E402
+
+
+class TestReviewerFallbackNormalMode(unittest.TestCase):
+    def test_clean_mcp_report_passes(self):
+        report = (
+            "Used gitea_list_prs and gitea_view_pr via MCP reviewer namespace.\n"
+            "Capability evidence: gitea_resolve_task_capability(review_pr)."
+        )
+        result = assess_reviewer_fallback_report(report, mcp_available=True)
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_profiles_json_access_blocks(self):
+        report = (
+            "Inspected ~/.config/gitea-tools/profiles.json for reviewer credentials.\n"
+            "Then used gitea_view_pr."
+        )
+        result = assess_reviewer_fallback_report(report, mcp_available=True)
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(any("profiles.json" in r for r in result["reasons"]))
+
+    def test_local_list_prs_script_blocks_when_mcp_available(self):
+        report = (
+            "Ran `python list_prs.py --remote prgs` with GITEA_MCP_CONFIG set.\n"
+            "MCP tools were also available."
+        )
+        result = assess_reviewer_fallback_report(
+            report,
+            mcp_available=True,
+            action_log=[{"command": "GITEA_MCP_PROFILE=prgs-reviewer python list_prs.py"}],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(result["script_violations"])
+
+    def test_action_log_profile_access_blocks(self):
+        result = assess_reviewer_fallback_report(
+            "Review complete via MCP.",
+            action_log=[{"path": "/Users/me/.config/gitea-tools/profiles.json", "action": "read"}],
+            mcp_available=True,
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+
+class TestReviewerFallbackRecoveryMode(unittest.TestCase):
+    def _recovery_report(self, **extra):
+        base = "\n".join([
+            "Recovery mode: MCP tools unavailable in this session.",
+            "Local fallback: ran python view_pr.py after capability proof.",
+            "Identity proof: sysadmin / prgs-reviewer",
+            "Profile proof: prgs-reviewer",
+            "Repo proof: Scaled-Tech-Consulting/Gitea-Tools",
+            "Capability proof: gitea.pr.review allowed for recovery command",
+            "Why MCP was unavailable: reviewer MCP namespace failed to start",
+        ])
+        return base + ("\n" + extra.get("append", "") if extra.get("append") else "")
+
+    def test_recovery_fallback_with_proof_passes(self):
+        result = assess_reviewer_fallback_report(
+            self._recovery_report(),
+            action_log=[{"command": "python view_pr.py --remote prgs"}],
+            mcp_available=False,
+            recovery_mode=True,
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_recovery_fallback_without_proof_blocks(self):
+        report = (
+            "Recovery mode engaged.\n"
+            "Local fallback: python list_prs.py\n"
+        )
+        result = assess_reviewer_fallback_report(
+            report,
+            mcp_available=False,
+            recovery_mode=True,
+            action_log=[{"command": "python list_prs.py"}],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("recovery" in r.lower() for r in result["reasons"])
+        )
+
+    def test_mcp_unavailable_allows_recovery_attempt(self):
+        result = assess_reviewer_fallback_report(
+            self._recovery_report(),
+            mcp_available=False,
+            recovery_mode=True,
+        )
+        self.assertTrue(result["proven"])
+
+
+class TestReviewerFallbackExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_reviewer_fallback_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_workspace_mutation_consistency.py

@@ -0,0 +1,165 @@
+"""Tests for workspace-vs-worktree mutation consistency verifier (#313).
+
+Final reports must not claim ``Workspace mutations: none`` when worktree
+mutations (reset --hard, checkout, clean, worktree add/remove, ...)
+occurred, and must report observed worktree / git ref mutations under the
+precise category fields.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import assess_workspace_mutation_consistency  # noqa: E402
+
+
+class TestWorkspaceNoneContradiction(unittest.TestCase):
+    def test_reset_hard_with_workspace_none_fails(self):
+        report = (
+            "Controller Handoff\n"
+            "- Workspace mutations: none (no local file edits)\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(result["downgraded"])
+        self.assertTrue(
+            any("workspace mutations" in r.lower() for r in result["reasons"])
+        )
+
+    def test_checkout_with_workspace_none_fails(self):
+        report = "- Workspace mutations: none\n"
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git checkout feat/some-branch"],
+        )
+        self.assertFalse(result["complete"])
+
+    def test_worktree_add_with_workspace_none_fails(self):
+        report = "- Workspace mutations: none\n"
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git worktree add /tmp/review-pr1 abc123"],
+        )
+        self.assertFalse(result["complete"])
+
+    def test_clean_with_workspace_none_fails(self):
+        report = "- Workspace mutations: none\n"
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git clean -fd"],
+        )
+        self.assertFalse(result["complete"])
+
+    def test_self_reported_worktree_mutation_contradicts_workspace_none(self):
+        # No observed command log; the report itself carries the
+        # contradiction (#313 observed behavior).
+        report = (
+            "- Worktree mutations: git reset --hard FETCH_HEAD\n"
+            "- Workspace mutations: none (no local file edits)\n"
+        )
+        result = assess_workspace_mutation_consistency(report)
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("workspace mutations" in r.lower() for r in result["reasons"])
+        )
+
+
+class TestPreciseCategoriesPass(unittest.TestCase):
+    def test_file_edits_none_with_worktree_mutation_listed_passes(self):
+        report = (
+            "- File edits by reviewer: none\n"
+            "- Worktree mutations: git reset --hard FETCH_HEAD\n"
+            "- Git ref mutations: git fetch prgs\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=[
+                "git fetch prgs",
+                "git reset --hard FETCH_HEAD",
+            ],
+        )
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["downgraded"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_noop_report_passes(self):
+        report = (
+            "- File edits by reviewer: none\n"
+            "- Worktree mutations: none\n"
+            "- Git ref mutations: none\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report, observed_commands=[]
+        )
+        self.assertTrue(result["complete"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_fetch_only_with_ref_mutation_reported_passes(self):
+        # Fetch is a git ref mutation, not a worktree mutation; a
+        # workspace-none claim is not contradicted by fetch alone.
+        report = (
+            "- Workspace mutations: none\n"
+            "- Git ref mutations: git fetch prgs master\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git fetch prgs master"],
+        )
+        self.assertTrue(result["complete"])
+
+
+class TestObservedMutationsMustBeReported(unittest.TestCase):
+    def test_reset_without_worktree_mutation_field_fails(self):
+        report = "- File edits by reviewer: none\n"
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("worktree mutation" in r.lower() for r in result["reasons"])
+        )
+
+    def test_reset_with_worktree_mutations_none_fails(self):
+        report = (
+            "- File edits by reviewer: none\n"
+            "- Worktree mutations: none\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["complete"])
+
+    def test_fetch_without_ref_mutation_field_fails(self):
+        report = (
+            "- File edits by reviewer: none\n"
+            "- Worktree mutations: none\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git fetch prgs master"],
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("git ref mutation" in r.lower() for r in result["reasons"])
+        )
+
+    def test_fetch_with_ref_mutations_none_fails(self):
+        report = (
+            "- Worktree mutations: none\n"
+            "- Git ref mutations: none\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git fetch prgs master"],
+        )
+        self.assertFalse(result["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()