Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 68 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request 'fix: mutation budget counts server-side changes only (Closes #617)' (#775) from fix/issue-617-mutation-budget-classifier into master

Browse Source
This commit was merged in pull request #775.
This commit is contained in:
sysadmin
2026-07-20 20:28:32 -05:00
parent 702ceb2480 296601647d
commit 52ded0ea71
3 changed files with 746 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
final_report_validator.py
+72
View File
@@ -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, ()):
mutation_budget_classifier.py
+286
View File
@@ -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"},
}
tests/test_mutation_budget_classifier.py
+388
View File
@@ -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))