fix: resolve conflicts for PR #371 against latest master

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
2026-07-07 09:31:44 -04:00
co-authored by Claude Opus 4.8
8 changed files with 1498 additions and 3 deletions
+151
View File
@@ -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()
+12
View File
@@ -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
+240 -3
View File
@@ -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()
+135
View File
@@ -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()
@@ -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()