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

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")),

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_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",
@@ -1324,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()