From 1320a55a7ea255375889f89fb102a13e74fc2573 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 11:28:34 -0400 Subject: [PATCH] feat: require validation cwd and HEAD proof in reviewer reports (Closes #398) Rebased onto current prgs/master; merged with #396 validation failure history by keeping both validator rules and workflow handoff fields. Co-Authored-By: Claude Opus 4.8 (1M context) --- final_report_validator.py | 37 +++ review_proofs.py | 7 + reviewer_validation_cwd_proof.py | 233 ++++++++++++++++++ .../workflows/review-merge-pr.md | 23 ++ tests/test_llm_workflow_split.py | 6 + tests/test_reviewer_validation_cwd_proof.py | 135 ++++++++++ 6 files changed, 441 insertions(+) create mode 100644 reviewer_validation_cwd_proof.py create mode 100644 tests/test_reviewer_validation_cwd_proof.py diff --git a/final_report_validator.py b/final_report_validator.py index 3fbac0f..a67ecba 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -490,6 +490,42 @@ def _rule_reviewer_validation_failure_history( ] +def _rule_reviewer_validation_cwd_proof( + report_text: str, + *, + validation_session: dict | None = None, +) -> list[dict[str, str]]: + from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report + + session = validation_session or {} + claims = ( + session.get("validation_ran") + or session.get("command") + or session.get("baseline_validation_ran") + ) + if not claims and "validation command:" not in (report_text or "").lower(): + return [] + + result = assess_validation_cwd_proof_report( + report_text, + validation_session=session, + ) + if result.get("proven") or not result.get("claims_validation"): + return [] + severity = "block" if result.get("violations") else "downgrade" + return [ + validator_finding( + "reviewer.validation_cwd_proof", + severity, + "Validation cwd/HEAD proof", + reason, + result.get("safe_next_action") + or "document pwd, HEAD SHA, and explicit cwd before validation", + ) + for reason in (result.get("violations") or result.get("reasons") or ["incomplete"]) + ] + + def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]: text = report_text or "" if not _BARE_PYTEST_RE.search(text): @@ -899,6 +935,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reviewer_git_fetch_readonly, _rule_reviewer_validation_command, _rule_reviewer_validation_failure_history, + _rule_reviewer_validation_cwd_proof, _rule_reviewer_validation_structured, _rule_reviewer_linked_issue, _rule_reviewer_baseline_on_failure, diff --git a/review_proofs.py b/review_proofs.py index 20cf89b..5bcd3c7 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -5512,6 +5512,13 @@ def assess_validation_failure_history_report(report_text, **kwargs): return _assess(report_text, **kwargs) +def assess_validation_cwd_proof_report(report_text, **kwargs): + """#398: validation commands require explicit worktree cwd and HEAD proof.""" + from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report as _assess + + return _assess(report_text, **kwargs) + + def assess_already_landed_classification_report(report_text, **kwargs): """#295: already-landed PRs are reconciliation-only, not review eligible.""" from reviewer_already_landed_classification import ( diff --git a/reviewer_validation_cwd_proof.py b/reviewer_validation_cwd_proof.py new file mode 100644 index 0000000..15cfb02 --- /dev/null +++ b/reviewer_validation_cwd_proof.py @@ -0,0 +1,233 @@ +"""Explicit worktree and cwd proof for PR review validation (#398).""" + +from __future__ import annotations + +import re +from typing import Any + +_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE) + +_PWD_RE = re.compile( + r"(?:^|\n)\s*(?:pwd|working\s+directory|cwd)\s*:\s*(\S+)", + re.IGNORECASE, +) +_HEAD_RE = re.compile( + r"(?:git\s+rev-parse\s+head|observed\s+head\s+sha)\s*:\s*([0-9a-f]{7,40})", + re.IGNORECASE | re.MULTILINE, +) +_EXPECTED_HEAD_RE = re.compile( + r"(?:expected\s+(?:pr\s+)?head\s+sha|candidate\s+head\s+sha|pinned\s+head)\s*:\s*([0-9a-f]{7,40})", + re.IGNORECASE, +) +_STATUS_RE = re.compile( + r"git\s+status\s+(?:--short\s+--branch|--short|-sb)\s*:\s*(.+)", + re.IGNORECASE, +) +_VALIDATION_CMD_RE = re.compile( + r"validation\s+command\s*:\s*(.+)", + re.IGNORECASE, +) +_GIT_C_CMD_RE = re.compile(r"git\s+-C\s+\S+", re.IGNORECASE) +_CD_CMD_RE = re.compile(r"(?:^|&&\s*)cd\s+\S+", re.IGNORECASE) +_BASELINE_CWD_RE = re.compile( + r"baseline\s+(?:worktree|working\s+directory|cwd)\s*:\s*(\S+)", + re.IGNORECASE, +) +_BASELINE_SHA_RE = re.compile( + r"baseline\s+(?:target\s+)?sha\s*:\s*([0-9a-f]{7,40})", + re.IGNORECASE, +) +_BASELINE_CMD_RE = re.compile( + r"baseline\s+validation\s+command\s*:\s*(.+)", + re.IGNORECASE, +) + + +def _normalize_path(path: str) -> str: + return (path or "").replace("\\", "/").rstrip("/") + + +def _path_under_branches(path: str, project_root: str | None = None) -> bool: + normalized = _normalize_path(path) + if not normalized: + return False + if "/branches/" in f"{normalized}/": + return True + if normalized.endswith("/branches"): + return True + if project_root: + root = _normalize_path(project_root) + if normalized.startswith(f"{root}/"): + rel = normalized[len(root) + 1 :] + return rel == "branches" or rel.startswith("branches/") + return False + + +def _expand_sha(sha: str) -> str: + return (sha or "").strip().lower() + + +def _sha_matches(expected: str, observed: str) -> bool: + exp = _expand_sha(expected) + obs = _expand_sha(observed) + if not exp or not obs: + return False + if len(exp) == 40 and len(obs) == 40: + return exp == obs + return obs.startswith(exp) or exp.startswith(obs) + + +def _command_has_explicit_cwd(command: str, cwd: str) -> bool: + text = (command or "").strip() + if not text: + return False + if _GIT_C_CMD_RE.search(text): + return True + if _CD_CMD_RE.search(text): + return True + if cwd and cwd in text: + return True + return False + + +def assess_validation_cwd_proof_report( + report_text: str, + *, + validation_session: dict | None = None, + project_root: str | None = None, +) -> dict[str, Any]: + """Require cwd/HEAD proof before reviewer validation claims (#398).""" + text = report_text or "" + session = dict(validation_session or {}) + reasons: list[str] = [] + violations: list[str] = [] + + claims_validation = bool( + session.get("validation_ran") + or _VALIDATION_CMD_RE.search(text) + or session.get("command") + ) + if not claims_validation: + return { + "proven": True, + "block": False, + "claims_validation": False, + "reasons": [], + "violations": [], + "safe_next_action": "proceed", + } + + expected_head = ( + session.get("expected_head_sha") + or session.get("candidate_head_sha") + or "" + ).strip() + if not expected_head: + match = _EXPECTED_HEAD_RE.search(text) + expected_head = (match.group(1) if match else "").strip() + + observed_head = (session.get("observed_head_sha") or "").strip() + if not observed_head: + match = _HEAD_RE.search(text) + observed_head = (match.group(1) if match else "").strip() + + cwd = ( + session.get("working_directory") + or session.get("cwd") + or session.get("pwd") + or "" + ).strip() + if not cwd: + match = _PWD_RE.search(text) + cwd = (match.group(1) if match else "").strip().rstrip(",.;") + + command = (session.get("command") or "").strip() + if not command: + match = _VALIDATION_CMD_RE.search(text) + command = (match.group(1) if match else "").strip().rstrip(".;") + + if not cwd: + reasons.append( + "validation claimed without pwd/working-directory proof (#398)" + ) + elif not _path_under_branches(cwd, project_root): + violations.append( + f"validation cwd {cwd!r} is not under branches/ (#398)" + ) + reasons.append( + "reviewer validation must run from a branches/ worktree, " + "not the main checkout (#398)" + ) + + if not observed_head: + reasons.append( + "validation claimed without git rev-parse HEAD / observed HEAD SHA " + "proof (#398)" + ) + elif expected_head and not _sha_matches(expected_head, observed_head): + violations.append( + f"observed HEAD {observed_head} does not match expected " + f"PR head {expected_head} (#398)" + ) + reasons.append("validation HEAD SHA must match pinned PR head (#398)") + + if not _STATUS_RE.search(text) and session.get("git_status") is None: + reasons.append( + "validation claimed without git status --short --branch proof (#398)" + ) + + if command and cwd and not _command_has_explicit_cwd(command, cwd): + if session.get("tool_working_directory") is not True: + reasons.append( + "validation command must use git -C , " + "cd && ..., or tool-provided cwd metadata (#398)" + ) + + baseline_ran = bool( + session.get("baseline_validation_ran") + or _BASELINE_CMD_RE.search(text) + ) + if baseline_ran: + baseline_cwd = (session.get("baseline_worktree_path") or "").strip() + if not baseline_cwd: + match = _BASELINE_CWD_RE.search(text) + baseline_cwd = (match.group(1) if match else "").strip().rstrip(",.;") + if not baseline_cwd or not _path_under_branches(baseline_cwd, project_root): + reasons.append( + "baseline validation claimed without baseline worktree cwd " + "under branches/ (#398)" + ) + baseline_sha = (session.get("baseline_target_sha") or "").strip() + if not baseline_sha: + match = _BASELINE_SHA_RE.search(text) + baseline_sha = (match.group(1) if match else "").strip() + if not baseline_sha: + reasons.append( + "baseline validation claimed without baseline target SHA (#398)" + ) + baseline_cmd = (session.get("baseline_command") or "").strip() + if not baseline_cmd: + match = _BASELINE_CMD_RE.search(text) + baseline_cmd = (match.group(1) if match else "").strip() + if not baseline_cmd: + reasons.append( + "baseline validation claimed without exact baseline command (#398)" + ) + + proven = not reasons and not violations + return { + "proven": proven, + "block": bool(violations) or not proven, + "claims_validation": True, + "expected_head_sha": expected_head or None, + "observed_head_sha": observed_head or None, + "working_directory": cwd or None, + "reasons": reasons, + "violations": violations, + "safe_next_action": ( + "before validation record pwd, git rev-parse HEAD, git status, " + "expected PR head SHA; run commands with git -C or cd in the same line" + if not proven + else "proceed" + ), + } \ No newline at end of file diff --git a/skills/llm-project-workflow/workflows/review-merge-pr.md b/skills/llm-project-workflow/workflows/review-merge-pr.md index 5b0e10a..5eda726 100644 --- a/skills/llm-project-workflow/workflows/review-merge-pr.md +++ b/skills/llm-project-workflow/workflows/review-merge-pr.md @@ -604,6 +604,28 @@ Do not claim “full-suite failures are pre-existing” unless baseline proof is ## 23. Validation command proof rule +Before any diff, test, or compile validation, record in the same command +transcript or final report: + +* `pwd` or explicit working directory +* `git rev-parse HEAD` +* `git status --short --branch` +* expected PR head SHA (candidate head SHA) + +Validation commands must use one of: + +* `git -C ...` +* `cd && ...` in the same command +* tool-provided explicit working-directory metadata + +Do not rely on inferred shell cwd from a prior command in a different block. + +`gitea_validate_review_final_report` rejects validation claims without +cwd/HEAD proof when `validation_session` is supplied. + +Baseline validation must document baseline worktree path, baseline target SHA, +cwd proof, exact baseline command, and baseline result using the same rules. + Report the exact validation command as executed. Report the working directory where validation ran. @@ -1127,6 +1149,7 @@ Controller Handoff: * Files reviewed: * Validation: * Validation failure history: +* Validation cwd/HEAD proof: * Official validation integrity status: * Terminal review mutation: * Review decision: diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 44d80ce..a5e5e33 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -213,6 +213,12 @@ def test_validation_failure_history_verifier_exported(): assert callable(assess_validation_failure_history_report) +def test_validation_cwd_proof_verifier_exported(): + from review_proofs import assess_validation_cwd_proof_report + + assert callable(assess_validation_cwd_proof_report) + + def test_prior_blocker_skip_verifier_exported(): from review_proofs import assess_prior_blocker_skip_proof diff --git a/tests/test_reviewer_validation_cwd_proof.py b/tests/test_reviewer_validation_cwd_proof.py new file mode 100644 index 0000000..59b81fc --- /dev/null +++ b/tests/test_reviewer_validation_cwd_proof.py @@ -0,0 +1,135 @@ +"""Tests for validation cwd/HEAD proof verifier (#398).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from final_report_validator import assess_final_report_validator # noqa: E402 +from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report # noqa: E402 + +ROOT = "/Users/jasonwalker/Development/Gitea-Tools" +WORKTREE = f"{ROOT}/branches/review-feat-issue-398" +HEAD = "f5953549aad5e822f14f52d3ea3c6d7990109384" + + +def _proof_backed_report() -> str: + return "\n".join([ + f"Candidate head SHA: {HEAD}", + f"pwd: {WORKTREE}", + f"git rev-parse HEAD: {HEAD}", + "git status --short --branch: ## feat/issue-398...prgs/master", + f"Validation command: cd {WORKTREE} && venv/bin/python -m pytest tests/ -q", + "Result: 1497 passed, 6 skipped", + ]) + + +class TestValidationCwdProof(unittest.TestCase): + def test_no_validation_claim_passes(self): + result = assess_validation_cwd_proof_report("Review decision: approve") + self.assertTrue(result["proven"]) + + def test_missing_cwd_proof_fails(self): + result = assess_validation_cwd_proof_report( + f"Validation command: pytest tests/\nCandidate head SHA: {HEAD}", + validation_session={"validation_ran": True, "expected_head_sha": HEAD}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_main_checkout_cwd_blocks(self): + result = assess_validation_cwd_proof_report( + "\n".join([ + f"Candidate head SHA: {HEAD}", + f"pwd: {ROOT}", + f"git rev-parse HEAD: {HEAD}", + "git status --short --branch: ## master", + "Validation command: pytest tests/ -q", + ]), + validation_session={"validation_ran": True, "expected_head_sha": HEAD}, + project_root=ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["violations"]) + + def test_wrong_head_blocks(self): + wrong = "a" * 40 + result = assess_validation_cwd_proof_report( + "\n".join([ + f"Candidate head SHA: {HEAD}", + f"pwd: {WORKTREE}", + f"git rev-parse HEAD: {wrong}", + "git status --short --branch: clean", + f"Validation command: cd {WORKTREE} && pytest -q", + ]), + validation_session={"validation_ran": True, "expected_head_sha": HEAD}, + project_root=ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["violations"]) + + def test_fully_proof_backed_passes(self): + result = assess_validation_cwd_proof_report( + _proof_backed_report(), + validation_session={"validation_ran": True, "expected_head_sha": HEAD}, + project_root=ROOT, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_baseline_without_cwd_fails(self): + result = assess_validation_cwd_proof_report( + "\n".join([ + _proof_backed_report(), + "Baseline validation command: pytest tests/ -q", + ]), + validation_session={ + "validation_ran": True, + "expected_head_sha": HEAD, + "baseline_validation_ran": True, + }, + project_root=ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue( + any("baseline" in r.lower() for r in result["reasons"]) + ) + + def test_baseline_with_full_proof_passes(self): + result = assess_validation_cwd_proof_report( + "\n".join([ + _proof_backed_report(), + f"Baseline worktree: {ROOT}/branches/baseline-master-pr376", + f"Baseline target SHA: {HEAD}", + f"Baseline validation command: cd {ROOT}/branches/baseline-master-pr376 && pytest -q", + ]), + validation_session={ + "validation_ran": True, + "expected_head_sha": HEAD, + "baseline_validation_ran": True, + }, + project_root=ROOT, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_final_report_validator_integration(self): + result = assess_final_report_validator( + "Validation command: pytest tests/ -q", + "review_pr", + validation_session={"validation_ran": True}, + ) + self.assertTrue(result["blocked"] or result["downgraded"]) + self.assertTrue( + any( + f.get("rule_id") == "reviewer.validation_cwd_proof" + for f in result.get("findings") or [] + ) + ) + + def test_exported_from_review_proofs(self): + from review_proofs import assess_validation_cwd_proof_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file