Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat: require validation cwd and HEAD proof in reviewer reports (Closes #398)

Browse Source 
Add assess_validation_cwd_proof_report so reviewer final reports must prove
pwd, git rev-parse HEAD, git status, and expected PR head SHA before
claiming validation ran in a branches/ worktree. Wire the verifier into
final_report_validator and gitea_validate_review_final_report.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-07 11:28:34 -04:00
co-authored by Claude Opus 4.8
parent 027a3cb92c
commit 8df795ef80
8 changed files with 447 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
final_report_validator.py
+39
View File
@@ -456,6 +456,42 @@ def _rule_reviewer_git_fetch_readonly(
return []
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):
@@ -864,6 +900,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
_rule_reviewer_mutation_categories,
_rule_reviewer_git_fetch_readonly,
_rule_reviewer_validation_command,
_rule_reviewer_validation_cwd_proof,
_rule_reviewer_validation_structured,
_rule_reviewer_linked_issue,
_rule_reviewer_baseline_on_failure,
@@ -970,6 +1007,7 @@ def assess_final_report_validator(
local_edits: bool = False,
issue_filing_lock: dict | None = None,
session_pr_opened: bool = False,
validation_session: dict | None = None,
) -> dict[str, Any]:
"""Validate final-report text against task-specific proof rules (#327).
@@ -1025,6 +1063,7 @@ def assess_final_report_validator(
"mutations_observed": mutations_observed,
"local_edits": local_edits,
"session_pr_opened": session_pr_opened,
"validation_session": validation_session,
}
for rule in _RULES_BY_TASK.get(normalized_kind, ()):
gitea_mcp_server.py
+2
View File
@@ -5506,6 +5506,7 @@ def gitea_validate_review_final_report(
action_log: list[dict] | None = None,
mutations_observed: bool = False,
local_edits: bool = False,
validation_session: dict | None = None,
) -> dict:
"""Read-only: machine-validate reviewer final report schema before output (#391).
@@ -5523,6 +5524,7 @@ def gitea_validate_review_final_report(
action_log=action_log,
mutations_observed=mutations_observed,
local_edits=local_edits,
validation_session=validation_session,
)
review_final_report_schema.py
+2
View File
@@ -301,6 +301,7 @@ def assess_review_final_report_schema(
action_log: list[dict] | None = None,
mutations_observed: bool = False,
local_edits: bool = False,
validation_session: dict | None = None,
) -> dict[str, Any]:
"""Validate reviewer final report text before session completion (#391)."""
base = assess_final_report_validator(
@@ -312,6 +313,7 @@ def assess_review_final_report_schema(
action_log=action_log,
mutations_observed=mutations_observed,
local_edits=local_edits,
validation_session=validation_session,
)
extra: list[dict[str, str]] = []
for rule in _SCHEMA_RULES:
review_proofs.py
+7
View File
@@ -5487,6 +5487,13 @@ def assess_validation_integrity_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 (
reviewer_validation_cwd_proof.py
+233
View File
@@ -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 <worktree>, "
"cd <worktree> && ..., 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"
),
}
skills/llm-project-workflow/workflows/review-merge-pr.md
+23
View File
@@ -574,6 +574,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 <review_worktree> ...`
* `cd <review_worktree> && ...` 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.
@@ -1096,6 +1118,7 @@ Controller Handoff:
* Baseline worktree path:
* Files reviewed:
* Validation:
* Validation cwd/HEAD proof:
* Official validation integrity status:
* Terminal review mutation:
* Review decision:
tests/test_llm_workflow_split.py
+6
View File
@@ -207,6 +207,12 @@ def test_validation_integrity_verifier_exported():
assert callable(assess_validation_integrity_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
tests/test_reviewer_validation_cwd_proof.py
+135
View File
@@ -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()