diff --git a/review_proofs.py b/review_proofs.py index 4ae699d..60945cb 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1242,6 +1242,102 @@ def assess_workspace_mutation_consistency(report_text, observed_commands=None): } +# ── Already-landed report wording (Issue #298) ─────────────────────────────── +# +# An already-landed PR is reconciliation-only: it must never be called +# eligible, its head SHA must never be called reviewed, and the legacy +# eligible/reviewed/scratch-worktree handoff fields must not appear. +# Reviewed-head claims in any report require validation + diff review to +# have actually passed. + +ALREADY_LANDED_FORBIDDEN_PHRASES = ( + "oldest eligible pr", + "next eligible pr", + "pinned reviewed head", + "scratch worktree used", +) + +ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED" + + +def assess_already_landed_report_wording(report_text, *, already_landed, + review_validated=False): + """#298: reject eligible/reviewed wording for already-landed PRs. + + *already_landed* states whether the already-landed gate fired for the + selected PR. *review_validated* states whether validation and diff + review actually passed this session; only then may a reviewed head + SHA be populated in a non-already-landed report. + + Returns {'complete', 'downgraded', 'reasons'}; fails closed. + """ + text_lower = (report_text or "").lower() + fields = _report_labeled_fields(report_text) + reasons = [] + + reviewed_head = fields.get("reviewed head sha") + reviewed_head_populated = bool( + reviewed_head and not reviewed_head.strip().lower().startswith("none") + ) + pinned_head_present = "pinned reviewed head" in fields + + if already_landed: + for phrase in ALREADY_LANDED_FORBIDDEN_PHRASES: + if phrase in text_lower: + reasons.append( + f"already-landed report must not use '{phrase}'; the PR " + "is reconciliation-only, not review/merge eligible" + ) + + eligibility = fields.get("eligibility class", "") + if ALREADY_LANDED_ELIGIBILITY_CLASS not in eligibility.upper(): + reasons.append( + "already-landed report missing 'Eligibility class: " + f"{ALREADY_LANDED_ELIGIBILITY_CLASS}'" + ) + + for required in ("oldest open pr requiring action", + "candidate head sha"): + if required not in fields: + reasons.append( + f"already-landed report missing '{required}' field" + ) + + if "reviewed head sha" not in fields: + reasons.append( + "already-landed report must state 'Reviewed head SHA: none'" + ) + elif reviewed_head_populated: + reasons.append( + "already-landed report must not claim a reviewed head SHA; " + "no validation or diff review passed for this PR" + ) + + worktree_used = fields.get("review worktree used", "") + if worktree_used.strip().lower() not in ("false", "no"): + reasons.append( + "already-landed report must state 'Review worktree used: " + "false'" + ) + elif not review_validated: + if pinned_head_present: + reasons.append( + "'Pinned reviewed head' populated before validation and " + "diff review passed" + ) + if reviewed_head_populated: + reasons.append( + "reviewed head SHA populated before validation and diff " + "review passed" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None, justification=None): """Assess reviewer/author role separation for blind queue workflows. @@ -1422,6 +1518,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination, report_text=None, review_decision_lock=None, controller_handoff=None, capability_proof=None, sweep_proof=None, worktree_proof=None, + validation_environment=None, queue_ordering=None, baseline_validation=None, project_root=None): """Required behavior 6 + acceptance criteria: one report, distinct proofs. @@ -1450,6 +1547,22 @@ def build_final_report(checkout_proof, inventory, validation, contamination, ) empty_queue_report = assess_empty_queue_report(report_text) + if validation_environment is None and report_text: + validation_env_proof = assess_validation_environment_proof( + report_text=report_text, + ) + elif validation_environment is not None: + validation_env_proof = assess_validation_environment_proof( + report_text=report_text, + validation_environment=validation_environment, + ) + else: + validation_env_proof = { + "proven": True, + "block": False, + "reasons": [], + "violations": [], + } if queue_ordering is None and report_text: queue_ordering_proof = assess_queue_ordering_proof( report_text=report_text, @@ -1622,6 +1735,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "empty-queue report missing or failed trust-gate proof (#198)" ) downgrade_reasons.extend(empty_queue_report.get("reasons", [])) + if not validation_env_proof.get("proven"): + downgrade_reasons.append( + "validation environment proof missing or failed (#311)" + ) + downgrade_reasons.extend(validation_env_proof.get("reasons", [])) if not queue_ordering_proof.get("proven"): downgrade_reasons.append( "queue ordering proof missing or failed (#321)" @@ -1658,11 +1776,13 @@ 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 validation_env_proof.get("proven") and queue_ordering_proof.get("proven") and baseline_validation.get("proven") ) violations = [] + violations.extend(validation_env_proof.get("violations", [])) violations.extend(queue_ordering_proof.get("violations", [])) violations.extend(baseline_validation.get("violations", [])) if merge_performed and not merge_allowed: @@ -1716,6 +1836,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination, else True ), "empty_queue_trust_gate_status": empty_queue_report.get("status"), + "validation_environment_proven": bool( + validation_env_proof.get("proven") + ), + "validation_environment_violations": list( + validation_env_proof.get("violations") or [] + ), "queue_ordering_proven": bool(queue_ordering_proof.get("proven")), "queue_ordering_policy": queue_ordering_proof.get("policy"), "queue_ordering_violations": list( @@ -3336,6 +3462,192 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict: } +# --------------------------------------------------------------------------- +# Validation environment proof (#311) +# --------------------------------------------------------------------------- + +_VALIDATION_CLAIM_RE = re.compile( + r"(?:validation\s+command|pytest\s+(?:executed|passed|failed)|" + r"\d+\s+passed|\d+\s+failed|tests?\s+passed)", + re.IGNORECASE, +) + +_VALIDATION_CMD_LINE_RE = re.compile( + r"validation\s+command\s*:\s*(.+)", + re.IGNORECASE, +) + +_VALIDATION_CWD_LINE_RE = re.compile( + r"working\s+directory\s*:\s*(\S+)", + re.IGNORECASE, +) + +_WHICH_PYTEST_LINE_RE = re.compile( + r"which\s+pytest\s*:\s*(.+)", + re.IGNORECASE, +) + +_PYTEST_VERSION_LINE_RE = re.compile( + r"pytest\s+--version\s*:\s*(.+)", + re.IGNORECASE, +) + +_BARE_PYTEST_CMD_RE = re.compile( + r"^(?:python\s+-m\s+)?pytest(?:\s|$)", + re.IGNORECASE, +) + +_VAGUE_PYTEST_CLAIM_RE = re.compile( + r"pytest\s+(?:executed|passed|failed)\b", + re.IGNORECASE, +) + + +def _parse_validation_command(report_text: str) -> str: + for line in (report_text or "").splitlines(): + match = _VALIDATION_CMD_LINE_RE.search(line) + if match: + return match.group(1).strip().rstrip(".") + return "" + + +def _parse_validation_cwd(report_text: str) -> str: + for line in (report_text or "").splitlines(): + match = _VALIDATION_CWD_LINE_RE.search(line) + if match: + return match.group(1).strip().rstrip(",.;") + return "" + + +def _is_bare_pytest_command(command: str) -> bool: + text = (command or "").strip() + if not text: + return False + if "/" in text or "\\" in text: + return False + return bool(_BARE_PYTEST_CMD_RE.match(text)) + + +def assess_validation_environment_proof( + *, + report_text: str | None = None, + validation_environment: dict | None = None, +) -> dict: + """#311: reviewer reports must include exact validation command/environment.""" + text = report_text or "" + lower = text.lower() + env = validation_environment or {} + reasons: list[str] = [] + violations: list[str] = [] + + claims_validation = bool(_VALIDATION_CLAIM_RE.search(text)) + if not claims_validation and not env.get("command"): + return { + "proven": True, + "block": False, + "claims_validation": False, + "reasons": [], + "violations": [], + "safe_next_action": "proceed", + } + + command = (env.get("command") or _parse_validation_command(text)).strip() + working_directory = ( + env.get("working_directory") or env.get("cwd") + or _parse_validation_cwd(text) + ).strip() + + if not command: + reasons.append( + "validation result claimed without exact validation command (#311)" + ) + elif _VAGUE_PYTEST_CLAIM_RE.search(text) and command == "": + violations.append( + "vague pytest summary without exact validation command (#311)" + ) + + if not working_directory: + reasons.append( + "validation result claimed without working directory (#311)" + ) + + has_result = any( + token in lower + for token in ("passed", "failed", "skipped", "error") + ) or any( + env.get(key) is not None + for key in ("passed", "failed", "skipped", "result") + ) + if not has_result: + reasons.append( + "validation result claimed without pass/fail result evidence (#311)" + ) + + if command and _is_bare_pytest_command(command): + which_pytest = ( + (env.get("which_pytest") or env.get("executable_path") or "") + .strip() + ) + if not which_pytest: + for line in text.splitlines(): + match = _WHICH_PYTEST_LINE_RE.search(line) + if match: + which_pytest = match.group(1).strip() + break + if not which_pytest: + reasons.append( + "bare pytest command requires which-pytest/executable proof (#311)" + ) + + pytest_version = (env.get("pytest_version") or "").strip() + if not pytest_version: + for line in text.splitlines(): + match = _PYTEST_VERSION_LINE_RE.search(line) + if match: + pytest_version = match.group(1).strip() + break + if not pytest_version: + reasons.append( + "bare pytest command requires pytest --version proof (#311)" + ) + + venv_resolved = env.get("venv_resolved") + if venv_resolved is None: + venv_resolved = any( + marker in lower + for marker in ( + "project venv", + "venv/bin/pytest", + "resolves to the project venv", + "venv resolved: yes", + ) + ) + if not venv_resolved and which_pytest and "venv" not in which_pytest: + reasons.append( + "bare pytest command requires proof whether executable " + "resolves to the project venv (#311)" + ) + + proven = not reasons and not violations + return { + "proven": proven, + "block": bool(violations), + "claims_validation": True, + "command": command or None, + "working_directory": working_directory or None, + "reasons": reasons, + "violations": violations, + "safe_next_action": ( + "report exact validation command, working directory, result, and " + "for bare pytest also which-pytest, pytest --version, and venv proof" + if not proven + else "proceed" + ), + } + + + + # --------------------------------------------------------------------------- # Reviewer queue ordering proof (#321) # --------------------------------------------------------------------------- @@ -3666,6 +3978,172 @@ def assess_reviewer_baseline_validation_proof( } +# --------------------------------------------------------------------------- +# Already-landed review gate (#292) +# --------------------------------------------------------------------------- + +ALREADY_LANDED_STATE = "ALREADY_LANDED_RECONCILE_REQUIRED" + +ALREADY_LANDED_HANDOFF_FIELDS = ( + "PR number/title", + "candidate head SHA", + "target branch", + "target branch SHA", + "ancestor proof", + "linked issue status", + "recommended reconciliation action", + "capability proof for close/comment mutations", +) + +_FORBIDDEN_LANDED_STATES_RE = re.compile( + r"review decision\s*:\s*approved?\b|" + r"merge result\s*:\s*merged\b|" + r"\bready[_ ]to[_ ]merge\b|" + r"\bmerge[d]?\s*:\s*success\b", + re.IGNORECASE, +) + + +def assess_already_landed_review_gate( + *, + pr_number: int | None, + candidate_head_sha: str | None, + target_branch: str | None, + target_branch_sha: str | None, + head_is_ancestor_of_target: bool | None, + live_head_sha: str | None = None, + real_blocker: bool = False, + mergeable: bool | None = None, +) -> dict: + """#292: hard gate before any review mutation on an already-landed PR. + + Runs after PR selection and head pinning, before approval, request- + changes, or the merge API. *head_is_ancestor_of_target* is the result + of an ancestry check of *candidate_head_sha* against the freshly + fetched target branch at *target_branch_sha* (e.g. ``git merge-base + --is-ancestor``). ``None`` means the check was not run — the gate then + fails closed. + + *live_head_sha*, when provided, must equal *candidate_head_sha*; a + changed head invalidates the pinned ancestry result until re-checked. + + *mergeable* is accepted for caller convenience but never consulted: + conflict/mergeability handling is a separate gate. + """ + del mergeable # ancestry gate only; mergeability is a separate gate + reasons: list[str] = [] + + if not isinstance(pr_number, int) or pr_number <= 0: + reasons.append("PR number missing or invalid (#292)") + if not _FULL_SHA.match((candidate_head_sha or "").strip()): + reasons.append( + "candidate head SHA is not a full 40-hex commit SHA (#292)" + ) + if not (target_branch or "").strip(): + reasons.append("target branch missing (#292)") + if not _FULL_SHA.match((target_branch_sha or "").strip()): + reasons.append( + "target branch SHA is not a full 40-hex commit SHA; fetch the " + "target branch and record its SHA before the ancestry check " + "(#292)" + ) + if live_head_sha is not None and candidate_head_sha and ( + live_head_sha.strip() != candidate_head_sha.strip() + ): + reasons.append( + "PR head changed since the ancestry check was pinned; re-fetch " + "and re-run the already-landed gate (#292)" + ) + if head_is_ancestor_of_target is None: + reasons.append( + "ancestry of the PR head against the target branch was not " + "checked; the already-landed gate must run before any review " + "mutation (#292)" + ) + + if reasons: + return { + "state": "GATE_NOT_PROVEN", + "approve_allowed": False, + "request_changes_allowed": False, + "merge_api_allowed": False, + "reconciliation_handoff_required": False, + "required_handoff_fields": (), + "reasons": reasons, + "safe_next_action": ( + "fetch the target branch, pin the live PR head, run the " + "ancestry check, then re-run this gate" + ), + } + + if head_is_ancestor_of_target: + return { + "state": ALREADY_LANDED_STATE, + "approve_allowed": False, + "request_changes_allowed": bool(real_blocker), + "merge_api_allowed": False, + "reconciliation_handoff_required": True, + "required_handoff_fields": ALREADY_LANDED_HANDOFF_FIELDS, + "reasons": [ + f"PR #{pr_number} head {candidate_head_sha} is already an " + f"ancestor of {target_branch} @ {target_branch_sha}; the PR " + "is reconciliation-only (#292)" + ], + "safe_next_action": ( + "stop before review mutation and emit an " + f"{ALREADY_LANDED_STATE} reconciliation handoff; any PR/issue " + "close or comment requires exact capability proof" + ), + } + + return { + "state": "NORMAL_REVIEW_CANDIDATE", + "approve_allowed": True, + "request_changes_allowed": True, + "merge_api_allowed": True, + "reconciliation_handoff_required": False, + "required_handoff_fields": (), + "reasons": [], + "safe_next_action": "proceed with the normal review workflow gates", + } + + +def assess_already_landed_report_state( + report_text: str | None, + *, + gate_fired: bool, +) -> dict: + """#292: reject APPROVED/MERGED/READY_TO_MERGE after the gate fired. + + *gate_fired* comes from the session's own gate result, so a report + that omits the already-landed markers entirely still fails closed — + unlike the text-only #327 wording rules this does not depend on the + report admitting the PR was already landed. + """ + if not gate_fired: + return {"complete": True, "block": False, "reasons": []} + + text = report_text or "" + reasons: list[str] = [] + + if _FORBIDDEN_LANDED_STATES_RE.search(text): + reasons.append( + "report claims APPROVED/MERGED/READY_TO_MERGE although the " + "already-landed gate fired (#292)" + ) + if ALREADY_LANDED_STATE.lower() not in text.lower(): + reasons.append( + f"report must state {ALREADY_LANDED_STATE} when the " + "already-landed gate fired (#292)" + ) + + return { + "complete": not reasons, + "block": bool(reasons), + "reasons": reasons, + } + + # --------------------------------------------------------------------------- # Full-suite failure approval gate (#323) # --------------------------------------------------------------------------- @@ -4251,6 +4729,13 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None): } +def assess_merge_simulation_report(report_text, **kwargs): + """#317: classify local merge simulation as worktree/index mutations.""" + from reviewer_merge_simulation import assess_merge_simulation_report as _assess + + return _assess(report_text, **kwargs) + + def assess_prior_blocker_skip_proof(report_text, **kwargs): """#318: require live blocker proof before skipping earlier open PRs.""" from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess @@ -4265,6 +4750,15 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs): return _assess(report_text, **kwargs) +def assess_validation_worktree_edit_report(report_text, **kwargs): + """#315: block edits in PR validation worktrees; require diagnostic separation.""" + from reviewer_validation_worktree_edits import ( + assess_validation_worktree_edit_report as _assess, + ) + + return _assess(report_text, **kwargs) + + def assess_already_landed_handoff_report(report_text, **kwargs): """#299: reject stale fields after the already-landed gate fires.""" from reviewer_already_landed_handoff import assess_already_landed_handoff_report as _assess diff --git a/reviewer_merge_simulation.py b/reviewer_merge_simulation.py new file mode 100644 index 0000000..5397c4f --- /dev/null +++ b/reviewer_merge_simulation.py @@ -0,0 +1,148 @@ +"""Merge-simulation worktree mutation verifier for reviewer reports (#317).""" + +from __future__ import annotations + +import re +from typing import Any + +_WORKTREE_INDEX_FIELD_RE = re.compile( + r"^\s*worktree/index mutations\s*:\s*(.+?)\s*$", + re.IGNORECASE | re.MULTILINE, +) +_READONLY_DIAG_RE = re.compile( + r"read[- ]only diagnostics\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_WORKTREE_PATH_RE = re.compile( + r"(?:worktree path|diagnostic worktree|simulation worktree)\s*:", + re.IGNORECASE, +) +_PRE_CLEAN_RE = re.compile( + r"(?:pre[- ]simulation|before simulation|clean before).*(?:clean|status)", + re.IGNORECASE, +) +_MERGE_RESULT_RE = re.compile( + r"(?:merge result|simulation result|conflict status|merge conflict)", + re.IGNORECASE, +) +_ABORT_RE = re.compile( + r"(?:merge --abort|abort/reset command|abort command|git merge --abort)", + re.IGNORECASE, +) +_POST_CLEAN_RE = re.compile( + r"(?:post[- ]abort|after abort|clean after).*(?:clean|status)", + re.IGNORECASE, +) + + +def _command_text(entry: Any) -> str: + if isinstance(entry, dict): + return str(entry.get("command") or "").strip() + return str(entry or "").strip() + + +def _git_subcommand_line(command: str) -> str | None: + tokens = command.split() + if not tokens or tokens[0] != "git": + return None + return " ".join(tokens[1:]) + + +def merge_simulation_commands(command_log: list | None) -> list[str]: + """Return logged commands that perform local merge simulation (#317).""" + out: list[str] = [] + for entry in command_log or []: + command = _command_text(entry) + rest = _git_subcommand_line(command) + if rest is None: + continue + lower = command.lower() + if rest.startswith("merge"): + if "--no-commit" in lower or ( + "--no-ff" in lower and "merge" in lower and "--commit" not in lower + ): + out.append(command) + elif "--abort" in lower: + out.append(command) + return out + + +def assess_merge_simulation_report( + report_text: str, + *, + command_log: list | None = None, +) -> dict[str, Any]: + """Reject merge simulation classified as read-only; require worktree proof (#317).""" + text = report_text or "" + lower = text.lower() + reasons: list[str] = [] + simulation_commands = merge_simulation_commands(command_log) + has_abort = any("--abort" in c.lower() for c in simulation_commands) + has_simulation = any( + "--no-commit" in c.lower() or ( + "--no-ff" in c.lower() and "--abort" not in c.lower() + ) + for c in simulation_commands + ) + + if not simulation_commands: + return { + "proven": True, + "block": False, + "reasons": [], + "simulation_commands": [], + "safe_next_action": "proceed", + } + + field = _WORKTREE_INDEX_FIELD_RE.search(text) + value = (field.group(1).strip().lower() if field else "") or "" + if not value or value in {"none", "n/a", "no"}: + reasons.append( + "merge simulation commands ran but report lacks " + "'Worktree/index mutations' entry" + ) + elif "merge" not in value and "simulation" not in value: + if not any(cmd.lower() in lower for cmd in simulation_commands): + reasons.append( + "Worktree/index mutations must document merge simulation commands" + ) + + readonly = _READONLY_DIAG_RE.search(text) + if readonly: + readonly_value = readonly.group(1) + for command in simulation_commands: + if command.lower() in readonly_value.lower(): + reasons.append( + "merge simulation may not be listed under read-only diagnostics" + ) + if has_simulation and "merge simulation" in readonly_value.lower(): + reasons.append( + "merge simulation may not be classified as read-only diagnostics" + ) + + if has_simulation: + if not _WORKTREE_PATH_RE.search(text): + reasons.append("merge simulation report missing worktree path") + if not _PRE_CLEAN_RE.search(text): + reasons.append("merge simulation report missing pre-simulation clean status") + if not _MERGE_RESULT_RE.search(text): + reasons.append("merge simulation report missing merge result/conflict status") + if has_abort or has_simulation: + if has_abort and not _ABORT_RE.search(text): + reasons.append("merge simulation report missing abort/reset command proof") + if has_abort and not _POST_CLEAN_RE.search(text): + reasons.append("merge simulation report missing post-abort clean status") + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "simulation_commands": simulation_commands, + "safe_next_action": ( + "classify merge simulation under Worktree/index mutations with full " + "pre/merge/abort/post-clean proof; never list as read-only diagnostics" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/reviewer_validation_worktree_edits.py b/reviewer_validation_worktree_edits.py new file mode 100644 index 0000000..c4e904f --- /dev/null +++ b/reviewer_validation_worktree_edits.py @@ -0,0 +1,184 @@ +"""PR validation worktree read-only edit verifier (#315).""" + +from __future__ import annotations + +import re +from typing import Any + +_VALIDATION_WORKTREE_RE = re.compile( + r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)", + re.IGNORECASE, +) +_DIAGNOSTIC_WORKTREE_RE = re.compile( + r"branches/(?:diagnostic[\w/-]*|scratch[\w/-]*)", + re.IGNORECASE, +) +_FILE_EDITS_NONE_RE = re.compile( + r"file edits by reviewer\s*:\s*none\b", + re.IGNORECASE, +) +_FILE_EDITS_FIELD_RE = re.compile( + r"file edits by reviewer\s*:\s*(.+)", + re.IGNORECASE, +) +_VALIDATION_WORKTREE_PATH_RE = re.compile( + r"(?:validation worktree path|review worktree path)\s*:\s*(\S+)", + re.IGNORECASE, +) +_DIAGNOSTIC_WORKTREE_PATH_RE = re.compile( + r"(?:diagnostic (?:scratch )?worktree path|diagnostic worktree)\s*:\s*(\S+)", + re.IGNORECASE, +) +_DIAGNOSTIC_LABEL_RE = re.compile( + r"diagnostic local experiment|not pr-head validation|diagnostic-only", + re.IGNORECASE, +) +_OFFICIAL_VALIDATION_RE = re.compile( + r"(?:official (?:pr-head )?validation|pr-head validation result)", + re.IGNORECASE, +) +_POST_EDIT_OFFICIAL_RE = re.compile( + r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit)", + re.IGNORECASE, +) + +_PERFORMED_EDIT_KINDS = frozenset({"file_edit", "edit", "write", "edited", "created", "wrote"}) + + +def _normalize_path(path: str) -> str: + return (path or "").replace("\\", "/").strip() + + +def _is_validation_worktree(path: str) -> bool: + return bool(_VALIDATION_WORKTREE_RE.search(_normalize_path(path))) + + +def _is_diagnostic_worktree(path: str) -> bool: + normalized = _normalize_path(path) + return bool( + _DIAGNOSTIC_WORKTREE_RE.search(normalized) + or "diagnostic" in normalized.lower() + ) + + +def _performed_edits(action_log: list[dict] | None) -> list[dict[str, str]]: + edits: list[dict[str, str]] = [] + for entry in action_log or []: + if entry.get("gated_rejected") or entry.get("performed") is False: + continue + kind = (entry.get("kind") or entry.get("action") or "file_edit").strip().lower() + if kind not in _PERFORMED_EDIT_KINDS: + continue + path = (entry.get("path") or "").strip() + if not path: + continue + worktree = _normalize_path(str(entry.get("worktree_path") or entry.get("cwd") or "")) + edits.append({"path": path, "worktree_path": worktree}) + return edits + + +def _extract_validation_path(text: str, session: dict) -> str: + path = _normalize_path(str(session.get("validation_worktree_path") or "")) + if path: + return path + match = _VALIDATION_WORKTREE_PATH_RE.search(text or "") + return _normalize_path(match.group(1)) if match else "" + + +def _extract_diagnostic_path(text: str, session: dict) -> str: + path = _normalize_path(str(session.get("diagnostic_worktree_path") or "")) + if path: + return path + match = _DIAGNOSTIC_WORKTREE_PATH_RE.search(text or "") + return _normalize_path(match.group(1)) if match else "" + + +def assess_validation_worktree_edit_report( + report_text: str, + *, + validation_session: dict | None = None, + action_log: list[dict] | None = None, +) -> dict[str, Any]: + """Block edits in PR validation worktrees; require diagnostic separation (#315).""" + text = report_text or "" + session = dict(validation_session or {}) + reasons: list[str] = [] + + validation_path = _extract_validation_path(text, session) + diagnostic_path = _extract_diagnostic_path(text, session) + validation_edits = list(session.get("validation_worktree_edits") or []) + diagnostic_edits = list(session.get("diagnostic_edits") or []) + edits = _performed_edits(action_log) + + if not validation_edits: + for entry in edits: + wt = entry.get("worktree_path") or validation_path + if wt and _is_validation_worktree(wt) and not _is_diagnostic_worktree(wt): + validation_edits.append(entry["path"]) + + if not diagnostic_edits: + for entry in edits: + wt = entry.get("worktree_path") or "" + if wt and _is_diagnostic_worktree(wt): + diagnostic_edits.append(entry["path"]) + elif entry["path"] and diagnostic_path and not validation_edits: + if _is_diagnostic_worktree(diagnostic_path): + diagnostic_edits.append(entry["path"]) + + claimed_none = bool(_FILE_EDITS_NONE_RE.search(text)) + any_edits = bool(validation_edits or diagnostic_edits or edits) + + if validation_edits: + reasons.append( + "reviewer must not edit files in the PR validation worktree; " + f"observed edits: {', '.join(validation_edits)}" + ) + + if diagnostic_edits or session.get("diagnostic_experiment"): + if not diagnostic_path and not _DIAGNOSTIC_WORKTREE_PATH_RE.search(text): + reasons.append( + "diagnostic experiments require a separate diagnostic scratch worktree path" + ) + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "diagnostic experiments must be labeled " + "'Diagnostic local experiment — not PR-head validation'" + ) + if not _FILE_EDITS_FIELD_RE.search(text) or claimed_none: + reasons.append( + "diagnostic edits must be reported under 'File edits by reviewer'" + ) + + if any_edits and claimed_none: + reasons.append( + "report claims 'File edits by reviewer: none' but reviewer file edits occurred" + ) + + if session.get("official_validation_after_edit") or _POST_EDIT_OFFICIAL_RE.search(text): + if validation_edits or (validation_path and diagnostic_edits): + reasons.append( + "post-edit test results cannot be reported as official PR-head validation" + ) + + if validation_edits and _OFFICIAL_VALIDATION_RE.search(text): + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "validation worktree was edited; official PR-head validation is no longer valid" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "validation_worktree_path": validation_path or None, + "diagnostic_worktree_path": diagnostic_path or None, + "validation_worktree_edits": validation_edits, + "diagnostic_edits": diagnostic_edits, + "safe_next_action": ( + "keep PR validation worktrees read-only; use a separate diagnostic " + "scratch worktree and report all reviewer file edits" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_already_landed_report_wording.py b/tests/test_already_landed_report_wording.py new file mode 100644 index 0000000..ac0379f --- /dev/null +++ b/tests/test_already_landed_report_wording.py @@ -0,0 +1,151 @@ +"""Tests for already-landed final-report wording validator (#298). + +An already-landed PR is reconciliation-only, never review/merge +eligible, and a head SHA may not be called reviewed unless validation +and diff review actually passed. Final reports and controller handoffs +must use the reconciliation wording, not the legacy eligible/reviewed +fields. +""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from review_proofs import assess_already_landed_report_wording # noqa: E402 + + +GOOD_ALREADY_LANDED = ( + "Controller Handoff\n" + "- Oldest open PR requiring action: PR #278\n" + "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n" + "- Candidate head SHA: 2a544b7\n" + "- Reviewed head SHA: none\n" + "- Review worktree used: false\n" +) + + +class TestAlreadyLandedWording(unittest.TestCase): + def test_correct_reconciliation_wording_passes(self): + result = assess_already_landed_report_wording( + GOOD_ALREADY_LANDED, already_landed=True + ) + self.assertTrue(result["complete"]) + self.assertFalse(result["downgraded"]) + self.assertEqual(result["reasons"], []) + + def test_oldest_eligible_wording_fails(self): + report = GOOD_ALREADY_LANDED + "- oldest eligible PR: PR #278\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("oldest eligible pr" in r.lower() for r in result["reasons"]) + ) + + def test_next_eligible_wording_fails(self): + report = GOOD_ALREADY_LANDED + "- next eligible PR: PR #278\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + + def test_pinned_reviewed_head_fails(self): + report = GOOD_ALREADY_LANDED + "- Pinned reviewed head: 2a544b7\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("pinned reviewed head" in r.lower() for r in result["reasons"]) + ) + + def test_scratch_worktree_field_fails(self): + report = GOOD_ALREADY_LANDED + "- Scratch worktree used: False\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + + def test_missing_eligibility_class_fails(self): + report = ( + "Controller Handoff\n" + "- Oldest open PR requiring action: PR #278\n" + "- Candidate head SHA: 2a544b7\n" + "- Reviewed head SHA: none\n" + "- Review worktree used: false\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("eligibility class" in r.lower() for r in result["reasons"]) + ) + + def test_populated_reviewed_head_sha_fails(self): + report = GOOD_ALREADY_LANDED.replace( + "- Reviewed head SHA: none\n", + "- Reviewed head SHA: 2a544b7\n", + ) + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("reviewed head" in r.lower() for r in result["reasons"]) + ) + + +class TestNonAlreadyLandedReports(unittest.TestCase): + def test_normal_reviewed_report_passes(self): + report = ( + "- Selected PR: 281\n" + "- Pinned reviewed head: abc1234\n" + "- Review decision: approve\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=True + ) + self.assertTrue(result["complete"]) + + def test_blocked_infra_report_with_pinned_head_fails(self): + report = ( + "- Selected PR: 281\n" + "- Pinned reviewed head: abc1234\n" + "- Current status: blocked on infra_stop\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("before validation" in r.lower() for r in result["reasons"]) + ) + + def test_validation_failed_report_with_reviewed_sha_fails(self): + report = ( + "- Selected PR: 281\n" + "- Reviewed head SHA: abc1234\n" + "- Validation: full suite failed\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertFalse(result["complete"]) + + def test_validation_failed_report_with_reviewed_none_passes(self): + report = ( + "- Selected PR: 281\n" + "- Reviewed head SHA: none\n" + "- Validation: full suite failed\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertTrue(result["complete"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 6521c6c..82c18a2 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -136,6 +136,12 @@ def test_create_issue_final_report_verifier_exported(): assert callable(assess_create_issue_final_report) +def test_merge_simulation_verifier_exported(): + from review_proofs import assess_merge_simulation_report + + assert callable(assess_merge_simulation_report) + + def test_prior_blocker_skip_verifier_exported(): from review_proofs import assess_prior_blocker_skip_proof @@ -148,6 +154,12 @@ def test_non_mergeable_skip_verifier_exported(): assert callable(assess_non_mergeable_skip_proof) +def test_validation_worktree_edit_verifier_exported(): + from review_proofs import assess_validation_worktree_edit_report + + assert callable(assess_validation_worktree_edit_report) + + def test_already_landed_handoff_verifier_exported(): from review_proofs import assess_already_landed_handoff_report diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 8721545..0d42359 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -23,7 +23,10 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par from review_proofs import ( # noqa: E402 ISSUE_SELECTION_CONTINUATION_EXPLICIT, ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR, + assess_already_landed_report_state, + assess_already_landed_review_gate, assess_author_pr_report, + assess_validation_environment_proof, assess_full_suite_failure_approval_gate, assess_queue_ordering_proof, assess_reviewer_baseline_validation_proof, @@ -2315,10 +2318,108 @@ class TestCreateIssueFinalReport(unittest.TestCase): self.assertTrue(result["downgraded"]) -class TestQueueOrderingProof(unittest.TestCase): - """Issue #321: reviewer queue selection must prove ordering.""" +class TestValidationEnvironmentProof(unittest.TestCase): + """Issue #311: exact validation command and execution environment.""" + + ROOT = "/repo/Gitea-Tools/branches/review-pr-280" + + def test_venv_pytest_with_full_report_passes(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: venv/bin/pytest tests/ --ignore=branches\n" + f"Working directory: {self.ROOT}\n" + "Result: 1014 passed, 6 skipped" + ), + ) + self.assertTrue(result["proven"]) + + def test_bare_pytest_without_path_proof_fails(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: pytest\n" + f"Working directory: {self.ROOT}\n" + "1014 passed, 6 skipped" + ), + ) + self.assertFalse(result["proven"]) + + def test_bare_pytest_with_path_and_version_proof_passes(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: pytest tests/\n" + f"Working directory: {self.ROOT}\n" + "which pytest: /repo/Gitea-Tools/venv/bin/pytest\n" + "pytest --version: pytest 8.3.4\n" + "Resolves to the project venv: yes\n" + "1014 passed, 6 skipped" + ), + ) + self.assertTrue(result["proven"]) + + def test_vague_pytest_summary_without_command_fails(self): + result = assess_validation_environment_proof( + report_text=( + f"Working directory: {self.ROOT}\n" + "pytest executed inside the worktree: 1014 passed, 6 skipped" + ), + ) + self.assertFalse(result["proven"]) + + def test_structured_environment_proof_passes(self): + result = assess_validation_environment_proof( + validation_environment={ + "command": "pytest tests/", + "working_directory": self.ROOT, + "which_pytest": "/repo/Gitea-Tools/venv/bin/pytest", + "pytest_version": "pytest 8.3.4", + "venv_resolved": True, + "passed": 1014, + "skipped": 6, + }, + ) + self.assertTrue(result["proven"]) + + def test_missing_working_directory_fails(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: venv/bin/pytest tests/\n" + "1014 passed, 6 skipped" + ), + ) + self.assertFalse(result["proven"]) + + def test_build_final_report_downgrades_missing_environment_proof(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=( + f"Working directory: {self.ROOT}\n" + "pytest executed: 1014 passed, 6 skipped" + ), + ) + self.assertNotEqual(report["grade"], "A") + self.assertFalse(report["validation_environment_proven"]) + - def test_next_eligible_claim_without_policy_fails(self): result = assess_queue_ordering_proof( report_text="Selected the next eligible PR: PR #286.", ) @@ -2540,6 +2641,139 @@ class TestReviewerBaselineValidationProof(unittest.TestCase): self.assertFalse(report["baseline_validation_proven"]) +class TestAlreadyLandedReviewGate(unittest.TestCase): + """Issue #292: PR head already on target blocks approval and merge.""" + + def _gate(self, **overrides): + kwargs = { + "pr_number": 278, + "candidate_head_sha": PINNED, + "target_branch": "master", + "target_branch_sha": OTHER, + "head_is_ancestor_of_target": True, + "live_head_sha": PINNED, + } + kwargs.update(overrides) + return assess_already_landed_review_gate(**kwargs) + + def test_already_landed_blocks_approval_and_merge(self): + result = self._gate() + self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED") + self.assertFalse(result["approve_allowed"]) + self.assertFalse(result["merge_api_allowed"]) + self.assertFalse(result["request_changes_allowed"]) + self.assertTrue(result["reconciliation_handoff_required"]) + + def test_already_landed_with_real_blocker_allows_request_changes(self): + result = self._gate(real_blocker=True) + self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED") + self.assertFalse(result["approve_allowed"]) + self.assertFalse(result["merge_api_allowed"]) + self.assertTrue(result["request_changes_allowed"]) + + def test_normal_candidate_passes_gate(self): + result = self._gate(head_is_ancestor_of_target=False) + self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE") + self.assertTrue(result["approve_allowed"]) + self.assertTrue(result["merge_api_allowed"]) + self.assertFalse(result["reconciliation_handoff_required"]) + + def test_non_mergeable_normal_pr_still_passes_this_gate(self): + # Mergeability/conflicts are separate gates; ancestry gate only + # decides already-landed vs normal candidate. + result = self._gate( + head_is_ancestor_of_target=False, mergeable=False) + self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE") + self.assertTrue(result["request_changes_allowed"]) + + def test_unchecked_ancestry_blocks_review_mutations(self): + result = self._gate(head_is_ancestor_of_target=None) + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + self.assertFalse(result["approve_allowed"]) + self.assertFalse(result["merge_api_allowed"]) + + def test_changed_head_since_pin_blocks_until_recheck(self): + result = self._gate( + head_is_ancestor_of_target=False, live_head_sha=OTHER) + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + self.assertFalse(result["approve_allowed"]) + self.assertTrue(any( + "head" in reason.lower() for reason in result["reasons"] + )) + + def test_invalid_shas_block_gate(self): + result = self._gate(candidate_head_sha="deadbeef") + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + self.assertFalse(result["approve_allowed"]) + + result = self._gate(target_branch_sha=None) + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + + def test_already_landed_handoff_lists_required_fields(self): + result = self._gate() + for field in ( + "PR number/title", + "candidate head SHA", + "target branch", + "target branch SHA", + "ancestor proof", + "linked issue status", + "recommended reconciliation action", + "capability proof for close/comment mutations", + ): + self.assertIn(field, result["required_handoff_fields"]) + + +class TestAlreadyLandedReportState(unittest.TestCase): + """Issue #292: reports cannot say APPROVED after the gate fired.""" + + GOOD_REPORT = ( + "Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n" + "Candidate head SHA: " + PINNED + "\n" + "Review decision: none\n" + "Merge result: none\n" + ) + + def test_gate_fired_with_reconcile_state_passes(self): + result = assess_already_landed_report_state( + self.GOOD_REPORT, gate_fired=True) + self.assertTrue(result["complete"]) + self.assertFalse(result["block"]) + + def test_gate_fired_with_approved_state_blocks(self): + report = self.GOOD_REPORT.replace( + "Review decision: none", "Review decision: APPROVED") + result = assess_already_landed_report_state(report, gate_fired=True) + self.assertTrue(result["block"]) + + def test_gate_fired_with_merged_state_blocks(self): + report = self.GOOD_REPORT.replace( + "Merge result: none", "Merge result: MERGED") + result = assess_already_landed_report_state(report, gate_fired=True) + self.assertTrue(result["block"]) + + def test_gate_fired_with_ready_to_merge_blocks(self): + result = assess_already_landed_report_state( + self.GOOD_REPORT + "Current status: READY_TO_MERGE\n", + gate_fired=True) + self.assertTrue(result["block"]) + + def test_gate_fired_without_reconcile_state_blocks(self): + result = assess_already_landed_report_state( + "Current status: review complete.\n", gate_fired=True) + self.assertTrue(result["block"]) + self.assertTrue(any( + "already_landed_reconcile_required" in reason.lower() + for reason in result["reasons"] + )) + + def test_gate_not_fired_reports_pass_untouched(self): + result = assess_already_landed_report_state( + "Review decision: APPROVED\nMerge result: MERGED\n", + gate_fired=False) + self.assertTrue(result["complete"]) + + class TestFullSuiteFailureApprovalGate(unittest.TestCase): """Issue #323: full-suite failure blocks approval without baseline proof.""" @@ -2707,3 +2941,6 @@ class TestFullSuiteFailureApprovalGate(unittest.TestCase): if __name__ == "__main__": unittest.main() + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_reviewer_merge_simulation.py b/tests/test_reviewer_merge_simulation.py new file mode 100644 index 0000000..e3101f8 --- /dev/null +++ b/tests/test_reviewer_merge_simulation.py @@ -0,0 +1,135 @@ +"""Tests for merge-simulation worktree mutation verifier (#317).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_merge_simulation import ( # noqa: E402 + assess_merge_simulation_report, + merge_simulation_commands, +) + + +def _good_simulation_report(**overrides) -> str: + lines = [ + "Worktree/index mutations: merge simulation in branches/review-pr281", + "Worktree path: branches/review-pr281", + "Pre-simulation clean status: clean (git status --porcelain empty)", + "Merge result: conflicts in review_proofs.py", + "Abort command: git merge --abort", + "Post-abort clean status: clean after abort (git status --porcelain empty)", + ] + text = "\n".join(lines) + for key, value in overrides.items(): + text = text.replace(f"{key}:", f"{key}: {value}") + return text + + +class TestMergeSimulationDetection(unittest.TestCase): + def test_detects_no_commit_merge(self): + commands = merge_simulation_commands([ + "git merge prgs/master --no-commit --no-ff", + "git status", + ]) + self.assertEqual(commands, ["git merge prgs/master --no-commit --no-ff"]) + + def test_detects_merge_abort(self): + commands = merge_simulation_commands([ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ]) + self.assertIn("git merge --abort", commands) + + +class TestMergeSimulationReport(unittest.TestCase): + def test_no_simulation_passes(self): + result = assess_merge_simulation_report( + "Review complete. Worktree/index mutations: none", + command_log=["git status --porcelain", "git diff prgs/master...HEAD"], + ) + self.assertTrue(result["proven"]) + + def test_documented_simulation_passes(self): + report = _good_simulation_report() + result = assess_merge_simulation_report( + report, + command_log=[ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_readonly_classification_blocks(self): + report = "\n".join([ + "Read-only diagnostics: git merge prgs/master --no-commit --no-ff, git status", + "Worktree/index mutations: none", + ]) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("read-only" in r.lower() for r in result["reasons"])) + + def test_missing_worktree_field_blocks(self): + report = "Merge simulation ran but only noted in prose." + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("worktree/index mutations" in r.lower() for r in result["reasons"])) + + def test_conflict_simulation_requires_merge_result(self): + report = _good_simulation_report().replace( + "Merge result: conflicts in review_proofs.py", + "", + ) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("merge result" in r.lower() for r in result["reasons"])) + + def test_aborted_simulation_requires_post_clean_proof(self): + report = _good_simulation_report().replace( + "Post-abort clean status: clean after abort (git status --porcelain empty)", + "", + ) + result = assess_merge_simulation_report( + report, + command_log=[ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("post-abort" in r.lower() for r in result["reasons"])) + + def test_successful_simulation_without_abort_omits_abort_fields(self): + report = "\n".join([ + "Worktree/index mutations: merge simulation completed cleanly", + "Worktree path: branches/review-pr281", + "Pre-simulation clean status: clean before simulation", + "Merge result: clean fast-forward simulation", + ]) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertTrue(result["proven"], result["reasons"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_merge_simulation_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_reviewer_validation_worktree_edits.py b/tests/test_reviewer_validation_worktree_edits.py new file mode 100644 index 0000000..adedaca --- /dev/null +++ b/tests/test_reviewer_validation_worktree_edits.py @@ -0,0 +1,134 @@ +"""Tests for PR validation worktree no-edit verifier (#315).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_validation_worktree_edits import ( # noqa: E402 + assess_validation_worktree_edit_report, +) + + +def _read_only_report() -> str: + return "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official PR-head validation on unmodified worktree.", + "Official validation result: failed (1 failed)", + "File edits by reviewer: none", + "Review decision: request_changes", + ]) + + +def _diagnostic_report() -> str: + return "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official PR-head validation on unmodified worktree.", + "Official validation result: failed (1 failed)", + "Review decision: request_changes", + "Diagnostic local experiment — not PR-head validation", + "Diagnostic scratch worktree path: branches/diagnostic-pr280-fix-test", + "File edits by reviewer: tests/test_agent_temp_artifacts.py (diagnostic only)", + "Diagnostic result: passed after local fix", + ]) + + +class TestValidationWorktreeEdits(unittest.TestCase): + def test_read_only_review_passes(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_validation_worktree_edit_blocks(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"], + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("validation worktree" in r.lower() for r in result["reasons"])) + + def test_file_edits_none_contradiction_blocks(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + }, + action_log=[ + { + "path": "tests/test_agent_temp_artifacts.py", + "kind": "file_edit", + "performed": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + } + ], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("file edits" in r.lower() for r in result["reasons"])) + + def test_diagnostic_scratch_with_reporting_passes(self): + result = assess_validation_worktree_edit_report( + _diagnostic_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "diagnostic_worktree_path": "branches/diagnostic-pr280-fix-test", + "diagnostic_edits": ["tests/test_agent_temp_artifacts.py"], + "diagnostic_experiment": True, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_diagnostic_without_scratch_worktree_blocks(self): + result = assess_validation_worktree_edit_report( + "File edits by reviewer: tests/test_example.py", + validation_session={ + "diagnostic_edits": ["tests/test_example.py"], + "diagnostic_experiment": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"])) + + def test_post_edit_official_validation_blocks(self): + report = "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official validation result: passed after local edit", + "File edits by reviewer: tests/test_agent_temp_artifacts.py", + ]) + result = assess_validation_worktree_edit_report( + report, + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"], + "official_validation_after_edit": True, + }, + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_validation_worktree_edit_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file