sysadminandClaude Opus 4.8
eff4572a91
Introduce thread_state_ledger_validator for the mandatory two-comment workflow: tagged [CONTROLLER HANDOFF] paired with [THREAD STATE LEDGER]. Wire validation into final_report_validator and gitea_create_issue_comment, add runbook docs, worked examples, and tests. Closes #507 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
543 lines
21 KiB
Python
543 lines
21 KiB
Python
"""Tests for composable final-report validator (#327)."""
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from final_report_validator import ( # noqa: E402
|
|
FINAL_REPORT_TASK_KINDS,
|
|
assess_final_report_validator,
|
|
validator_finding,
|
|
)
|
|
|
|
|
|
def _review_handoff(**overrides):
|
|
lines = [
|
|
"## Controller Handoff",
|
|
"",
|
|
"- Task: review PR #203",
|
|
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
|
"- Role: reviewer",
|
|
"- Identity: sysadmin / prgs-reviewer",
|
|
"- Issue/PR: #182 / PR #203",
|
|
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
|
"- Files changed: review_proofs.py",
|
|
"- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
|
|
"- Mutations: review only",
|
|
"- File edits by reviewer: none",
|
|
"- Worktree/index mutations: none",
|
|
"- Git ref mutations: git fetch prgs master",
|
|
"- MCP/Gitea mutations: review submitted",
|
|
"- Review mutations: submitted request_changes review",
|
|
"- Merge mutations: none",
|
|
"- Cleanup mutations: none",
|
|
"- External-state mutations: none",
|
|
"- Read-only diagnostics: issue and PR metadata inspected",
|
|
"- Current status: PR open",
|
|
"- Blockers: none",
|
|
"- Next: await author",
|
|
"- Safety: no self-review; no self-merge",
|
|
"- Selected PR: #203",
|
|
"- Reviewer eligibility: eligible",
|
|
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
|
"- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
|
"- Final live head SHA before approval: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
|
"- Final live head SHA before merge: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
|
"- Push occurred during validation: no",
|
|
"- Worktree path: branches/review-203",
|
|
"- Worktree dirty: clean",
|
|
"- Scratch worktree used: yes (branches/review-203)",
|
|
"- Unrelated local mutations: none",
|
|
"- Review decision: request_changes",
|
|
"- Merge result: not attempted",
|
|
"- Linked issue: #182",
|
|
"- Linked issue status: open",
|
|
"- Cleanup status: none",
|
|
]
|
|
text = "\n".join(lines)
|
|
for key, value in overrides.items():
|
|
text = text.replace(f"- {key}:", f"- {key}: {value}")
|
|
return text
|
|
|
|
|
|
def _reconcile_handoff(drop=(), **overrides):
|
|
lines = [
|
|
"## Controller Handoff",
|
|
"",
|
|
"- Task: reconcile already-landed PR #99",
|
|
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
|
"- Role/profile: reconciler / prgs-reconciler",
|
|
"- Identity: sysadmin",
|
|
"- Selected PR: #99",
|
|
"- PR live state: open",
|
|
"- Candidate head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
|
"- Target branch: master",
|
|
"- Target branch SHA: 5e023dc71b0e2b813a0b1eee6b0f42630c47a95c",
|
|
"- Ancestor proof: candidate head is ancestor of target branch SHA",
|
|
"- Linked issue: #98",
|
|
"- Linked issue live status: not verified in this session",
|
|
"- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
|
|
"- Capabilities proven: gitea.read, gitea.pr.comment",
|
|
"- Missing capabilities: gitea.pr.close",
|
|
"- PR comments posted: 1 reconciliation comment on PR #99",
|
|
"- Issue comments posted: none",
|
|
"- PRs closed: none",
|
|
"- Issues closed: none",
|
|
"- File edits by reconciler: none",
|
|
"- Git ref mutations: git fetch prgs master",
|
|
"- Worktree mutations: none",
|
|
"- MCP/Gitea mutations: PR comment posted",
|
|
"- External-state mutations: none",
|
|
"- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
|
|
"- Reconciliation mutations: PR comment posted",
|
|
"- Current status: reconciliation complete",
|
|
"- Blocker: missing gitea.pr.close capability",
|
|
"- Safe next action: hand off to a profile with gitea.pr.close",
|
|
"- No review/merge confirmation: confirmed",
|
|
]
|
|
kept = [
|
|
line
|
|
for line in lines
|
|
if not any(line.startswith(f"- {name}:") for name in drop)
|
|
]
|
|
text = "\n".join(kept)
|
|
for key, value in overrides.items():
|
|
if f"- {key}:" in text:
|
|
text = text.replace(f"- {key}:", f"- {key}: {value}")
|
|
return text
|
|
|
|
|
|
class TestTwoCommentWorkflowIntegration(unittest.TestCase):
|
|
def test_tagged_pair_produces_no_two_comment_findings(self):
|
|
from thread_state_ledger_validator import findings_for_final_report # noqa: E402
|
|
from tests.test_thread_state_ledger_validator import ( # noqa: E402
|
|
_minimal_handoff,
|
|
_minimal_ledger,
|
|
)
|
|
|
|
report = _minimal_handoff() + "\n\n" + _minimal_ledger()
|
|
self.assertEqual(findings_for_final_report(report), [])
|
|
|
|
def test_tagged_handoff_without_ledger_blocks_via_validator(self):
|
|
from thread_state_ledger_validator import findings_for_final_report # noqa: E402
|
|
from tests.test_thread_state_ledger_validator import _minimal_handoff # noqa: E402
|
|
|
|
findings = findings_for_final_report(_minimal_handoff())
|
|
self.assertTrue(findings)
|
|
self.assertTrue(
|
|
any(f["severity"] == "block" for f in findings)
|
|
)
|
|
|
|
|
|
class TestValidatorFinding(unittest.TestCase):
|
|
def test_finding_shape(self):
|
|
finding = validator_finding(
|
|
"reviewer.test",
|
|
"block",
|
|
"Validation",
|
|
"missing proof",
|
|
"repair validation proof",
|
|
)
|
|
self.assertEqual(finding["rule_id"], "reviewer.test")
|
|
self.assertEqual(finding["severity"], "block")
|
|
self.assertEqual(finding["field"], "Validation")
|
|
|
|
|
|
class TestReviewPrRules(unittest.TestCase):
|
|
def test_clean_reviewer_report_passes(self):
|
|
result = assess_final_report_validator(
|
|
_review_handoff(),
|
|
"review_pr",
|
|
linked_issue_lock={"issue_number": 182, "pr_number": 203},
|
|
)
|
|
self.assertEqual(result["grade"], "A")
|
|
self.assertFalse(result["blocked"])
|
|
self.assertEqual(result["findings"], [])
|
|
|
|
def test_legacy_workspace_mutations_blocks(self):
|
|
report = _review_handoff() + "\n- Workspace mutations: none"
|
|
result = assess_final_report_validator(
|
|
report,
|
|
"review_pr",
|
|
mutations_observed=True,
|
|
)
|
|
self.assertEqual(result["grade"], "blocked")
|
|
rule_ids = [f["rule_id"] for f in result["findings"]]
|
|
self.assertIn("reviewer.legacy_workspace_mutations", rule_ids)
|
|
|
|
def test_vague_mutations_none_blocks(self):
|
|
report = _review_handoff().replace(
|
|
"- Mutations: review only",
|
|
"- Mutations: none",
|
|
)
|
|
result = assess_final_report_validator(
|
|
report,
|
|
"review_pr",
|
|
mutations_observed=True,
|
|
)
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(f["rule_id"] == "reviewer.vague_mutations_none" for f in result["findings"])
|
|
)
|
|
|
|
def test_bare_pytest_without_cwd_downgrades(self):
|
|
report = _review_handoff().replace(
|
|
"- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
|
|
"- Validation: pytest passed",
|
|
)
|
|
result = assess_final_report_validator(report, "review_pr")
|
|
self.assertEqual(result["grade"], "downgraded")
|
|
self.assertTrue(
|
|
any(f["rule_id"] == "reviewer.validation_command_proof" for f in result["findings"])
|
|
)
|
|
|
|
def test_linked_issue_mismatch_blocks(self):
|
|
report = _review_handoff().replace("- Linked issue: #182", "- Linked issue: #999")
|
|
result = assess_final_report_validator(
|
|
report,
|
|
"review_pr",
|
|
linked_issue_lock={"issue_number": 182, "pr_number": 203},
|
|
)
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(f["rule_id"] == "reviewer.linked_issue_mismatch" for f in result["findings"])
|
|
)
|
|
|
|
def test_approval_after_full_suite_failure_blocks_without_baseline(self):
|
|
report = (
|
|
_review_handoff()
|
|
.replace("- Review decision: request_changes", "- Review decision: approve")
|
|
+ "\nSubmitted 'approve' after full suite failed with 3 failed tests."
|
|
)
|
|
result = assess_final_report_validator(report, "review_pr")
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reviewer.baseline_on_full_suite_failure"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_same_as_master_without_baseline_blocks(self):
|
|
report = _review_handoff() + "\nFailures are same as master baseline."
|
|
result = assess_final_report_validator(report, "review_pr")
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(f["rule_id"] == "reviewer.main_checkout_baseline" for f in result["findings"])
|
|
)
|
|
|
|
def test_already_landed_eligible_wording_blocks(self):
|
|
report = (
|
|
_review_handoff()
|
|
+ "\nPR is already landed and eligible for review next."
|
|
)
|
|
result = assess_final_report_validator(report, "review_pr")
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reviewer.already_landed_eligible_wording"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_already_landed_oldest_eligible_pr_blocks(self):
|
|
report = (
|
|
_review_handoff()
|
|
+ "\nAlready landed.\nOldest eligible PR: PR #278."
|
|
)
|
|
result = assess_final_report_validator(report, "review_pr")
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reviewer.already_landed_eligible_wording"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_already_landed_gate_blocks_review_terminal_states(self):
|
|
cases = {
|
|
"APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"),
|
|
"MERGED": ("- Merge result: not attempted", "- Merge result: MERGED"),
|
|
"READY_TO_MERGE": ("- Current status: PR open", "- Current status: READY_TO_MERGE"),
|
|
}
|
|
for state, (old, new) in cases.items():
|
|
with self.subTest(state=state):
|
|
report = (
|
|
_review_handoff()
|
|
.replace("- Reviewer eligibility: eligible", "- Reviewer eligibility: blocked")
|
|
.replace(old, new)
|
|
+ "\n- Already-landed gate: fired"
|
|
+ "\n- Ancestor proof: passed"
|
|
+ "\n- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED"
|
|
)
|
|
result = assess_final_report_validator(report, "review_pr")
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reviewer.already_landed_review_state"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_target_branch_up_to_date_requires_fetch_and_sha(self):
|
|
report = (
|
|
_review_handoff()
|
|
.replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none")
|
|
+ "\nTarget branch up to date."
|
|
)
|
|
result = assess_final_report_validator(report, "review_pr")
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reviewer.target_branch_freshness_proof"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_target_branch_up_to_date_with_fetch_and_sha_passes(self):
|
|
report = (
|
|
_review_handoff()
|
|
+ "\n- Target branch SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
|
|
+ "\nTarget branch up to date."
|
|
)
|
|
result = assess_final_report_validator(report, "review_pr")
|
|
self.assertEqual(result["grade"], "A")
|
|
def test_git_fetch_must_not_be_readonly_only(self):
|
|
report = (
|
|
_review_handoff()
|
|
.replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none")
|
|
.replace(
|
|
"- Read-only diagnostics: issue and PR metadata inspected",
|
|
"- Read-only diagnostics: git fetch prgs master",
|
|
)
|
|
)
|
|
result = assess_final_report_validator(
|
|
report,
|
|
"review_pr",
|
|
action_log=[{"command": "git fetch prgs master", "performed": True}],
|
|
)
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(f["rule_id"] == "reviewer.git_fetch_readonly" for f in result["findings"])
|
|
)
|
|
|
|
|
|
class TestReconciliationRules(unittest.TestCase):
|
|
def test_clean_reconcile_report_passes(self):
|
|
result = assess_final_report_validator(
|
|
_reconcile_handoff(),
|
|
"reconcile_already_landed",
|
|
)
|
|
self.assertEqual(result["grade"], "A")
|
|
|
|
def test_stale_author_field_blocks(self):
|
|
report = _reconcile_handoff() + "\n- PR number opened: #12"
|
|
result = assess_final_report_validator(report, "reconcile_already_landed")
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_eligible_wording_for_landed_pr_blocks(self):
|
|
report = _reconcile_handoff() + "\nPR already landed and eligible for merge."
|
|
result = assess_final_report_validator(report, "reconcile_already_landed")
|
|
self.assertTrue(result["blocked"])
|
|
|
|
def test_linked_issue_status_without_live_proof_downgrades(self):
|
|
report = _reconcile_handoff().replace(
|
|
"- Linked issue live status: not verified in this session",
|
|
"- Linked issue live status: closed",
|
|
)
|
|
result = assess_final_report_validator(report, "reconcile_already_landed")
|
|
self.assertEqual(result["grade"], "downgraded")
|
|
self.assertTrue(
|
|
any(f["rule_id"] == "reconcile.linked_issue_live_status" for f in result["findings"])
|
|
)
|
|
|
|
def test_inventory_complete_requires_pagination_proof(self):
|
|
report = _reconcile_handoff() + "\nInventory complete for open PRs."
|
|
result = assess_final_report_validator(report, "reconcile_already_landed")
|
|
self.assertEqual(result["grade"], "downgraded")
|
|
self.assertTrue(
|
|
any(f["rule_id"] == "reconcile.inventory_pagination_proof" for f in result["findings"])
|
|
)
|
|
|
|
|
|
class TestCanonicalReconcileSchema(unittest.TestCase):
|
|
"""Issue #307: reconciliation workflows emit only the canonical schema."""
|
|
|
|
def test_comment_only_reconciliation_passes(self):
|
|
result = assess_final_report_validator(
|
|
_reconcile_handoff(),
|
|
"reconcile_already_landed",
|
|
)
|
|
self.assertEqual(result["grade"], "A")
|
|
self.assertEqual(result["findings"], [])
|
|
|
|
def test_successful_close_reconciliation_passes(self):
|
|
report = _reconcile_handoff().replace(
|
|
"- Capabilities proven: gitea.read, gitea.pr.comment",
|
|
"- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close",
|
|
).replace(
|
|
"- Missing capabilities: gitea.pr.close",
|
|
"- Missing capabilities: none",
|
|
).replace(
|
|
"- PRs closed: none",
|
|
"- PRs closed: #99",
|
|
).replace(
|
|
"- Blocker: missing gitea.pr.close capability",
|
|
"- Blocker: none",
|
|
)
|
|
result = assess_final_report_validator(report, "reconcile_already_landed")
|
|
self.assertEqual(result["grade"], "A")
|
|
|
|
def test_blocked_reconciliation_passes(self):
|
|
report = _reconcile_handoff().replace(
|
|
"- Capabilities proven: gitea.read, gitea.pr.comment",
|
|
"- Capabilities proven: gitea.read",
|
|
).replace(
|
|
"- Missing capabilities: gitea.pr.close",
|
|
"- Missing capabilities: gitea.pr.close, gitea.pr.comment",
|
|
).replace(
|
|
"- PR comments posted: 1 reconciliation comment on PR #99",
|
|
"- PR comments posted: none",
|
|
).replace(
|
|
"- MCP/Gitea mutations: PR comment posted",
|
|
"- MCP/Gitea mutations: none",
|
|
).replace(
|
|
"- Reconciliation mutations: PR comment posted",
|
|
"- Reconciliation mutations: none",
|
|
)
|
|
result = assess_final_report_validator(report, "reconcile_already_landed")
|
|
self.assertEqual(result["grade"], "A")
|
|
|
|
def test_missing_capabilities_field_required(self):
|
|
report = _reconcile_handoff(drop=("Missing capabilities",))
|
|
result = assess_final_report_validator(report, "reconcile_already_landed")
|
|
self.assertEqual(result["grade"], "downgraded")
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reconcile.controller_handoff"
|
|
and "Missing capabilities" in f["reason"]
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_missing_ancestor_proof_and_candidate_sha_downgrade(self):
|
|
report = _reconcile_handoff(drop=("Ancestor proof", "Candidate head SHA"))
|
|
result = assess_final_report_validator(report, "reconcile_already_landed")
|
|
self.assertEqual(result["grade"], "downgraded")
|
|
reasons = " ".join(f["reason"] for f in result["findings"])
|
|
self.assertIn("Ancestor proof", reasons)
|
|
self.assertIn("Candidate head SHA", reasons)
|
|
|
|
def test_stale_issue_lock_proof_blocks(self):
|
|
report = _reconcile_handoff() + "\n- Issue lock proof: locked before diff"
|
|
result = assess_final_report_validator(report, "reconcile_already_landed")
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
|
|
and f["field"] == "issue lock proof"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_stale_claim_comment_status_blocks(self):
|
|
report = _reconcile_handoff() + "\n- Claim/comment status: claimed"
|
|
result = assess_final_report_validator(report, "reconcile_already_landed")
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
|
|
and f["field"] == "claim/comment status"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_stale_workspace_mutations_blocks_without_observed_mutations(self):
|
|
report = _reconcile_handoff() + "\n- Workspace mutations: None"
|
|
result = assess_final_report_validator(report, "reconcile_already_landed")
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
|
|
and f["field"] == "workspace mutations"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_pr_number_opened_allowed_when_session_opened_pr(self):
|
|
report = _reconcile_handoff() + "\n- PR number opened: #12"
|
|
result = assess_final_report_validator(
|
|
report,
|
|
"reconcile_already_landed",
|
|
session_pr_opened=True,
|
|
)
|
|
self.assertFalse(
|
|
any(
|
|
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
|
|
and f["field"] == "pr number opened"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_pr_number_opened_still_blocked_without_session_proof(self):
|
|
report = _reconcile_handoff() + "\n- PR number opened: #12"
|
|
result = assess_final_report_validator(
|
|
report,
|
|
"reconcile_already_landed",
|
|
session_pr_opened=False,
|
|
)
|
|
self.assertTrue(result["blocked"])
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
|
|
and f["field"] == "pr number opened"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
def test_action_log_fetch_requires_git_ref_mutation_entry(self):
|
|
report = _reconcile_handoff().replace(
|
|
"- Git ref mutations: git fetch prgs master",
|
|
"- Git ref mutations: none",
|
|
)
|
|
result = assess_final_report_validator(
|
|
report,
|
|
"reconcile_already_landed",
|
|
action_log=[{"command": "git fetch prgs master", "performed": True}],
|
|
)
|
|
self.assertTrue(
|
|
any(
|
|
f["rule_id"] == "reviewer.git_fetch_readonly"
|
|
for f in result["findings"]
|
|
)
|
|
)
|
|
|
|
|
|
class TestEntryPoint(unittest.TestCase):
|
|
def test_unknown_task_kind_blocks(self):
|
|
result = assess_final_report_validator("report", "unknown_mode")
|
|
self.assertTrue(result["blocked"])
|
|
self.assertEqual(result["findings"][0]["rule_id"], "shared.unknown_task_kind")
|
|
|
|
def test_review_proofs_reexport(self):
|
|
from review_proofs import assess_final_report_validator as exported
|
|
|
|
self.assertTrue(callable(exported))
|
|
result = exported(_review_handoff(), "review_pr")
|
|
self.assertIn("grade", result)
|
|
|
|
def test_supported_task_kinds_include_review_and_reconcile(self):
|
|
self.assertIn("review_pr", FINAL_REPORT_TASK_KINDS)
|
|
self.assertIn("reconcile_already_landed", FINAL_REPORT_TASK_KINDS)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|