feat: block edits in PR validation worktrees (Closes #315) #363

Merged
sysadmin merged 3 commits from feat/issue-315-validation-worktree-no-edits into master 2026-07-07 08:22:58 -05:00
4 changed files with 334 additions and 1 deletions
+9
View File
@@ -4251,6 +4251,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
+184
View File
@@ -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"
),
}
+7 -1
View File
@@ -148,7 +148,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)
@@ -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()