From 1ead7c94720417d2a71641cf7615006f11cb7080 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 18 Jul 2026 03:41:07 +0800 Subject: [PATCH] fix(review): scope decisions to exact head Co-Authored-By: Codex --- prompts/pr_review.md | 1 + scripts/run_codex_pr_review.py | 1049 ++--------------------------- tests/test_run_codex_pr_review.py | 942 +++----------------------- 3 files changed, 163 insertions(+), 1829 deletions(-) diff --git a/prompts/pr_review.md b/prompts/pr_review.md index 7f5240d..7c79bef 100644 --- a/prompts/pr_review.md +++ b/prompts/pr_review.md @@ -23,6 +23,7 @@ You are reviewing a pull request for a **production quantitative trading and dat - Assign **critical** or **high** only when the supplied PR context proves an exact changed path/line, a current caller or entry point proven by the supplied PR context whether pre-existing or introduced by this PR or an explicitly declared public untrusted boundary, reachability under current configuration and inputs, and concrete correctness, security, or data-integrity impact. State the reachability and impact in the description. If any element is missing, downgrade it to medium or low or omit it. - Do not block on a hypothetical future consumer, forged internal object state, or generic defense-in-depth concern. Do not request a new parser, store, registry, or event-persistence layer unless the changed code already exposes that current boundary and the defect is reachable through it. - Review the entire diff holistically and report all independent actionable findings in one response. Do not stop after the first blocking issue. +- Review only the supplied current exact-head diff. Do not inherit findings, fingerprints, retry counts, or arbitration state from any prior head. - Do not invent backward-compatibility requirements that are absent from the repository and PR contract. If both explicitly define a clean-slate namespace, check for accidental legacy fallback instead of requesting dual-read or migration. This never overrides security or data-integrity findings. - Only for a public JSON/wire contract proven by the reachability rule above, check optional-key presence versus explicit null, recursive JSON-safe types, every identity-bearing integer range, one canonical timestamp representation, deterministic round-trips and digests, immutability, and identifier/path safety. diff --git a/scripts/run_codex_pr_review.py b/scripts/run_codex_pr_review.py index f69229b..6fc5419 100644 --- a/scripts/run_codex_pr_review.py +++ b/scripts/run_codex_pr_review.py @@ -56,27 +56,12 @@ # Risk → block mapping BLOCK_SEVERITIES = frozenset({"critical", "high"}) COMMENT_SEVERITIES = frozenset({"critical", "high", "medium", "low"}) -STREAK_MARKER_PREFIX = "" -FINGERPRINT_MARKER_PREFIX = "" -FINGERPRINTS_MARKER_PREFIX = "" HEAD_SHA_MARKER_PREFIX = "" -FINDING_HISTORY_MARKER_PREFIX = "" -CONTRACT_CONFLICT_MARKER_PREFIX = "" DECISION_MARKER_SUFFIX = " -->" -FINDING_HISTORY_MAX_ROUNDS = 4 -FINDING_HISTORY_MAX_BYTES = 8192 -FINDING_HISTORY_MAX_ENCODED_BYTES = ((FINDING_HISTORY_MAX_BYTES + 2) // 3) * 4 -FINDING_HISTORY_TEXT_LIMIT = 500 -ARBITRATION_REPEAT_THRESHOLD = 2 class ReviewError(RuntimeError): @@ -328,9 +313,10 @@ def build_review_prompt(diff: str, pr_title: str, pr_body: str, repo: str) -> st 4. Assign **critical** or **high** only when the supplied PR context proves all of the following: an exact changed path and line; a current caller or entry point proven by the supplied PR context, whether pre-existing or introduced by this PR, or an explicitly declared public untrusted boundary; reachability under the current configuration and inputs; and a concrete correctness, security, or data-integrity impact. State that reachability and impact in the finding description. If any element is missing, downgrade it to medium or low or omit it. 5. Do not block on a hypothetical future consumer, forged internal object state, or generic defense-in-depth concern. Do not request a new parser, store, registry, or event-persistence layer unless the changed code already exposes that current boundary and the defect is reachable through it. 6. Review the entire diff holistically and report all independent actionable findings in one response. Do not stop after the first blocking issue. -7. Do not invent backward-compatibility requirements that are absent from the repository and PR contract. When the repository and PR explicitly define a clean-slate namespace with legacy compatibility out of scope, review that boundary for accidental fallback instead of requesting dual-read or migration. This never overrides security or data-integrity findings. -8. Only for a public JSON/wire contract proven by rule 4, check optional-key presence versus explicit null, recursive JSON-safe types, every identity-bearing integer range, one canonical timestamp representation, deterministic round-trips and digests, immutability, and identifier/path safety. -9. For each finding, classify its severity: +7. Review only the supplied current exact-head diff. Do not inherit findings, fingerprints, retry counts, or arbitration state from any prior head. +8. Do not invent backward-compatibility requirements that are absent from the repository and PR contract. When the repository and PR explicitly define a clean-slate namespace with legacy compatibility out of scope, review that boundary for accidental fallback instead of requesting dual-read or migration. This never overrides security or data-integrity findings. +9. Only for a public JSON/wire contract proven by rule 4, check optional-key presence versus explicit null, recursive JSON-safe types, every identity-bearing integer range, one canonical timestamp representation, deterministic round-trips and digests, immutability, and identifier/path safety. +10. For each finding, classify its severity: - **critical**: security vulnerability, data loss, production crash - **high**: logic error that produces wrong results, API break, memory/connection leak - **medium**: missing error handling, performance degradation, race condition @@ -784,163 +770,6 @@ def parse_review_output( raise ReviewError(f"Failed to parse Codex review output as JSON: {stripped[:500]}") -def parse_arbitration_output( - text: str, *, require_contract_conflict: bool = False -) -> dict[str, Any]: - """Parse the independent arbiter's constrained verdict.""" - payload = parse_review_output(text, require_findings=False, required_keys=("verdict", "reason")) - verdict = str(payload.get("verdict") or "").strip().lower() - if verdict not in {"clear", "block", "ambiguous"}: - raise ReviewError("Arbitration output verdict must be clear, block, or ambiguous") - reason = str(payload.get("reason") or "").strip() - if not reason: - raise ReviewError("Arbitration output reason is required") - contract_conflict = payload.get("contract_conflict", False) - if require_contract_conflict and "contract_conflict" not in payload: - raise ReviewError("Arbitration output contract_conflict is required") - if not isinstance(contract_conflict, bool): - raise ReviewError("Arbitration output contract_conflict must be a boolean") - result: dict[str, Any] = { - "verdict": verdict, - "reason": reason, - } - if "contract_conflict" in payload: - result["contract_conflict"] = contract_conflict - return result - - -def blocking_finding_fingerprint(findings: list[dict[str, Any]]) -> str: - """Return a stable arbitration-candidate identifier despite wording drift.""" - normalized: list[dict[str, str]] = [] - for finding in findings: - if not isinstance(finding, dict): - continue - normalized.append( - { - "category": str(finding.get("category") or "").strip().lower(), - "file": str(finding.get("file") or "").strip(), - "severity": str(finding.get("severity") or "").strip().lower(), - } - ) - if not normalized: - return "" - payload = json.dumps(sorted(normalized, key=lambda item: json.dumps(item, sort_keys=True)), sort_keys=True) - return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:20] - - -def blocking_finding_fingerprints(findings: list[dict[str, Any]]) -> tuple[str, ...]: - """Return per-finding keys so unrelated findings do not reset arbitration state.""" - return tuple( - sorted( - { - fingerprint - for finding in findings - if isinstance(finding, dict) - for fingerprint in [blocking_finding_fingerprint([finding])] - if fingerprint - } - ) - ) - - -def build_arbitration_prompt( - *, - repo: str, - pr_title: str, - diff: str, - findings: list[dict[str, Any]], - previous_findings: list[dict[str, Any]] | None = None, - previous_head_sha: str = "", - history_state: str = "", -) -> str: - """Ask an independent Codex pass to adjudicate repeated or conflicting findings.""" - diff_limited = _truncate_lines(diff, DEFAULT_MAX_CONTEXT_LINES * 3) - findings_json = json.dumps(findings, ensure_ascii=False, indent=2) - previous_findings_json = json.dumps(previous_findings or [], ensure_ascii=False, indent=2) - return f"""You are the independent Codex review arbiter for a production quantitative codebase. - -The primary reviewer raised the current blocking findings below. Compare them with the prior blocking findings when present, then decide whether every current finding remains valid against the cumulative PR diff. Do not defer to either review round. - -A contract conflict means that following the current suggestion would reverse or contradict the behavior required by a prior finding for the same file/category/severity. Wording drift that preserves the same required behavior is not a conflict. Unrelated findings are not conflicts. - -Treat public interfaces, schemas, tests, and documentation in the base/current source as contract evidence. Use `clear` only when that source of truth proves every current finding false, obsolete, or fixed. Use `block` when a current finding is proven valid. Use `ambiguous` when the source of truth is insufficient. Never clear solely because two reviewers disagree. - -When current blocking findings are empty but the trusted history state is `active_blocking_history`, decide whether the prior blocking findings are demonstrably fixed by the current source-of-truth diff. A clean primary review alone is not evidence. - -Repository: {repo} -PR title: {pr_title} - -## Prior reviewed head -{previous_head_sha or "not available"} - -## Trusted history state -{history_state or "normal"} - -## Prior blocking findings -{previous_findings_json} - -## Current blocking findings -{findings_json} - -## Current PR diff -{diff_limited} - -Return exactly one JSON object: -{{ - "verdict": "clear" | "block" | "ambiguous", - "contract_conflict": true | false, - "reason": "Concrete evidence for the verdict." -}} - -Do not discuss style or generic test coverage. -""" - - -def apply_arbitration_result( - decision: dict[str, Any], arbitration: dict[str, Any] -) -> dict[str, Any]: - """Apply an arbiter verdict without allowing contract-conflict remediation churn.""" - result = dict(decision) - contract_conflict = bool(arbitration.get("contract_conflict")) - if arbitration.get("verdict") == "clear": - result["blocked"] = False - result["cleared_blocking_findings"] = list( - result.get("blocking_findings") or [] - ) - result["blocking_findings"] = [] - result["summary"] = ( - "✅ **Merge allowed**: blocking findings were cleared by independent Codex arbitration" - ) - result["contract_conflict"] = contract_conflict - result["auto_fix_allowed"] = not contract_conflict - result["next_action"] = ( - "contract_arbitration" - if contract_conflict - else "auto_remediation" if result.get("blocked") else "none" - ) - return result - - -def apply_arbitration_failure( - decision: dict[str, Any], error: ReviewError -) -> dict[str, Any]: - """Fail closed when history-aware arbitration cannot establish the contract.""" - result = dict(decision) - result.update( - { - "blocked": True, - "contract_conflict": True, - "auto_fix_allowed": False, - "next_action": "contract_arbitration", - "summary": ( - "🚫 **Merge blocked**: contract arbitration failed closed; " - "automatic remediation is disabled" - ), - } - ) - return result - - # --------------------------------------------------------------------------- # Findings evaluation # --------------------------------------------------------------------------- @@ -1033,279 +862,18 @@ def evaluate_findings( # --------------------------------------------------------------------------- -def _sanitize_history_text(value: Any, limit: int = FINDING_HISTORY_TEXT_LIMIT) -> str: - text = re.sub(r"[\x00-\x1f\x7f]+", " ", str(value or "")).strip() - text = re.sub(r"-----BEGIN [^-]+-----.*?-----END [^-]+-----", "[REDACTED]", text, flags=re.IGNORECASE) - text = re.sub(r"\b(?:AKIA|ASIA|AIDA|AROA)[A-Z0-9]{16}\b", "[REDACTED]", text) - text = re.sub(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b", "[REDACTED]", text) - text = re.sub(r"\bglpat-[A-Za-z0-9_-]{10,}\b", "[REDACTED]", text) - text = re.sub( - r"\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b", - "[REDACTED]", - text, - ) - text = re.sub( - r"\b[a-z][a-z0-9+.-]*://[^\s/@:]+:[^\s/@]+@[^\s]+", - "[REDACTED_DSN]", - text, - flags=re.IGNORECASE, - ) - text = re.sub( - r"(?i)\b(token|secret|password|api[ _-]?key|authorization|cookie)\b\s*[:=]\s*\S+", - r"\1=[REDACTED]", - text, - ) - text = re.sub(r"(?i)\bbearer\s+\S+", "Bearer [REDACTED]", text) - text = re.sub(r"\b(?:gh[pousr]_[A-Za-z0-9_]{20,}|sk-[A-Za-z0-9_-]{20,})\b", "[REDACTED]", text) - text = re.sub( - r"(? dict[str, str]: - return { - "severity": _sanitize_history_text(finding.get("severity"), 20).lower(), - "category": _sanitize_history_text(finding.get("category"), 80).lower(), - "file": _sanitize_history_path(finding.get("file")), - "description": _sanitize_history_text(finding.get("description")), - "suggestion": _sanitize_history_text(finding.get("suggestion")), - } - - -def _sanitize_history_path(value: Any) -> str: - """Preserve stable PR paths while excluding control characters and excess length.""" - return re.sub(r"[\x00-\x1f\x7f]+", "", str(value or "")).strip()[:300] - - -def build_finding_history_marker( - history: list[dict[str, Any]], - findings: list[dict[str, Any]], - head_sha: str, - *, - status: str | None = None, -) -> str: - """Serialize a bounded, sanitized blocking-finding history into a bot marker.""" - sanitized_findings = [_history_finding(finding) for finding in findings if isinstance(finding, dict)] - will_append = bool(sanitized_findings or status) - history_limit = FINDING_HISTORY_MAX_ROUNDS - 1 if will_append else FINDING_HISTORY_MAX_ROUNDS - rounds = list(history[-history_limit:]) - normalized_head_sha = str(head_sha or "").strip().lower() - if will_append: - if not re.fullmatch(r"[0-9a-f]{7,64}", normalized_head_sha): - return build_invalid_finding_history_marker(normalized_head_sha) - rounds.append( - { - "head_sha": normalized_head_sha, - "findings": sanitized_findings, - "status": status or "blocking", - } - ) - payload = {"version": 1, "rounds": rounds[-FINDING_HISTORY_MAX_ROUNDS:]} - raw = json.dumps(payload, ensure_ascii=True, separators=(",", ":")).encode("utf-8") - while len(raw) > FINDING_HISTORY_MAX_BYTES and len(rounds) > 1: - rounds.pop(0) - payload["rounds"] = rounds - raw = json.dumps(payload, ensure_ascii=True, separators=(",", ":")).encode("utf-8") - if len(raw) > FINDING_HISTORY_MAX_BYTES: - overflow_head_sha = normalized_head_sha or str(rounds[-1].get("head_sha") or "") - if not re.fullmatch(r"[0-9a-f]{7,64}", overflow_head_sha): - return build_invalid_finding_history_marker(overflow_head_sha) - overflow_payload = { - "version": 1, - "rounds": [ - { - "head_sha": overflow_head_sha, - "findings": [], - "status": "overflow", - } - ], - } - raw = json.dumps(overflow_payload, separators=(",", ":")).encode("utf-8") - encoded = base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") - return f"{FINDING_HISTORY_MARKER_PREFIX}{encoded}{FINDING_HISTORY_MARKER_SUFFIX}" - - -def build_invalid_finding_history_marker(head_sha: str = "") -> str: - normalized_head_sha = str(head_sha or "").strip().lower() - if re.fullmatch(r"[0-9a-f]{7,64}", normalized_head_sha): - payload = { - "version": 1, - "rounds": [ - { - "head_sha": normalized_head_sha, - "findings": [], - "status": "invalid_history", - } - ], - } - raw = json.dumps(payload, separators=(",", ":")).encode("utf-8") - else: - raw = b'{"version":1,"invalid":true}' - encoded = base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") - return f"{FINDING_HISTORY_MARKER_PREFIX}{encoded}{FINDING_HISTORY_MARKER_SUFFIX}" - - -def parse_finding_history(body: str) -> tuple[list[dict[str, Any]], bool]: - """Recover trusted history; legacy absence is valid, malformed state fails closed.""" - if FINDING_HISTORY_MARKER_PREFIX not in (body or ""): - return [], True - match = re.search( - rf"{re.escape(FINDING_HISTORY_MARKER_PREFIX)}([A-Za-z0-9_-]+){re.escape(FINDING_HISTORY_MARKER_SUFFIX)}", - body or "", - ) - if not match: - return [], False - encoded = match.group(1) - if len(encoded) > FINDING_HISTORY_MAX_ENCODED_BYTES: - return [], False - try: - padding = "=" * (-len(encoded) % 4) - raw = base64.b64decode(encoded + padding, altchars=b"-_", validate=True) - if len(raw) > FINDING_HISTORY_MAX_BYTES: - return [], False - payload = json.loads(raw) - except (ValueError, json.JSONDecodeError): - return [], False - if not isinstance(payload, dict) or payload.get("version") != 1: - return [], False - rounds = payload.get("rounds") - if not isinstance(rounds, list) or len(rounds) > FINDING_HISTORY_MAX_ROUNDS: - return [], False - validated: list[dict[str, Any]] = [] - field_limits = { - "severity": 20, - "category": 80, - "file": 300, - "description": FINDING_HISTORY_TEXT_LIMIT, - "suggestion": FINDING_HISTORY_TEXT_LIMIT, - } - for round_state in rounds: - if not isinstance(round_state, dict): - return [], False - head_sha = round_state.get("head_sha") - findings = round_state.get("findings") - status = round_state.get("status", "blocking") - if not isinstance(head_sha, str) or not re.fullmatch(r"[0-9a-f]{7,64}", head_sha): - return [], False - if not isinstance(findings, list): - return [], False - if status not in {"blocking", "clear", "cleared", "overflow", "invalid_history"}: - return [], False - if set(round_state) not in ({"head_sha", "findings"}, {"head_sha", "findings", "status"}): - return [], False - checked_findings: list[dict[str, str]] = [] - for finding in findings: - if not isinstance(finding, dict) or set(finding) != set(field_limits): - return [], False - if any( - not isinstance(finding[field], str) or len(finding[field]) > limit - for field, limit in field_limits.items() - ): - return [], False - checked_findings.append(dict(finding)) - validated.append( - {"head_sha": head_sha, "findings": checked_findings, "status": status} - ) - return validated, True - - -def previous_matching_findings( - history: list[dict[str, Any]], current_findings: list[dict[str, Any]] -) -> list[dict[str, Any]]: - """Return the latest unresolved historical finding for every current key.""" - current_keys = set(blocking_finding_fingerprints(current_findings)) - matched: dict[str, dict[str, Any]] = {} - resolved: set[str] = set() - for round_state in reversed(history): - prior = round_state.get("findings") - if not isinstance(prior, list): - continue - status = round_state.get("status", "blocking") - if status in {"clear", "cleared"} and not prior: - break - for finding in prior: - if not isinstance(finding, dict): - continue - key = blocking_finding_fingerprint([finding]) - if key not in current_keys or key in resolved or key in matched: - continue - if status in {"clear", "cleared"}: - resolved.add(key) - continue - matched[key] = { - **finding, - "history_head_sha": str(round_state.get("head_sha") or ""), - } - if current_keys.issubset(resolved | set(matched)): - break - return [matched[key] for key in sorted(matched)] - - -def previous_matching_round( - history: list[dict[str, Any]], current_findings: list[dict[str, Any]] -) -> dict[str, Any] | None: - """Compatibility wrapper returning aggregated matches and their newest head.""" - findings = previous_matching_findings(history, current_findings) - if not findings: - return None - heads = [str(finding.get("history_head_sha") or "") for finding in findings] - return {"head_sha": next((head for head in heads if head), ""), "findings": findings} - - -def unresolved_history_findings(history: list[dict[str, Any]]) -> list[dict[str, Any]]: - """Aggregate every latest unresolved finding key in the bounded history.""" - all_findings = [ - finding - for round_state in history - for finding in (round_state.get("findings") or []) - if isinstance(finding, dict) - ] - return previous_matching_findings(history, all_findings) - - -def has_active_blocking_history(history: list[dict[str, Any]]) -> bool: - if not history: - return False - latest = history[-1] - status = latest.get("status", "blocking") - return finding_history_requires_confirmation(history) or ( - status == "blocking" and bool(latest.get("findings")) - ) - - -def finding_history_requires_confirmation(history: list[dict[str, Any]]) -> bool: - return bool( - history - and history[-1].get("status") in {"overflow", "invalid_history"} - ) - - def build_pr_comment( decision: dict[str, Any], pr_url: str, *, - blocking_streak: int = 0, - finding_fingerprint: str = "", - finding_fingerprints: tuple[str, ...] = (), reviewed_head_sha: str = "", - arbitration: dict[str, Any] | None = None, - finding_history_marker: str = "", + review_completed: bool = True, ) -> str: - """Build a markdown comment to post on the PR.""" + """Build a comment containing only the current exact-head result.""" lines = [ "", - f"{STREAK_MARKER_PREFIX}{int(blocking_streak)}{STREAK_MARKER_SUFFIX}", - f"{FINGERPRINT_MARKER_PREFIX}{finding_fingerprint}{FINGERPRINT_MARKER_SUFFIX}", - f"{FINGERPRINTS_MARKER_PREFIX}{','.join(finding_fingerprints)}{FINGERPRINTS_MARKER_SUFFIX}", f"{HEAD_SHA_MARKER_PREFIX}{reviewed_head_sha}{HEAD_SHA_MARKER_SUFFIX}", - finding_history_marker, - f"{CONTRACT_CONFLICT_MARKER_PREFIX}{str(bool(decision.get('contract_conflict'))).lower()}{DECISION_MARKER_SUFFIX}", - f"{AUTO_FIX_ALLOWED_MARKER_PREFIX}{str(bool(decision.get('auto_fix_allowed', True))).lower()}{DECISION_MARKER_SUFFIX}", - f"{NEXT_ACTION_MARKER_PREFIX}{decision.get('next_action', 'none')}{DECISION_MARKER_SUFFIX}", + f"{REVIEW_COMPLETED_MARKER_PREFIX}{str(review_completed).lower()}{DECISION_MARKER_SUFFIX}", f"{IMPLEMENTATION_MARKER_PREFIX}{review_implementation_digest()}{IMPLEMENTATION_MARKER_SUFFIX}", "## 🤖 Codex PR Review", "", @@ -1313,28 +881,12 @@ def build_pr_comment( "", ] - if arbitration: - verdict = arbitration.get("verdict", "") - reason = arbitration.get("reason", "") - emoji = "✅" if verdict == "clear" else "🚫" if verdict == "block" else "⚠️" - lines.extend( - [ - "### ⚖️ Codex Review Arbitration", - "", - f"{emoji} **{verdict or 'error'}**: {reason}", - "", - ] - ) blocking = decision["blocking_findings"] if blocking: lines.extend([ "### 🚫 Blocking Issues", "", - ( - "These primary findings were cleared by independent Codex arbitration:" - if arbitration and arbitration.get("verdict") == "clear" - else "These issues must be fixed before this PR can be merged:" - ), + "These issues must be fixed before this PR can be merged:", "", ]) for i, f in enumerate(blocking, 1): @@ -1437,22 +989,6 @@ def _is_trusted_review_comment(comment: Any) -> bool: return True -def trusted_review_comment_provenance(comment: Any) -> str: - """Derive provenance from API record fields, never from comment markdown.""" - if not _is_trusted_review_comment(comment): - return "" - user = comment["user"] - record = { - "comment_id": comment["id"], - "author_id": user.get("id"), - "author_login": str(user.get("login") or "").casefold(), - "created_at": comment["created_at"], - "updated_at": comment.get("updated_at"), - } - raw = json.dumps(record, sort_keys=True, separators=(",", ":")).encode("utf-8") - return hashlib.sha256(raw).hexdigest()[:24] - - def parse_review_implementation_digest(body: str) -> str: match = re.search( rf"{re.escape(IMPLEMENTATION_MARKER_PREFIX)}([0-9a-f]{{24}}){re.escape(IMPLEMENTATION_MARKER_SUFFIX)}", @@ -1461,83 +997,6 @@ def parse_review_implementation_digest(body: str) -> str: return match.group(1) if match else "" -def parse_blocking_streak(body: str) -> int: - """Read consecutive blocking-round counter from a prior review comment.""" - match = re.search( - rf"{re.escape(STREAK_MARKER_PREFIX)}(\d+){re.escape(STREAK_MARKER_SUFFIX)}", - body or "", - ) - if not match: - return 0 - try: - return max(0, int(match.group(1))) - except ValueError: - return 0 - - -def parse_blocking_fingerprint(body: str) -> str: - """Read the primary finding fingerprint from a prior review comment.""" - match = re.search( - rf"{re.escape(FINGERPRINT_MARKER_PREFIX)}([0-9a-f]*){re.escape(FINGERPRINT_MARKER_SUFFIX)}", - body or "", - ) - return match.group(1) if match else "" - - -def parse_blocking_fingerprints(body: str) -> tuple[str, ...]: - """Read per-finding keys, including comments written before the new marker.""" - match = re.search( - rf"{re.escape(FINGERPRINTS_MARKER_PREFIX)}([0-9a-f,]*){re.escape(FINGERPRINTS_MARKER_SUFFIX)}", - body or "", - ) - if match: - return tuple(sorted({item for item in match.group(1).split(",") if re.fullmatch(r"[0-9a-f]{20}", item)})) - - legacy_findings: list[dict[str, str]] = [] - for severity, category, file_path in re.findall( - r"^#### \d+\. .*?\[([A-Z]+)\] (.+?) in `([^`]+)`$", - body or "", - flags=re.MULTILINE, - ): - legacy_findings.append( - {"severity": severity.lower(), "category": category.lower(), "file": file_path} - ) - return blocking_finding_fingerprints(legacy_findings) - - -def parse_reviewed_head_sha(body: str) -> str: - """Read the reviewed pull-request head SHA from a prior trusted comment.""" - match = re.search( - rf"{re.escape(HEAD_SHA_MARKER_PREFIX)}([0-9a-f]{{7,64}}){re.escape(HEAD_SHA_MARKER_SUFFIX)}", - body or "", - ) - return match.group(1) if match else "" - - -def next_blocking_streak( - previous_streak: int, - *, - blocked: bool, - previous_fingerprint: str = "", - current_fingerprint: str = "", - previous_head_sha: str = "", - current_head_sha: str = "", -) -> int: - """Advance only when the same finding is reviewed on a new PR head.""" - if not blocked: - return 0 - if previous_fingerprint and current_fingerprint and previous_fingerprint == current_fingerprint: - if previous_head_sha and current_head_sha and previous_head_sha != current_head_sha: - return previous_streak + 1 - return max(1, previous_streak) - return 1 - - -def should_arbitrate(*, blocked: bool, streak: int, repeated: bool, new_head: bool) -> bool: - """Arbitrate only after the same finding survives a new author commit.""" - return bool(blocked and repeated and new_head and streak >= ARBITRATION_REPEAT_THRESHOLD) - - def upsert_pr_comment( token: str, repo: str, pr_number: int, body: str ) -> None: @@ -1572,18 +1031,15 @@ def write_decision_outputs(decision_payload: dict[str, Any]) -> None: github_output = os.environ.get("GITHUB_OUTPUT") if github_output: - arbitration = decision_payload.get("arbitration") with open(github_output, "a", encoding="utf-8") as f: f.write(f"blocked={'true' if decision_payload['blocked'] else 'false'}\n") f.write(f"total_findings={decision_payload['total_findings']}\n") f.write(f"blocking_count={len(decision_payload['blocking_findings'])}\n") - f.write(f"blocking_streak={decision_payload.get('blocking_streak', 0)}\n") - f.write(f"repeated_finding={'true' if decision_payload.get('repeated_finding') else 'false'}\n") - f.write(f"new_head={'true' if decision_payload.get('new_head') else 'false'}\n") - f.write(f"arbitration_verdict={arbitration.get('verdict', '') if arbitration else ''}\n") - f.write(f"contract_conflict={'true' if decision_payload.get('contract_conflict') else 'false'}\n") - f.write(f"auto_fix_allowed={'true' if decision_payload.get('auto_fix_allowed') else 'false'}\n") - f.write(f"next_action={decision_payload.get('next_action', 'none')}\n") + f.write( + f"review_completed={'true' if decision_payload['review_completed'] else 'false'}\n" + ) + f.write(f"reviewed_head_sha={decision_payload['reviewed_head_sha']}\n") + f.write(f"failure_kind={decision_payload.get('failure_kind', '')}\n") def publish_review_decision( @@ -1594,20 +1050,10 @@ def publish_review_decision( decision: dict[str, Any], *, exit_code: int, - blocking_streak: int = 0, - finding_fingerprint: str = "", - finding_fingerprints: tuple[str, ...] = (), - repeated_fingerprints: tuple[str, ...] = (), - repeated_finding: bool = False, - previous_head_sha: str = "", current_head_sha: str = "", - reviewed_head_sha: str = "", - new_head: bool = False, - arbitration: dict[str, Any] | None = None, - finding_history_marker: str = "", - history_valid: bool = True, + review_completed: bool = True, ) -> int: - """Publish one consistent comment, artifact, and GitHub step-output decision.""" + """Publish one current-head comment, artifact, and step-output decision.""" upsert_pr_comment( token, repo, @@ -1615,32 +1061,52 @@ def publish_review_decision( build_pr_comment( decision, pr_url, - blocking_streak=blocking_streak, - finding_fingerprint=finding_fingerprint, - finding_fingerprints=finding_fingerprints, - reviewed_head_sha=reviewed_head_sha, - arbitration=arbitration, - finding_history_marker=finding_history_marker, + reviewed_head_sha=current_head_sha, + review_completed=review_completed, ), ) write_decision_outputs( { **decision, - "blocking_streak": blocking_streak, - "finding_fingerprint": finding_fingerprint, - "finding_fingerprints": finding_fingerprints, - "repeated_fingerprints": repeated_fingerprints, - "repeated_finding": repeated_finding, - "previous_head_sha": previous_head_sha, + "review_completed": review_completed, + "reviewed_head_sha": current_head_sha if review_completed else "", "current_head_sha": current_head_sha, - "new_head": new_head, - "arbitration": arbitration, - "history_valid": history_valid, + "review_implementation_digest": review_implementation_digest(), } ) return exit_code +def publish_retryable_review_failure( + token: str, + repo: str, + pr_number: int, + pr_url: str, + current_head_sha: str, +) -> int: + """Record an external retryable failure without marking the head reviewed.""" + return publish_review_decision( + token, + repo, + pr_number, + pr_url, + { + "blocked": False, + "blocking_findings": [], + "non_blocking_findings": [], + "total_findings": 0, + "summary": ( + "⚠️ **Review incomplete**: external retryable failure; " + "this head has not been successfully reviewed." + ), + "failure_kind": "external_retryable", + }, + exit_code=1, + current_head_sha=current_head_sha, + review_completed=False, + ) + + # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- @@ -1675,6 +1141,9 @@ def main() -> int: pr_url = str(pr.get("html_url", "")) head = pr.get("head") if isinstance(pr.get("head"), dict) else {} current_head_sha = str(head.get("sha") or "").strip().lower() + if not re.fullmatch(r"[0-9a-f]{7,64}", current_head_sha): + print("::error::Pull request exact head SHA is missing or invalid", file=sys.stderr) + return 1 print(f"Reviewing PR #{pr_number}: {pr_title}") @@ -1697,85 +1166,10 @@ def main() -> int: if policy.get("policy_errors"): print(f"::warning::Policy errors: {policy['policy_errors']}") - history_source_valid = True - try: - _existing_id, previous_comment = find_existing_review_comment(token, repo, pr_number) - except ReviewError as exc: - print(f"::warning::Failed to fetch prior review comment: {exc}") - previous_comment = "" - history_source_valid = False - previous_streak = parse_blocking_streak(previous_comment) - previous_fingerprint = parse_blocking_fingerprint(previous_comment) - previous_fingerprints = parse_blocking_fingerprints(previous_comment) - previous_head_sha = parse_reviewed_head_sha(previous_comment) - finding_history, history_valid = parse_finding_history(previous_comment) - history_valid = history_source_valid and history_valid - if history_valid and finding_history: - latest_history = finding_history[-1] - if not previous_head_sha: - previous_head_sha = str(latest_history.get("head_sha") or "") - if not previous_fingerprints: - previous_fingerprints = blocking_finding_fingerprints( - unresolved_history_findings(finding_history) - ) - active_blocking_history = has_active_blocking_history(finding_history) - legacy_blocking_state = bool( - not finding_history and (previous_streak > 0 or previous_fingerprints) - ) - - if not history_valid: - decision = { - "blocked": True, - "blocking_findings": [], - "non_blocking_findings": [], - "total_findings": 0, - "summary": ( - "🚫 **Merge blocked**: trusted review history is malformed or oversized; " - "automatic remediation is disabled pending contract arbitration" - ), - "contract_conflict": True, - "auto_fix_allowed": False, - "next_action": "contract_arbitration", - } - if not history_source_valid: - write_decision_outputs( - { - **decision, - "blocking_streak": previous_streak, - "previous_head_sha": previous_head_sha, - "current_head_sha": current_head_sha, - "new_head": False, - "history_valid": False, - "history_source_valid": False, - } - ) - return 1 - return publish_review_decision( - token, - repo, - pr_number, - pr_url, - decision, - exit_code=1, - blocking_streak=previous_streak, - previous_head_sha=previous_head_sha, - current_head_sha=current_head_sha, - reviewed_head_sha=current_head_sha, - new_head=bool(previous_head_sha and current_head_sha != previous_head_sha), - finding_history_marker=build_invalid_finding_history_marker( - current_head_sha - ), - history_valid=False, - ) - - # First pass: classify files. If all files are low-risk, skip review. + # Each invocation reviews only the event's exact head. Prior PR comments are + # presentation records, never input to the current application decision. all_low_risk = changed_files_are_low_risk(changed_paths, policy) - if ( - all_low_risk - and changed_paths - and not active_blocking_history - and not legacy_blocking_state - ): + if all_low_risk and changed_paths: print("All changed files are low-risk (docs/tests). Skipping Codex review.") decision = { "blocked": False, @@ -1783,9 +1177,6 @@ def main() -> int: "non_blocking_findings": [], "total_findings": 0, "summary": "✅ **Merge allowed**: All changes are in docs/tests — Codex review skipped.", - "contract_conflict": False, - "auto_fix_allowed": True, - "next_action": "none", } return publish_review_decision( token, @@ -1795,23 +1186,18 @@ def main() -> int: decision, exit_code=0, current_head_sha=current_head_sha, - reviewed_head_sha=current_head_sha, - finding_history_marker=build_finding_history_marker( - finding_history, [], current_head_sha, status="clear" - ), ) - # Fetch PR diff diff = fetch_pr_diff(token, repo, pr_number) print(f"Fetched diff: {len(diff)} chars, {len(diff.splitlines())} lines") - # Build review prompt prompt = build_review_prompt(diff, pr_title, pr_body, repo) print(f"Built review prompt: {len(prompt)} chars") - # Run Codex review try: - complexity = _estimate_review_complexity(diff, changed_paths, title=pr_title, body=pr_body) + complexity = _estimate_review_complexity( + diff, changed_paths, title=pr_title, body=pr_body + ) output = run_codex_review_with_fallback( prompt, DEFAULT_TIMEOUT_MINUTES, @@ -1820,129 +1206,18 @@ def main() -> int: changed_line_count=len(diff.splitlines()), ) except ReviewError as exc: - if _review_capacity_is_unavailable(exc): - print(f"::warning::Codex review unavailable due to quota or capacity: {exc}") - if active_blocking_history or legacy_blocking_state: - decision = { - "blocked": True, - "blocking_findings": [], - "non_blocking_findings": [], - "total_findings": 0, - "summary": ( - "🚫 **Merge blocked**: review capacity is unavailable while " - "blocking contract history is active" - ), - "contract_conflict": True, - "auto_fix_allowed": False, - "next_action": "contract_arbitration", - } - return publish_review_decision( - token, - repo, - pr_number, - pr_url, - decision, - exit_code=1, - blocking_streak=previous_streak, - finding_fingerprints=previous_fingerprints, - previous_head_sha=previous_head_sha, - current_head_sha=current_head_sha, - reviewed_head_sha=previous_head_sha, - new_head=bool( - previous_head_sha and current_head_sha != previous_head_sha - ), - finding_history_marker=build_finding_history_marker( - finding_history, [], current_head_sha - ), - ) - decision = { - "blocked": False, - "blocking_findings": [], - "non_blocking_findings": [], - "total_findings": 0, - "summary": ( - "⚠️ **Review unavailable**: Codex review quota or capacity is unavailable. " - "Required CI checks remain the merge gate." - ), - "contract_conflict": False, - "auto_fix_allowed": False, - "next_action": "review_retry", - } - return publish_review_decision( - token, - repo, - pr_number, - pr_url, - decision, - exit_code=0, - previous_head_sha=previous_head_sha, - current_head_sha=current_head_sha, - finding_history_marker=build_finding_history_marker( - finding_history, [], current_head_sha - ), - ) - print(f"::error::Codex review failed: {exc}", file=sys.stderr) - decision = { - "blocked": True, - "blocking_findings": [], - "non_blocking_findings": [], - "total_findings": 0, - "summary": "🚫 **Merge blocked**: The Codex review could not be completed.", - "contract_conflict": active_blocking_history, - "auto_fix_allowed": False, - "next_action": "contract_arbitration" if active_blocking_history else "review_retry", - } - return publish_review_decision( - token, - repo, - pr_number, - pr_url, - decision, - exit_code=1, - blocking_streak=previous_streak, - finding_fingerprints=previous_fingerprints, - previous_head_sha=previous_head_sha, - current_head_sha=current_head_sha, - reviewed_head_sha=previous_head_sha, - new_head=bool(previous_head_sha and current_head_sha != previous_head_sha), - finding_history_marker=build_finding_history_marker( - finding_history, [], current_head_sha - ), + print(f"::warning::Codex review external retryable failure: {exc}") + return publish_retryable_review_failure( + token, repo, pr_number, pr_url, current_head_sha ) print(f"Codex output: {len(output)} chars") - - # Parse review output try: review = parse_review_output(output) except ReviewError as exc: - print(f"::warning::Failed to parse review output: {exc}") - decision = { - "blocked": True, - "blocking_findings": [], - "non_blocking_findings": [], - "total_findings": 0, - "summary": "🚫 **Merge blocked**: review output could not be parsed safely.", - "contract_conflict": active_blocking_history, - "auto_fix_allowed": False, - "next_action": "contract_arbitration" if active_blocking_history else "review_retry", - } - return publish_review_decision( - token, - repo, - pr_number, - pr_url, - decision, - exit_code=1, - blocking_streak=previous_streak, - finding_fingerprints=previous_fingerprints, - previous_head_sha=previous_head_sha, - current_head_sha=current_head_sha, - reviewed_head_sha=previous_head_sha, - new_head=bool(previous_head_sha and current_head_sha != previous_head_sha), - finding_history_marker=build_finding_history_marker( - finding_history, [], current_head_sha - ), + print(f"::warning::Codex review parse failure is retryable: {exc}") + return publish_retryable_review_failure( + token, repo, pr_number, pr_url, current_head_sha ) findings = review.get("findings", []) @@ -1950,187 +1225,11 @@ def main() -> int: findings = [] print(f"Found {len(findings)} issue(s)") - # Evaluate findings decision = evaluate_findings(findings, changed_files, policy) - decision.update( - { - "contract_conflict": False, - "auto_fix_allowed": True, - "next_action": "auto_remediation" if decision["blocked"] else "none", - } - ) - history_requires_confirmation = finding_history_requires_confirmation( - finding_history - ) or legacy_blocking_state - active_history_clearance_required = bool( - active_blocking_history and not decision["blocked"] - ) - if history_requires_confirmation: - decision.update( - { - "blocked": True, - "contract_conflict": True, - "auto_fix_allowed": False, - "next_action": "contract_arbitration", - } - ) - elif active_history_clearance_required: - decision.update( - { - "blocked": True, - "contract_conflict": False, - "auto_fix_allowed": False, - "next_action": "contract_arbitration", - "summary": ( - "🚫 **Merge blocked**: active blocking history requires " - "independent source-of-truth clearance" - ), - } - ) - finding_fingerprint = blocking_finding_fingerprint(decision["blocking_findings"]) - finding_fingerprints = blocking_finding_fingerprints(decision["blocking_findings"]) - repeated_fingerprints = tuple(sorted(set(finding_fingerprints).intersection(previous_fingerprints))) - repeated_finding = bool( - decision["blocked"] - and repeated_fingerprints - ) - new_head = bool(previous_head_sha and current_head_sha and previous_head_sha != current_head_sha) - blocking_streak = next_blocking_streak( - previous_streak, - blocked=bool(decision["blocked"]), - previous_fingerprint=repeated_fingerprints[0] if repeated_fingerprints else previous_fingerprint, - current_fingerprint=repeated_fingerprints[0] if repeated_fingerprints else finding_fingerprint, - previous_head_sha=previous_head_sha, - current_head_sha=current_head_sha, - ) - if active_history_clearance_required and finding_history: - previous_findings = unresolved_history_findings(finding_history) - else: - previous_findings = previous_matching_findings( - finding_history, decision["blocking_findings"] - ) - matched_history_heads = sorted( - { - str(finding.get("history_head_sha") or "") - for finding in previous_findings - if finding.get("history_head_sha") - } - ) - arbitration: dict[str, Any] | None = None - confirmation_arbitration_required = bool( - history_requires_confirmation and previous_findings - ) - history_arbitration_required = bool(decision["blocked"] and previous_findings) - repeated_arbitration_required = should_arbitrate( - blocked=bool(decision["blocked"]), - streak=blocking_streak, - repeated=repeated_finding, - new_head=new_head, - ) and not (history_requires_confirmation and not previous_findings) - if confirmation_arbitration_required or history_arbitration_required or repeated_arbitration_required: - arbitration_prompt = build_arbitration_prompt( - repo=repo, - pr_title=pr_title, - diff=diff, - findings=decision["blocking_findings"], - previous_findings=previous_findings, - previous_head_sha=( - ", ".join(matched_history_heads) if matched_history_heads else previous_head_sha - ), - history_state=( - str(finding_history[-1].get("status") or "") - if confirmation_arbitration_required - else "active_blocking_history" if active_history_clearance_required else "" - ), - ) - try: - arbitration = parse_arbitration_output( - run_codex_review_with_fallback( - arbitration_prompt, - DEFAULT_TIMEOUT_MINUTES, - complexity=TASK_COMPLEXITY_HIGH, - changed_file_count=len(changed_paths), - changed_line_count=len(diff.splitlines()), - ), - require_contract_conflict=bool(previous_findings), - ) - except ReviewError as exc: - arbitration = { - "verdict": "ambiguous", - "reason": f"Arbitration failed closed: {exc}", - "contract_conflict": bool(previous_findings), - } - if previous_findings or confirmation_arbitration_required: - decision = apply_arbitration_failure(decision, exc) - else: - decision = apply_arbitration_result(decision, arbitration) - if ( - confirmation_arbitration_required - and arbitration.get("verdict") != "clear" - ): - arbitration["contract_conflict"] = True - decision = apply_arbitration_failure( - decision, ReviewError("history confirmation was not cleared") - ) - elif ( - active_history_clearance_required - and arbitration.get("verdict") != "clear" - ): - decision.update( - { - "blocked": True, - "auto_fix_allowed": False, - "next_action": "contract_arbitration", - } - ) - elif previous_findings and arbitration.get("verdict") == "ambiguous": - arbitration["contract_conflict"] = True - decision = apply_arbitration_failure( - decision, ReviewError("contract evidence is ambiguous") - ) - if arbitration.get("verdict") == "clear": - blocking_streak = 0 - arbitration_cleared = bool(arbitration and arbitration.get("verdict") == "clear") - if (history_requires_confirmation or active_history_clearance_required) and not arbitration_cleared: - finding_history_marker = build_finding_history_marker( - finding_history, - [], - current_head_sha, - status="invalid_history" if legacy_blocking_state and not finding_history else None, - ) - else: - history_status = ( - "cleared" - if arbitration and arbitration.get("verdict") == "clear" - else "blocking" if decision["blocked"] else "clear" - ) - finding_history_marker = build_finding_history_marker( - finding_history, - ( - decision.get("cleared_blocking_findings") or [] - if arbitration_cleared - else decision["blocking_findings"] - ), - current_head_sha, - status=history_status, - ) - serialized_history, serialized_history_valid = parse_finding_history( - finding_history_marker - ) - serialized_history_requires_confirmation = finding_history_requires_confirmation( - serialized_history - ) - if not serialized_history_valid or serialized_history_requires_confirmation: - decision = apply_arbitration_failure( - decision, ReviewError("blocking finding history exceeds its safe bound") - ) if decision["blocked"]: - print( - f"::error::Merge blocked: serious issues found " - f"(repeated-finding streak {blocking_streak})" - ) + print("::error::Merge blocked: serious issues found on current exact head") else: - print("Review passed: no blocking issues") + print("Review passed: no blocking issues on current exact head") return publish_review_decision( token, repo, @@ -2138,20 +1237,8 @@ def main() -> int: pr_url, decision, exit_code=1 if decision["blocked"] else 0, - blocking_streak=blocking_streak, - finding_fingerprint=finding_fingerprint if decision["blocked"] else "", - finding_fingerprints=finding_fingerprints if decision["blocked"] else (), - repeated_fingerprints=repeated_fingerprints, - repeated_finding=repeated_finding, - previous_head_sha=previous_head_sha, current_head_sha=current_head_sha, - new_head=new_head, - arbitration=arbitration, - finding_history_marker=finding_history_marker, - reviewed_head_sha=current_head_sha, - history_valid=serialized_history_valid, ) - if __name__ == "__main__": raise SystemExit(main()) diff --git a/tests/test_run_codex_pr_review.py b/tests/test_run_codex_pr_review.py index f91ee76..77ee10c 100644 --- a/tests/test_run_codex_pr_review.py +++ b/tests/test_run_codex_pr_review.py @@ -26,6 +26,41 @@ def _write_event(self, tmpdir: str, files: list[str]) -> str: ) return str(path) + def _run_main_with_review( + self, tmpdir: str, output: str | Exception, previous_comment: str = "" + ) -> tuple[int, object, object, dict[str, object]]: + env = { + "GH_TOKEN": "token", + "GITHUB_REPOSITORY": "org/repo", + "GITHUB_EVENT_PATH": self._write_event( + tmpdir, ["scripts/run_codex_pr_review.py"] + ), + } + backend_patch = ( + patch("scripts.run_codex_pr_review.run_codex_review_with_fallback", side_effect=output) + if isinstance(output, Exception) + else patch("scripts.run_codex_pr_review.run_codex_review_with_fallback", return_value=output) + ) + previous_cwd = os.getcwd() + try: + os.chdir(tmpdir) + with ( + patch.dict(os.environ, env, clear=True), + patch("scripts.run_codex_pr_review.fetch_pr_files", return_value=[{"filename": "scripts/run_codex_pr_review.py"}]), + patch("scripts.run_codex_pr_review.fetch_pr_diff", return_value="current diff"), + patch("scripts.run_codex_pr_review.load_policy", return_value=run_codex_pr_review._default_policy()), + patch("scripts.run_codex_pr_review.find_existing_review_comment", return_value=(99, previous_comment)), + backend_patch as backend, + patch("scripts.run_codex_pr_review.upsert_pr_comment") as comment, + ): + result = run_codex_pr_review.main() + finally: + os.chdir(previous_cwd) + decision = json.loads( + (Path(tmpdir) / "data/output/codex_pr_review/decision.json").read_text(encoding="utf-8") + ) + return result, backend, comment, decision + def test_changed_files_are_low_risk_only_for_docs_and_tests(self) -> None: policy = run_codex_pr_review.load_policy() self.assertTrue(run_codex_pr_review.changed_files_are_low_risk(["docs/guide.md", "tests/test_x.py"], policy)) @@ -45,6 +80,7 @@ def test_review_prompt_requires_holistic_contract_review(self) -> None: ) self.assertIn("report all independent actionable findings in one response", prompt) self.assertIn("Do not stop after the first blocking issue", prompt) + self.assertIn("current exact-head diff", prompt) self.assertIn("clean-slate", prompt) self.assertIn("optional-key presence versus explicit null", prompt) self.assertIn("every identity-bearing integer", prompt) @@ -91,84 +127,6 @@ def test_isolated_review_runtime_imports_from_the_trusted_bridge(self) -> None: self.assertIn("GH_TOKEN or GITHUB_TOKEN is required", result.stderr) self.assertNotIn("ModuleNotFoundError", result.stderr) - def test_repeated_findings_are_fingerprinted_before_arbitration(self) -> None: - findings = [ - { - "severity": "high", - "category": "logic", - "file": "service/review.py", - "line": 11, - "description": "Leaves a failing review check green after retry.", - "suggestion": "Return a non-zero result.", - } - ] - reordered = [dict(findings[0], line=42)] - reworded = [dict(findings[0], description="The retry path incorrectly returns success.")] - other = [dict(findings[0], file="service/auth.py")] - reclassified = [dict(findings[0], severity="critical")] - - fingerprint = run_codex_pr_review.blocking_finding_fingerprint(findings) - fingerprints = run_codex_pr_review.blocking_finding_fingerprints(findings) - self.assertEqual(fingerprint, run_codex_pr_review.blocking_finding_fingerprint(reordered)) - self.assertEqual(fingerprint, run_codex_pr_review.blocking_finding_fingerprint(reworded)) - self.assertNotEqual(fingerprint, run_codex_pr_review.blocking_finding_fingerprint(other)) - self.assertNotEqual(fingerprint, run_codex_pr_review.blocking_finding_fingerprint(reclassified)) - self.assertEqual(fingerprints, run_codex_pr_review.blocking_finding_fingerprints(reworded)) - self.assertEqual( - run_codex_pr_review.next_blocking_streak( - 1, - blocked=True, - previous_fingerprint=fingerprint, - current_fingerprint=fingerprint, - previous_head_sha="deadbeef", - current_head_sha="feedface", - ), - 2, - ) - self.assertEqual( - run_codex_pr_review.next_blocking_streak( - 2, - blocked=True, - previous_fingerprint=fingerprint, - current_fingerprint=run_codex_pr_review.blocking_finding_fingerprint(other), - previous_head_sha="deadbeef", - current_head_sha="feedface", - ), - 1, - ) - self.assertEqual( - run_codex_pr_review.next_blocking_streak( - 1, - blocked=True, - previous_fingerprint=fingerprint, - current_fingerprint=fingerprint, - previous_head_sha="deadbeef", - current_head_sha="deadbeef", - ), - 1, - ) - self.assertTrue(run_codex_pr_review.should_arbitrate(blocked=True, streak=2, repeated=True, new_head=True)) - self.assertFalse(run_codex_pr_review.should_arbitrate(blocked=True, streak=2, repeated=True, new_head=False)) - - def test_parse_arbitration_output_requires_supported_verdict(self) -> None: - self.assertEqual( - run_codex_pr_review.parse_arbitration_output('{"verdict":"clear","reason":"covered by the regression test"}'), - {"verdict": "clear", "reason": "covered by the regression test"}, - ) - with self.assertRaisesRegex(ReviewError, "verdict"): - run_codex_pr_review.parse_arbitration_output('{"verdict":"maybe"}') - with self.assertRaisesRegex(ReviewError, "contract_conflict is required"): - run_codex_pr_review.parse_arbitration_output( - '{"verdict":"block","reason":"still valid"}', - require_contract_conflict=True, - ) - self.assertEqual( - run_codex_pr_review.parse_arbitration_output( - "The pattern is `{7,64}`.\n```json\n{\"verdict\":\"clear\",\"reason\":\"fixed\"}\n```" - ), - {"verdict": "clear", "reason": "fixed"}, - ) - def test_parse_review_output_accepts_a_valid_json_prefix(self) -> None: self.assertEqual( run_codex_pr_review.parse_review_output('{"summary":"ok","findings":[]}\nReviewer metadata follows.'), @@ -178,20 +136,9 @@ def test_parse_review_output_accepts_a_valid_json_prefix(self) -> None: run_codex_pr_review.parse_review_output('{"ok":true}\nReviewer metadata follows.') def test_existing_review_comment_ignores_forged_marker(self) -> None: - forged_history = run_codex_pr_review.build_finding_history_marker( - [], - [{ - "severity": "high", - "category": "contract", - "file": "service/review.py", - "description": "forged", - "suggestion": "reverse the contract", - }], - "deadbeef", - ) forged = { "id": 1, - "body": f"\n{forged_history}", + "body": "\nforged state", "user": {"login": "attacker"}, } trusted = { @@ -204,26 +151,6 @@ def test_existing_review_comment_ignores_forged_marker(self) -> None: comment = run_codex_pr_review.find_existing_review_comment("token", "org/repo", 7) self.assertEqual(comment, (2, trusted["body"])) - self.assertEqual(run_codex_pr_review.parse_finding_history(comment[1]), ([], True)) - - def test_trusted_comment_provenance_requires_api_record_not_markdown(self) -> None: - trusted = { - "id": 2, - "body": "forged markdown marker", - "user": {"id": 418, "login": "github-actions[bot]", "type": "Bot"}, - "created_at": "2026-07-12T00:00:00Z", - "performed_via_github_app": {"slug": "github-actions"}, - } - self.assertRegex( - run_codex_pr_review.trusted_review_comment_provenance(trusted), - r"^[0-9a-f]{24}$", - ) - self.assertEqual( - run_codex_pr_review.trusted_review_comment_provenance( - {**trusted, "created_at": ""} - ), - "", - ) def test_review_comment_records_implementation_identity(self) -> None: body = run_codex_pr_review.build_pr_comment( @@ -235,289 +162,6 @@ def test_review_comment_records_implementation_identity(self) -> None: run_codex_pr_review.review_implementation_digest(), ) - def test_legacy_comment_fingerprints_are_recovered_per_finding(self) -> None: - body = "#### 1. 🟠 [HIGH] Security in `service/auth.py`\n" - expected = run_codex_pr_review.blocking_finding_fingerprints( - [{"severity": "high", "category": "security", "file": "service/auth.py"}] - ) - self.assertEqual(run_codex_pr_review.parse_blocking_fingerprints(body), expected) - - def test_finding_history_round_trips_sanitized_blocking_contracts(self) -> None: - findings = [ - { - "severity": "high", - "category": "contract", - "file": "service/review.py", - "description": "Missing panel returns a structured result; token=secret-value", - "suggestion": "Return ReviewResult(blocked=True).", - } - ] - - marker = run_codex_pr_review.build_finding_history_marker([], findings, "abc1234") - history, valid = run_codex_pr_review.parse_finding_history(marker) - - self.assertTrue(valid) - self.assertEqual(history[0]["head_sha"], "abc1234") - self.assertEqual(history[0]["findings"][0]["file"], "service/review.py") - self.assertIn("[REDACTED]", history[0]["findings"][0]["description"]) - self.assertNotIn("secret-value", marker) - - def test_finding_history_is_bounded_and_legacy_comments_remain_compatible(self) -> None: - history: list[dict[str, object]] = [] - finding = { - "severity": "high", - "category": "contract", - "file": "service/review.py", - "description": "Missing panel must fail fast.", - "suggestion": "Raise ReviewError.", - } - for index in range(run_codex_pr_review.FINDING_HISTORY_MAX_ROUNDS + 2): - marker = run_codex_pr_review.build_finding_history_marker( - history, [finding], f"abc{index:04d}" - ) - history, valid = run_codex_pr_review.parse_finding_history(marker) - self.assertTrue(valid) - - self.assertEqual(len(history), run_codex_pr_review.FINDING_HISTORY_MAX_ROUNDS) - republished = run_codex_pr_review.build_finding_history_marker( - history, [], "feedface" - ) - republished_history, valid = run_codex_pr_review.parse_finding_history( - republished - ) - self.assertTrue(valid) - self.assertEqual(republished_history, history) - self.assertEqual(run_codex_pr_review.parse_finding_history("legacy comment"), ([], True)) - - def test_finding_history_scrubs_common_credential_formats(self) -> None: - secrets = ( - "AKIAIOSFODNN7EXAMPLE", - "xoxb-123456789012-secretvalue", - "glpat-exampleSecret123", - "eyJhbGciOiJIUzI1NiJ9.payload.signature", - "postgres://reviewer:p%40ssword@db.example.test/audit", - ) - marker = run_codex_pr_review.build_finding_history_marker( - [], - [{ - "severity": "high", - "category": "security", - "file": "service/v2/review.py", - "description": " ".join(secrets), - "suggestion": "Remove quoted credentials.", - }], - "deadbeef", - ) - history, valid = run_codex_pr_review.parse_finding_history(marker) - serialized = json.dumps(history) - - self.assertTrue(valid) - for secret in secrets: - self.assertNotIn(secret, serialized) - self.assertIn("REDACTED", serialized) - self.assertEqual(history[0]["findings"][0]["file"], "service/v2/review.py") - - def test_cleared_history_retains_semantics_without_remaining_active(self) -> None: - head_sha = "7eed3550854ee498edc378f3658e6a8f536299cc" - finding = { - "severity": "high", - "category": "contract", - "file": "service/review.py", - "description": "Missing panel must fail fast.", - "suggestion": "Raise ReviewError.", - } - blocking_marker = run_codex_pr_review.build_finding_history_marker( - [], [finding], "deadbeef" - ) - history, valid = run_codex_pr_review.parse_finding_history(blocking_marker) - marker = run_codex_pr_review.build_finding_history_marker( - history, [finding], head_sha, status="cleared" - ) - history, valid = run_codex_pr_review.parse_finding_history(marker) - - self.assertTrue(valid) - self.assertEqual(history[-1]["head_sha"], head_sha) - self.assertEqual(history[-1]["status"], "cleared") - self.assertFalse(run_codex_pr_review.has_active_blocking_history(history)) - self.assertIsNone( - run_codex_pr_review.previous_matching_round(history, [finding]) - ) - - def test_malformed_or_oversized_history_fails_closed(self) -> None: - malformed = "" - oversized = ( - "" - ) - - self.assertEqual(run_codex_pr_review.parse_finding_history(malformed), ([], False)) - self.assertEqual(run_codex_pr_review.parse_finding_history(oversized), ([], False)) - generated = run_codex_pr_review.build_finding_history_marker( - [], - [{ - "severity": "high", - "category": "contract", - "file": f"service/review_{index}.py", - "description": "contract word " * 50, - "suggestion": "behavior word " * 50, - } for index in range(20)], - "deadbeef", - ) - history, valid = run_codex_pr_review.parse_finding_history(generated) - self.assertTrue(valid) - self.assertEqual(history[0]["status"], "overflow") - self.assertTrue(run_codex_pr_review.has_active_blocking_history(history)) - self.assertTrue( - run_codex_pr_review.finding_history_requires_confirmation(history) - ) - recovered_marker = run_codex_pr_review.build_finding_history_marker( - history, [], "feedface", status="clear" - ) - recovered, valid = run_codex_pr_review.parse_finding_history(recovered_marker) - self.assertTrue(valid) - self.assertFalse(run_codex_pr_review.has_active_blocking_history(recovered)) - - def test_invalid_history_sentinel_is_fail_closed_but_recoverable(self) -> None: - marker = run_codex_pr_review.build_invalid_finding_history_marker("deadbeef") - history, valid = run_codex_pr_review.parse_finding_history(marker) - - self.assertTrue(valid) - self.assertEqual(history[0]["status"], "invalid_history") - self.assertTrue(run_codex_pr_review.has_active_blocking_history(history)) - recovered_marker = run_codex_pr_review.build_finding_history_marker( - history, [], "feedface", status="clear" - ) - recovered, valid = run_codex_pr_review.parse_finding_history(recovered_marker) - self.assertTrue(valid) - self.assertFalse(run_codex_pr_review.has_active_blocking_history(recovered)) - - def test_matching_history_aggregates_keys_from_multiple_rounds(self) -> None: - first = { - "severity": "high", - "category": "contract", - "file": "service/review.py", - "description": "Return a structured result.", - "suggestion": "Return ReviewResult.", - } - second = dict(first, file="service/auth.py", description="Reject invalid auth.") - first_marker = run_codex_pr_review.build_finding_history_marker( - [], [first], "deadbeef" - ) - history, _valid = run_codex_pr_review.parse_finding_history(first_marker) - second_marker = run_codex_pr_review.build_finding_history_marker( - history, [second], "feedface" - ) - history, _valid = run_codex_pr_review.parse_finding_history(second_marker) - - matched = run_codex_pr_review.previous_matching_findings( - history, [first, second] - ) - self.assertEqual(len(matched), 2) - self.assertEqual( - {finding["history_head_sha"] for finding in matched}, - {"deadbeef", "feedface"}, - ) - self.assertEqual( - run_codex_pr_review.unresolved_history_findings(history), matched - ) - - def test_history_aware_arbitration_distinguishes_conflict_from_repetition(self) -> None: - prior = [{ - "severity": "high", - "category": "contract", - "file": "service/review.py", - "description": "Missing panel must return a structured blocked result.", - "suggestion": "Return ReviewResult(blocked=True).", - }] - current = [dict( - prior[0], - description="Missing panel must fail fast.", - suggestion="Raise ReviewError instead of returning a result.", - )] - prompt = run_codex_pr_review.build_arbitration_prompt( - repo="org/repo", - pr_title="fix contract", - diff="diff", - findings=current, - previous_findings=prior, - previous_head_sha="deadbeef", - ) - - self.assertIn("contract_conflict", prompt) - self.assertIn("Return ReviewResult(blocked=True)", prompt) - self.assertIn("Raise ReviewError", prompt) - self.assertEqual( - run_codex_pr_review.parse_arbitration_output( - '{"verdict":"block","reason":"opposite contract",' - '"contract_conflict":true}' - ), - {"verdict": "block", "reason": "opposite contract", "contract_conflict": True}, - ) - self.assertEqual( - run_codex_pr_review.parse_arbitration_output( - '{"verdict":"block","reason":"same behavior",' - '"contract_conflict":false}' - )["contract_conflict"], - False, - ) - - def test_contract_arbitration_result_is_fail_closed(self) -> None: - for arbitration, expected_blocked in ( - ({"verdict": "clear", "reason": "tests prove the contract", "contract_conflict": True}, False), - ({"verdict": "block", "reason": "public API requires fail fast", "contract_conflict": True}, True), - ({"verdict": "ambiguous", "reason": "source of truth is unclear", "contract_conflict": True}, True), - ): - result = run_codex_pr_review.apply_arbitration_result( - {"blocked": True, "summary": "blocked"}, arbitration - ) - self.assertEqual(result["blocked"], expected_blocked) - self.assertTrue(result["contract_conflict"]) - self.assertFalse(result["auto_fix_allowed"]) - self.assertEqual(result["next_action"], "contract_arbitration") - - failed = run_codex_pr_review.apply_arbitration_failure( - {"blocked": True, "summary": "blocked"}, ReviewError("backend failed") - ) - self.assertTrue(failed["blocked"]) - self.assertTrue(failed["contract_conflict"]) - self.assertFalse(failed["auto_fix_allowed"]) - self.assertEqual(failed["next_action"], "contract_arbitration") - cleared = run_codex_pr_review.apply_arbitration_result( - {"blocked": True, "summary": "blocked", "blocking_findings": [{"file": "x.py"}]}, - {"verdict": "clear", "reason": "false positive", "contract_conflict": False}, - ) - self.assertEqual(cleared["blocking_findings"], []) - self.assertEqual(cleared["cleared_blocking_findings"], [{"file": "x.py"}]) - - def test_same_contract_wording_drift_and_unrelated_findings_do_not_conflict(self) -> None: - prior = { - "severity": "high", - "category": "contract", - "file": "service/review.py", - "description": "Return a blocked result when the panel is missing.", - "suggestion": "Return ReviewResult(blocked=True).", - } - marker = run_codex_pr_review.build_finding_history_marker([], [prior], "deadbeef") - history, valid = run_codex_pr_review.parse_finding_history(marker) - reworded = [dict( - prior, - description="A missing panel should produce the structured blocked response.", - )] - unrelated = [dict(prior, file="service/auth.py")] - - self.assertTrue(valid) - self.assertEqual(len(run_codex_pr_review.previous_matching_findings(history, reworded)), 1) - self.assertEqual(run_codex_pr_review.previous_matching_findings(history, unrelated), []) - result = run_codex_pr_review.apply_arbitration_result( - {"blocked": True, "summary": "blocked"}, - {"verdict": "block", "reason": "same behavior", "contract_conflict": False}, - ) - self.assertTrue(result["blocked"]) - self.assertFalse(result["contract_conflict"]) - self.assertTrue(result["auto_fix_allowed"]) - self.assertEqual(result["next_action"], "auto_remediation") - def test_repository_policy_has_no_bypass_fields(self) -> None: policy = run_codex_pr_review.load_policy() self.assertTrue( @@ -721,9 +365,53 @@ def test_main_blocks_high_risk_on_review_infra_error(self) -> None: comment.assert_called_once() backend.assert_called_once() - self.assertIn("Merge blocked", comment.call_args.args[3]) + self.assertIn("Review incomplete", comment.call_args.args[3]) + + def test_main_does_not_carry_blockers_across_heads(self) -> None: + prior = """ + + + +""" + with tempfile.TemporaryDirectory() as tmpdir: + result, backend, comment, _decision = self._run_main_with_review( + tmpdir, '{"summary":"clear","findings":[]}', prior + ) + + self.assertEqual(result, 0) + backend.assert_called_once() + body = comment.call_args.args[3] + self.assertIn("codex-pr-review-head-sha:abc1234", body) + for historical_marker in ("streak", "fingerprint", "history"): + self.assertNotIn(f"codex-pr-review-{historical_marker}", body) + + def test_backend_failure_is_retryable_and_not_a_successful_review(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + result, _backend, comment, decision = self._run_main_with_review( + tmpdir, ReviewError("quota_or_capacity_failure") + ) + + self.assertEqual(result, 1) + self.assertFalse(decision["review_completed"]) + body = comment.call_args.args[3] + self.assertIn("codex-pr-review-completed:false", body) + self.assertIn("external retryable failure", body) + self.assertNotIn("Merge allowed", body) - def test_main_allows_required_ci_to_gate_when_review_quota_is_unavailable(self) -> None: + def test_parse_failure_records_external_retry_without_review_success(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + result, _backend, comment, decision = self._run_main_with_review( + tmpdir, "not json" + ) + + self.assertEqual(result, 1) + self.assertFalse(decision["review_completed"]) + self.assertEqual(decision["reviewed_head_sha"], "") + self.assertEqual(decision["current_head_sha"], "abc1234") + self.assertEqual(decision["failure_kind"], "external_retryable") + self.assertIn("codex-pr-review-completed:false", comment.call_args.args[3]) + + def test_main_retries_when_review_quota_is_unavailable(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: event_path = self._write_event(tmpdir, ["src/app.py"]) env = { @@ -744,13 +432,13 @@ def test_main_allows_required_ci_to_gate_when_review_quota_is_unavailable(self) ), patch("scripts.run_codex_pr_review.upsert_pr_comment") as comment, ): - self.assertEqual(run_codex_pr_review.main(), 0) + self.assertEqual(run_codex_pr_review.main(), 1) comment.assert_called_once() - self.assertIn("Review unavailable", comment.call_args.args[3]) + self.assertIn("Review incomplete", comment.call_args.args[3]) self.assertNotIn("Merge blocked", comment.call_args.args[3]) - def test_main_allows_required_ci_to_gate_when_daily_budget_is_exhausted(self) -> None: + def test_main_retries_when_daily_budget_is_exhausted(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: event_path = self._write_event(tmpdir, ["src/app.py"]) env = { @@ -774,12 +462,12 @@ def test_main_allows_required_ci_to_gate_when_daily_budget_is_exhausted(self) -> ), patch("scripts.run_codex_pr_review.upsert_pr_comment") as comment, ): - self.assertEqual(run_codex_pr_review.main(), 0) + self.assertEqual(run_codex_pr_review.main(), 1) - self.assertIn("Review unavailable", comment.call_args.args[3]) + self.assertIn("Review incomplete", comment.call_args.args[3]) self.assertNotIn("Merge blocked", comment.call_args.args[3]) - def test_main_allows_required_ci_to_gate_when_codex_exec_failure_is_unclassified(self) -> None: + def test_main_retries_when_codex_exec_failure_is_unclassified(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: event_path = self._write_event(tmpdir, ["src/app.py"]) env = { @@ -800,9 +488,9 @@ def test_main_allows_required_ci_to_gate_when_codex_exec_failure_is_unclassified ), patch("scripts.run_codex_pr_review.upsert_pr_comment") as comment, ): - self.assertEqual(run_codex_pr_review.main(), 0) + self.assertEqual(run_codex_pr_review.main(), 1) - self.assertIn("Review unavailable", comment.call_args.args[3]) + self.assertIn("Review incomplete", comment.call_args.args[3]) def test_main_skips_low_risk_docs_before_calling_review_backend(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: @@ -830,198 +518,6 @@ def test_main_skips_low_risk_docs_before_calling_review_backend(self) -> None: comment.assert_called_once() backend.assert_not_called() - def test_main_does_not_skip_low_risk_changes_with_legacy_blocking_state(self) -> None: - with tempfile.TemporaryDirectory() as tmpdir: - event_path = self._write_event(tmpdir, ["docs/guide.md"]) - finding = { - "severity": "high", - "category": "logic", - "file": "scripts/run_codex_pr_review.py", - "description": "Prior source defect.", - "suggestion": "Fix the source defect.", - } - fingerprint = run_codex_pr_review.blocking_finding_fingerprint([finding]) - prior_comment = ( - "\n" - "\n" - f"\n" - "\n" - ) - env = { - "GH_TOKEN": "token", - "GITHUB_REPOSITORY": "org/repo", - "GITHUB_EVENT_PATH": event_path, - } - with ( - patch.dict(os.environ, env, clear=True), - patch( - "scripts.run_codex_pr_review.fetch_pr_files", - return_value=[{"filename": "docs/guide.md"}], - ), - patch("scripts.run_codex_pr_review.fetch_pr_diff", return_value="docs diff"), - patch( - "scripts.run_codex_pr_review.load_policy", - return_value=run_codex_pr_review._default_policy(), - ), - patch( - "scripts.run_codex_pr_review.find_existing_review_comment", - return_value=(99, prior_comment), - ), - patch( - "scripts.run_codex_pr_review.run_codex_review_with_fallback", - return_value='{"summary":"clear","findings":[]}', - ) as backend, - patch("scripts.run_codex_pr_review.upsert_pr_comment") as comment, - ): - self.assertEqual(run_codex_pr_review.main(), 1) - - backend.assert_called_once() - body = comment.call_args.args[3] - self.assertIn("Merge blocked", body) - history, valid = run_codex_pr_review.parse_finding_history(body) - self.assertTrue(valid) - self.assertEqual(history[-1]["status"], "invalid_history") - - def test_main_keeps_legacy_blocker_closed_when_review_backend_unavailable(self) -> None: - with tempfile.TemporaryDirectory() as tmpdir: - event_path = self._write_event(tmpdir, ["scripts/run_codex_pr_review.py"]) - prior_comment = ( - "\n" - "\n" - "\n" - "\n" - ) - env = { - "GH_TOKEN": "token", - "GITHUB_REPOSITORY": "org/repo", - "GITHUB_EVENT_PATH": event_path, - } - with ( - patch.dict(os.environ, env, clear=True), - patch( - "scripts.run_codex_pr_review.fetch_pr_files", - return_value=[{"filename": "scripts/run_codex_pr_review.py"}], - ), - patch("scripts.run_codex_pr_review.fetch_pr_diff", return_value="source diff"), - patch( - "scripts.run_codex_pr_review.load_policy", - return_value=run_codex_pr_review._default_policy(), - ), - patch( - "scripts.run_codex_pr_review.find_existing_review_comment", - return_value=(99, prior_comment), - ), - patch( - "scripts.run_codex_pr_review.run_codex_review_with_fallback", - side_effect=ReviewError("Codex service job timed out"), - ), - patch("scripts.run_codex_pr_review.upsert_pr_comment") as comment, - ): - self.assertEqual(run_codex_pr_review.main(), 1) - - self.assertIn("Merge blocked", comment.call_args.args[3]) - - def test_main_clears_active_history_only_after_independent_arbitration(self) -> None: - with tempfile.TemporaryDirectory() as tmpdir: - event_path = self._write_event(tmpdir, ["scripts/run_codex_pr_review.py"]) - finding = { - "severity": "high", - "category": "logic", - "file": "scripts/run_codex_pr_review.py", - "description": "Prior source defect.", - "suggestion": "Fix the source defect.", - } - prior_comment = ( - "\n" - "\n" - "\n" - + run_codex_pr_review.build_finding_history_marker( - [], [finding], "deadbeef" - ) - ) - env = { - "GH_TOKEN": "token", - "GITHUB_REPOSITORY": "org/repo", - "GITHUB_EVENT_PATH": event_path, - } - with ( - patch.dict(os.environ, env, clear=True), - patch( - "scripts.run_codex_pr_review.fetch_pr_files", - return_value=[{"filename": "scripts/run_codex_pr_review.py"}], - ), - patch("scripts.run_codex_pr_review.fetch_pr_diff", return_value="source diff"), - patch( - "scripts.run_codex_pr_review.load_policy", - return_value=run_codex_pr_review._default_policy(), - ), - patch( - "scripts.run_codex_pr_review.find_existing_review_comment", - return_value=(99, prior_comment), - ), - patch( - "scripts.run_codex_pr_review.run_codex_review_with_fallback", - side_effect=[ - '{"summary":"clear","findings":[]}', - '{"verdict":"clear","reason":"source diff fixes the prior defect",' - '"contract_conflict":false}', - ], - ) as backend, - patch("scripts.run_codex_pr_review.upsert_pr_comment") as comment, - ): - self.assertEqual(run_codex_pr_review.main(), 0) - - self.assertEqual(backend.call_count, 2) - body = comment.call_args.args[3] - history, valid = run_codex_pr_review.parse_finding_history(body) - self.assertTrue(valid) - self.assertEqual(history[-1]["status"], "cleared") - - def test_main_does_not_arbitrate_confirmation_history_without_prior_findings(self) -> None: - with tempfile.TemporaryDirectory() as tmpdir: - event_path = self._write_event(tmpdir, ["scripts/run_codex_pr_review.py"]) - prior_comment = ( - "\n" - "\n" - + run_codex_pr_review.build_invalid_finding_history_marker("deadbeef") - ) - env = { - "GH_TOKEN": "token", - "GITHUB_REPOSITORY": "org/repo", - "GITHUB_EVENT_PATH": event_path, - } - with ( - patch.dict(os.environ, env, clear=True), - patch( - "scripts.run_codex_pr_review.fetch_pr_files", - return_value=[{"filename": "scripts/run_codex_pr_review.py"}], - ), - patch("scripts.run_codex_pr_review.fetch_pr_diff", return_value="source diff"), - patch( - "scripts.run_codex_pr_review.load_policy", - return_value=run_codex_pr_review._default_policy(), - ), - patch( - "scripts.run_codex_pr_review.find_existing_review_comment", - return_value=(99, prior_comment), - ), - patch( - "scripts.run_codex_pr_review.run_codex_review_with_fallback", - return_value='{"summary":"clear","findings":[]}', - ) as backend, - patch("scripts.run_codex_pr_review.upsert_pr_comment") as comment, - ): - self.assertEqual(run_codex_pr_review.main(), 1) - - backend.assert_called_once() - body = comment.call_args.args[3] - history, valid = run_codex_pr_review.parse_finding_history(body) - self.assertTrue(valid) - self.assertEqual(history[-1]["status"], "invalid_history") - self.assertNotIn("Codex Review Arbitration", body) - self.assertIn("codex-pr-review-auto-fix-allowed:false", body) - - def test_main_blocks_unconfigured_backend_even_with_legacy_opt_in(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: event_path = self._write_event(tmpdir, ["scripts/run_codex_pr_review.py"]) @@ -1044,257 +540,7 @@ def test_main_blocks_unconfigured_backend_even_with_legacy_opt_in(self) -> None: self.assertEqual(run_codex_pr_review.main(), 1) comment.assert_called_once() - self.assertIn("Merge blocked", comment.call_args.args[3]) - - def test_blocking_streak_requires_a_matching_fingerprint(self) -> None: - self.assertEqual(run_codex_pr_review.parse_blocking_streak(""), 0) - self.assertEqual( - run_codex_pr_review.parse_blocking_streak( - "\n\n" - ), - 2, - ) - self.assertEqual( - run_codex_pr_review.next_blocking_streak( - 2, - blocked=True, - previous_fingerprint="same", - current_fingerprint="same", - previous_head_sha="deadbeef", - current_head_sha="feedface", - ), - 3, - ) - self.assertEqual( - run_codex_pr_review.next_blocking_streak( - 2, - blocked=True, - previous_fingerprint="old", - current_fingerprint="new", - ), - 1, - ) - self.assertEqual(run_codex_pr_review.next_blocking_streak(2, blocked=False), 0) - - def test_main_clears_repeated_finding_only_after_arbitration(self) -> None: - with tempfile.TemporaryDirectory() as tmpdir: - event = { - "pull_request": { - "number": 7, - "title": "feat: risky", - "body": "", - "html_url": "https://example.test/pr/7", - "labels": [], - "head": {"sha": "abc1234"}, - "base": {"sha": "base123", "repo": {"full_name": "org/repo"}}, - } - } - event_path = Path(tmpdir) / "event.json" - event_path.write_text(json.dumps(event), encoding="utf-8") - env = { - "GH_TOKEN": "token", - "GITHUB_REPOSITORY": "org/repo", - "GITHUB_EVENT_PATH": str(event_path), - "GITHUB_EVENT_NAME": "pull_request", - } - review_json = json.dumps( - { - "summary": "blocking issue", - "findings": [ - { - "severity": "high", - "category": "security", - "file": "scripts/run_codex_pr_review.py", - "line": 1, - "description": "example blocking finding", - "suggestion": "fix it", - } - ], - } - ) - fingerprint = run_codex_pr_review.blocking_finding_fingerprint(json.loads(review_json)["findings"]) - fingerprints = run_codex_pr_review.blocking_finding_fingerprints(json.loads(review_json)["findings"]) - prior_comment = ( - "\n" - "\n" - f"\n" - f"\n" - "\n" - "## prior\n" - ) - with ( - patch.dict(os.environ, env, clear=True), - patch( - "scripts.run_codex_pr_review.fetch_pr_files", - return_value=[{"filename": "scripts/run_codex_pr_review.py"}], - ), - patch( - "scripts.run_codex_pr_review.fetch_pr_diff", - return_value="diff --git a/scripts/run_codex_pr_review.py b/scripts/run_codex_pr_review.py", - ), - patch("scripts.run_codex_pr_review.load_policy", return_value=run_codex_pr_review._default_policy()), - patch( - "scripts.run_codex_pr_review.find_existing_review_comment", - return_value=(99, prior_comment), - ), - patch( - "scripts.run_codex_pr_review.run_codex_review_with_fallback", - side_effect=[review_json, '{"verdict":"clear","reason":"The regression test covers the reported behavior."}'], - ), - patch("scripts.run_codex_pr_review.upsert_pr_comment") as comment, - ): - self.assertEqual(run_codex_pr_review.main(), 1) - - comment.assert_called_once() - self.assertIn("Merge blocked", comment.call_args.args[3]) - body = comment.call_args.args[3] - self.assertIn("codex-pr-review-streak:2", body) - self.assertIn("codex-pr-review-fingerprints:", body) - self.assertNotIn("Codex Review Arbitration", body) - history, valid = run_codex_pr_review.parse_finding_history(body) - self.assertTrue(valid) - self.assertEqual(history[-1]["status"], "invalid_history") - - def test_main_contract_conflict_is_consistent_across_comment_artifact_and_outputs(self) -> None: - with tempfile.TemporaryDirectory() as tmpdir: - event = { - "pull_request": { - "number": 7, - "title": "fix missing panel contract", - "body": "", - "html_url": "https://example.test/pr/7", - "head": {"sha": "abc1234"}, - "base": {"sha": "base123", "repo": {"full_name": "org/repo"}}, - } - } - event_path = Path(tmpdir) / "event.json" - event_path.write_text(json.dumps(event), encoding="utf-8") - github_output = Path(tmpdir) / "github-output.txt" - prior = { - "severity": "high", - "category": "contract", - "file": "service/review.py", - "description": "Missing panel must return a structured blocked result.", - "suggestion": "Return ReviewResult(blocked=True).", - } - current = dict( - prior, - description="Missing panel must fail fast.", - suggestion="Raise ReviewError instead of returning a result.", - ) - fingerprint = run_codex_pr_review.blocking_finding_fingerprint([prior]) - prior_comment = ( - "\n" - "\n" - f"\n" - f"\n" - "\n" - + run_codex_pr_review.build_finding_history_marker([], [prior], "abc1234") - ) - env = { - "GH_TOKEN": "token", - "GITHUB_REPOSITORY": "org/repo", - "GITHUB_EVENT_PATH": str(event_path), - "GITHUB_OUTPUT": str(github_output), - } - previous_cwd = os.getcwd() - try: - os.chdir(tmpdir) - with ( - patch.dict(os.environ, env, clear=True), - patch( - "scripts.run_codex_pr_review.fetch_pr_files", - return_value=[{"filename": "service/review.py"}], - ), - patch( - "scripts.run_codex_pr_review.fetch_pr_diff", - return_value="diff --git a/service/review.py b/service/review.py", - ), - patch( - "scripts.run_codex_pr_review.load_policy", - return_value=run_codex_pr_review._default_policy(), - ), - patch( - "scripts.run_codex_pr_review.find_existing_review_comment", - return_value=(99, prior_comment), - ), - patch( - "scripts.run_codex_pr_review.run_codex_review_with_fallback", - side_effect=[ - json.dumps({"summary": "block", "findings": [current]}), - json.dumps({ - "verdict": "block", - "contract_conflict": True, - "reason": "The suggestions require opposite missing-panel behavior.", - }), - ], - ), - patch("scripts.run_codex_pr_review.upsert_pr_comment") as comment, - ): - self.assertEqual(run_codex_pr_review.main(), 1) - finally: - os.chdir(previous_cwd) - - decision = json.loads( - (Path(tmpdir) / "data/output/codex_pr_review/decision.json").read_text( - encoding="utf-8" - ) - ) - outputs = github_output.read_text(encoding="utf-8") - body = comment.call_args.args[3] - - self.assertTrue(decision["contract_conflict"]) - self.assertFalse(decision["auto_fix_allowed"]) - self.assertEqual(decision["next_action"], "contract_arbitration") - self.assertIn("codex-pr-review-contract-conflict:true", body) - self.assertIn("codex-pr-review-auto-fix-allowed:false", body) - self.assertIn("codex-pr-review-next-action:contract_arbitration", body) - self.assertIn("contract_conflict=true", outputs) - self.assertIn("auto_fix_allowed=false", outputs) - self.assertIn("next_action=contract_arbitration", outputs) - - def test_main_malformed_trusted_history_blocks_before_review(self) -> None: - with tempfile.TemporaryDirectory() as tmpdir: - event_path = self._write_event(tmpdir, ["docs/guide.md"]) - env = { - "GH_TOKEN": "token", - "GITHUB_REPOSITORY": "org/repo", - "GITHUB_EVENT_PATH": event_path, - } - previous_cwd = os.getcwd() - try: - os.chdir(tmpdir) - with ( - patch.dict(os.environ, env, clear=True), - patch( - "scripts.run_codex_pr_review.fetch_pr_files", - return_value=[{"filename": "docs/guide.md"}], - ), - patch( - "scripts.run_codex_pr_review.load_policy", - return_value=run_codex_pr_review._default_policy(), - ), - patch( - "scripts.run_codex_pr_review.find_existing_review_comment", - return_value=(99, "\n"), - ), - patch("scripts.run_codex_pr_review.run_codex_review_with_fallback") as backend, - patch("scripts.run_codex_pr_review.upsert_pr_comment") as comment, - ): - self.assertEqual(run_codex_pr_review.main(), 1) - finally: - os.chdir(previous_cwd) - - decision = json.loads( - (Path(tmpdir) / "data/output/codex_pr_review/decision.json").read_text( - encoding="utf-8" - ) - ) - - backend.assert_not_called() - self.assertFalse(decision["history_valid"]) - self.assertFalse(decision["auto_fix_allowed"]) - self.assertIn("automatic remediation is disabled", comment.call_args.args[3]) + self.assertIn("Review incomplete", comment.call_args.args[3]) def test_main_ignores_legacy_bypass_policy_fields(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: @@ -1391,7 +637,7 @@ def test_main_fails_closed_on_unconfigured_backend_without_opt_in(self) -> None: self.assertEqual(run_codex_pr_review.main(), 1) comment.assert_called_once() - self.assertIn("Merge blocked", comment.call_args.args[3]) + self.assertIn("Review incomplete", comment.call_args.args[3]) def test_main_fails_closed_on_infrastructure_failure_when_policy_requires_it(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: @@ -1428,7 +674,7 @@ def test_main_fails_closed_on_infrastructure_failure_when_policy_requires_it(sel self.assertEqual(run_codex_pr_review.main(), 1) comment.assert_called_once() - self.assertIn("Merge blocked", comment.call_args.args[3]) + self.assertIn("Review incomplete", comment.call_args.args[3]) def test_service_timeout_does_not_fall_back_to_direct_api(self) -> None: with (