diff --git a/review_proofs.py b/review_proofs.py index e55f38a..b11545f 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1231,7 +1231,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination, role_boundary=None, review_mutation=None, report_text=None, review_decision_lock=None, controller_handoff=None, capability_proof=None, - sweep_proof=None, worktree_proof=None): + sweep_proof=None, worktree_proof=None, + baseline_validation=None, project_root=None): """Required behavior 6 + acceptance criteria: one report, distinct proofs. Combines the individual proof verdicts into the final-report fields the @@ -1258,11 +1259,28 @@ def build_final_report(checkout_proof, inventory, validation, contamination, ) empty_queue_report = assess_empty_queue_report(report_text) + queue_status_report = ( + assess_queue_status_report(report_text) + if report_text and _QUEUE_STATUS_REPORT_HINT.search(report_text) + else {"proven": True, "block": False, "reasons": [], "violations": []} + ) proof_wording = ( assess_proof_wording(report_text) if report_text else {"proven": True, "block": False, "reasons": [], "violations": []} ) + if baseline_validation is None and report_text: + baseline_validation = assess_reviewer_baseline_validation_proof( + report_text=report_text, + project_root=project_root, + ) + elif baseline_validation is None: + baseline_validation = { + "proven": True, + "block": False, + "reasons": [], + "violations": [], + } contamination_status = contamination.get("status", "unknown") checkout_proven = bool(checkout_proof.get("proven")) @@ -1397,6 +1415,16 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "unsupported proof wording in final report (#330)" ) downgrade_reasons.extend(proof_wording.get("reasons", [])) + if not queue_status_report.get("proven"): + downgrade_reasons.append( + "queue-status report violates loaded workflow proof gates (#339)" + ) + downgrade_reasons.extend(queue_status_report.get("reasons", [])) + if not baseline_validation.get("proven"): + downgrade_reasons.append( + "reviewer baseline validation proof missing or failed (#325)" + ) + downgrade_reasons.extend(baseline_validation.get("reasons", [])) merge_allowed = ( identity_eligible @@ -1408,9 +1436,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, # #179: no merge without a proven final live-state recheck. and live_state_proven and worktree_proven + and baseline_validation.get("proven") ) violations = [] + violations.extend(baseline_validation.get("violations", [])) if merge_performed and not merge_allowed: violations.append( "merge was performed/claimed although the proofs did not allow " @@ -1464,6 +1494,14 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "empty_queue_trust_gate_status": empty_queue_report.get("status"), "proof_wording_proven": bool(proof_wording.get("proven")), "proof_wording_violations": list(proof_wording.get("violations") or []), + "queue_status_report_proven": bool(queue_status_report.get("proven")), + "queue_status_violations": list( + queue_status_report.get("violations") or [] + ), + "baseline_validation_proven": bool(baseline_validation.get("proven")), + "baseline_validation_violations": list( + baseline_validation.get("violations") or [] + ), } @@ -3032,6 +3070,180 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict: } +# --------------------------------------------------------------------------- +# Reviewer baseline validation (#325) +# --------------------------------------------------------------------------- + +_TEST_VALIDATION_COMMAND_RE = re.compile( + r"(?:^|\s)(?:venv/)?(?:bin/)?(?:python\s+-m\s+)?" + r"(?:pytest|make\s+test|npm\s+test|cargo\s+test|go\s+test\b)", + re.IGNORECASE, +) + +_PREEXISTING_FAILURE_CLAIM_RE = re.compile( + r"(?:\bsame\s+as\s+master\b|" + r"\bpre-?existing(?:\s+(?:on\s+)?master)?\b|" + r"\bfailures?\s+(?:are|is)\s+pre-?existing\b|" + r"\bmaster\s+also\s+fails?\b|" + r"\bfull-?suite\s+failures?\s+(?:are|is)\s+pre-?existing\b)", + re.IGNORECASE, +) + +_REPORT_VALIDATION_CWD_RE = re.compile( + r"(?:working\s+directory|cwd|directory)\s*:\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: + """True when *path* is inside the project's ``branches/`` directory.""" + 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 _is_test_validation_command(command: str) -> bool: + return bool(_TEST_VALIDATION_COMMAND_RE.search((command or "").strip())) + + +def _is_full_sha(value: str | None) -> bool: + return bool(value and _FULL_SHA.match((value or "").strip())) + + +def assess_reviewer_baseline_validation_proof( + *, + validation_runs: list[dict] | None = None, + baseline_proof: dict | None = None, + report_text: str | None = None, + project_root: str | None = None, +) -> dict: + """#325: reviewer validation and baseline comparison must use ``branches/``. + + *validation_runs* entries: ``command``, ``working_directory`` (or ``cwd``), + optional ``project_root``. + + *baseline_proof* keys when claiming pre-existing master failures: + ``worktree_path``, ``baseline_target_sha``, ``pr_head_sha``, + ``baseline_failures``, ``pr_failures``, ``failure_signatures_match``, + ``clean_before``, ``clean_after``. + """ + reasons: list[str] = [] + violations: list[str] = [] + text = report_text or "" + + runs = list(validation_runs or []) + pending_command = None + for raw_line in text.splitlines(): + line = raw_line.strip().lstrip("-*").strip() + lower = line.lower() + if lower.startswith("validation command:"): + pending_command = line.split(":", 1)[1].strip() + continue + cwd_match = _REPORT_VALIDATION_CWD_RE.search(line) + if cwd_match: + cwd = cwd_match.group(1).strip().rstrip(",.;") + command = pending_command or line + if _is_test_validation_command(command): + runs.append( + { + "command": command, + "working_directory": cwd, + "project_root": project_root, + } + ) + pending_command = None + + for run in runs: + command = (run.get("command") or "").strip() + if not command or not _is_test_validation_command(command): + continue + cwd = (run.get("working_directory") or run.get("cwd") or "").strip() + root = run.get("project_root") or project_root + if not cwd: + reasons.append( + "test validation command stated without working directory; " + "cannot prove branches-only execution (#325)" + ) + continue + if not _path_under_branches(cwd, root): + violations.append( + f"test validation ran outside branches/ worktree: cwd={cwd!r}" + ) + reasons.append( + "reviewer workflow must not run test suites in the main " + f"checkout; cwd {cwd!r} is not under branches/ (#325)" + ) + + claims_preexisting = bool(_PREEXISTING_FAILURE_CLAIM_RE.search(text)) + if claims_preexisting: + proof = baseline_proof or {} + worktree = (proof.get("worktree_path") or "").strip() + if not worktree or not _path_under_branches(worktree, project_root): + reasons.append( + "pre-existing master failure claimed without a baseline " + "worktree path under branches/ (#325)" + ) + if not _is_full_sha(proof.get("baseline_target_sha")): + reasons.append( + "pre-existing master failure claimed without " + "baseline_target_sha proof (#325)" + ) + if not _is_full_sha(proof.get("pr_head_sha")): + reasons.append( + "pre-existing master failure claimed without pr_head_sha " + "proof (#325)" + ) + baseline_failures = proof.get("baseline_failures") + pr_failures = proof.get("pr_failures") + if baseline_failures is None or pr_failures is None: + reasons.append( + "pre-existing master failure claimed without baseline and " + "PR failure listings (#325)" + ) + if proof.get("failure_signatures_match") is not True: + reasons.append( + "pre-existing master failure claimed without proven matching " + "failure signatures (#325)" + ) + if proof.get("clean_before") is not True: + reasons.append( + "baseline worktree clean-before validation not proven (#325)" + ) + if proof.get("clean_after") is not True: + reasons.append( + "baseline worktree clean-after validation not proven (#325)" + ) + + proven = not reasons and not violations + return { + "proven": proven, + "block": bool(violations), + "claims_preexisting": claims_preexisting, + "reasons": reasons, + "violations": violations, + "safe_next_action": ( + "create a clean baseline worktree under branches/, e.g. " + "branches/baseline-master-pr, and rerun validation there" + if not proven + else "proceed" + ), + } + + # --------------------------------------------------------------------------- # Identity disclosure (#305) # --------------------------------------------------------------------------- @@ -3113,9 +3325,22 @@ def assess_email_disclosure( } +_QUEUE_STATUS_REPORT_HINT = re.compile( + r"queue[- ]status|selected pr:\s*none|no pr selected|" + r"queue[- ]status[- ]only", + re.I, +) + +_GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I) + +_NOT_APPLICABLE_VALUE = re.compile( + r"\b(?:none|not applicable|n/?a|not run|not verified|—|-)\b", + re.I, +) + _PRIOR_PROOF_LABEL = re.compile( r"prior (?:blocker|proof|request[- ]changes|feedback)|" - r"head sha unchanged since|prior blocker reused", + r"head sha unchanged since|prior blocker reused|labeled as prior", re.I, ) @@ -3123,7 +3348,8 @@ _PAGINATION_FINALITY_EVIDENCE = re.compile( r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|" r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|" r"no next page|final[- ]page|pr_inventory_trust_gate\.status|" - r"total_count\s*:|pagination.*(?:final|complete)", + r"total_count\s*:|pagination.*(?:final|complete)|" + r"inventory pagination proof:", re.I, ) @@ -3186,6 +3412,48 @@ _PROOF_WORDING_RULES: tuple[dict, ...] = ( }, ) +_QUEUE_STATUS_GATES_NO_PR = ( + "already-landed gate", + "author-safety result", + "merge preflight", +) + +_REVIEW_WORKTREE_DETAIL_FIELDS = ( + "review worktree dirty before validation", + "review worktree dirty after validation", + "review worktree head state", +) + + +def _controller_handoff_field_map(report_text: str | None) -> dict[str, str]: + """Parse ``- Field: value`` lines from the Controller Handoff section.""" + section = _handoff_section_lines(report_text) + if section is None: + return {} + fields: dict[str, str] = {} + for line in section: + stripped = line.strip().lstrip("-*").strip() + if ":" not in stripped: + continue + key, value = stripped.split(":", 1) + fields[key.strip().lower()] = value.strip() + return fields + + +def _queue_status_only_run(fields: dict[str, str], report_text: str) -> bool: + selected = fields.get("selected pr", "") + if selected and _NOT_APPLICABLE_VALUE.search(selected): + return True + if re.search(r"\bnone\b", selected, re.I): + return True + if re.search( + r"no pr selected|queue[- ]status[- ]only|selected pr:\s*none", + report_text or "", + re.I, + ): + return True + return not selected + def assess_proof_wording( report_text: str | None, @@ -3237,6 +3505,93 @@ def assess_proof_wording( } +def assess_queue_status_report( + report_text: str | None, + *, + session_evidence: dict | None = None, +) -> dict: + """Issue #339: reject contradictory reviewer queue-status-only reports.""" + text = report_text or "" + fields = _controller_handoff_field_map(text) + violations: list[str] = [] + + proof = assess_proof_wording(text, session_evidence=session_evidence) + violations.extend(proof.get("violations", [])) + + if re.search(r"prior diagnostic", text, re.I) and not _PRIOR_PROOF_LABEL.search( + text + ): + violations.append( + "prior diagnostic worktree simulation used as current conflict proof" + ) + + queue_only = _queue_status_only_run(fields, text) + if queue_only: + for gate in _QUEUE_STATUS_GATES_NO_PR: + value = fields.get(gate, "") + if value and _GATE_PASSED_VALUE.search(value): + if not _NOT_APPLICABLE_VALUE.search(value): + violations.append( + f"{gate} cannot be 'passed' when no PR is selected (#339)" + ) + + worktree_used = fields.get("review worktree used", "") + if worktree_used and re.search(r"\bfalse\b", worktree_used, re.I): + for detail_field in _REVIEW_WORKTREE_DETAIL_FIELDS: + detail_value = fields.get(detail_field, "") + if ( + detail_value + and not _NOT_APPLICABLE_VALUE.search(detail_value) + ): + violations.append( + f"{detail_field} must be 'not applicable' or 'none' " + "when no review worktree was created" + ) + + blockers = fields.get("blockers", "") + if blockers and re.search(r"\bnone\b", blockers, re.I): + if re.search( + r"all (?:open )?prs? (?:are |is )?(?:conflicted|blocked|unverified)|" + r"every (?:open )?pr (?:is )?(?:conflicted|blocked|unverified)", + text, + re.I, + ): + violations.append( + "Blockers: none contradicts report stating all PRs are " + "conflicted, blocked, or unverified" + ) + + if re.search(r"skipped (?:pr|#)|earlier pr.*skipped", text, re.I): + has_skip_proof = bool( + re.search( + r"current[- ]session|prior blocker reused|conflict proof:|" + r"gitea_view_pr|review[- ]feedback proof|unchanged head", + text, + re.I, + ) + or (session_evidence or {}).get("skip_proof_per_pr") + ) + if not has_skip_proof: + violations.append( + "skipped PRs listed without current-session or labeled prior proof" + ) + + if re.search(r"workflows/review-merge-pr\.md", text, re.I) and violations: + violations.append( + "canonical workflow cited but mandatory queue-status proof gates violated" + ) + + deduped = list(dict.fromkeys(violations)) + proven = not deduped + return { + "proven": proven, + "block": not proven, + "queue_status_only": queue_only, + "violations": deduped, + "reasons": deduped, + } + + # --------------------------------------------------------------------------- # Git ref mutations (#297) # --------------------------------------------------------------------------- @@ -3351,3 +3706,10 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None): "ref_mutating_commands": ref_commands, "fetch_targets": fetch_targets, } + + +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 + + return _assess(report_text, **kwargs) diff --git a/reviewer_mergeability_skip.py b/reviewer_mergeability_skip.py new file mode 100644 index 0000000..565de40 --- /dev/null +++ b/reviewer_mergeability_skip.py @@ -0,0 +1,159 @@ +"""Non-mergeable PR skip conflict-proof verifier for reviewer reports (#322).""" + +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) +_MERGEABILITY_FALSE_RE = re.compile( + r"(?:mergeable\s*:\s*false|not mergeable|non[- ]mergeable|mergeability\s*:\s*false|" + r"mergeability result\s*:\s*false)", + re.IGNORECASE, +) +_CONFLICT_PROOF_RE = re.compile( + r"(?:merge-tree|git merge-tree|gitea_view_pr|gitea_check_pr_eligibility|" + r"conflict proof|mergeability tool|merge simulation|conflicting files?)", + re.IGNORECASE, +) +_CONFLICTING_FILES_RE = re.compile( + r"(?:conflicting files?|conflict files?)\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_HEAD_CHANGED_RE = re.compile( + r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|" + r"head changed since|head unchanged since)", + re.IGNORECASE, +) +_MERGEABILITY_UNVERIFIED_RE = re.compile( + r"\bMERGEABILITY_UNVERIFIED\b", +) + + +def _pr_section(text: str, pr_number: int) -> str: + """Return report lines likely describing one skipped PR.""" + 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_non_mergeable_skip_proof( + report_text: str, + *, + skipped_prs: list[dict] | None = None, +) -> dict[str, Any]: + """Validate skipped non-mergeable PR documentation in reviewer reports (#322).""" + 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) + mergeable = entry.get("mergeable") + verified = entry.get("mergeability_verified") + classification = (entry.get("classification") or "").strip().upper() + head_sha = (entry.get("head_sha") or "").strip() or None + + item_reasons: list[str] = [] + + 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 mergeable is False or verified is False: + if not _MERGEABILITY_UNVERIFIED_RE.search(section) and verified is False: + if "MERGEABILITY_UNVERIFIED" not in classification: + item_reasons.append( + f"PR #{pr_number} conflict proof unavailable; report must classify " + "MERGEABILITY_UNVERIFIED" + ) + elif verified is not False: + if not _MERGEABILITY_FALSE_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing mergeability:false proof" + ) + if not _has_head_sha(section, head_sha): + item_reasons.append( + f"PR #{pr_number} skip missing current head SHA" + ) + if not _CONFLICT_PROOF_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing conflict proof command/tool" + ) + if entry.get("head_changed_since_prior_blocker") is not None: + if not _HEAD_CHANGED_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing head-changed-since-blocker proof" + ) + + if ( + entry.get("conflicting_files") + and not _CONFLICTING_FILES_RE.search(section) + and not any( + path.lower() in section.lower() + for path in (entry.get("conflicting_files") or []) + ) + ): + item_reasons.append( + f"PR #{pr_number} has conflicting files in session proof but " + "report omits file-level conflict proof" + ) + + proven = not item_reasons + assessments.append({ + "pr_number": pr_number, + "proven": proven, + "classification": classification or ( + "MERGEABILITY_UNVERIFIED" if verified is False else "NON_MERGEABLE_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": ( + "document each skipped non-mergeable PR with head SHA, mergeability result, " + "conflict proof command, conflicting files, and head-change status" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/skills/llm-project-workflow/schemas/review-merge-final-report.md b/skills/llm-project-workflow/schemas/review-merge-final-report.md index 237d2f2..08c094b 100644 --- a/skills/llm-project-workflow/schemas/review-merge-final-report.md +++ b/skills/llm-project-workflow/schemas/review-merge-final-report.md @@ -77,6 +77,18 @@ When eligibility class is `ALREADY_LANDED_RECONCILE_REQUIRED`: Identity format: `username / profile` (not personal email unless required — #305). +### Queue-status-only runs (no selected PR) + +When the run inventories the queue but selects no PR for review: + +- Selected PR: `none` +- Already-landed gate, Author-safety result, Merge preflight: `not applicable` or `not run` — never `passed` +- Review worktree detail fields: `not applicable` or `none` when Review worktree used is `false` +- Blockers must not be `none` if the narrative says all open PRs are conflicted, blocked, or unverified +- Inventory pagination proof must cite final-page metadata, not default page-size assumptions + +Verifier: `review_proofs.assess_queue_status_report()`. + Narrative final report and controller handoff must agree on eligibility class, candidate/reviewed head SHA, mutation state, worktree usage, review decision, terminal review mutation, merge result, and linked issue status. \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index f8d3b04..7e0fe23 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -121,4 +121,10 @@ def test_start_issue_template_references_work_issue_workflow(): def test_create_issue_final_report_verifier_exported(): from review_proofs import assess_create_issue_final_report - assert callable(assess_create_issue_final_report) \ No newline at end of file + assert callable(assess_create_issue_final_report) + + +def test_non_mergeable_skip_verifier_exported(): + from review_proofs import assess_non_mergeable_skip_proof + + assert callable(assess_non_mergeable_skip_proof) \ No newline at end of file diff --git a/tests/test_queue_status_report.py b/tests/test_queue_status_report.py new file mode 100644 index 0000000..c8b23aa --- /dev/null +++ b/tests/test_queue_status_report.py @@ -0,0 +1,91 @@ +"""Doc-contract checks for queue-status report verifier (#339).""" +import unittest + +from review_proofs import assess_queue_status_report, build_final_report + +BAD_QUEUE_STATUS_REPORT = """ +## PR Queue Status + +Loaded workflows/review-merge-pr.md. Inventory complete because gitea_list_prs +returned 6 open PRs, fewer than the default Gitea page size. + +PR #286 skipped: prior diagnostic worktree simulation showed merge conflicts. +Live blocker proof for all skipped PRs. + +All open PRs are conflicted or blocked. Blockers: none. + +## Controller Handoff + +- Task: review queue status +- Repo: Scaled-Tech-Consulting/Gitea-Tools +- Role: reviewer +- Identity: reviewer1 / prgs-reviewer +- Selected PR: none +- Inventory pagination proof: assumed complete (6 < 50) +- Already-landed gate: passed +- Author-safety result: passed +- Merge preflight: passed +- Review worktree used: false +- Review worktree dirty before validation: false +- Blockers: none +- Current status: queue inspected; no PR selected +""" + + +class TestQueueStatusReport(unittest.TestCase): + def test_bad_fixture_fails_closed(self): + result = assess_queue_status_report(BAD_QUEUE_STATUS_REPORT) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(result["queue_status_only"]) + joined = " ".join(result["violations"]).lower() + self.assertIn("already-landed gate", joined) + self.assertIn("author-safety", joined) + self.assertIn("merge preflight", joined) + self.assertIn("prior diagnostic", joined) + self.assertIn("blockers", joined) + + def test_passes_with_not_applicable_gates(self): + report = """ +## Controller Handoff +- Selected PR: none +- Inventory pagination proof: is_final_page: true, has_more: false +- Already-landed gate: not applicable +- Author-safety result: not run +- Merge preflight: not applicable +- Review worktree used: false +- Review worktree dirty before validation: not applicable +- Blockers: all PRs skipped pending reviewer +""" + result = assess_queue_status_report(report) + self.assertTrue(result["proven"]) + + def test_rejects_passed_gates_without_selected_pr(self): + report = """ +## Controller Handoff +- Selected PR: none +- Already-landed gate: passed +- Blockers: PR #1 conflicted +""" + result = assess_queue_status_report(report) + self.assertFalse(result["proven"]) + self.assertIn("already-landed gate", " ".join(result["violations"]).lower()) + + def test_build_final_report_downgrades_bad_queue_status(self): + report = build_final_report( + {"proven": True}, + {"complete": True}, + {"claimable": True, "verdict": "strong"}, + {"status": "clean"}, + True, + False, + True, + report_text=BAD_QUEUE_STATUS_REPORT, + ) + self.assertEqual(report["grade"], "downgraded") + self.assertFalse(report["queue_status_report_proven"]) + self.assertTrue(report["queue_status_violations"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index b4683dc..ec411d2 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_reviewer_baseline_validation_proof, assess_capability_evidence, assess_capability_proof, assess_contradictory_no_pr_claim, @@ -2160,5 +2161,125 @@ class TestCreateIssueFinalReport(unittest.TestCase): self.assertTrue(result["downgraded"]) +class TestReviewerBaselineValidationProof(unittest.TestCase): + """Issue #325: baseline validation must use branches/ worktrees.""" + + ROOT = "/repo/Gitea-Tools" + + def _good_baseline_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, + } + proof.update(overrides) + return proof + + def test_main_checkout_test_run_blocks(self): + result = assess_reviewer_baseline_validation_proof( + validation_runs=[ + { + "command": "venv/bin/pytest tests/", + "working_directory": self.ROOT, + "project_root": self.ROOT, + } + ], + project_root=self.ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(result["violations"]) + + def test_branches_worktree_test_run_passes(self): + result = assess_reviewer_baseline_validation_proof( + validation_runs=[ + { + "command": "venv/bin/pytest tests/", + "working_directory": f"{self.ROOT}/branches/review-pr-280", + "project_root": self.ROOT, + } + ], + project_root=self.ROOT, + ) + self.assertTrue(result["proven"]) + + def test_preexisting_claim_without_baseline_proof_fails(self): + result = assess_reviewer_baseline_validation_proof( + report_text=( + "Validation: full-suite failures are pre-existing on master." + ), + project_root=self.ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["claims_preexisting"]) + + def test_preexisting_claim_with_complete_baseline_proof_passes(self): + result = assess_reviewer_baseline_validation_proof( + report_text="Failures are pre-existing on master.", + baseline_proof=self._good_baseline_proof(), + project_root=self.ROOT, + ) + self.assertTrue(result["proven"]) + + def test_mismatched_failures_block_preexisting_claim(self): + result = assess_reviewer_baseline_validation_proof( + report_text="Same as master.", + baseline_proof=self._good_baseline_proof( + failure_signatures_match=False, + pr_failures=["tests/test_other.py::test_other"], + ), + project_root=self.ROOT, + ) + self.assertFalse(result["proven"]) + + def test_report_parsed_main_checkout_cwd_blocks(self): + result = assess_reviewer_baseline_validation_proof( + report_text=( + "- Validation command: venv/bin/pytest tests/\n" + f"- Working directory: {self.ROOT}\n" + ), + project_root=self.ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_build_final_report_downgrades_main_checkout_validation(self): + report = build_final_report( + checkout_proof=_good_checkout(), + inventory=_good_inventory(), + validation=_good_validation(), + contamination=_good_contamination(), + identity_eligible=True, + merge_performed=False, + issue_status_verified=True, + capability_evidence=_good_capability_evidence(), + sweep=_good_sweep(), + live_state=_good_live_state(), + role_boundary=_good_role_boundary(), + review_mutation=_good_review_mutation(), + controller_handoff=_good_handoff(), + capability_proof=_good_capability_proof(), + sweep_proof=_good_secret_sweep(), + worktree_proof={ + "worktree_path": "/repo/branches/review-feat-issue-224", + "porcelain_status": "", + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": True, + }, + report_text=( + "- Validation command: venv/bin/pytest tests/\n" + f"- Working directory: {self.ROOT}\n" + ), + project_root=self.ROOT, + ) + self.assertNotEqual(report["grade"], "A") + self.assertFalse(report["baseline_validation_proven"]) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_reviewer_mergeability_skip.py b/tests/test_reviewer_mergeability_skip.py new file mode 100644 index 0000000..98fcc2d --- /dev/null +++ b/tests/test_reviewer_mergeability_skip.py @@ -0,0 +1,150 @@ +"""Tests for non-mergeable skip conflict-proof verifier (#322).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_mergeability_skip import assess_non_mergeable_skip_proof # noqa: E402 + +HEAD_A = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9" +HEAD_B = "1111111111111111111111111111111111111111" + + +def _good_skip_report(pr_number: int, head: str, files: str = "review_proofs.py") -> str: + return "\n".join([ + f"Skipped PR #{pr_number} (non-mergeable).", + f"- PR number: #{pr_number}", + f"- Current head SHA: {head}", + "- Mergeability result: false", + "- Conflict proof command: git merge-tree $(git merge-base prgs/master prgs/feat) prgs/master prgs/feat", + f"- Conflicting files: {files}", + "- Head unchanged since prior blocker", + "- Blocking category: merge conflict", + ]) + + +class TestNonMergeableSkipProof(unittest.TestCase): + def test_no_skips_passes(self): + report = "Selected PR #203 for review. No earlier PRs skipped." + result = assess_non_mergeable_skip_proof(report, skipped_prs=[]) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_documented_non_mergeable_skip_passes(self): + report = _good_skip_report(282, HEAD_A) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "head_sha": HEAD_A, + "mergeability_verified": True, + "conflicting_files": ["review_proofs.py"], + "head_changed_since_prior_blocker": False, + }], + ) + self.assertTrue(result["proven"]) + + def test_skip_without_conflict_proof_blocks(self): + report = "\n".join([ + "Skipped PR #284 because it was not mergeable.", + f"Head SHA: {HEAD_B}", + ]) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 284, + "mergeable": False, + "head_sha": HEAD_B, + "mergeability_verified": True, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("conflict proof" in r for r in result["reasons"])) + + def test_missing_head_sha_blocks(self): + report = _good_skip_report(282, HEAD_A).replace(HEAD_A, "") + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "head_sha": HEAD_A, + "mergeability_verified": True, + }], + ) + self.assertFalse(result["proven"]) + + def test_mergeability_unverified_classification_passes(self): + report = "\n".join([ + "Skipped PR #282.", + "Classification: MERGEABILITY_UNVERIFIED", + "Conflict proof could not be retrieved in this session.", + ]) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "mergeability_verified": False, + }], + ) + self.assertTrue(result["proven"]) + + def test_unverified_without_classification_blocks(self): + report = "Skipped PR #282 due to conflicts." + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "mergeability_verified": False, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("MERGEABILITY_UNVERIFIED" in r for r in result["reasons"])) + + def test_non_mergeable_without_file_details_still_passes_with_command(self): + report = _good_skip_report(284, HEAD_B, files="not available") + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 284, + "mergeable": False, + "head_sha": HEAD_B, + "mergeability_verified": True, + "conflicting_files": [], + }], + ) + self.assertTrue(result["proven"]) + + def test_stale_head_requires_head_change_field(self): + report = _good_skip_report(282, HEAD_A).replace( + "- Head unchanged since prior blocker", + "", + ) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "head_sha": HEAD_A, + "mergeability_verified": True, + "head_changed_since_prior_blocker": True, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("head-changed" in r for r in result["reasons"])) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_non_mergeable_skip_proof as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file