diff --git a/final_report_validator.py b/final_report_validator.py index a487cf6..50ee2e0 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -728,6 +728,65 @@ def _rule_reviewer_stale_head_proof(report_text: str) -> list[dict[str, str]]: ) +_MUTATION_ACCOUNTING_PATTERNS = { + "local_failed_attempts": re.compile( + r"local\s+failed\s+attempts\s*:\s*(\d+)", re.IGNORECASE + ), + "blocked_api_attempts": re.compile( + r"blocked\s+api\s+attempts\s*:\s*(\d+)", re.IGNORECASE + ), + "successful_server_mutations": re.compile( + r"successful\s+server(?:[-\s]side)?\s+mutations\s*:\s*(\d+)", re.IGNORECASE + ), +} + +_READBACK_VERIFIED_PATTERN = re.compile( + r"read[-\s]?after[-\s]?write\s+verified\s*:\s*(yes|true)", re.IGNORECASE +) + + +def _rule_shared_mutation_budget_accounting( + report_text: str, + *, + mutation_attempt_ledger: list[dict] | None = None, +) -> list[dict[str, str]]: + """#617: mutation budget counts server-side changes only. + + No-op unless the session supplies an attempt ledger. When it does, the + report's three attempt categories must match the ledger exactly, so a + pre-API validator rejection can never be reported as a Gitea mutation and + a real mutation can never be hidden. + """ + if mutation_attempt_ledger is None: + return [] + + from mutation_budget_classifier import assess_final_report_mutation_accounting + + text = report_text or "" + claimed: dict[str, Any] = {} + for field, pattern in _MUTATION_ACCOUNTING_PATTERNS.items(): + match = pattern.search(text) + if match: + claimed[field] = int(match.group(1)) + if _READBACK_VERIFIED_PATTERN.search(text): + claimed["readback_verified"] = True + + result = assess_final_report_mutation_accounting(claimed, mutation_attempt_ledger) + if result.get("valid"): + return [] + return _findings_from_reasons( + "shared.mutation_budget_accounting", + result.get("reasons") or [], + field="Mutation accounting", + severity="block", + safe_next_action=( + "report 'Local failed attempts:', 'Blocked API attempts:', and " + "'Successful server-side mutations:' with counts matching the " + "attempt ledger; pre-API rejections are not Gitea mutations" + ), + ) + + def _rule_conflict_fix_classification_proof(report_text: str) -> list[dict[str, str]]: from conflict_fix_classification import ( assess_conflict_fix_classification_final_report, @@ -1584,6 +1643,10 @@ _SHARED_CANONICAL_COMMENT_RULES = ( _rule_shared_canonical_comment_post_claim, ) +_SHARED_MUTATION_BUDGET_RULES = ( + _rule_shared_mutation_budget_accounting, +) + _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { "review_pr": [ _rule_shared_controller_handoff, @@ -1591,6 +1654,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_shared_email_disclosure, *_SHARED_TWO_COMMENT_RULES, *_SHARED_CANONICAL_COMMENT_RULES, + *_SHARED_MUTATION_BUDGET_RULES, *_SHARED_ISSUE_LOCK_RULES, _rule_reviewer_legacy_workspace_mutations, _rule_reviewer_vague_mutations_none, @@ -1636,6 +1700,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_shared_email_disclosure, *_SHARED_TWO_COMMENT_RULES, *_SHARED_CANONICAL_COMMENT_RULES, + *_SHARED_MUTATION_BUDGET_RULES, *_SHARED_ISSUE_LOCK_RULES, *_SHARED_CLEANUP_PROOF_RULES, _rule_reconcile_stale_author_fields, @@ -1655,6 +1720,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_shared_email_disclosure, *_SHARED_TWO_COMMENT_RULES, *_SHARED_CANONICAL_COMMENT_RULES, + *_SHARED_MUTATION_BUDGET_RULES, *_SHARED_ISSUE_LOCK_RULES, _rule_reviewer_vague_mutations_none, ], @@ -1664,6 +1730,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_shared_email_disclosure, *_SHARED_TWO_COMMENT_RULES, *_SHARED_CANONICAL_COMMENT_RULES, + *_SHARED_MUTATION_BUDGET_RULES, *_SHARED_ISSUE_LOCK_RULES, _rule_shared_issue_acceptance_gate, _rule_reviewer_vague_mutations_none, @@ -1677,6 +1744,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_shared_email_disclosure, *_SHARED_TWO_COMMENT_RULES, *_SHARED_CANONICAL_COMMENT_RULES, + *_SHARED_MUTATION_BUDGET_RULES, *_SHARED_ISSUE_LOCK_RULES, ], "inventory": [ @@ -1685,6 +1753,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_shared_email_disclosure, *_SHARED_TWO_COMMENT_RULES, *_SHARED_CANONICAL_COMMENT_RULES, + *_SHARED_MUTATION_BUDGET_RULES, *_SHARED_ISSUE_LOCK_RULES, _rule_reconcile_pagination_proof, ], @@ -1694,6 +1763,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_shared_email_disclosure, *_SHARED_TWO_COMMENT_RULES, *_SHARED_CANONICAL_COMMENT_RULES, + *_SHARED_MUTATION_BUDGET_RULES, *_SHARED_ISSUE_LOCK_RULES, ], # Controller issue closure (#529): a closure report must not bury an @@ -1766,6 +1836,7 @@ def assess_final_report_validator( session_pr_opened: bool = False, validation_session: dict | None = None, reconciler_close_lock: dict | None = None, + mutation_attempt_ledger: list[dict] | None = None, ) -> dict[str, Any]: """Validate final-report text against task-specific proof rules (#327). @@ -1829,6 +1900,7 @@ def assess_final_report_validator( "session_pr_opened": session_pr_opened, "validation_session": validation_session, "reconciler_close_lock": reconciler_close_lock, + "mutation_attempt_ledger": mutation_attempt_ledger, } for rule in _RULES_BY_TASK.get(normalized_kind, ()): diff --git a/mutation_budget_classifier.py b/mutation_budget_classifier.py new file mode 100644 index 0000000..930ff48 --- /dev/null +++ b/mutation_budget_classifier.py @@ -0,0 +1,286 @@ +"""Mutation-budget classification for auto-mode attempts (#617). + +Mutation budget must count only *server-side* Gitea state changes. A tool call +that fails closed before the Gitea API is reached changed nothing on the +server, so it must not consume the budget that protects against repeated real +mutations. + +The classifier separates four outcome classes plus an explicit ambiguous class: + +``local_validator_rejection`` + A canonical-content validator (for example the ``[THREAD STATE LEDGER]`` or + ``## Canonical Issue State`` blocks) rejected the payload before any API + call. No server-side state exists. + +``capability_gate_rejection`` + A profile/permission gate refused the operation before any API call. + +``transport_failure_before_api`` + The request never reached the Gitea API (transport/EOF/connection error). + +``server_side_mutation`` + The API succeeded and returned proof of durable state (comment id, review + id, merge commit, label result, or an issue/PR state change). + +``ambiguous_requires_readback`` + The API *was* reached but the result carries no usable proof either way. + This fails closed: the attempt is treated as budget-consuming until a + read-after-write check proves otherwise, so #617 never weakens the guard + that prevents repeated real mutations. + +Only ``server_side_mutation`` consumes budget outright. Every attempt — failed +or not — is still recorded in the local attempt ledger so a final report can +show local failed attempts, blocked API attempts, and successful server-side +mutations separately. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +LOCAL_VALIDATOR_REJECTION = "local_validator_rejection" +CAPABILITY_GATE_REJECTION = "capability_gate_rejection" +TRANSPORT_FAILURE_BEFORE_API = "transport_failure_before_api" +SERVER_SIDE_MUTATION = "server_side_mutation" +AMBIGUOUS_REQUIRES_READBACK = "ambiguous_requires_readback" + +CLASSIFICATIONS = ( + LOCAL_VALIDATOR_REJECTION, + CAPABILITY_GATE_REJECTION, + TRANSPORT_FAILURE_BEFORE_API, + SERVER_SIDE_MUTATION, + AMBIGUOUS_REQUIRES_READBACK, +) + +#: Result fields that prove durable server-side state was created. +MUTATION_PROOF_FIELDS = ( + "comment_id", + "review_id", + "merge_commit_sha", + "label_result", + "state_change", + "created_pr_number", +) + +#: Classes that never consume server-side mutation budget. +PRE_API_CLASSIFICATIONS = ( + LOCAL_VALIDATOR_REJECTION, + CAPABILITY_GATE_REJECTION, + TRANSPORT_FAILURE_BEFORE_API, +) + +FINAL_REPORT_REQUIRED_FIELDS = ( + "local_failed_attempts", + "blocked_api_attempts", + "successful_server_mutations", +) + + +def _clean(value: Any) -> str: + return (value or "").strip() if isinstance(value, str) else str(value or "").strip() + + +def _proof_fields_present(result: dict) -> list[str]: + """Return the mutation-proof fields carrying a usable value.""" + present: list[str] = [] + for field in MUTATION_PROOF_FIELDS: + value = result.get(field) + if value is None or value is False: + continue + if isinstance(value, str) and not value.strip(): + continue + present.append(field) + return present + + +def _decision( + classification: str, + *, + budget_consumed: bool, + requires_readback: bool, + reasons: list[str], + proof_fields: list[str], + api_called: bool | None, +) -> dict: + return { + "classification": classification, + "budget_consumed": budget_consumed, + "requires_readback": requires_readback, + "pre_api": classification in PRE_API_CLASSIFICATIONS, + "api_called": api_called, + "proof_fields": proof_fields, + "reasons": reasons, + } + + +def classify_mutation_attempt(result: dict | None) -> dict: + """Classify one mutation attempt and decide whether it consumes budget. + + ``result`` is the raw dict a Gitea MCP tool returned. The caller does not + pre-interpret it: classification is driven by the explicit ``api_called`` + signal plus the proof fields the tool reports. + """ + data = dict(result or {}) + success = bool(data.get("success")) + proof_fields = _proof_fields_present(data) + api_called = data.get("api_called") + + # An unambiguous success carrying durable proof is a real mutation however + # the attempt was labelled upstream. + if success and proof_fields: + return _decision( + SERVER_SIDE_MUTATION, + budget_consumed=True, + requires_readback=False, + reasons=[ + "API reported success with durable proof field(s): " + + ", ".join(proof_fields) + ], + proof_fields=proof_fields, + api_called=True, + ) + + if api_called is False: + # Nothing reached the server; pick the precise pre-API class. + if data.get("transport_error") or data.get("transport_failed"): + return _decision( + TRANSPORT_FAILURE_BEFORE_API, + budget_consumed=False, + requires_readback=False, + reasons=["transport failed before the Gitea API was reached"], + proof_fields=[], + api_called=False, + ) + if data.get("permission_report") or data.get("capability_blocked"): + return _decision( + CAPABILITY_GATE_REJECTION, + budget_consumed=False, + requires_readback=False, + reasons=["capability/permission gate refused before any API call"], + proof_fields=[], + api_called=False, + ) + return _decision( + LOCAL_VALIDATOR_REJECTION, + budget_consumed=False, + requires_readback=False, + reasons=[ + "local validator rejected the payload before any API call; " + "no server-side state was created" + ], + proof_fields=[], + api_called=False, + ) + + if api_called is True: + if success: + reason = ( + "API reported success but returned no durable proof field; " + "read-after-write verification required before counting budget" + ) + else: + reason = ( + "API was reached and the outcome carries no durable proof; " + "read-after-write verification required before counting budget" + ) + return _decision( + AMBIGUOUS_REQUIRES_READBACK, + budget_consumed=True, + requires_readback=True, + reasons=[reason], + proof_fields=proof_fields, + api_called=True, + ) + + # ``api_called`` was not reported at all. Fail closed rather than assuming + # nothing happened. + return _decision( + AMBIGUOUS_REQUIRES_READBACK, + budget_consumed=True, + requires_readback=True, + reasons=[ + "attempt did not report 'api_called'; cannot prove the request " + "stopped before the Gitea API, so the attempt fails closed" + ], + proof_fields=proof_fields, + api_called=None, + ) + + +def record_attempt( + ledger: list[dict] | None, + result: dict | None, + *, + operation: str = "", + timestamp: str | None = None, +) -> dict: + """Append one classified attempt to the local ledger and return the entry. + + Every attempt is recorded, including the ones that consume no budget: the + point of #617 is that failed local attempts stay visible without being + miscounted as Gitea mutations. + """ + entries = ledger if isinstance(ledger, list) else [] + entry = { + "operation": _clean(operation), + "timestamp": _clean(timestamp) or datetime.now(timezone.utc).isoformat(), + **classify_mutation_attempt(result), + } + entries.append(entry) + return entry + + +def summarize_attempt_ledger(ledger: list[dict] | None) -> dict: + """Summarize a ledger into the categories a final report must show.""" + entries = [e for e in (ledger or []) if isinstance(e, dict)] + + def _count(*classifications: str) -> int: + return sum(1 for e in entries if e.get("classification") in classifications) + + return { + "total_attempts": len(entries), + "local_failed_attempts": _count(LOCAL_VALIDATOR_REJECTION), + "blocked_api_attempts": _count( + CAPABILITY_GATE_REJECTION, TRANSPORT_FAILURE_BEFORE_API + ), + "successful_server_mutations": _count(SERVER_SIDE_MUTATION), + "ambiguous_attempts": _count(AMBIGUOUS_REQUIRES_READBACK), + "budget_consumed": sum(1 for e in entries if e.get("budget_consumed")), + "requires_readback": any(e.get("requires_readback") for e in entries), + "entries": entries, + } + + +def assess_final_report_mutation_accounting( + report: dict | None, + ledger: list[dict] | None, +) -> dict: + """Fail closed when a report's mutation accounting contradicts the ledger.""" + data = dict(report or {}) + summary = summarize_attempt_ledger(ledger) + reasons: list[str] = [] + + for field in FINAL_REPORT_REQUIRED_FIELDS: + if field not in data: + reasons.append(f"final report omits required field '{field}'") + continue + claimed = data.get(field) + actual = summary[field] + if claimed != actual: + reasons.append( + f"final report claims {field}={claimed} but the attempt ledger " + f"shows {actual}" + ) + + if summary["requires_readback"] and not data.get("readback_verified"): + reasons.append( + "ledger contains an ambiguous attempt; final report must record " + "'readback_verified' proof before claiming mutation accounting" + ) + + return { + "valid": not reasons, + "reasons": reasons, + "ledger_summary": {k: v for k, v in summary.items() if k != "entries"}, + } diff --git a/tests/test_mutation_budget_classifier.py b/tests/test_mutation_budget_classifier.py new file mode 100644 index 0000000..8e2b8b9 --- /dev/null +++ b/tests/test_mutation_budget_classifier.py @@ -0,0 +1,388 @@ +"""Tests for the #617 mutation-budget classifier. + +Covers every acceptance criterion on issue #617: + +* AC1 — the classifier distinguishes local validator rejection, capability-gate + rejection, transport failure before API, and successful server-side mutation. +* AC2 — pre-API validator failures do not consume server-side mutation budget. +* AC3 — failed attempts are still logged in the local attempt ledger. +* AC4 — the final report separately shows local failed attempts, blocked API + attempts, and successful server-side mutations. +* AC5 — the six named scenarios, including the #615 reproduction where two + local validator rejections precede one successful comment. +""" + +from __future__ import annotations + +import unittest + +from mutation_budget_classifier import ( + AMBIGUOUS_REQUIRES_READBACK, + CAPABILITY_GATE_REJECTION, + LOCAL_VALIDATOR_REJECTION, + SERVER_SIDE_MUTATION, + TRANSPORT_FAILURE_BEFORE_API, + assess_final_report_mutation_accounting, + classify_mutation_attempt, + record_attempt, + summarize_attempt_ledger, +) + +# The two pre-API rejections observed on the #615 comment flow. +MISSING_LEDGER_BLOCK = { + "success": False, + "performed": False, + "api_called": False, + "reasons": ["missing [THREAD STATE LEDGER] block"], +} + +MISSING_CANONICAL_STATE = { + "success": False, + "performed": False, + "api_called": False, + "reasons": ["missing ## Canonical Issue State block"], +} + +# The corrected comment that actually landed as #615 comment 9137. +SUCCESSFUL_COMMENT = { + "success": True, + "performed": True, + "api_called": True, + "comment_id": 9137, + "issue_number": 615, +} + + +class TestAC1Classification(unittest.TestCase): + """AC1: the four outcome classes are distinguished.""" + + def test_local_validator_rejection_is_its_own_class(self): + result = classify_mutation_attempt(MISSING_LEDGER_BLOCK) + self.assertEqual(result["classification"], LOCAL_VALIDATOR_REJECTION) + self.assertTrue(result["pre_api"]) + + def test_capability_gate_rejection_is_its_own_class(self): + result = classify_mutation_attempt( + { + "success": False, + "api_called": False, + "permission_report": {"missing_permission": "gitea.issue.comment"}, + } + ) + self.assertEqual(result["classification"], CAPABILITY_GATE_REJECTION) + self.assertTrue(result["pre_api"]) + + def test_transport_failure_before_api_is_its_own_class(self): + result = classify_mutation_attempt( + {"success": False, "api_called": False, "transport_error": "EOF"} + ) + self.assertEqual(result["classification"], TRANSPORT_FAILURE_BEFORE_API) + self.assertTrue(result["pre_api"]) + + def test_successful_server_mutation_is_its_own_class(self): + result = classify_mutation_attempt(SUCCESSFUL_COMMENT) + self.assertEqual(result["classification"], SERVER_SIDE_MUTATION) + self.assertFalse(result["pre_api"]) + + def test_each_class_is_distinct(self): + classes = { + classify_mutation_attempt(payload)["classification"] + for payload in ( + MISSING_LEDGER_BLOCK, + {"success": False, "api_called": False, "capability_blocked": True}, + {"success": False, "api_called": False, "transport_failed": True}, + SUCCESSFUL_COMMENT, + ) + } + self.assertEqual(len(classes), 4) + + +class TestAC2BudgetAccounting(unittest.TestCase): + """AC2: pre-API failures never consume server-side mutation budget.""" + + def test_missing_thread_state_ledger_not_counted_as_mutation(self): + result = classify_mutation_attempt(MISSING_LEDGER_BLOCK) + self.assertFalse(result["budget_consumed"]) + self.assertIs(result["api_called"], False) + + def test_missing_canonical_issue_state_not_counted_as_mutation(self): + result = classify_mutation_attempt(MISSING_CANONICAL_STATE) + self.assertFalse(result["budget_consumed"]) + self.assertIs(result["api_called"], False) + + def test_transport_failure_before_api_not_counted_as_mutation(self): + result = classify_mutation_attempt( + { + "success": False, + "api_called": False, + "transport_error": "connection reset", + } + ) + self.assertFalse(result["budget_consumed"]) + + def test_capability_gate_block_not_counted_as_mutation(self): + result = classify_mutation_attempt( + { + "success": False, + "api_called": False, + "permission_report": {"missing_permission": "gitea.pr.merge"}, + } + ) + self.assertFalse(result["budget_consumed"]) + + def test_successful_comment_with_comment_id_counts_as_one_mutation(self): + result = classify_mutation_attempt(SUCCESSFUL_COMMENT) + self.assertTrue(result["budget_consumed"]) + self.assertEqual(result["proof_fields"], ["comment_id"]) + + +class TestAC2FailsClosed(unittest.TestCase): + """AC2 must not become a loophole: ambiguity still fails closed.""" + + def test_api_reached_without_proof_is_ambiguous_and_consumes_budget(self): + result = classify_mutation_attempt({"success": True, "api_called": True}) + self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK) + self.assertTrue(result["budget_consumed"]) + self.assertTrue(result["requires_readback"]) + + def test_missing_api_called_signal_fails_closed(self): + result = classify_mutation_attempt({"success": False}) + self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK) + self.assertTrue(result["budget_consumed"]) + self.assertIsNone(result["api_called"]) + + def test_empty_and_none_results_fail_closed(self): + for payload in ({}, None): + result = classify_mutation_attempt(payload) + self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK) + self.assertTrue(result["budget_consumed"]) + + def test_success_with_proof_counts_even_when_api_called_absent(self): + result = classify_mutation_attempt({"success": True, "comment_id": 13320}) + self.assertEqual(result["classification"], SERVER_SIDE_MUTATION) + self.assertTrue(result["budget_consumed"]) + + def test_blank_proof_field_is_not_proof(self): + result = classify_mutation_attempt( + {"success": True, "api_called": True, "merge_commit_sha": " "} + ) + self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK) + + +class TestAC3AttemptLedger(unittest.TestCase): + """AC3: failed attempts are still logged locally.""" + + def test_failed_attempts_are_recorded(self): + ledger: list[dict] = [] + record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="create_issue_comment") + record_attempt(ledger, MISSING_CANONICAL_STATE, operation="create_issue_comment") + self.assertEqual(len(ledger), 2) + self.assertTrue( + all(e["classification"] == LOCAL_VALIDATOR_REJECTION for e in ledger) + ) + + def test_recorded_entry_carries_operation_and_timestamp(self): + ledger: list[dict] = [] + entry = record_attempt( + ledger, + SUCCESSFUL_COMMENT, + operation="create_issue_comment", + timestamp="2026-07-20T18:15:04+00:00", + ) + self.assertEqual(entry["operation"], "create_issue_comment") + self.assertEqual(entry["timestamp"], "2026-07-20T18:15:04+00:00") + + def test_timestamp_is_generated_when_omitted(self): + ledger: list[dict] = [] + entry = record_attempt(ledger, SUCCESSFUL_COMMENT) + self.assertTrue(entry["timestamp"]) + + +class TestAC5CorrectedCommentAllowed(unittest.TestCase): + """AC5: the #615 reproduction — two local rejections then one success.""" + + def _replay_615_flow(self) -> list[dict]: + ledger: list[dict] = [] + record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="create_issue_comment") + record_attempt(ledger, MISSING_CANONICAL_STATE, operation="create_issue_comment") + record_attempt(ledger, SUCCESSFUL_COMMENT, operation="create_issue_comment") + return ledger + + def test_corrected_comment_after_two_rejections_is_allowed(self): + summary = summarize_attempt_ledger(self._replay_615_flow()) + # The regression: budget must show ONE mutation, not three attempts. + self.assertEqual(summary["successful_server_mutations"], 1) + self.assertEqual(summary["budget_consumed"], 1) + + def test_all_three_attempts_remain_visible(self): + summary = summarize_attempt_ledger(self._replay_615_flow()) + self.assertEqual(summary["total_attempts"], 3) + self.assertEqual(summary["local_failed_attempts"], 2) + + def test_no_readback_required_for_clean_flow(self): + summary = summarize_attempt_ledger(self._replay_615_flow()) + self.assertFalse(summary["requires_readback"]) + + +class TestAC4FinalReportAccounting(unittest.TestCase): + """AC4: the report must show the three categories, and match the ledger.""" + + def _mixed_ledger(self) -> list[dict]: + ledger: list[dict] = [] + record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="comment") + record_attempt(ledger, MISSING_CANONICAL_STATE, operation="comment") + record_attempt( + ledger, + {"success": False, "api_called": False, "transport_error": "EOF"}, + operation="comment", + ) + record_attempt( + ledger, + {"success": False, "api_called": False, "capability_blocked": True}, + operation="merge", + ) + record_attempt(ledger, SUCCESSFUL_COMMENT, operation="comment") + return ledger + + def test_summary_separates_the_three_categories(self): + summary = summarize_attempt_ledger(self._mixed_ledger()) + self.assertEqual(summary["local_failed_attempts"], 2) + self.assertEqual(summary["blocked_api_attempts"], 2) + self.assertEqual(summary["successful_server_mutations"], 1) + + def test_matching_report_is_valid(self): + result = assess_final_report_mutation_accounting( + { + "local_failed_attempts": 2, + "blocked_api_attempts": 2, + "successful_server_mutations": 1, + }, + self._mixed_ledger(), + ) + self.assertTrue(result["valid"], result["reasons"]) + + def test_omitted_category_fails_closed(self): + result = assess_final_report_mutation_accounting( + {"local_failed_attempts": 2, "blocked_api_attempts": 2}, + self._mixed_ledger(), + ) + self.assertFalse(result["valid"]) + self.assertTrue( + any("successful_server_mutations" in r for r in result["reasons"]) + ) + + def test_inflated_mutation_count_fails_closed(self): + # The #617 bug shape: claiming three mutations when only one landed. + result = assess_final_report_mutation_accounting( + { + "local_failed_attempts": 2, + "blocked_api_attempts": 2, + "successful_server_mutations": 3, + }, + self._mixed_ledger(), + ) + self.assertFalse(result["valid"]) + self.assertTrue( + any("successful_server_mutations=3" in r for r in result["reasons"]) + ) + + def test_ambiguous_attempt_requires_readback_proof(self): + ledger: list[dict] = [] + record_attempt(ledger, {"success": True, "api_called": True}, operation="comment") + report = { + "local_failed_attempts": 0, + "blocked_api_attempts": 0, + "successful_server_mutations": 0, + } + blocked = assess_final_report_mutation_accounting(report, ledger) + self.assertFalse(blocked["valid"]) + self.assertTrue(any("readback_verified" in r for r in blocked["reasons"])) + + allowed = assess_final_report_mutation_accounting( + {**report, "readback_verified": True}, ledger + ) + self.assertTrue(allowed["valid"], allowed["reasons"]) + + def test_ledger_summary_is_returned_without_raw_entries(self): + result = assess_final_report_mutation_accounting({}, self._mixed_ledger()) + self.assertNotIn("entries", result["ledger_summary"]) + self.assertEqual(result["ledger_summary"]["total_attempts"], 5) + + +class TestEmptyLedger(unittest.TestCase): + def test_empty_ledger_summarizes_to_zero(self): + summary = summarize_attempt_ledger([]) + self.assertEqual(summary["total_attempts"], 0) + self.assertEqual(summary["successful_server_mutations"], 0) + self.assertFalse(summary["requires_readback"]) + + def test_none_ledger_is_tolerated(self): + self.assertEqual(summarize_attempt_ledger(None)["total_attempts"], 0) + + +if __name__ == "__main__": + unittest.main() + + +class TestValidatorIntegration(unittest.TestCase): + """The classifier is wired into the shared final-report validator (AC4).""" + + def _ledger_two_rejections_one_success(self) -> list[dict]: + ledger: list[dict] = [] + record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="comment") + record_attempt(ledger, MISSING_CANONICAL_STATE, operation="comment") + record_attempt(ledger, SUCCESSFUL_COMMENT, operation="comment") + return ledger + + def test_rule_is_noop_without_a_ledger(self): + from final_report_validator import assess_final_report_validator + + result = assess_final_report_validator("some report", "review_pr") + self.assertFalse( + any( + f["rule_id"] == "shared.mutation_budget_accounting" + for f in result["findings"] + ) + ) + + def test_report_matching_ledger_produces_no_finding(self): + from final_report_validator import assess_final_report_validator + + report = ( + "Local failed attempts: 2\n" + "Blocked API attempts: 0\n" + "Successful server-side mutations: 1\n" + ) + result = assess_final_report_validator( + report, + "review_pr", + mutation_attempt_ledger=self._ledger_two_rejections_one_success(), + ) + self.assertFalse( + any( + f["rule_id"] == "shared.mutation_budget_accounting" + for f in result["findings"] + ) + ) + + def test_counting_rejections_as_mutations_is_blocked(self): + from final_report_validator import assess_final_report_validator + + # The #617 bug: three attempts reported as three server-side mutations. + report = ( + "Local failed attempts: 0\n" + "Blocked API attempts: 0\n" + "Successful server-side mutations: 3\n" + ) + result = assess_final_report_validator( + report, + "review_pr", + mutation_attempt_ledger=self._ledger_two_rejections_one_success(), + ) + findings = [ + f + for f in result["findings"] + if f["rule_id"] == "shared.mutation_budget_accounting" + ] + self.assertTrue(findings) + self.assertTrue(all(f["severity"] == "block" for f in findings))