diff --git a/review_proofs.py b/review_proofs.py index 62344f0..586398a 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3666,6 +3666,113 @@ def assess_reviewer_baseline_validation_proof( } +# --------------------------------------------------------------------------- +# Full-suite failure approval gate (#323) +# --------------------------------------------------------------------------- + + +def assess_full_suite_failure_approval_gate( + *, + review_decision: str, + full_suite_passed: bool | None, + baseline_validation: dict | None = None, + baseline_proof: dict | None = None, + report_text: str | None = None, + project_root: str | None = None, +) -> dict: + """#323: approving a PR whose full suite fails requires baseline proof. + + Default action on a full-suite failure is REQUEST_CHANGES. Approval is + allowed only when the #325 baseline proof passes in full (branches/ + worktree, pinned SHAs, matching failure signatures) and the extra #323 + fields prove the failures are pre-existing: ``pr_suite_command``, + ``baseline_suite_command``, ``new_tests_passed``, and a non-empty + ``unrelated_explanation``. + + *review_decision* is the decision the workflow intends to submit; only + ``approve`` can be blocked. ``block`` is True when an approval was + requested that the proofs do not allow. + """ + decision = (review_decision or "").strip().lower().replace("-", "_") + wants_approval = decision == "approve" + reasons: list[str] = [] + + if full_suite_passed is True: + return { + "approve_allowed": True, + "block": False, + "default_action": "proceed", + "reasons": [], + } + + default_action = "request_changes" + + if full_suite_passed is None: + reasons.append( + "full-suite result not proven; approval requires a recorded " + "full-suite command and result (#323)" + ) + else: + if baseline_validation is None: + baseline_validation = assess_reviewer_baseline_validation_proof( + baseline_proof=baseline_proof, + report_text=report_text, + project_root=project_root, + ) + proof = baseline_proof or {} + + if not proof: + reasons.append( + "full suite failed but no baseline proof was provided; " + "default action is REQUEST_CHANGES (#323)" + ) + if not baseline_validation.get("proven"): + reasons.append( + "baseline validation proof missing or failed (#323/#325)" + ) + reasons.extend(baseline_validation.get("reasons", [])) + + if proof: + worktree = (proof.get("worktree_path") or "").strip() + if not _path_under_branches(worktree, project_root): + reasons.append( + "baseline comparison must run in a clean baseline " + "worktree under branches/, not the main checkout (#323)" + ) + if not (proof.get("pr_suite_command") or "").strip(): + reasons.append( + "PR full-suite command/result missing from baseline " + "proof (#323)" + ) + if not (proof.get("baseline_suite_command") or "").strip(): + reasons.append( + "baseline full-suite command/result missing from " + "baseline proof (#323)" + ) + if proof.get("new_tests_passed") is not True: + reasons.append( + "proof that new/changed tests passed is missing (#323)" + ) + if not (proof.get("unrelated_explanation") or "").strip(): + reasons.append( + "explanation why failures are unrelated to the PR is " + "missing (#323)" + ) + if proof.get("failure_signatures_match") is not True: + reasons.append( + "PR and baseline failure signatures do not match; " + "request changes (#323)" + ) + + approve_allowed = not reasons + return { + "approve_allowed": approve_allowed, + "block": wants_approval and not approve_allowed, + "default_action": "proceed" if approve_allowed else default_action, + "reasons": reasons, + } + + # --------------------------------------------------------------------------- # Identity disclosure (#305) # --------------------------------------------------------------------------- @@ -4169,3 +4276,10 @@ def assess_reconcile_inventory_report(report_text, **kwargs): from reviewer_reconcile_inventory import assess_reconcile_inventory_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 + + 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_agent_temp_artifacts.py b/tests/test_agent_temp_artifacts.py index 2e14385..903eab8 100644 --- a/tests/test_agent_temp_artifacts.py +++ b/tests/test_agent_temp_artifacts.py @@ -62,7 +62,24 @@ class TestPreflightWarnings(unittest.TestCase): self.assertIn("_encode_commit_payload.py", status["preflight_warnings"][0]) +# Issue-write tools are profile-gated (#69); gitea_lock_issue requires +# gitea.issue.comment (see task_capability_map), so the gate must be +# seeded exactly like tests/test_mcp_server.py::TestIssueLocking (#359). +ISSUE_WRITE_ENV = { + "GITEA_ALLOWED_OPERATIONS": ( + "gitea.issue.create,gitea.issue.close,gitea.issue.comment" + ), +} + + class TestIssueLockArtifactWarning(unittest.TestCase): + def setUp(self): + self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True) + self._env_patcher.start() + + def tearDown(self): + self._env_patcher.stop() + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server._auth", return_value="token x") @patch("mcp_server._resolve", return_value=("h", "o", "r")) @@ -72,6 +89,9 @@ class TestIssueLockArtifactWarning(unittest.TestCase): mock_state.return_value = { "current_branch": "master", "porcelain_status": "?? _emit_payload.py\n", + "base_equivalent": True, + "inspected_git_root": "/scratch/wt", + "base_branch": "origin/master", } result = mcp_server.gitea_lock_issue( issue_number=261, diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index cfb916a..e207fda 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -152,3 +152,9 @@ def test_reconcile_inventory_verifier_exported(): from review_proofs import assess_reconcile_inventory_report assert callable(assess_reconcile_inventory_report) + + +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_review_proofs.py b/tests/test_review_proofs.py index 3a3ea7a..8721545 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -24,6 +24,7 @@ from review_proofs import ( # noqa: E402 ISSUE_SELECTION_CONTINUATION_EXPLICIT, ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR, assess_author_pr_report, + assess_full_suite_failure_approval_gate, assess_queue_ordering_proof, assess_reviewer_baseline_validation_proof, assess_capability_evidence, @@ -2539,5 +2540,170 @@ class TestReviewerBaselineValidationProof(unittest.TestCase): self.assertFalse(report["baseline_validation_proven"]) +class TestFullSuiteFailureApprovalGate(unittest.TestCase): + """Issue #323: full-suite failure blocks approval without baseline proof.""" + + ROOT = "/repo/Gitea-Tools" + + def _good_proof(self, **overrides): + proof = { + "worktree_path": f"{self.ROOT}/branches/baseline-master-pr280", + "baseline_target_sha": PINNED, + "pr_head_sha": OTHER, + "baseline_failures": ["tests/test_foo.py::test_bar"], + "pr_failures": ["tests/test_foo.py::test_bar"], + "failure_signatures_match": True, + "clean_before": True, + "clean_after": True, + "pr_suite_command": "venv/bin/pytest tests/", + "baseline_suite_command": "venv/bin/pytest tests/", + "new_tests_passed": True, + "unrelated_explanation": ( + "failures touch agent_temp_artifacts only; PR changes " + "review_proofs handoff fields" + ), + } + proof.update(overrides) + return proof + + def _good_report(self): + return ( + "Full suite failed on PR worktree and on baseline.\n" + "- Validation command: venv/bin/pytest tests/\n" + f"- Working directory: {self.ROOT}/branches/review-pr-280\n" + "Failures are pre-existing on master.\n" + ) + + def test_full_suite_pass_allows_approval_without_baseline(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=True, + project_root=self.ROOT, + ) + self.assertTrue(result["approve_allowed"]) + self.assertFalse(result["block"]) + self.assertEqual(result["default_action"], "proceed") + + def test_full_suite_failure_defaults_to_request_changes(self): + result = assess_full_suite_failure_approval_gate( + review_decision="request_changes", + full_suite_passed=False, + project_root=self.ROOT, + ) + self.assertFalse(result["block"]) + self.assertEqual(result["default_action"], "request_changes") + + def test_approval_with_failure_and_no_proof_blocks(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(result["block"]) + self.assertEqual(result["default_action"], "request_changes") + + def test_approval_with_matching_baseline_proof_allowed(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof(), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertTrue(result["approve_allowed"]) + self.assertFalse(result["block"]) + + def test_mismatched_failure_signatures_block_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof( + failure_signatures_match=False, + pr_failures=["tests/test_other.py::test_other"], + ), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(result["block"]) + + def test_main_checkout_baseline_blocks_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof(worktree_path=self.ROOT), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(result["block"]) + + def test_missing_new_tests_proof_blocks_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof(new_tests_passed=None), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(any( + "new" in reason.lower() and "test" in reason.lower() + for reason in result["reasons"] + )) + + def test_missing_unrelated_explanation_blocks_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof(unrelated_explanation=""), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + + def test_missing_suite_commands_block_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof( + pr_suite_command="", baseline_suite_command=""), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(any( + "command" in reason.lower() for reason in result["reasons"] + )) + + def test_unknown_suite_result_blocks_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=None, + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(result["block"]) + + def test_vague_same_as_master_claim_without_proof_blocks(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + report_text="Validation: failures same as master; approving.", + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(result["block"]) + + def test_non_approve_decision_never_blocks(self): + result = assess_full_suite_failure_approval_gate( + review_decision="comment", + full_suite_passed=None, + project_root=self.ROOT, + ) + self.assertFalse(result["block"]) + + if __name__ == "__main__": unittest.main() 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