From ce3bd784c026618a9765a92e03be43b8e68b7430 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:26:37 -0400 Subject: [PATCH] Add worktree ownership verifier for reviewer reset safety (#312) Require session-owned review worktrees or documented safe-reuse proof before destructive reset/validation. Block contradictory workspace-none claims when git reset --hard occurred. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 7 + reviewer_worktree_ownership.py | 207 ++++++++++++++++++++++ tests/test_llm_workflow_split.py | 8 +- tests/test_reviewer_worktree_ownership.py | 137 ++++++++++++++ 4 files changed, 358 insertions(+), 1 deletion(-) create mode 100644 reviewer_worktree_ownership.py create mode 100644 tests/test_reviewer_worktree_ownership.py diff --git a/review_proofs.py b/review_proofs.py index 60cca02..d6c6bc6 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3816,3 +3816,10 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs): from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess return _assess(report_text, **kwargs) + + +def assess_worktree_ownership_report(report_text, **kwargs): + """#312: prove session-owned or safe-reuse worktree before reset/validation.""" + from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_worktree_ownership.py b/reviewer_worktree_ownership.py new file mode 100644 index 0000000..d30c48d --- /dev/null +++ b/reviewer_worktree_ownership.py @@ -0,0 +1,207 @@ +"""Reviewer worktree ownership and safe-reuse proof verifier (#312).""" + +from __future__ import annotations + +import re +from typing import Any + +_SESSION_OWNED_RE = re.compile( + r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)", + re.IGNORECASE, +) +_WORKTREE_PATH_RE = re.compile( + r"(?:review worktree path|worktree path|session-owned worktree)\s*:\s*(\S+)", + re.IGNORECASE, +) +_BRANCHES_PATH_RE = re.compile(r"/branches/|\bbranches/", re.IGNORECASE) +_MAIN_CHECKOUT_RE = re.compile( + r"main checkout|not (?:the )?main checkout|outside branches/", + re.IGNORECASE, +) +_SAFE_REUSE_RE = re.compile(r"safe[- ]reuse proof", re.IGNORECASE) +_REUSE_POLICY_RE = re.compile( + r"(?:project policy|policy) (?:allows|allowing) (?:reuse|reset)", + re.IGNORECASE, +) +_CLEAN_TRACKED_RE = re.compile( + r"(?:clean tracked state|tracked state\s*:\s*clean|no uncommitted tracked)", + re.IGNORECASE, +) +_CLEAN_UNTRACKED_RE = re.compile( + r"(?:clean untracked state|untracked state\s*:\s*clean|no untracked files)", + re.IGNORECASE, +) +_NOT_OTHER_SESSION_RE = re.compile( + r"not owned by (?:another|other) (?:active )?(?:task|session)", + re.IGNORECASE, +) +_BRANCH_BEFORE_RESET_RE = re.compile( + r"(?:branch/head before reset|head before reset|branch before reset)\s*:\s*(\S+)", + re.IGNORECASE, +) +_RESET_TARGET_RE = re.compile( + r"(?:reset target sha|reset target)\s*:\s*([0-9a-f]{7,40})", + re.IGNORECASE, +) +_DESTRUCTIVE_CMD_RE = re.compile( + r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?" + r"(?:reset\s+--hard|clean(?:\s+-[A-Za-z]+)*|checkout\s+(?!HEAD\s+--)|switch\s+)", + re.IGNORECASE, +) +_WORKSPACE_NONE_RE = re.compile(r"workspace mutations\s*:\s*none", re.IGNORECASE) +_WORKTREE_MUTATION_RE = re.compile( + r"(?:worktree(?:/index)? mutations|destructive reset)\s*:\s*(?!none\b).+", + re.IGNORECASE, +) +_RESET_IN_REPORT_RE = re.compile(r"reset\s+--hard", re.IGNORECASE) + + +def _command_text(entry) -> str: + if isinstance(entry, dict): + return str(entry.get("command") or "").strip() + return str(entry or "").strip() + + +def _destructive_commands(command_log: list | None) -> list[str]: + return [ + cmd + for cmd in (_command_text(entry) for entry in (command_log or [])) + if cmd and _DESTRUCTIVE_CMD_RE.search(cmd) + ] + + +def _extract_worktree_path(text: str, session: dict) -> str: + path = (session.get("worktree_path") or "").strip() + if path: + return path + match = _WORKTREE_PATH_RE.search(text or "") + return match.group(1).strip() if match else "" + + +def _is_session_owned_path(path: str) -> bool: + normalized = (path or "").replace("\\", "/") + return bool(_SESSION_OWNED_RE.search(normalized)) + + +def _safe_reuse_fields_present(text: str) -> list[str]: + missing: list[str] = [] + if not _WORKTREE_PATH_RE.search(text): + missing.append("exact worktree path") + if not _BRANCHES_PATH_RE.search(text): + missing.append("worktree inside branches/") + if not _MAIN_CHECKOUT_RE.search(text): + missing.append("worktree is not the main checkout") + if not _NOT_OTHER_SESSION_RE.search(text): + missing.append("worktree not owned by another active session") + if not _CLEAN_TRACKED_RE.search(text): + missing.append("clean tracked state") + if not _CLEAN_UNTRACKED_RE.search(text): + missing.append("clean untracked state") + if not _BRANCH_BEFORE_RESET_RE.search(text): + missing.append("branch/head before reset") + if not _RESET_TARGET_RE.search(text): + missing.append("reset target SHA") + if not _REUSE_POLICY_RE.search(text): + missing.append("explicit project policy allowing reuse/reset") + return missing + + +def assess_worktree_ownership_report( + report_text: str, + *, + ownership_session: dict | None = None, + command_log: list | None = None, +) -> dict[str, Any]: + """Prove reviewer worktree ownership before reset or validation (#312).""" + text = report_text or "" + session = dict(ownership_session or {}) + reasons: list[str] = [] + + worktree_path = _extract_worktree_path(text, session) + destructive = _destructive_commands(command_log or session.get("command_log")) + if not destructive and session.get("destructive_reset"): + destructive = ["git reset --hard (session)"] + + session_owned = bool( + session.get("session_owned") + or (worktree_path and _is_session_owned_path(worktree_path)) + ) + safe_reuse = bool(session.get("safe_reuse") or _SAFE_REUSE_RE.search(text)) + main_checkout = bool(session.get("main_checkout")) + dirty_tracked = session.get("dirty_tracked") + dirty_untracked = session.get("dirty_untracked") + other_session_owned = bool(session.get("other_session_owned")) + + if worktree_path or destructive or session.get("validation_ran"): + if not worktree_path: + reasons.append("report missing exact review worktree path") + elif main_checkout or "branches/" not in worktree_path.replace("\\", "/"): + reasons.append("review worktree must be inside branches/, not the main checkout") + elif not _BRANCHES_PATH_RE.search(worktree_path.replace("\\", "/")): + reasons.append("review worktree path must be under branches/") + + if other_session_owned and not safe_reuse: + reasons.append( + "reused worktree appears owned by another active task/session; " + "safe-reuse proof required" + ) + + if dirty_tracked: + reasons.append( + "worktree has uncommitted tracked changes before reset or validation" + ) + if dirty_untracked and safe_reuse: + reasons.append("safe-reuse proof requires clean untracked state") + + if safe_reuse and not session_owned: + reasons.extend( + f"safe-reuse proof missing {field}" + for field in _safe_reuse_fields_present(text) + ) + + if destructive: + if not session_owned and not safe_reuse: + reasons.append( + "destructive worktree commands forbidden without session-owned " + "worktree or safe-reuse proof" + ) + if _WORKSPACE_NONE_RE.search(text) and ( + _RESET_IN_REPORT_RE.search(text) or any("reset" in c.lower() for c in destructive) + ): + reasons.append( + "final report must not claim 'Workspace mutations: none' when " + "git reset --hard occurred" + ) + if not _WORKTREE_MUTATION_RE.search(text) and not _RESET_IN_REPORT_RE.search(text): + reasons.append( + "destructive reset must be reported under Worktree/index mutations " + "or destructive reset operations" + ) + + if ( + worktree_path + and not session_owned + and not safe_reuse + and (destructive or session.get("validation_ran")) + and not _is_session_owned_path(worktree_path) + ): + reasons.append( + "reused branch-named worktree requires safe-reuse proof before reset or validation" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "worktree_path": worktree_path or None, + "session_owned": session_owned, + "safe_reuse": safe_reuse, + "destructive_commands": destructive, + "safe_next_action": ( + "use a fresh session-owned review worktree under branches/review-pr-* " + "or document full safe-reuse proof before destructive reset" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 7e0fe23..443f88b 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -127,4 +127,10 @@ def test_create_issue_final_report_verifier_exported(): def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof - assert callable(assess_non_mergeable_skip_proof) \ No newline at end of file + assert callable(assess_non_mergeable_skip_proof) + + +def test_worktree_ownership_verifier_exported(): + from review_proofs import assess_worktree_ownership_report + + assert callable(assess_worktree_ownership_report) \ No newline at end of file diff --git a/tests/test_reviewer_worktree_ownership.py b/tests/test_reviewer_worktree_ownership.py new file mode 100644 index 0000000..3b9dad3 --- /dev/null +++ b/tests/test_reviewer_worktree_ownership.py @@ -0,0 +1,137 @@ +"""Tests for reviewer worktree ownership verifier (#312).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_worktree_ownership import assess_worktree_ownership_report # noqa: E402 + + +def _fresh_review_report() -> str: + return "\n".join([ + "Review worktree path: branches/review-pr280-feat-issue-275-claim", + "Worktree is inside branches/ and not the main checkout.", + "Tracked state: clean; untracked state: clean.", + "Official PR-head validation on session-owned worktree.", + "Worktree mutations: none", + ]) + + +def _safe_reuse_report() -> str: + return "\n".join([ + "Safe-reuse proof for existing review worktree.", + "Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks", + "Worktree inside branches/; not the main checkout.", + "Not owned by another active task/session.", + "Clean tracked state; clean untracked state.", + "Branch/head before reset: feat/issue-260-example", + "Reset target SHA: abc123def4567890abcdef1234567890abcdef12", + "Project policy allows reuse/reset for this clean review worktree.", + "Worktree mutations: git reset --hard FETCH_HEAD", + "Destructive reset operations: git reset --hard FETCH_HEAD", + ]) + + +class TestWorktreeOwnership(unittest.TestCase): + def test_fresh_session_owned_passes(self): + result = assess_worktree_ownership_report( + _fresh_review_report(), + ownership_session={ + "validation_ran": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + self.assertTrue(result["session_owned"]) + + def test_safe_reuse_with_reset_passes(self): + result = assess_worktree_ownership_report( + _safe_reuse_report(), + ownership_session={ + "safe_reuse": True, + "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_destructive_reset_without_proof_blocks(self): + report = "\n".join([ + "Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks", + "Worktree mutations: git reset --hard FETCH_HEAD", + ]) + result = assess_worktree_ownership_report( + report, + ownership_session={ + "validation_ran": True, + "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_reused_dirty_tracked_blocks(self): + result = assess_worktree_ownership_report( + _fresh_review_report(), + ownership_session={ + "validation_ran": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + "dirty_tracked": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("tracked" in r.lower() for r in result["reasons"])) + + def test_safe_reuse_dirty_untracked_blocks(self): + result = assess_worktree_ownership_report( + _safe_reuse_report(), + ownership_session={ + "safe_reuse": True, + "dirty_untracked": True, + "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("untracked" in r.lower() for r in result["reasons"])) + + def test_workspace_none_with_reset_blocks(self): + report = "\n".join([ + "Review worktree path: branches/review-pr280-feat-issue-275-claim", + "Workspace mutations: none", + "Worktree mutations: git reset --hard FETCH_HEAD", + ]) + result = assess_worktree_ownership_report( + report, + ownership_session={ + "session_owned": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"])) + + def test_main_checkout_blocks(self): + result = assess_worktree_ownership_report( + "Review worktree path: /Users/dev/Gitea-Tools", + ownership_session={ + "validation_ran": True, + "main_checkout": True, + "worktree_path": "/Users/dev/Gitea-Tools", + }, + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_worktree_ownership_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file -- 2.43.7