diff --git a/review_proofs.py b/review_proofs.py index db52131..df2366f 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1052,6 +1052,15 @@ def _prs_from_list_response(list_prs_response: list | dict | None) -> list | Non return None +# ── Linked-issue consistency (Issue #314) ──────────────────────────────────── +# +# Stale linked-issue text from a previous PR must not leak into a final +# report: the "Linked issue status" field must match the issue(s) the +# selected PR actually links, verified live this session, and no other +# "Issue #N" mention may appear unless it is a verified linked issue. + +_ISSUE_MENTION_RE = re.compile(r"\bissue\s+#(\d+)", re.IGNORECASE) + # ── Workspace-vs-worktree mutation consistency (Issue #313) ────────────────── # # A worktree reset/checkout/clean is a workspace mutation even when the @@ -1092,6 +1101,68 @@ def _report_labeled_fields(report_text): return fields +def _issue_numbers_mentioned(text): + """Return the set of ints referenced as 'Issue #N' in *text*.""" + return {int(n) for n in _ISSUE_MENTION_RE.findall(text or "")} + + +def assess_linked_issue_consistency(report_text, *, selected_pr=None, + linked_issues=None): + """#314: linked-issue status must match the selected PR, live-verified. + + *linked_issues* is the list of issue numbers the selected PR was + live-verified to link this session, or None when no live verification + happened. Without live proof the report may only say the status was + not verified; with proof, the ``Linked issue status`` field must name + every linked issue and no stale issue number may appear anywhere in + the report. + + Returns {'complete', 'downgraded', 'reasons'}; fails closed. + """ + fields = _report_labeled_fields(report_text) + status_value = fields.get("linked issue status") + + reasons = [] + + if status_value is None: + reasons.append( + "final report missing 'Linked issue status' field for the " + "selected PR" + ) + elif linked_issues is None: + if not status_value.strip().lower().startswith("not verified"): + reasons.append( + "linked issue status claimed without live proof; report " + "'Linked issue status: not verified in this session' or " + "verify the linked issue live" + ) + else: + allowed = set(linked_issues) + status_mentions = _issue_numbers_mentioned(status_value) + for missing in sorted(allowed - status_mentions): + reasons.append( + f"linked issue #{missing} of selected PR " + f"#{selected_pr} not reported in 'Linked issue status'" + ) + for stale in sorted(status_mentions - allowed): + reasons.append( + f"'Linked issue status' names issue #{stale}, which is " + f"not a live-verified linked issue of selected PR " + f"#{selected_pr}" + ) + for stale in sorted(_issue_numbers_mentioned(report_text) - allowed): + reasons.append( + f"stale issue #{stale} mentioned in final report but not " + f"live-verified as linked to selected PR #{selected_pr}" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + def _field_claims_none(fields, label): """True when *label* is present and its value is empty or 'none...'.""" value = fields.get(label) @@ -3595,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) # --------------------------------------------------------------------------- @@ -4073,6 +4251,13 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None): } +def assess_prior_blocker_skip_proof(report_text, **kwargs): + """#318: require live blocker proof before skipping earlier open PRs.""" + from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess + + return _assess(report_text, **kwargs) + + def assess_non_mergeable_skip_proof(report_text, **kwargs): """#322: require conflict proof when skipping non-mergeable PRs.""" from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess @@ -4085,3 +4270,10 @@ def assess_inventory_worktree_report(report_text, **kwargs): from reviewer_inventory_worktree import assess_inventory_worktree_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_blocker_skip.py b/reviewer_blocker_skip.py new file mode 100644 index 0000000..a95d368 --- /dev/null +++ b/reviewer_blocker_skip.py @@ -0,0 +1,164 @@ +"""Prior-blocker skip proof verifier for reviewer queue reports (#318).""" + +from __future__ import annotations + +import re +from typing import Any + +_FULL_SHA = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE) +_SHORT_SHA = re.compile(r"\b[0-9a-f]{7,39}\b", re.IGNORECASE) +_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE) +_BLOCKING_DECISION_RE = re.compile( + r"(?:blocking review decision|review decision|blocking decision)\s*:\s*request[_ ]changes|" + r"request[_ ]changes", + re.IGNORECASE, +) +_BLOCKING_HEAD_RE = re.compile( + r"(?:blocking review head(?:\s+sha)?|blocker head(?:\s+sha)?|review head at blocker)", + re.IGNORECASE, +) +_HEAD_CHANGED_RE = re.compile( + r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|" + r"head changed (?:after|since) (?:the )?blocker|head unchanged since blocker)", + re.IGNORECASE, +) +_BLOCKER_REASON_RE = re.compile( + r"(?:reason (?:it )?remains blocked|remains blocked because|blocking category)", + re.IGNORECASE, +) +_LIVE_BLOCKER_PROOF_RE = re.compile( + r"(?:blocker revalidated live|live proof|gitea_get_pr_review_feedback|" + r"gitea_view_pr|review feedback fetched|current review state)", + re.IGNORECASE, +) +_BLOCKER_UNVERIFIED_RE = re.compile(r"\bBLOCKER_STATUS_UNVERIFIED\b") + + +def _pr_section(text: str, pr_number: int) -> str: + lines = (text or "").splitlines() + chunks: list[str] = [] + capture = False + token = f"#{pr_number}" + for line in lines: + lower = line.lower() + if token in lower or f"pr {pr_number}" in lower or f"pr#{pr_number}" in lower.replace(" ", ""): + capture = True + chunks.append(line) + continue + if capture: + if _PR_NUMBER_RE.search(line) and token not in line.lower(): + break + if line.strip() == "" and len(chunks) > 3: + break + chunks.append(line) + if chunks: + return "\n".join(chunks) + return text or "" + + +def _has_head_sha(text: str, head_sha: str | None) -> bool: + if not head_sha: + return bool(_FULL_SHA.search(text) or _SHORT_SHA.search(text)) + head = head_sha.strip().lower() + if head in text.lower(): + return True + if len(head) >= 7 and head[:7] in text.lower(): + return True + return bool(_FULL_SHA.search(text)) + + +def assess_prior_blocker_skip_proof( + report_text: str, + *, + skipped_prs: list[dict] | None = None, +) -> dict[str, Any]: + """Validate live blocker proof for skipped earlier open PRs (#318).""" + text = report_text or "" + reasons: list[str] = [] + assessments: list[dict[str, Any]] = [] + + for entry in skipped_prs or []: + pr_number = entry.get("pr_number") + if pr_number is None: + reasons.append("skipped PR entry missing pr_number") + continue + pr_number = int(pr_number) + section = _pr_section(text, pr_number) + verified = entry.get("blocker_verified") + head_changed = entry.get("head_changed_since_blocker") + blocking_head = (entry.get("blocking_review_head_sha") or "").strip() or None + current_head = (entry.get("head_sha") or "").strip() or None + skip_reason = (entry.get("skip_reason") or entry.get("blocking_decision") or "").lower() + + item_reasons: list[str] = [] + + if head_changed is True: + item_reasons.append( + f"PR #{pr_number} head changed after blocker; it cannot be skipped " + "based on stale REQUEST_CHANGES" + ) + + if f"#{pr_number}" not in text.lower() and f"pr {pr_number}" not in text.lower(): + item_reasons.append(f"skipped PR #{pr_number} not documented in final report") + + if "request_changes" in skip_reason or entry.get("blocking_decision") == "request_changes": + if verified is False: + if not _BLOCKER_UNVERIFIED_RE.search(section): + item_reasons.append( + f"PR #{pr_number} blocker proof unavailable; report must classify " + "BLOCKER_STATUS_UNVERIFIED" + ) + else: + if not _BLOCKING_DECISION_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing blocking review decision proof" + ) + if not _has_head_sha(section, current_head): + item_reasons.append( + f"PR #{pr_number} skip missing current head SHA" + ) + if blocking_head and not _BLOCKING_HEAD_RE.search(section): + if blocking_head.lower() not in section.lower(): + item_reasons.append( + f"PR #{pr_number} skip missing blocking review head SHA" + ) + if head_changed is False and not _HEAD_CHANGED_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing head-changed-since-blocker proof" + ) + if not _BLOCKER_REASON_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing reason-it-remains-blocked" + ) + if not _LIVE_BLOCKER_PROOF_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing live blocker proof for this session" + ) + + proven = not item_reasons + assessments.append({ + "pr_number": pr_number, + "proven": proven, + "classification": ( + "BLOCKER_STATUS_UNVERIFIED" + if verified is False + else "BLOCKED_SKIPPED" + ), + "reasons": item_reasons, + }) + reasons.extend(item_reasons) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "downgraded": False, + "assessments": assessments, + "reasons": reasons, + "safe_next_action": ( + "fetch live review feedback and document blocker proof for each skipped PR, " + "or classify BLOCKER_STATUS_UNVERIFIED" + if reasons + else "proceed" + ), + } \ No newline at end of file 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_linked_issue_consistency.py b/tests/test_linked_issue_consistency.py new file mode 100644 index 0000000..cf325bd --- /dev/null +++ b/tests/test_linked_issue_consistency.py @@ -0,0 +1,118 @@ +"""Tests for linked-issue consistency verifier (#314). + +Reviewer reports must verify the linked issue live for the selected PR; +stale issue numbers from a previous PR must not leak into the final +report, and linked-issue status must never be claimed without live proof. +""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from review_proofs import assess_linked_issue_consistency # noqa: E402 + + +class TestCorrectLinkedIssue(unittest.TestCase): + def test_matching_linked_issue_passes(self): + report = ( + "Selected PR: 280\n" + "Linked issue status: Issue #275 open, closes on merge\n" + ) + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275] + ) + self.assertTrue(result["complete"]) + self.assertFalse(result["downgraded"]) + self.assertEqual(result["reasons"], []) + + def test_multiple_linked_issues_all_reported_passes(self): + report = ( + "Linked issue status: Issue #275 and Issue #276 both open\n" + ) + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275, 276] + ) + self.assertTrue(result["complete"]) + + def test_multiple_linked_issues_one_missing_fails(self): + report = "Linked issue status: Issue #275 open\n" + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275, 276] + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("276" in r for r in result["reasons"]) + ) + + +class TestStaleIssueLeak(unittest.TestCase): + def test_stale_issue_in_status_field_fails(self): + # #314 observed behavior: PR #280 links Issue #275 but the report + # carries Issue #260 from the previous PR. + report = "Linked issue status: Issue #260 open\n" + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275] + ) + self.assertFalse(result["complete"]) + self.assertTrue(any("260" in r for r in result["reasons"])) + + def test_stale_issue_in_diagnostics_fails(self): + report = ( + "Linked issue status: Issue #275 open\n" + "Read-only diagnostics: viewed Issue #260 status\n" + ) + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275] + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("stale" in r.lower() and "260" in r for r in result["reasons"]) + ) + + def test_selected_pr_number_mention_is_not_stale(self): + # "PR #280" mentions must not be confused with issue mentions. + report = ( + "Selected PR: 280\n" + "Validation: ran tests at PR #280 head\n" + "Linked issue status: Issue #275 open\n" + ) + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275] + ) + self.assertTrue(result["complete"]) + + +class TestMissingProof(unittest.TestCase): + def test_status_claim_without_live_proof_fails(self): + report = "Linked issue status: Issue #275 open\n" + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=None + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("live proof" in r.lower() for r in result["reasons"]) + ) + + def test_unverified_wording_without_proof_passes(self): + report = ( + "Linked issue status: not verified in this session\n" + ) + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=None + ) + self.assertTrue(result["complete"]) + + def test_missing_field_fails(self): + report = "Selected PR: 280\n" + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275] + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("linked issue status" in r.lower() for r in result["reasons"]) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 24ea6fa..0a241a4 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -136,6 +136,12 @@ def test_create_issue_final_report_verifier_exported(): assert callable(assess_create_issue_final_report) +def test_prior_blocker_skip_verifier_exported(): + from review_proofs import assess_prior_blocker_skip_proof + + assert callable(assess_prior_blocker_skip_proof) + + def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof @@ -146,3 +152,9 @@ def test_inventory_worktree_verifier_exported(): from review_proofs import assess_inventory_worktree_report assert callable(assess_inventory_worktree_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_blocker_skip.py b/tests/test_reviewer_blocker_skip.py new file mode 100644 index 0000000..b037ede --- /dev/null +++ b/tests/test_reviewer_blocker_skip.py @@ -0,0 +1,139 @@ +"""Tests for prior-blocker skip proof verifier (#318).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_blocker_skip import assess_prior_blocker_skip_proof # noqa: E402 + +HEAD_CURRENT = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9" +HEAD_BLOCKER = "1111111111111111111111111111111111111111" + + +def _good_blocker_skip_report(pr_number: int) -> str: + return "\n".join([ + f"Skipped PR #{pr_number} due to prior REQUEST_CHANGES.", + f"- PR number: #{pr_number}", + f"- Current head SHA: {HEAD_CURRENT}", + "- Blocking review decision: request_changes", + f"- Blocking review head SHA: {HEAD_CURRENT}", + "- Head unchanged since blocker", + "- Reason remains blocked: unresolved review feedback at same head", + "- Blocker revalidated live in this session via gitea_get_pr_review_feedback", + ]) + + +class TestPriorBlockerSkipProof(unittest.TestCase): + def test_no_skips_passes(self): + result = assess_prior_blocker_skip_proof( + "Selected PR #281 for review.", + skipped_prs=[], + ) + self.assertTrue(result["proven"]) + + def test_blocked_unchanged_with_live_proof_passes(self): + report = _good_blocker_skip_report(280) + result = assess_prior_blocker_skip_proof( + report, + skipped_prs=[{ + "pr_number": 280, + "head_sha": HEAD_CURRENT, + "blocking_decision": "request_changes", + "blocking_review_head_sha": HEAD_CURRENT, + "head_changed_since_blocker": False, + "blocker_verified": True, + }], + ) + self.assertTrue(result["proven"]) + + def test_head_changed_after_blocker_blocks_skip(self): + result = assess_prior_blocker_skip_proof( + _good_blocker_skip_report(280), + skipped_prs=[{ + "pr_number": 280, + "head_sha": HEAD_CURRENT, + "blocking_decision": "request_changes", + "blocking_review_head_sha": HEAD_BLOCKER, + "head_changed_since_blocker": True, + "blocker_verified": True, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("head changed" in r for r in result["reasons"])) + + def test_missing_live_proof_blocks(self): + report = _good_blocker_skip_report(280).replace( + "- Blocker revalidated live in this session via gitea_get_pr_review_feedback", + "", + ) + result = assess_prior_blocker_skip_proof( + report, + skipped_prs=[{ + "pr_number": 280, + "head_sha": HEAD_CURRENT, + "blocking_decision": "request_changes", + "blocking_review_head_sha": HEAD_CURRENT, + "head_changed_since_blocker": False, + "blocker_verified": True, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("live blocker proof" in r for r in result["reasons"])) + + def test_blocker_unverified_classification_passes(self): + report = "\n".join([ + "Skipped PR #280.", + "Classification: BLOCKER_STATUS_UNVERIFIED", + "Review feedback could not be fetched live in this session.", + ]) + result = assess_prior_blocker_skip_proof( + report, + skipped_prs=[{ + "pr_number": 280, + "blocking_decision": "request_changes", + "blocker_verified": False, + }], + ) + self.assertTrue(result["proven"]) + + def test_unverified_without_classification_blocks(self): + result = assess_prior_blocker_skip_proof( + "Skipped PR #280 based on memory from last session.", + skipped_prs=[{ + "pr_number": 280, + "blocking_decision": "request_changes", + "blocker_verified": False, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("BLOCKER_STATUS_UNVERIFIED" in r for r in result["reasons"])) + + def test_missing_blocking_head_sha_blocks(self): + report = _good_blocker_skip_report(280).replace( + f"- Blocking review head SHA: {HEAD_CURRENT}", + "", + ) + result = assess_prior_blocker_skip_proof( + report, + skipped_prs=[{ + "pr_number": 280, + "head_sha": HEAD_CURRENT, + "blocking_decision": "request_changes", + "blocking_review_head_sha": HEAD_CURRENT, + "head_changed_since_blocker": False, + "blocker_verified": True, + }], + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_prior_blocker_skip_proof as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file 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