Merge pull request 'feat(reviewer): fail closed on dirty worktrees and forbidden git cleanup (closes #233)' (#234) from feat/issue-233-reviewer-worktree-gate into master

Add reviewer_worktree proofs that block stash/reset/checkout -- cleanup of
unrelated local files, require scratch worktree reporting when the main tree
is dirty outside PR scope, and integrate the gate into final report grading
and Controller Handoff review fields.

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

review_proofs.py

@@ -17,6 +17,7 @@ here weakens or replaces them.
 import re
 
 import issue_duplicate_gate
+from reviewer_worktree import assess_reviewer_worktree_proof
 
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
 
@@ -637,7 +638,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
                        role_boundary=None, review_mutation=None,
                        report_text=None, review_decision_lock=None,
                        controller_handoff=None, capability_proof=None,
-                       sweep_proof=None):
+                       sweep_proof=None, worktree_proof=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
 
     Combines the individual proof verdicts into the final-report fields the
@@ -707,6 +708,14 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "downgraded": True,
         "reasons": ["review mutation proof not provided (#211)"],
     }
+    if worktree_proof is not None:
+        worktree = assess_reviewer_worktree_proof(worktree_proof)
+    else:
+        worktree = {
+            "proven": False,
+            "block": True,
+            "reasons": ["reviewer worktree proof not provided (#233)"],
+        }
 
     capability_proven = bool(capability_evidence.get("proven"))
     sweep_proven = bool(sweep.get("proven"))
@@ -719,6 +728,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "reasons": ["review mutation proof missing"],
     }
     review_mutation_complete = bool(review_mutation.get("complete"))
+    worktree_proven = bool(worktree.get("proven"))
 
     downgrade_reasons = []
     if not identity_eligible:
@@ -772,6 +782,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
     if not review_mutation_complete:
         downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
         downgrade_reasons.extend(review_mutation.get("reasons", []))
+    if not worktree_proven:
+        downgrade_reasons.append(
+            "reviewer worktree safety proof missing or failed (#233)"
+        )
+        downgrade_reasons.extend(worktree.get("reasons", []))
 
     merge_allowed = (
         identity_eligible
@@ -782,6 +797,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         and validation.get("verdict") != "invalid"
         # #179: no merge without a proven final live-state recheck.
         and live_state_proven
+        and worktree_proven
     )
 
     violations = []
@@ -825,6 +841,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "live_state_recheck_proven": live_state_proven,
         "role_boundary_clean": role_boundary_clean,
         "review_mutation_complete": review_mutation_complete,
+        "worktree_proof_proven": worktree_proven,
+        "worktree_scratch_used": bool(worktree.get("scratch_used")),
+        "unrelated_mutations_avoided": bool(
+            worktree.get("unrelated_mutations_avoided")
+        ),
     }
 
 
@@ -971,6 +992,12 @@ HANDOFF_ROLE_FIELDS = {
         ("Selected PR", ("selected pr",)),
         ("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
         ("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
+        ("Worktree path", ("worktree path", "starting worktree path")),
+        ("Worktree dirty", ("worktree dirty", "whether worktree was dirty")),
+        ("Scratch worktree used", ("scratch worktree used", "scratch clone used",
+                                   "scratch worktree")),
+        ("Unrelated local mutations", ("unrelated local mutations",
+                                       "unrelated files modified")),
         ("Review decision", ("review decision", "decision")),
         ("Merge result", ("merge result",)),
         ("Linked issue status", ("linked issue status", "linked issue")),
@@ -978,6 +1005,7 @@ HANDOFF_ROLE_FIELDS = {
     ),
     "author": (
         ("Selected issue", ("selected issue",)),
+        ("Issue lock proof", ("issue lock proof", "lock before diff")),
         ("Claim/comment status", ("claim/comment status", "claim status",
                                   "claim")),
         ("PR number opened", ("pr number opened", "pr opened", "pr number")),

reviewer_worktree.py

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

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

@@ -20,6 +20,15 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
 
 Rules (llm-project-workflow):
 - Review in a SEPARATE detached review worktree, never the author's folder.
+- Worktree safety (#233): before checkout, diff, validation, review, or merge,
+  report the starting worktree path and whether it was dirty. If unrelated
+  tracked files exist outside the PR scope, STOP or run
+  `scripts/worktree-review <pr-head-branch>` and validate in the scratch path.
+  NEVER run `git stash`, `git stash pop/drop`, `git checkout --`, `git reset`,
+  or `git clean` to manage another session's dirty files.
+- Final report must state: Worktree path, Worktree dirty (yes/no),
+  Scratch worktree used (yes/no + path if yes), and confirm no unrelated local
+  files were modified, stashed, reset, or dropped.
 - You must NOT be the PR author. If the authenticated user == PR author, stop.
   A different LLM-Agent-SHA does NOT make you a different actor — only a
   different authenticated Gitea user does (docs/llm-agent-sha.md).

tests/test_mcp_server.py

@@ -65,6 +65,20 @@ CREATE_PR_ENV = {
     ),
 }
 
+ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
+
+
+def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
+    record = {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "remote": "dadeschools",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+    }
+    record.update(overrides)
+    return record
+
 
 # ---------------------------------------------------------------------------
 # Create Issue
@@ -127,15 +141,19 @@ class TestCreatePR(unittest.TestCase):
     @patch("os.path.exists", return_value=True)
     @patch("builtins.open")
     def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
-        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
+        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
+        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertEqual(result["number"], 3)
         self.assertNotIn("url", result)
+        mock_exists.assert_called_with(ISSUE_LOCK_FILE)
+        mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
         payload = mock_api.call_args[0][3]
         self.assertEqual(payload["head"], "feat/x")
         self.assertEqual(payload["base"], "main")
+        self.assertIn("Closes #123", payload["title"])
 
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            return_value=(True, []))
@@ -144,13 +162,28 @@ class TestCreatePR(unittest.TestCase):
     @patch("os.path.exists", return_value=True)
     @patch("builtins.open")
     def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
-        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
+        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
+        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
         with patch.dict(os.environ, env, clear=True):
             result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertIn("pulls/3", result["url"])
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    @patch("os.path.exists", return_value=True)
+    @patch("builtins.open")
+    def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
+        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
+        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            with self.assertRaises(ValueError) as ctx:
+                gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
+        self.assertIn("Closes #123", str(ctx.exception))
+        mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
+
 
 # ---------------------------------------------------------------------------
 # Close Issue
@@ -2770,8 +2803,8 @@ class TestIssueLocking(unittest.TestCase):
     """Test issue locking and PR gating constraints."""
 
     def tearDown(self):
-        if os.path.exists("/tmp/gitea_issue_lock.json"):
-            os.remove("/tmp/gitea_issue_lock.json")
+        if os.path.exists(ISSUE_LOCK_FILE):
+            os.remove(ISSUE_LOCK_FILE)
 
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -2779,7 +2812,7 @@ class TestIssueLocking(unittest.TestCase):
         mock_api.return_value = [] # no open PRs
         res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertTrue(res["success"])
-        self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
+        self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
 
     def test_lock_issue_mismatch_branch_fails(self):
         with self.assertRaises(ValueError) as ctx:
@@ -2816,8 +2849,8 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_missing_lock_fails(self, _auth, _role):
-        if os.path.exists("/tmp/gitea_issue_lock.json"):
-            os.remove("/tmp/gitea_issue_lock.json")
+        if os.path.exists(ISSUE_LOCK_FILE):
+            os.remove(ISSUE_LOCK_FILE)
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(RuntimeError) as ctx:
                 gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
@@ -2827,8 +2860,9 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_branch_mismatch_fails(self, _auth, _role):
-        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump(_sample_issue_lock(
+                issue_number=196, branch_name="feat/issue-196-mutations"), f)
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(ValueError) as ctx:
                 gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
@@ -2838,8 +2872,9 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_forbidden_terms_fails(self, _auth, _role):
-        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump(_sample_issue_lock(
+                issue_number=196, branch_name="feat/issue-196-mutations"), f)
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             for term in ("equivalent to #196", "related to #196", "same as #196"):
                 with self.assertRaises(ValueError) as ctx:
@@ -2850,8 +2885,9 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
-        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump(_sample_issue_lock(
+                issue_number=196, branch_name="feat/issue-196-mutations"), f)
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(ValueError) as ctx:
                 gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")

tests/test_review_proofs.py

@@ -183,6 +183,24 @@ def _good_review_mutation():
     return assess_review_mutation_final_report(report, lock)
 
 
+def _good_worktree(**overrides):
+    proof = {
+        "worktree_path": "/repo/branches/review-feat-issue-224",
+        "porcelain_status": "",
+        "pr_scope_files": ["docs/wiki/Repositories.md"],
+        "scratch_used": True,
+        "scratch_path": "/repo/branches/review-feat-issue-224",
+        "git_commands": [
+            "git fetch prgs master feat/issue-224-wiki-proof-refresh",
+            "git diff prgs/master...prgs/feat/issue-224-wiki-proof-refresh",
+        ],
+    }
+    proof.update(overrides)
+    from reviewer_worktree import assess_reviewer_worktree_proof  # noqa: E402
+
+    return assess_reviewer_worktree_proof(proof)
+
+
 def _good_role_boundary_179(**overrides):
     kwargs = {
         "task_role": "reviewer",
@@ -595,6 +613,13 @@ class TestFinalReport(unittest.TestCase):
             "controller_handoff": _good_handoff(),
             "capability_proof": _good_capability_proof(),
             "sweep_proof": _good_secret_sweep(),
+            "worktree_proof": {
+                "worktree_path": "/repo/branches/review-feat-issue-224",
+                "porcelain_status": "",
+                "pr_scope_files": ["docs/wiki/Repositories.md"],
+                "scratch_used": True,
+                "scratch_path": "/repo/branches/review-feat-issue-224",
+            },
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -899,12 +924,17 @@ class TestControllerHandoff(unittest.TestCase):
         result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
         self.assertEqual(result["verdict"], "incomplete")
         self.assertIn("Pinned reviewed head", result["missing_fields"])
+        self.assertIn("Worktree path", result["missing_fields"])
         self.assertIn("Merge result", result["missing_fields"])
 
         complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected PR: #999",
             "- Reviewer eligibility: passed",
             "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+            "- Worktree path: /repo/branches/review-pr-999",
+            "- Worktree dirty: no",
+            "- Scratch worktree used: yes (/repo/branches/review-pr-999)",
+            "- Unrelated local mutations: none",
             "- Review decision: approve",
             "- Merge result: merged",
             "- Linked issue status: closed",
@@ -913,24 +943,46 @@ class TestControllerHandoff(unittest.TestCase):
         result = assess_controller_handoff(complete, role="review")
         self.assertEqual(result["verdict"], "complete")
 
-    def test_author_role_requires_author_fields(self):
-        complete = self.BASE_HANDOFF + "\n" + "\n".join([
-            "- Selected issue: #182",
+    def _author_role_fields(self, issue_number=182, pr_number=999):
+        return [
+            f"- Selected issue: #{issue_number}",
+            "- Issue lock proof: lock before diff on feat/x @ master",
             "- Claim/comment status: comment-claimed",
-            "- PR number opened: #999",
+            f"- PR number opened: #{pr_number}",
             "- No review/merge: confirmed",
-        ])
+        ]
+
+    def test_handoff_role_fields_author_includes_issue_lock_proof(self):
+        from review_proofs import HANDOFF_ROLE_FIELDS
+        names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]]
+        self.assertIn("Issue lock proof", names)
+
+    def test_author_role_requires_author_fields(self):
+        complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields())
         result = assess_controller_handoff(complete, role="author")
         self.assertEqual(result["verdict"], "complete")
 
         result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
         self.assertEqual(result["verdict"], "incomplete")
+        self.assertIn("Issue lock proof", result["missing_fields"])
         self.assertIn("No review/merge confirmation", result["missing_fields"])
 
+    def test_author_role_requires_issue_lock_proof(self):
+        without_lock = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: #182",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #999",
+            "- No review/merge: confirmed",
+        ])
+        result = assess_controller_handoff(without_lock, role="author")
+        self.assertEqual(result["verdict"], "incomplete")
+        self.assertIn("Issue lock proof", result["missing_fields"])
+
     def test_author_role_rejects_equivalent_or_multiple_issues(self):
         # 1. equivalent reference blocked
         incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected issue: Issue #194 / #196 equivalent",
+            "- Issue lock proof: lock before diff on feat/x @ master",
             "- Claim/comment status: comment-claimed",
             "- PR number opened: #999",
             "- No review/merge: confirmed",
@@ -942,6 +994,7 @@ class TestControllerHandoff(unittest.TestCase):
         # 2. multiple issues blocked
         incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected issue: #194, #196",
+            "- Issue lock proof: lock before diff on feat/x @ master",
             "- Claim/comment status: comment-claimed",
             "- PR number opened: #999",
             "- No review/merge: confirmed",
@@ -953,6 +1006,7 @@ class TestControllerHandoff(unittest.TestCase):
     def test_author_role_rejects_fuzzy_pr_number(self):
         incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected issue: #196",
+            "- Issue lock proof: lock before diff on feat/issue-196 @ master",
             "- Claim/comment status: comment-claimed",
             "- PR number opened: PR #203 / #204 equivalent",
             "- No review/merge: confirmed",
@@ -984,23 +1038,17 @@ class TestControllerHandoff(unittest.TestCase):
 
     def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
         # 1. Workspace mutations: none is rejected when local_edits is True
-        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
-            "- Selected issue: #196",
-            "- Claim/comment status: comment-claimed",
-            "- PR number opened: #203",
-            "- No review/merge: confirmed",
-        ])
+        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
+            self._author_role_fields(issue_number=196, pr_number=203))
         res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
         self.assertEqual(res["verdict"], "incomplete")
         self.assertIn("Workspace mutations", res["missing_fields"])
 
         # 2. Workspace mutations: edited files is allowed when local_edits is True
-        complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
-            "- Selected issue: #196",
-            "- Claim/comment status: comment-claimed",
-            "- PR number opened: #203",
-            "- No review/merge: confirmed",
-        ])
+        complete_eq = self.BASE_HANDOFF.replace(
+            "- Workspace mutations: none",
+            "- Workspace mutations: edited review_proofs.py",
+        ) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
         res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
         self.assertEqual(res2["verdict"], "complete")
 
@@ -1306,6 +1354,13 @@ class TestFinalReport179Bar(unittest.TestCase):
             "controller_handoff": _good_handoff(),
             "capability_proof": _good_capability_proof(),
             "sweep_proof": _good_secret_sweep(),
+            "worktree_proof": {
+                "worktree_path": "/repo/branches/review-feat-issue-224",
+                "porcelain_status": "",
+                "pr_scope_files": ["docs/wiki/Repositories.md"],
+                "scratch_used": True,
+                "scratch_path": "/repo/branches/review-feat-issue-224",
+            },
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)

tests/test_reviewer_worktree.py

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