fix: resolve conflicts for PR #382 against latest master
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -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.
|
||||
@@ -3890,6 +3986,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)
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -4505,6 +4767,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_worktree_ownership_report(report_text, **kwargs):
|
||||
"""#312: prove session-owned or safe-reuse worktree before reset/validation."""
|
||||
from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess
|
||||
|
||||
@@ -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"
|
||||
),
|
||||
}
|
||||
@@ -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()
|
||||
@@ -154,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_worktree_ownership_verifier_exported():
|
||||
from review_proofs import assess_worktree_ownership_report
|
||||
|
||||
|
||||
@@ -23,6 +23,8 @@ 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,
|
||||
@@ -2639,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."""
|
||||
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user