diff --git a/review_proofs.py b/review_proofs.py index 43dfbd3..89e3701 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -4474,6 +4474,15 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs): return _assess(report_text, **kwargs) +def assess_validation_worktree_edit_report(report_text, **kwargs): + """#315: block edits in PR validation worktrees; require diagnostic separation.""" + from reviewer_validation_worktree_edits import ( + assess_validation_worktree_edit_report 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 diff --git a/reviewer_validation_worktree_edits.py b/reviewer_validation_worktree_edits.py new file mode 100644 index 0000000..c4e904f --- /dev/null +++ b/reviewer_validation_worktree_edits.py @@ -0,0 +1,184 @@ +"""PR validation worktree read-only edit verifier (#315).""" + +from __future__ import annotations + +import re +from typing import Any + +_VALIDATION_WORKTREE_RE = re.compile( + r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)", + re.IGNORECASE, +) +_DIAGNOSTIC_WORKTREE_RE = re.compile( + r"branches/(?:diagnostic[\w/-]*|scratch[\w/-]*)", + re.IGNORECASE, +) +_FILE_EDITS_NONE_RE = re.compile( + r"file edits by reviewer\s*:\s*none\b", + re.IGNORECASE, +) +_FILE_EDITS_FIELD_RE = re.compile( + r"file edits by reviewer\s*:\s*(.+)", + re.IGNORECASE, +) +_VALIDATION_WORKTREE_PATH_RE = re.compile( + r"(?:validation worktree path|review worktree path)\s*:\s*(\S+)", + re.IGNORECASE, +) +_DIAGNOSTIC_WORKTREE_PATH_RE = re.compile( + r"(?:diagnostic (?:scratch )?worktree path|diagnostic worktree)\s*:\s*(\S+)", + re.IGNORECASE, +) +_DIAGNOSTIC_LABEL_RE = re.compile( + r"diagnostic local experiment|not pr-head validation|diagnostic-only", + re.IGNORECASE, +) +_OFFICIAL_VALIDATION_RE = re.compile( + r"(?:official (?:pr-head )?validation|pr-head validation result)", + re.IGNORECASE, +) +_POST_EDIT_OFFICIAL_RE = re.compile( + r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit)", + re.IGNORECASE, +) + +_PERFORMED_EDIT_KINDS = frozenset({"file_edit", "edit", "write", "edited", "created", "wrote"}) + + +def _normalize_path(path: str) -> str: + return (path or "").replace("\\", "/").strip() + + +def _is_validation_worktree(path: str) -> bool: + return bool(_VALIDATION_WORKTREE_RE.search(_normalize_path(path))) + + +def _is_diagnostic_worktree(path: str) -> bool: + normalized = _normalize_path(path) + return bool( + _DIAGNOSTIC_WORKTREE_RE.search(normalized) + or "diagnostic" in normalized.lower() + ) + + +def _performed_edits(action_log: list[dict] | None) -> list[dict[str, str]]: + edits: list[dict[str, str]] = [] + for entry in action_log or []: + if entry.get("gated_rejected") or entry.get("performed") is False: + continue + kind = (entry.get("kind") or entry.get("action") or "file_edit").strip().lower() + if kind not in _PERFORMED_EDIT_KINDS: + continue + path = (entry.get("path") or "").strip() + if not path: + continue + worktree = _normalize_path(str(entry.get("worktree_path") or entry.get("cwd") or "")) + edits.append({"path": path, "worktree_path": worktree}) + return edits + + +def _extract_validation_path(text: str, session: dict) -> str: + path = _normalize_path(str(session.get("validation_worktree_path") or "")) + if path: + return path + match = _VALIDATION_WORKTREE_PATH_RE.search(text or "") + return _normalize_path(match.group(1)) if match else "" + + +def _extract_diagnostic_path(text: str, session: dict) -> str: + path = _normalize_path(str(session.get("diagnostic_worktree_path") or "")) + if path: + return path + match = _DIAGNOSTIC_WORKTREE_PATH_RE.search(text or "") + return _normalize_path(match.group(1)) if match else "" + + +def assess_validation_worktree_edit_report( + report_text: str, + *, + validation_session: dict | None = None, + action_log: list[dict] | None = None, +) -> dict[str, Any]: + """Block edits in PR validation worktrees; require diagnostic separation (#315).""" + text = report_text or "" + session = dict(validation_session or {}) + reasons: list[str] = [] + + validation_path = _extract_validation_path(text, session) + diagnostic_path = _extract_diagnostic_path(text, session) + validation_edits = list(session.get("validation_worktree_edits") or []) + diagnostic_edits = list(session.get("diagnostic_edits") or []) + edits = _performed_edits(action_log) + + if not validation_edits: + for entry in edits: + wt = entry.get("worktree_path") or validation_path + if wt and _is_validation_worktree(wt) and not _is_diagnostic_worktree(wt): + validation_edits.append(entry["path"]) + + if not diagnostic_edits: + for entry in edits: + wt = entry.get("worktree_path") or "" + if wt and _is_diagnostic_worktree(wt): + diagnostic_edits.append(entry["path"]) + elif entry["path"] and diagnostic_path and not validation_edits: + if _is_diagnostic_worktree(diagnostic_path): + diagnostic_edits.append(entry["path"]) + + claimed_none = bool(_FILE_EDITS_NONE_RE.search(text)) + any_edits = bool(validation_edits or diagnostic_edits or edits) + + if validation_edits: + reasons.append( + "reviewer must not edit files in the PR validation worktree; " + f"observed edits: {', '.join(validation_edits)}" + ) + + if diagnostic_edits or session.get("diagnostic_experiment"): + if not diagnostic_path and not _DIAGNOSTIC_WORKTREE_PATH_RE.search(text): + reasons.append( + "diagnostic experiments require a separate diagnostic scratch worktree path" + ) + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "diagnostic experiments must be labeled " + "'Diagnostic local experiment — not PR-head validation'" + ) + if not _FILE_EDITS_FIELD_RE.search(text) or claimed_none: + reasons.append( + "diagnostic edits must be reported under 'File edits by reviewer'" + ) + + if any_edits and claimed_none: + reasons.append( + "report claims 'File edits by reviewer: none' but reviewer file edits occurred" + ) + + if session.get("official_validation_after_edit") or _POST_EDIT_OFFICIAL_RE.search(text): + if validation_edits or (validation_path and diagnostic_edits): + reasons.append( + "post-edit test results cannot be reported as official PR-head validation" + ) + + if validation_edits and _OFFICIAL_VALIDATION_RE.search(text): + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "validation worktree was edited; official PR-head validation is no longer valid" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "validation_worktree_path": validation_path or None, + "diagnostic_worktree_path": diagnostic_path or None, + "validation_worktree_edits": validation_edits, + "diagnostic_edits": diagnostic_edits, + "safe_next_action": ( + "keep PR validation worktrees read-only; use a separate diagnostic " + "scratch worktree and report all reviewer file edits" + 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 0c2b89a..550c007 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -154,7 +154,13 @@ def test_non_mergeable_skip_verifier_exported(): assert callable(assess_non_mergeable_skip_proof) +def test_validation_worktree_edit_verifier_exported(): + from review_proofs import assess_validation_worktree_edit_report + + assert callable(assess_validation_worktree_edit_report) + + def test_worktree_ownership_verifier_exported(): from review_proofs import assess_worktree_ownership_report - assert callable(assess_worktree_ownership_report) + assert callable(assess_worktree_ownership_report) \ No newline at end of file diff --git a/tests/test_reviewer_validation_worktree_edits.py b/tests/test_reviewer_validation_worktree_edits.py new file mode 100644 index 0000000..adedaca --- /dev/null +++ b/tests/test_reviewer_validation_worktree_edits.py @@ -0,0 +1,134 @@ +"""Tests for PR validation worktree no-edit verifier (#315).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_validation_worktree_edits import ( # noqa: E402 + assess_validation_worktree_edit_report, +) + + +def _read_only_report() -> str: + return "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official PR-head validation on unmodified worktree.", + "Official validation result: failed (1 failed)", + "File edits by reviewer: none", + "Review decision: request_changes", + ]) + + +def _diagnostic_report() -> str: + return "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official PR-head validation on unmodified worktree.", + "Official validation result: failed (1 failed)", + "Review decision: request_changes", + "Diagnostic local experiment — not PR-head validation", + "Diagnostic scratch worktree path: branches/diagnostic-pr280-fix-test", + "File edits by reviewer: tests/test_agent_temp_artifacts.py (diagnostic only)", + "Diagnostic result: passed after local fix", + ]) + + +class TestValidationWorktreeEdits(unittest.TestCase): + def test_read_only_review_passes(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_validation_worktree_edit_blocks(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"], + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("validation worktree" in r.lower() for r in result["reasons"])) + + def test_file_edits_none_contradiction_blocks(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + }, + action_log=[ + { + "path": "tests/test_agent_temp_artifacts.py", + "kind": "file_edit", + "performed": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + } + ], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("file edits" in r.lower() for r in result["reasons"])) + + def test_diagnostic_scratch_with_reporting_passes(self): + result = assess_validation_worktree_edit_report( + _diagnostic_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "diagnostic_worktree_path": "branches/diagnostic-pr280-fix-test", + "diagnostic_edits": ["tests/test_agent_temp_artifacts.py"], + "diagnostic_experiment": True, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_diagnostic_without_scratch_worktree_blocks(self): + result = assess_validation_worktree_edit_report( + "File edits by reviewer: tests/test_example.py", + validation_session={ + "diagnostic_edits": ["tests/test_example.py"], + "diagnostic_experiment": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"])) + + def test_post_edit_official_validation_blocks(self): + report = "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official validation result: passed after local edit", + "File edits by reviewer: tests/test_agent_temp_artifacts.py", + ]) + result = assess_validation_worktree_edit_report( + report, + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"], + "official_validation_after_edit": True, + }, + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_validation_worktree_edit_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file