From c4724bed8fb697275cbe21d0402f5aef934809b2 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 20:44:00 +0000 Subject: [PATCH 1/7] fix(permits): decide every budget write in the database, not from a stale read Four defects in permits.py, all the same shape: a spent_credits value read in one statement and written back in a later one, serialized only by SELECT ... FOR UPDATE. That lock is a silent no-op on SQLite, and nothing in this repo forbids SQLite in production -- config.py lists it and trust_mode.py has no engine guardrail -- so these were live on a supported configuration, not theoretical. reserve_budget gated on status and the cap but not on expiry. A permit past its expires_at is still status="active" until the sweeper runs, which is the steady state for every permit between its deadline and that sweep, so it kept funding work after its authority ended. The expiry is now in the statement's own predicate, matching authorize_and_reserve. reserve_budget also reported every guarded-write failure as permit_budget_exceeded. A revoked or expired permit was therefore described as out of money, which sends an operator to top up a permit that more money cannot revive. It now classifies status, then expiry, then budget. release_dispatch_budget_once decided whether it had already run by reading budget_released_at and then writing it. Two callers both observed NULL, both passed the check, and the reservation was released twice -- the permit ends up under-spent and can then exceed the cap it exists to enforce. The checkpoint is now claimed by a guarded UPDATE before any budget moves, so the database picks the single winner; the trailing ORM write of the same field is gone, since re-issuing it from a row loaded before that UPDATE is the bug in miniature. reconcile_budgets recomputed an absolute total from receipts and wrote it back unconditionally. A reservation landing between the receipt scan and that write was erased. An absolute set has no relative form, so it is now guarded on the value the pass actually observed: if the row moved, the repair is skipped and the permit is re-examined next pass rather than having a live reservation overwritten. The module comment claimed "Every spent_credits mutation is applied as a single guarded UPDATE rather than a read-modify-write." That was false -- reconcile_budgets did a plain ORM assignment. Rather than weaken the claim, the code now makes it true, and the comment states both forms the guard takes and the rule they encode. Each fix ships a regression test confirmed to fail against the unfixed source by reverting the service, not the test: - expired permit: DID NOT RAISE at all - revoked permit: reported permit_budget_exceeded, not permit_revoked - double release: both callers returned True - reconciler: clobbered a concurrent reservation of 9 back to 0 The interleaves are forced deterministically through an instrumented session rather than raced, so they cannot pass by scheduling luck. Validation: 1777 passed / 32 skipped (app + b2a_sdk); ruff and mypy 2.3.1 clean; doc references 272/242; PostgreSQL 16 -- permit concurrency 11 passed, two-process crash recovery 13 passed; trust release gate passed. --- CHANGELOG.md | 24 +++ app/services/permits.py | 126 ++++++++++-- tests/test_mcp_dispatch_reconciliation.py | 83 ++++++++ tests/test_permit_budget_integrity.py | 223 ++++++++++++++++++++++ 4 files changed, 438 insertions(+), 18 deletions(-) create mode 100644 tests/test_permit_budget_integrity.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e9e278f..b90c7453 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -317,6 +317,30 @@ full release gate; do not backfill a final `v1.2.0` tag. ### πŸ”’ Security +- **Every permit budget write is now decided by the database, not by a value + this process read earlier.** Three paths in `permits.py` wrote + `spent_credits` from a stale read, serialized only by `SELECT ... FOR + UPDATE` β€” which is a silent no-op on SQLite, and nothing in this repository + forbids SQLite in production. All three are fixed, and each ships a + regression test confirmed to fail against the unfixed source: + - `reserve_budget` gated on `status` and the cap but **not on expiry**, so a + permit past its `expires_at` kept funding work until some other process + happened to flip its status β€” authority outliving its own deadline. The + expiry is now in the statement's predicate. + - `reserve_budget` also collapsed every guarded-write failure to + `permit_budget_exceeded`, telling an operator to top up a permit that had + actually been revoked or had expired. It now classifies status, then + expiry, then budget, matching `authorize_and_reserve`. + - `release_dispatch_budget_once` decided whether it had already run by + reading `budget_released_at` and then writing it. Two callers both saw + `NULL`, both passed, and the reservation was released **twice**, leaving + the permit under-spent and able to exceed its own cap. The checkpoint is + now claimed by a guarded UPDATE before any budget moves. + - `reconcile_budgets` recomputed an absolute total from receipts and wrote + it back unconditionally, erasing any reservation that landed mid-pass. It + now commits only where the stored spend still equals what that pass read, + and skips rather than clobbering β€” optimistic concurrency control, with + the permit re-examined on the next pass. - **The governed AWI HTTP charge is now keyed to its idempotency record.** `awi_http_governance.py` called `money.charge()` with no `operation_key`, so the debit had neither the `uq_ledger_wallet_operation_key` constraint nor the diff --git a/app/services/permits.py b/app/services/permits.py index 30332ff3..f0bce5d6 100644 --- a/app/services/permits.py +++ b/app/services/permits.py @@ -67,6 +67,24 @@ def _stamp(value: datetime | None) -> str | None: # transient "database is locked"/"snapshot" conflict on the second writer; we # retry that with a small backoff. On PostgreSQL the row lock blocks instead of # raising, so the retry never triggers there. +# +# The guard takes one of two forms, and both are decided by the database rather +# than by this process: +# +# * **Relative** β€” reservations and releases increment or decrement in place +# (``spent_credits + amount``), with the cap, the status and the expiry in +# the statement's own predicate. A row count other than 1 means the write +# lost, and the caller re-reads to classify *why* before reporting a reason. +# * **Conditional on the observed value** β€” ``reconcile_budgets`` recomputes an +# absolute total from receipts, which has no relative form, so it commits +# only where the stored spend still equals the value that pass read. A +# concurrent reservation makes that predicate false and the repair is skipped +# rather than overwriting it. This is optimistic concurrency control: detect +# the lost race by affected-row count instead of holding a lock. +# +# The rule this encodes: no ``spent_credits`` value may be written from a number +# this process read in an earlier statement without the database re-checking +# that the number still holds. _PERMIT_WRITE_MAX_ATTEMPTS = 40 @@ -741,7 +759,11 @@ async def _once() -> None: async with session.begin(): now = utc_now() # Atomic guarded reserve (see authorize_and_reserve): the cap - # is enforced by the WHERE clause, not a read-then-write. + # *and the expiry* are enforced by the WHERE clause, not a + # read-then-write. Without the expiry predicate a permit that + # is past ``expires_at`` but still ``active`` -- the sweeper + # flips status lazily -- keeps accepting reservations, which + # is authority outliving its own deadline. reserved = await session.execute( sa_update(PermitModel) .where( @@ -753,6 +775,10 @@ async def _once() -> None: ColumnElement[bool], PermitModel.status == "active", ), + cast( + ColumnElement[bool], + PermitModel.expires_at > now, + ), cast( ColumnElement[bool], PermitModel.spent_credits + amount @@ -769,6 +795,15 @@ async def _once() -> None: exists = await session.get(PermitModel, permit_id) if exists is None: raise PermitError("permit_not_found") + # Classify in the same order as authorize_and_reserve: + # status, then expiry, then budget. Reporting an expired + # or revoked permit as "out of money" sends the operator + # to top up a permit that more money cannot revive. + await session.refresh(exists) + if exists.status != "active": + raise PermitError(f"permit_{exists.status}") + if to_naive_utc(exists.expires_at) <= now: + raise PermitError("permit_expired") raise PermitError("permit_budget_exceeded") # Budget percentage alerts, recomputed from the committed @@ -857,18 +892,47 @@ async def release_dispatch_budget_once(self, attempt_id: str) -> bool: async def _once() -> bool: async with factory() as session: async with session.begin(): - attempt = await session.get( - McpDispatchAttemptModel, - attempt_id, - with_for_update=True, - ) + attempt = await session.get(McpDispatchAttemptModel, attempt_id) if attempt is None: raise PermitError("dispatch_attempt_not_found") - if attempt.state != "returned_error": - raise PermitError("dispatch_budget_release_state_invalid") - if attempt.budget_released_at is not None: - return False now = utc_now() + # Claim the checkpoint with a guarded UPDATE *before* moving + # any budget. Reading ``budget_released_at`` and then writing + # it is a read-modify-write: it was serialized only by + # ``SELECT ... FOR UPDATE``, which is a silent no-op on + # SQLite, so two concurrent callers both saw NULL, both + # passed the check, and the reservation was released twice -- + # leaving the permit under-spent and able to over-spend its + # cap later. The database now decides the winner: exactly one + # transaction can flip NULL -> now. + claimed = await session.execute( + sa_update(McpDispatchAttemptModel) + .where( + cast( + ColumnElement[bool], + McpDispatchAttemptModel.attempt_id == attempt_id, + ), + cast( + ColumnElement[bool], + McpDispatchAttemptModel.state == "returned_error", + ), + cast( + ColumnElement[bool], + cast( + Any, McpDispatchAttemptModel.budget_released_at + ).is_(None), + ), + ) + .values(budget_released_at=now, updated_at=now) + .execution_options(synchronize_session=False) + ) + if (cast(Any, claimed).rowcount or 0) != 1: + # Re-read to tell "already released" (a benign duplicate + # call, False) from "wrong state" (a caller bug, raise). + await session.refresh(attempt) + if attempt.state != "returned_error": + raise PermitError("dispatch_budget_release_state_invalid") + return False # Atomic clamped decrement so a concurrent reservation on the # same permit is not clobbered by a read-modify-write here. released = await session.execute( @@ -899,9 +963,9 @@ async def _once() -> bool: ) if (cast(Any, released).rowcount or 0) == 0: raise PermitError("permit_not_found") - attempt.budget_released_at = now - attempt.updated_at = now - session.add(attempt) + # The checkpoint was already stamped by the guarded claim + # above; writing it again through the ORM would re-issue it + # from a row this session loaded before that UPDATE. return True return await self._run_with_write_retry(_once) @@ -1127,11 +1191,37 @@ async def reconcile_budgets(self, *, idle_seconds: int = 900) -> int: ), Decimal("0"), ) - if permit.spent_credits != consumed_decimal: - permit.spent_credits = consumed_decimal - permit.updated_at = utc_now() - session.add(permit) - corrected += 1 + observed = permit.spent_credits + if observed != consumed_decimal: + # This is an absolute set recomputed from receipts, not + # an increment, so it cannot be expressed as a relative + # guarded UPDATE. Guard it on the value this pass + # actually observed instead: if a reservation landed + # between the receipt scan and this write, the row no + # longer matches and we correct nothing rather than + # erasing that reservation. ``with_for_update()`` above + # does not cover this on SQLite, where it is a no-op. + # The skipped permit is simply re-examined next pass. + repaired = await session.execute( + sa_update(PermitModel) + .where( + cast( + ColumnElement[bool], + PermitModel.permit_id == permit.permit_id, + ), + cast( + ColumnElement[bool], + PermitModel.spent_credits == observed, + ), + ) + .values( + spent_credits=consumed_decimal, + updated_at=utc_now(), + ) + .execution_options(synchronize_session=False) + ) + if (cast(Any, repaired).rowcount or 0) == 1: + corrected += 1 await session.commit() return corrected diff --git a/tests/test_mcp_dispatch_reconciliation.py b/tests/test_mcp_dispatch_reconciliation.py index c25fb6ef..a567c08d 100644 --- a/tests/test_mcp_dispatch_reconciliation.py +++ b/tests/test_mcp_dispatch_reconciliation.py @@ -1010,3 +1010,86 @@ async def test_lost_commit_ack_recovery_no_double_charge( permit_after_retry = await get_permit_service().get_permit(permit["permit_id"]) assert permit_after_retry is not None assert permit_after_retry.spent_credits == spent_after_first + + +@pytest.mark.anyio +async def test_dispatch_budget_release_is_once_only_under_a_stale_read( + client: AsyncClient, + clean_database, + monkeypatch, +) -> None: + """Two callers that both observe an unreleased attempt must release once. + + ``release_dispatch_budget_once`` decided whether it had already run by + reading ``budget_released_at`` and then writing it β€” a read-modify-write + serialized only by ``SELECT ... FOR UPDATE``. That lock is a silent no-op + on SQLite, so both callers saw NULL, both passed the check, and the + reservation was released twice: the permit ends up *under*-spent and can + then exceed the very cap it is meant to enforce. + + The interleave is forced rather than raced. The first caller's read is + allowed to happen, a second caller then runs to completion, and only then + does the first caller reach its write β€” exactly the ordering the row lock + was supposed to prevent and does not on SQLite. + """ + seed = await _seed_attempt( + client, + suffix="release-once-only", + state="returned_error", + result_payload={"error": "confirmed"}, + error_code="upstream_returned_error", + ) + permits = get_permit_service() + spent_before = (await permits.get_permit(seed.permit_id)).spent_credits + assert spent_before == CREDITS + + import app.services.permits as permits_module + + real_factory = get_session_factory() + state = {"fired": False, "second_result": None} + + class _StaleReadSession: + def __init__(self, inner): + self._inner = inner + + def __getattr__(self, name): + return getattr(self._inner, name) + + async def execute(self, *args, **kwargs): + # Fires after this caller's `session.get` of the attempt row and + # before its first write β€” the read-modify-write window. + if not state["fired"]: + state["fired"] = True + monkeypatch.setattr( + permits_module, "get_session_factory", lambda: real_factory + ) + state["second_result"] = await permits.release_dispatch_budget_once( + seed.attempt_id + ) + return await self._inner.execute(*args, **kwargs) + + class _StaleReadFactory: + def __init__(self, cm): + self._cm = cm + + async def __aenter__(self): + return _StaleReadSession(await self._cm.__aenter__()) + + async def __aexit__(self, *exc): + return await self._cm.__aexit__(*exc) + + monkeypatch.setattr( + permits_module, + "get_session_factory", + lambda: (lambda: _StaleReadFactory(real_factory())), + ) + + first_result = await permits.release_dispatch_budget_once(seed.attempt_id) + + assert state["fired"], "the interleave never ran β€” the test proved nothing" + # Exactly one caller may claim the release. + assert [first_result, state["second_result"]].count(True) == 1 + # And the budget moved exactly once, not twice. + permit = await permits.get_permit(seed.permit_id) + assert permit is not None + assert permit.spent_credits == spent_before - CREDITS diff --git a/tests/test_permit_budget_integrity.py b/tests/test_permit_budget_integrity.py new file mode 100644 index 00000000..ac6ffd48 --- /dev/null +++ b/tests/test_permit_budget_integrity.py @@ -0,0 +1,223 @@ +"""Budget mutations must be decided by the database, not by a stale read. + +Every test here targets a path that wrote ``spent_credits`` from a value this +process had read in an earlier statement. On PostgreSQL the surrounding +``SELECT ... FOR UPDATE`` hid the problem; on SQLite that lock is a silent +no-op, and nothing in this repository forbids SQLite in production +(``config.py`` lists it and ``trust_mode.py`` has no engine guardrail), so +these are live defects on a supported configuration rather than theory. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from decimal import Decimal + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.core.time import utc_now +from app.db.database import get_session_factory +from app.db.models import PermitModel +from app.main import app +from app.schemas.trust import PermitCreateRequest +from app.services.permits import PermitError, get_permit_service +from tests.test_trust_helpers import provision_agent_wallet + + +@pytest.fixture +async def client(): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield c + + +async def _permit_with_budget(client, *, max_credits: str = "10"): + """A live, active permit with room in its cap.""" + provisioned = await provision_agent_wallet(client) + permit = await get_permit_service().create_permit( + PermitCreateRequest( + issuer_wallet_id=provisioned["agent_wallet_id"], + subject_wallet_id=provisioned["agent_wallet_id"], + subject_key_id=provisioned["key_id"], + allowed_tools=["budget-integrity-tool"], + scopes=["tool:budget-integrity-tool:invoke"], + max_credits=Decimal(max_credits), + expires_at=( + datetime.now(timezone.utc) + timedelta(minutes=30) + ).replace(tzinfo=None), + ) + ) + return provisioned, permit + + +async def _spent(permit_id: str) -> Decimal: + factory = get_session_factory() + async with factory() as session: + model = await session.get(PermitModel, permit_id) + assert model is not None + return model.spent_credits + + +async def _backdate_expiry(permit_id: str) -> None: + """Push ``expires_at`` into the past while leaving ``status`` active. + + This is the real steady state, not a contrived one: expiry is a timestamp + comparison and the sweeper flips ``status`` lazily, so between the deadline + and the sweep every permit in the table looks exactly like this. + """ + factory = get_session_factory() + async with factory() as session: + async with session.begin(): + model = await session.get(PermitModel, permit_id) + assert model is not None + assert model.status == "active" + model.expires_at = utc_now() - timedelta(minutes=5) + session.add(model) + + +@pytest.mark.anyio +async def test_reserve_budget_refuses_a_permit_past_its_deadline( + client, clean_database, enforce_naive_utc_datetime_columns +): + """An expired permit must not fund anything, even with budget remaining. + + ``reserve_budget`` gated only on ``status == "active"`` and the cap, so a + permit whose ``expires_at`` had passed kept reserving until some other + process happened to flip its status. That is authority outliving its own + deadline β€” the one bound a permit exists to carry. + """ + _, permit = await _permit_with_budget(client) + await _backdate_expiry(permit.permit_id) + + with pytest.raises(PermitError) as denied: + await get_permit_service().reserve_budget(permit.permit_id, Decimal("1")) + + assert denied.value.reason == "permit_expired" + # And no budget moved on the way to the denial. + assert await _spent(permit.permit_id) == Decimal("0") + + +@pytest.mark.anyio +async def test_reserve_budget_reports_revocation_rather_than_exhaustion( + client, clean_database, enforce_naive_utc_datetime_columns +): + """A revoked permit is not an out-of-money permit. + + The guarded UPDATE fails for several distinct reasons and the old code + collapsed all of them to ``permit_budget_exceeded``, which tells an + operator to top up a permit that more money cannot revive. + """ + _, permit = await _permit_with_budget(client) + await get_permit_service().revoke_permit(permit.permit_id) + + with pytest.raises(PermitError) as denied: + await get_permit_service().reserve_budget(permit.permit_id, Decimal("1")) + + assert denied.value.reason == "permit_revoked" + assert await _spent(permit.permit_id) == Decimal("0") + + +@pytest.mark.anyio +async def test_reserve_budget_still_denies_a_genuine_cap_breach( + client, clean_database, enforce_naive_utc_datetime_columns +): + """The added predicates must not swallow the reason they sit in front of.""" + _, permit = await _permit_with_budget(client, max_credits="5") + + await get_permit_service().reserve_budget(permit.permit_id, Decimal("4")) + with pytest.raises(PermitError) as denied: + await get_permit_service().reserve_budget(permit.permit_id, Decimal("2")) + + assert denied.value.reason == "permit_budget_exceeded" + assert await _spent(permit.permit_id) == Decimal("4") + + +@pytest.mark.anyio +async def test_reconcile_budgets_does_not_erase_a_concurrent_reservation( + client, clean_database, enforce_naive_utc_datetime_columns, monkeypatch +): + """The repair is an absolute overwrite and must not clobber a live write. + + ``reconcile_budgets`` recomputes ``spent_credits`` from receipts and writes + the total back. Between the receipt scan and that write, a reservation can + land. The old code overwrote it unconditionally β€” the reservation vanished, + the permit under-counted its own spend, and the cap it was meant to enforce + could then be exceeded by exactly the erased amount. + + The interleave is forced deterministically rather than raced: the service + calls ``utc_now()`` once before the scan and again while building the + repair statement, so the second call is a seam that sits after the value + has been observed and before the UPDATE executes. + """ + _, permit = await _permit_with_budget(client, max_credits="100") + + # Make the permit look orphaned: spend on the row, no receipts to back it, + # and idle long enough for the reconciler to pick it up. + factory = get_session_factory() + async with factory() as session: + async with session.begin(): + model = await session.get(PermitModel, permit.permit_id) + assert model is not None + model.spent_credits = Decimal("5") + model.status = "revoked" + model.updated_at = utc_now() - timedelta(hours=1) + session.add(model) + + import app.services.permits as permits_module + + real_factory = get_session_factory() + + async def _concurrent_reservation() -> None: + """Another caller reserves while the reconcile pass is mid-flight.""" + async with real_factory() as s: + async with s.begin(): + row = await s.get(PermitModel, permit.permit_id) + assert row is not None + row.spent_credits = Decimal("9") + s.add(row) + + fired = {"done": False, "executes": 0} + + class _InterleavingSession: + """Delegates to the real session, firing the hook once mid-pass.""" + + def __init__(self, inner): + self._inner = inner + + def __getattr__(self, name): + return getattr(self._inner, name) + + async def execute(self, *args, **kwargs): + fired["executes"] += 1 + # Fire *after* the first statement β€” the stale-permit SELECT that + # loads the row and fixes the value this pass will reason about. + # Firing before it would land the reservation ahead of the + # observation, which is not the race and proves nothing. + if fired["executes"] == 2 and not fired["done"]: + fired["done"] = True + await _concurrent_reservation() + return await self._inner.execute(*args, **kwargs) + + class _InterleavingFactory: + def __init__(self, cm): + self._cm = cm + + async def __aenter__(self): + return _InterleavingSession(await self._cm.__aenter__()) + + async def __aexit__(self, *exc): + return await self._cm.__aexit__(*exc) + + monkeypatch.setattr( + permits_module, + "get_session_factory", + lambda: (lambda: _InterleavingFactory(real_factory())), + ) + + corrected = await get_permit_service().reconcile_budgets(idle_seconds=1) + + # The concurrent value survives; the stale repair is skipped, not applied. + assert fired["done"], "the interleave never ran β€” the test proved nothing" + assert await _spent(permit.permit_id) == Decimal("9") + assert corrected == 0 From 8d56a97a53f3550550342b457314c31befc79ef9 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 21:06:03 +0000 Subject: [PATCH 2/7] fix(billing,wallets): decide every wallet money move in the database Same shape as the permit fixes in the previous commit, now in the paths that hold actual credits. Each read a balance or counter in one statement and wrote it back in a later one, serialized only by SELECT ... FOR UPDATE -- which is a silent no-op on SQLite, a configuration this repo permits in production. charge() checked wallet.balance and then, several statements later, wrote back balance - amount computed from that read. A charge committing in the gap is overwritten: the ledger records the debit but the balance never reflects it, so the request was served for free and the books stop balancing. The debit is now a single guarded UPDATE repeating balance >= amount as its own predicate; losing that predicate is reported exactly as failing the up-front check, because in both cases nothing moved. reclaim_child_wallet read the child's balance, zeroed the child, and credited the parent by the amount it had read. Two concurrent reclaims each credited the parent in full, so the parent gained twice what the child ever held -- credits created from nothing. The child's balance is now claimed by a guarded UPDATE before any of it reaches the parent, and the parent is credited relatively. Refund credits (refund_charge and the operator repair in refund_reconciliation) wrote absolute totals, so one of two concurrent refunds was lost outright. That is the customer's money, not a metric. The velocity counters accumulated by read-modify-write. Under-counting there is not a skewed statistic: hourly_spent and daily_spent are what the spend cap and the anomaly auto-freeze are measured against, so both controls silently stopped firing precisely when spend was most concurrent. The Stripe cumulative-refund clawback wrote a balance derived from an earlier read. The cumulative arithmetic already made the event idempotent, but the write itself could still erase a concurrent charge -- against fiat Stripe has already returned. refund_reconciliation also mutated permit.spent_credits directly, outside the module whose header now guarantees every such write is guarded. That guarantee is only as good as its weakest caller. app/db/sql_expressions.py holds the shared clamped-decrement expression so the safe form is defined once instead of re-derived at each call site. Both regression tests were confirmed to fail against the unfixed source by reverting the services, not the tests: - conservation: balance 8.50 while the ledger showed 3.00 debited from 10 -- one charge served free - reclaim: parent 140 instead of 100, from a child holding 40 The interleaves are forced deterministically through an instrumented session, firing after a statement returns so the seam is identical across the fixed and unfixed code, which issue different numbers of statements. Each test asserts the interleave actually fired before asserting anything else, so a harness that misses the window fails loudly instead of passing. Validation: 1779 passed / 32 skipped (app + b2a_sdk); ruff and mypy 2.3.1 clean over 173 files; doc references 277/243; PostgreSQL 16 -- permit concurrency 11 passed, two-process crash recovery 13 passed; trust release gate passed. --- CHANGELOG.md | 29 ++++ app/db/sql_expressions.py | 33 ++++ app/services/billing_engine.py | 92 +++++++++-- app/services/refund_reconciliation.py | 74 ++++++--- app/services/stripe_integration.py | 31 +++- app/services/velocity_monitor.py | 33 +++- app/services/wallet_engine.py | 73 ++++++++- tests/test_wallet_ledger_integrity.py | 228 ++++++++++++++++++++++++++ 8 files changed, 544 insertions(+), 49 deletions(-) create mode 100644 app/db/sql_expressions.py create mode 100644 tests/test_wallet_ledger_integrity.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b90c7453..f00624b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -317,6 +317,35 @@ full release gate; do not backfill a final `v1.2.0` tag. ### πŸ”’ Security +- **Every wallet money move is now decided by the database too.** The same + read-modify-write shape as the permit fixes below, in the paths that hold + actual credits. All were serialized only by `SELECT ... FOR UPDATE`, a + silent no-op on SQLite: + - `charge` read the balance, checked it, and several statements later wrote + back `balance - amount` computed from that read. A charge committing in + between was **overwritten**: the ledger recorded the debit but the balance + never reflected it, so that request was served for free and the books + stopped balancing. The debit is now one guarded UPDATE that repeats + `balance >= amount` as its own predicate. + - `reclaim_child_wallet` read the child's balance, zeroed the child, and + credited the parent by the amount it had read. Two concurrent reclaims + each credited the parent in full β€” **minting credits that never existed**. + The child's balance is now claimed by a guarded UPDATE before any of it + reaches the parent. + - Refund credits (`refund_charge`, and the operator repair in + `refund_reconciliation`) wrote absolute totals, so one of two concurrent + refunds was lost outright β€” the customer's money. + - The velocity counters `hourly_spent` / `daily_spent` accumulated by + read-modify-write. Under-counting there does not skew a metric; it is what + the spend cap and the anomaly auto-freeze are measured against, so **both + controls silently stopped firing** exactly when spend was most concurrent. + - The Stripe cumulative-refund clawback wrote a balance derived from an + earlier read, against fiat Stripe had already returned. + - `refund_reconciliation` also mutated `permit.spent_credits` directly, + outside the module that guarantees every such write is guarded. That + guarantee is only as good as its weakest caller. + - New `app/db/sql_expressions.py` holds the shared clamped-decrement + expression, so the safe form is defined once rather than re-derived. - **Every permit budget write is now decided by the database, not by a value this process read earlier.** Three paths in `permits.py` wrote `spent_credits` from a stale read, serialized only by `SELECT ... FOR diff --git a/app/db/sql_expressions.py b/app/db/sql_expressions.py new file mode 100644 index 00000000..858223d8 --- /dev/null +++ b/app/db/sql_expressions.py @@ -0,0 +1,33 @@ +"""Reusable SQL expressions for writes that must not lose a concurrent update. + +The money paths in this application mutate counters that several requests can +touch at once. The unsafe form is always the same: read a column into Python, +compute a new total, write the total back. That is a read-modify-write, and it +is serialized only by whatever row lock happens to be held β€” which is nothing +at all on SQLite, where ``SELECT ... FOR UPDATE`` parses and does nothing. + +Expressing the arithmetic against the *column* instead of against a value this +process read keeps the whole operation relative, so the database applies it to +whatever is stored at write time. +""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Any, cast + +from sqlalchemy import case +from sqlalchemy.sql.elements import ColumnElement + + +def clamped_decrement(column: Any, amount: Decimal) -> Any: + """``max(0, column - amount)``, evaluated by the database. + + The Python spelling of this clamp (``max(Decimal("0"), col - amount)``) + has to read the column first, which reintroduces the lost update the + relative form exists to prevent. + """ + return case( + (cast(ColumnElement[bool], column - amount < Decimal("0")), Decimal("0")), + else_=column - amount, + ) diff --git a/app/services/billing_engine.py b/app/services/billing_engine.py index 74b0fa74..fed5bad2 100644 --- a/app/services/billing_engine.py +++ b/app/services/billing_engine.py @@ -13,7 +13,7 @@ from decimal import Decimal from typing import Any, cast -from sqlalchemy import select +from sqlalchemy import select, update as sa_update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql.elements import ColumnElement @@ -21,6 +21,7 @@ from ..core.config import Settings from ..core.time import utc_now from ..db.converters import billing_alert_model_to_schema, ledger_entry_model_to_schema +from ..db.sql_expressions import clamped_decrement from ..db.models import ( BillingAlertModel, IdempotencyRecordModel, @@ -353,9 +354,50 @@ def reverse_velocity_record() -> None: top_up_url="/v1/billing/top-up/prepare", ) - wallet.balance -= charge_amount - wallet.lifetime_debits += charge_amount - wallet.updated_at = utc_now() + # Apply the debit as a single guarded UPDATE. The balance check above is + # a fast path for the common case; it is *not* what makes the debit + # safe, because reading the balance in one statement and writing it in + # another is a read-modify-write. ``SELECT ... FOR UPDATE`` above is a + # silent no-op on SQLite, so two concurrent charges both read the same + # balance, both clear the check, and both decrement -- the wallet is + # overdrawn by the second charge and the service is delivered twice. + # Repeating ``balance >= charge_amount`` inside the statement makes the + # database, not this process, decide whether the funds were there. + debited = await session.execute( + sa_update(WalletModel) + .where( + cast(ColumnElement[bool], WalletModel.wallet_id == wallet_id), + cast(ColumnElement[bool], WalletModel.balance >= charge_amount), + ) + .values( + balance=WalletModel.balance - charge_amount, + lifetime_debits=WalletModel.lifetime_debits + charge_amount, + updated_at=utc_now(), + ) + .execution_options(synchronize_session=False) + ) + if (cast(Any, debited).rowcount or 0) != 1: + # Lost the race: another charge took the funds after our check. + # Nothing was debited, so this is the same outcome as failing the + # check above and is reported identically. + await session.refresh(wallet) + shortfall = charge_amount - wallet.balance + reverse_velocity_record() + return InsufficientFundsResponse( + wallet_id=wallet_id, + current_balance=float(wallet.balance), + current_balance_exact=str(wallet.balance), + required_amount=float(charge_amount), + required_amount_exact=str(charge_amount), + shortfall=float(shortfall), + shortfall_exact=str(shortfall), + top_up_url="/v1/billing/top-up/prepare", + ) + # The guarded UPDATE bypassed the identity map, so the in-memory wallet + # still holds the pre-debit balance. Re-read it before anything derives + # a number from it -- the ledger's ``balance_after`` in particular has + # to be the balance this debit actually produced. + await session.refresh(wallet) entry = LedgerEntryModel( entry_id=str(uuid.uuid4()), @@ -608,18 +650,25 @@ async def refund_charge( return ledger_entry_model_to_schema(existing_refund) refund_amount = abs(charge_entry.amount) - wallet.balance += refund_amount - wallet.lifetime_debits = max( - Decimal("0"), - wallet.lifetime_debits - refund_amount, - ) + # Credit relatively, in one statement, for the same reason the + # debit does. Two refunds for *different* charges landing + # together each read the same balance and each write their own + # total, so one credit is silently lost -- and a lost credit is + # the customer's money. The duplicate-refund guard above does + # not help here: these are distinct, legitimate refunds. + refund_values: dict[str, Any] = { + "balance": WalletModel.balance + refund_amount, + "lifetime_debits": clamped_decrement( + WalletModel.lifetime_debits, refund_amount + ), + "updated_at": utc_now(), + } if _timestamp_in_current_period( charge_entry.timestamp, wallet.hourly_reset_at, ): - wallet.hourly_spent = max( - Decimal("0"), - wallet.hourly_spent - refund_amount, + refund_values["hourly_spent"] = clamped_decrement( + WalletModel.hourly_spent, refund_amount ) if _timestamp_in_current_period( charge_entry.timestamp, @@ -627,11 +676,22 @@ async def refund_charge( ): # daily_spent is incremented exactly once per charge (by the # velocity monitor), so reverse exactly one increment. - wallet.daily_spent = max( - Decimal("0"), - wallet.daily_spent - refund_amount, + refund_values["daily_spent"] = clamped_decrement( + WalletModel.daily_spent, refund_amount ) - wallet.updated_at = utc_now() + await session.execute( + sa_update(WalletModel) + .where( + cast( + ColumnElement[bool], WalletModel.wallet_id == wallet_id + ) + ) + .values(**refund_values) + .execution_options(synchronize_session=False) + ) + # ``balance_after`` below must be the balance this refund + # produced, not the one read before it. + await session.refresh(wallet) entry = LedgerEntryModel( entry_id=refund_entry_id, diff --git a/app/services/refund_reconciliation.py b/app/services/refund_reconciliation.py index 769574fd..3ed69a32 100644 --- a/app/services/refund_reconciliation.py +++ b/app/services/refund_reconciliation.py @@ -8,7 +8,7 @@ import json from typing import Any, Literal, cast -from sqlalchemy import select +from sqlalchemy import select, update as sa_update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql.elements import ColumnElement @@ -22,6 +22,7 @@ ReceiptModel, WalletModel, ) +from app.db.sql_expressions import clamped_decrement from app.schemas.trust import RefundReconciliationItem from app.schemas.trust import ReceiptResponse from app.services.permits import get_permit_service @@ -633,7 +634,7 @@ async def _retry_locked( refund = existing_refund if refund is None: - self._apply_refund(wallet, charge, item.credits) + await self._apply_refund(session, wallet, charge, item.credits) refund = LedgerEntryModel( entry_id=item.refund_entry_id, wallet_id=item.wallet_id, @@ -658,16 +659,35 @@ async def _retry_locked( ), correlation_id=item.ledger_entry_id, ) - session.add(wallet) session.add(refund) - permit.spent_credits -= item.credits - permit.updated_at = utc_now() + # Release the permit reservation with a clamped relative UPDATE + # for the same reason: ``permits.py`` guarantees that every + # ``spent_credits`` write is decided by the database, and that + # guarantee is only as good as its weakest caller. This one sat + # outside that module and wrote back a value read earlier. + await session.execute( + sa_update(PermitModel) + .where( + cast( + ColumnElement[bool], + PermitModel.permit_id == permit.permit_id, + ) + ) + .values( + spent_credits=clamped_decrement( + PermitModel.spent_credits, item.credits + ), + updated_at=utc_now(), + ) + .execution_options(synchronize_session=False) + ) + await session.refresh(permit) resolved_at = utc_now().isoformat() state = cast(dict[str, Any], payload["refund_reconciliation"]) state["status"] = "resolved" state["resolved_at"] = resolved_at record.response_json = json.dumps(payload, default=str) - session.add(permit) + # `permit` was updated by the guarded statement above, not the ORM. session.add(record) await session.flush() @@ -678,28 +698,44 @@ async def _retry_locked( ) return resolved, False - def _apply_refund( + async def _apply_refund( self, + session: AsyncSession, wallet: WalletModel, charge: LedgerEntryModel, amount: Decimal, ) -> None: - wallet.balance += amount - wallet.lifetime_debits = max( - Decimal("0"), - wallet.lifetime_debits - amount, - ) + """Credit a refund back to the wallet in a single relative statement. + + Reading each counter and writing back readΒ±amount is a read-modify-write + serialized only by a row lock, and ``SELECT ... FOR UPDATE`` is a silent + no-op on SQLite. Two refunds reconciled together each computed a total + from the same observed balance, so one credit was lost outright β€” and a + lost credit here is the customer's money, not a metric. + """ + values: dict[str, Any] = { + "balance": WalletModel.balance + amount, + "lifetime_debits": clamped_decrement(WalletModel.lifetime_debits, amount), + "updated_at": utc_now(), + } if _timestamp_in_current_period(charge.timestamp, wallet.hourly_reset_at): - wallet.hourly_spent = max( - Decimal("0"), - wallet.hourly_spent - amount, + values["hourly_spent"] = clamped_decrement( + WalletModel.hourly_spent, amount ) if _timestamp_in_current_period(charge.timestamp, wallet.daily_reset_at): - wallet.daily_spent = max( - Decimal("0"), - wallet.daily_spent - amount, + values["daily_spent"] = clamped_decrement(WalletModel.daily_spent, amount) + await session.execute( + sa_update(WalletModel) + .where( + cast( + ColumnElement[bool], + WalletModel.wallet_id == wallet.wallet_id, + ) ) - wallet.updated_at = utc_now() + .values(**values) + .execution_options(synchronize_session=False) + ) + await session.refresh(wallet) _service: RefundReconciliationService | None = None diff --git a/app/services/stripe_integration.py b/app/services/stripe_integration.py index ba19c9c3..b33ad480 100644 --- a/app/services/stripe_integration.py +++ b/app/services/stripe_integration.py @@ -13,12 +13,13 @@ import json import logging from decimal import Decimal, InvalidOperation -from typing import Any, Optional +from typing import Any, Optional, cast from uuid import uuid4 import stripe from sqlalchemy.exc import IntegrityError -from sqlalchemy import select +from sqlalchemy import select, update as sa_update +from sqlalchemy.sql.elements import ColumnElement from ..db.database import get_session_factory from ..db.models import BillingAlertModel, LedgerEntryModel, WalletModel @@ -402,8 +403,30 @@ async def _handle_refund(self, charge: Any, event_id: str | None = None) -> None payment_intent_id, ) return - wallet.balance -= refund_delta - wallet.lifetime_debits += refund_delta + # Apply the clawback relatively, in one statement. The + # cumulative arithmetic above already makes *this* event + # idempotent, but the write itself was still a + # read-modify-write: a charge landing on the same sponsor + # wallet between the read and the write is silently erased, + # and here that discrepancy is against real fiat Stripe has + # already returned. + await session.execute( + sa_update(WalletModel) + .where( + cast( + ColumnElement[bool], + WalletModel.wallet_id == credit_entry.wallet_id, + ) + ) + .values( + balance=WalletModel.balance - refund_delta, + lifetime_debits=WalletModel.lifetime_debits + refund_delta, + ) + .execution_options(synchronize_session=False) + ) + # The liability and freeze decision below turn on the + # balance this clawback produced, so read it back first. + await session.refresh(wallet) refund_liability = max(Decimal("0"), -wallet.balance) wallet_status_before_refund = wallet.status wallet_frozen_for_liability = False diff --git a/app/services/velocity_monitor.py b/app/services/velocity_monitor.py index a20ad193..82417785 100644 --- a/app/services/velocity_monitor.py +++ b/app/services/velocity_monitor.py @@ -14,7 +14,7 @@ from decimal import Decimal from typing import Optional, cast -from sqlalchemy import select +from sqlalchemy import select, update as sa_update from sqlalchemy.sql.elements import ColumnElement from ..core.time import to_naive_utc, utc_now @@ -109,13 +109,38 @@ async def check_and_record_charge( now = utc_now() self._reset_if_needed(wallet, now) + # Land any period reset before the relative increment below, so + # the increment applies on top of the reset value rather than + # racing it inside the same statement. + await session.flush() hourly_limit = wallet.hourly_limit or self._default_hourly_limit daily_limit = wallet.daily_limit or self._default_daily_limit - wallet.hourly_spent += charge_amount - wallet.daily_spent += charge_amount - wallet.last_charge_at = now + # Accumulate relatively, in one statement. Reading a counter and + # writing back read+charge is a read-modify-write serialized + # only by the ``SELECT ... FOR UPDATE`` above, which is a silent + # no-op on SQLite: concurrent charges each add to the same + # observed total and all but one increment is lost. This counter + # is what the spend cap and the anomaly auto-freeze are measured + # against, so under-counting does not merely skew a metric -- it + # silently disables both controls, and it does so precisely when + # spend is most concurrent. + await session.execute( + sa_update(WalletModel) + .where( + cast(ColumnElement[bool], WalletModel.wallet_id == wallet_id) + ) + .values( + hourly_spent=WalletModel.hourly_spent + charge_amount, + daily_spent=WalletModel.daily_spent + charge_amount, + last_charge_at=now, + ) + .execution_options(synchronize_session=False) + ) + # The limit check below must see the totals this charge produced, + # not the ones read before it. + await session.refresh(wallet) velocity_result = self._check_limits( wallet=wallet, diff --git a/app/services/wallet_engine.py b/app/services/wallet_engine.py index f2175485..9c30af5f 100644 --- a/app/services/wallet_engine.py +++ b/app/services/wallet_engine.py @@ -11,9 +11,9 @@ from collections.abc import Callable from datetime import datetime, timedelta, timezone from decimal import Decimal -from typing import cast +from typing import Any, cast -from sqlalchemy import select +from sqlalchemy import select, update as sa_update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.sql.elements import ColumnElement @@ -500,10 +500,69 @@ async def reclaim_child_wallet(self, child_wallet_id: str) -> dict: reclaim_amount = child.balance + # Claim the child's balance with a guarded UPDATE before any of + # it reaches the parent. Reading ``child.balance`` and then + # writing both wallets is a read-modify-write serialized only by + # the row locks taken above, and ``SELECT ... FOR UPDATE`` is a + # silent no-op on SQLite: two concurrent reclaims of the same + # child both read the full balance, both zero it, and both + # credit the parent -- minting credits that never existed. The + # database now admits exactly one reclaim, because only one + # transaction can move this row off the balance it was observed + # holding while it is still open. + claimed = await session.execute( + sa_update(WalletModel) + .where( + cast( + ColumnElement[bool], + WalletModel.wallet_id == child_wallet_id, + ), + cast( + ColumnElement[bool], + WalletModel.status != WalletStatus.CLOSED.value, + ), + cast( + ColumnElement[bool], + WalletModel.balance == reclaim_amount, + ), + ) + .values( + balance=Decimal("0"), + status=WalletStatus.CLOSED.value, + updated_at=utc_now(), + ) + .execution_options(synchronize_session=False) + ) + if (cast(Any, claimed).rowcount or 0) != 1: + # Someone else reclaimed this child, or its balance moved + # under us. Either way no credits may be created here. + await session.refresh(child) + if child.status == WalletStatus.CLOSED.value: + raise ValueError("Wallet already closed") + raise ValueError("Child wallet balance changed during reclaim") + if reclaim_amount > Decimal("0"): - child.balance = Decimal("0") - parent.balance += reclaim_amount - parent.lifetime_credits += reclaim_amount + # Credit the parent relatively, so a concurrent credit to + # the same parent is not lost. + await session.execute( + sa_update(WalletModel) + .where( + cast( + ColumnElement[bool], + WalletModel.wallet_id == parent_wallet_id, + ) + ) + .values( + balance=WalletModel.balance + reclaim_amount, + lifetime_credits=WalletModel.lifetime_credits + + reclaim_amount, + updated_at=utc_now(), + ) + .execution_options(synchronize_session=False) + ) + # The parent's ledger entry records the balance this credit + # produced, so it must be read after the write. + await session.refresh(parent) # Ledger entries session.add( @@ -527,7 +586,9 @@ async def reclaim_child_wallet(self, child_wallet_id: str) -> dict: ) ) - child.status = WalletStatus.CLOSED.value + # The guarded claim above already closed the child and zeroed + # its balance; re-reading keeps the returned view honest. + await session.refresh(child) await session.commit() diff --git a/tests/test_wallet_ledger_integrity.py b/tests/test_wallet_ledger_integrity.py new file mode 100644 index 00000000..eec6ac7c --- /dev/null +++ b/tests/test_wallet_ledger_integrity.py @@ -0,0 +1,228 @@ +"""Wallet money moves must be decided by the database, not by a stale read. + +Each path here read a balance or counter in one statement and wrote it back in +a later one, serialized only by ``SELECT ... FOR UPDATE``. That lock is a +silent no-op on SQLite, and nothing in this repository forbids SQLite in +production, so these are live on a supported configuration. + +The interleaves are forced deterministically through an instrumented session +rather than raced, so they cannot pass by scheduling luck. Each test asserts +the interleave actually fired before asserting anything else. +""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.db.database import get_session_factory +from sqlalchemy import select + +from app.db.models import LedgerEntryModel, WalletModel +from app.main import app +from app.schemas.billing import ServiceCategory +from app.services.agent_money import get_agent_money +from tests.test_trust_helpers import provision_agent_wallet + + +@pytest.fixture +async def client(): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield c + + +async def _balance(wallet_id: str) -> Decimal: + factory = get_session_factory() + async with factory() as session: + wallet = await session.get(WalletModel, wallet_id) + assert wallet is not None + return wallet.balance + + +async def _debits(wallet_id: str) -> Decimal: + """Total of the debits actually written to this wallet's ledger.""" + factory = get_session_factory() + async with factory() as session: + rows = ( + await session.execute( + select(LedgerEntryModel.amount).where( + LedgerEntryModel.wallet_id == wallet_id, # type: ignore[arg-type] + LedgerEntryModel.action == "debit", # type: ignore[arg-type] + ) + ) + ).scalars().all() + return sum((abs(amount) for amount in rows), Decimal("0")) + + +async def _set_balance(wallet_id: str, amount: Decimal) -> None: + factory = get_session_factory() + async with factory() as session: + async with session.begin(): + wallet = await session.get(WalletModel, wallet_id) + assert wallet is not None + wallet.balance = amount + session.add(wallet) + + +def _interleaving_factory(real_factory, hook, state, *, fire_on: int = 1): + """Wrap a session factory so `hook` runs once, mid-transaction. + + Fires after ``fire_on`` statements have been issued on the wrapped session, + which places it after the balance has been read and before it is written. + """ + + class _Session: + def __init__(self, inner): + self._inner = inner + + def __getattr__(self, name): + return getattr(self._inner, name) + + async def execute(self, *args, **kwargs): + state["executes"] = state.get("executes", 0) + 1 + n = state["executes"] + # Fire *after* the statement returns, so the hook lands once the + # value has been read but before anything is written back. Firing + # beforehand would place the concurrent writer ahead of the read, + # which is not the race. Counting completed reads also keeps the + # seam identical across the fixed and unfixed code, which differ in + # how many statements they issue. + result = await self._inner.execute(*args, **kwargs) + if n == fire_on and not state.get("fired"): + state["fired"] = True + await hook() + return result + + class _CM: + def __init__(self, cm): + self._cm = cm + + async def __aenter__(self): + return _Session(await self._cm.__aenter__()) + + async def __aexit__(self, *exc): + return await self._cm.__aexit__(*exc) + + return lambda: (lambda: _CM(real_factory())) + + +@pytest.mark.anyio +async def test_a_concurrent_charge_cannot_vanish_from_the_balance( + client, clean_database, monkeypatch +): + """The balance must equal the opening balance less every debit recorded. + + ``charge`` read ``wallet.balance``, and several statements later wrote back + ``balance - charge_amount`` computed from that read. The row lock between + them does nothing on SQLite, so a charge committing in between is + overwritten: the ledger records the debit, the balance never reflects it, + and that request was served for free. Conservation β€” balance equals opening + balance minus the debits on the books β€” is the invariant that catches it, + and it is the one the ledger's own credibility rests on. + """ + provisioned = await provision_agent_wallet(client) + wallet_id = provisioned["agent_wallet_id"] + # Exactly enough for one charge, not two. + await _set_balance(wallet_id, Decimal("10")) + + money = get_agent_money() + import app.services.billing_engine as billing_module + + real_factory = get_session_factory() + state: dict = {} + + async def _concurrent_charge() -> None: + # A second charge commits in full while the first is mid-transaction. + monkeypatch.setattr( + billing_module, "get_session_factory", lambda: real_factory, raising=False + ) + state["second"] = await money.charge( + wallet_id=wallet_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/concurrent-b", + ) + + monkeypatch.setattr( + money._billing_engine, + "_session_factory", + _interleaving_factory(real_factory, _concurrent_charge, state, fire_on=1), + ) + + state["first"] = await money.charge( + wallet_id=wallet_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/concurrent-a", + ) + + assert state.get("fired"), "the interleave never ran β€” the test proved nothing" + # Whatever mix of success and refusal the two charges produced, the balance + # must account for exactly the debits the ledger actually recorded. + assert await _balance(wallet_id) == Decimal("10") - await _debits(wallet_id) + # And it may never go negative. + assert await _balance(wallet_id) >= Decimal("0") + + +@pytest.mark.anyio +async def test_reclaiming_a_child_twice_cannot_mint_credits( + client, clean_database, monkeypatch +): + """A child's balance may be reclaimed to its parent exactly once. + + ``reclaim_child_wallet`` read ``child.balance``, zeroed the child, and + credited the parent by the amount it had read. Both halves were protected + only by row locks. Two concurrent reclaims of the same child therefore each + credited the parent the full balance β€” the parent gained twice what the + child ever held, and the extra credits were created from nothing. + """ + money = get_agent_money() + # Only agent (or child) wallets may spawn children. + provisioned = await provision_agent_wallet(client) + parent_id = provisioned["agent_wallet_id"] + await _set_balance(parent_id, Decimal("100")) + child = await money.create_child_wallet( + parent_wallet_id=parent_id, + child_agent_id="reclaim-integrity-child", + budget_credits=Decimal("40"), + max_spend=Decimal("40"), + task_description="reclaim-integrity", + ) + child_id = child.wallet_id + + parent_before = await _balance(parent_id) + child_before = await _balance(child_id) + assert child_before == Decimal("40") + + import app.services.wallet_engine as wallet_module # noqa: F401 + + real_factory = get_session_factory() + state: dict = {} + + async def _concurrent_reclaim() -> None: + monkeypatch.setattr( + money._wallet_engine, "_session_factory", lambda: real_factory, raising=False + ) + try: + state["second"] = await money.reclaim_child_wallet(child_id) + except Exception as exc: # a refusal is a correct outcome here + state["second"] = exc + + monkeypatch.setattr( + money._wallet_engine, + "_session_factory", + _interleaving_factory(real_factory, _concurrent_reclaim, state, fire_on=1), + ) + + try: + state["first"] = await money.reclaim_child_wallet(child_id) + except Exception as exc: + state["first"] = exc + + assert state.get("fired"), "the interleave never ran β€” the test proved nothing" + # The parent may gain exactly what the child held, never twice. + assert await _balance(parent_id) == parent_before + child_before + assert await _balance(child_id) == Decimal("0") From 2e1c11b0de570dbc6c87c65a7514379944024ddf Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 21:28:20 +0000 Subject: [PATCH 3/7] fix(idempotency,billing): surface the orphaned local debit; cap the child wallet in SQL Closes the last of the fourteen audit findings, plus one raised by review. reconcile_stuck_records had a hole that is engine-independent -- it bites on PostgreSQL too, unlike the read-modify-write defects fixed in the previous commits. money.charge() commits in its own transaction and mark_charged() runs in a separate one, so a crash between them leaves the wallet debited while the idempotency record still reads ledger_entry_id IS NULL. That record matched neither pass: the effect-free cleanup requires operation_kind == "upstream_mcp", and the charged-but-stuck repair requires a non-NULL ledger_entry_id. So it sat forever, and because nothing counted it, needs_review stayed 0 -- money had moved and no operator was ever told. The debit is recoverable because governed charges carry operation_key = record_id. A third pass now finds these records, adopts the checkpoint the crash lost, and counts them for review so the next pass can resolve them into a real replay. Records with no committed debit are left alone: the effect-free deletion above is deliberately scoped to the one identity whose ordering makes deletion provably safe, and widening it here would be guessing. Review on #305 then caught that the child-wallet lifetime cap had the same shape as the balance: lifetime_debits + charge_amount <= max_spend was checked against an earlier read while the increment landed later, so two concurrent charges both cleared a cap only one of them fits inside. The balance guard did not carry the cap -- they are separate predicates. Both now travel with the debit, and a lost race is classified so a cap breach is still reported as a cap breach rather than as an empty balance. Documentation corrected where it described something the code does not do: - invention-inventory.md listed the debit ledger entry as the fourth member of prepare()'s atomic unit. It is not; prepare() writes no ledger entry, and its own docstring names the idempotency record, permit row, reservation, and attempt identity. The correction matters because the gap between those commits is the very window the reconciler fix closes. - complete_awi_http_governed's docstring said "Charge wallet, reserve permit budget"; the code reserves first. Anyone reasoning about crash compensation from that reading had the required rollback backwards. - PROOF_MATRIX.md said the coverage gate covers 20 test files across 18 modules (actually 24 and 22) and that CI runs "the same six tests" in postgres_permit_concurrency (actually 13 crash-consistency tests then 11 row-lock tests). Both new tests were confirmed to fail against the unfixed source by reverting the services, not the tests: - orphaned local debit: needs_review == 0, exactly the silence described - child cap: 3.00 spent against a cap of 2 Validation: 1781 passed / 32 skipped (app + b2a_sdk); ruff and mypy 2.3.1 clean over 173 files; doc references 281/243; OpenAPI parity matches; PostgreSQL 16 -- permit concurrency 11 passed, two-process crash recovery 13 passed; trust release gate passed. --- CHANGELOG.md | 19 +++++ app/services/awi_http_governance.py | 10 ++- app/services/billing_engine.py | 51 ++++++++++-- app/services/idempotency.py | 94 ++++++++++++++++++++++ docs/PROOF_MATRIX.md | 7 +- docs/invention-inventory.md | 20 +++-- tests/test_mcp_trust.py | 108 ++++++++++++++++++++++++++ tests/test_wallet_ledger_integrity.py | 59 ++++++++++++++ 8 files changed, 355 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f00624b1..4bcc7c43 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -317,6 +317,25 @@ full release gate; do not backfill a final `v1.2.0` tag. ### πŸ”’ Security +- **An orphaned local debit is no longer invisible.** `money.charge()` commits + in its own transaction and `mark_charged()` runs in a separate one, so a + crash between them leaves the wallet debited while the idempotency record + still reads `ledger_entry_id IS NULL`. Such a record matched **neither** + reconciler pass β€” the effect-free cleanup requires + `operation_kind == "upstream_mcp"`, the charged-but-stuck repair requires a + non-NULL `ledger_entry_id` β€” so it sat forever and `needs_review` stayed 0. + Money had moved and no operator was ever told. **Unlike the read-modify-write + fixes below, this one does not depend on a row lock being a no-op: it is + exposed on PostgreSQL too.** Governed charges carry + `operation_key = record_id`, so the debit is still findable; the reconciler + now adopts it and raises it for review. +- **The child-wallet lifetime spend cap is enforced by the database.** + `lifetime_debits + charge_amount <= max_spend` was checked against an earlier + read while the increment happened later, so concurrent charges could push a + delegated wallet past the ceiling its parent granted. The cap now travels + with the debit as its own predicate. (Raised by review on #305 β€” the balance + guard and the cap are separate predicates, and fixing one did not carry the + other.) - **Every wallet money move is now decided by the database too.** The same read-modify-write shape as the permit fixes below, in the paths that hold actual credits. All were serialized only by `SELECT ... FOR UPDATE`, a diff --git a/app/services/awi_http_governance.py b/app/services/awi_http_governance.py index c18d6ef4..cb8f75ee 100644 --- a/app/services/awi_http_governance.py +++ b/app/services/awi_http_governance.py @@ -237,7 +237,15 @@ async def complete_awi_http_governed( response_payload: dict[str, Any], money: AgentMoney | None = None, ) -> dict[str, Any]: - """Charge wallet, reserve permit budget, write receipt, complete idempotency.""" + """Reserve permit budget, charge wallet, write receipt, complete idempotency. + + The order is load-bearing and is stated here in the order the code runs. + The reservation is taken **first**, so every later failure path compensates + by releasing it β€” which is exactly what the error handlers below do. + An earlier revision of this line named the charge first; anyone reasoning + about crash compensation from that reading would have had the direction of + the required rollback backwards. + """ if ctx.replay_response is not None: return ctx.replay_response diff --git a/app/services/billing_engine.py b/app/services/billing_engine.py index fed5bad2..ad10dcf4 100644 --- a/app/services/billing_engine.py +++ b/app/services/billing_engine.py @@ -13,7 +13,7 @@ from decimal import Decimal from typing import Any, cast -from sqlalchemy import select, update as sa_update +from sqlalchemy import or_, select, update as sa_update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql.elements import ColumnElement @@ -368,6 +368,28 @@ def reverse_velocity_record() -> None: .where( cast(ColumnElement[bool], WalletModel.wallet_id == wallet_id), cast(ColumnElement[bool], WalletModel.balance >= charge_amount), + # The delegated child-wallet lifetime cap has to travel with the + # debit for the same reason the balance does. Checking + # ``lifetime_debits + charge_amount <= max_spend`` above and + # incrementing here would let two concurrent charges both clear + # a cap only one of them fits inside, pushing a delegated wallet + # past the spend ceiling its parent granted it. Non-child + # wallets, and children with no cap set, are unconstrained. + cast( + ColumnElement[bool], + or_( + cast( + ColumnElement[bool], + WalletModel.wallet_type != WalletType.CHILD.value, + ), + cast(Any, WalletModel.max_spend).is_(None), + cast( + ColumnElement[bool], + WalletModel.lifetime_debits + charge_amount + <= cast(Any, WalletModel.max_spend), + ), + ), + ), ) .values( balance=WalletModel.balance - charge_amount, @@ -377,12 +399,31 @@ def reverse_velocity_record() -> None: .execution_options(synchronize_session=False) ) if (cast(Any, debited).rowcount or 0) != 1: - # Lost the race: another charge took the funds after our check. - # Nothing was debited, so this is the same outcome as failing the - # check above and is reported identically. + # Lost the race: a concurrent charge took the funds, or consumed the + # child's remaining cap, after our checks passed. Nothing was + # debited, so each case is reported exactly as failing the + # corresponding check above would have reported it. await session.refresh(wallet) - shortfall = charge_amount - wallet.balance reverse_velocity_record() + max_spend = wallet.max_spend + if ( + wallet.wallet_type == WalletType.CHILD.value + and max_spend is not None + and wallet.lifetime_debits + charge_amount > max_spend + ): + remaining = max_spend - wallet.lifetime_debits + return InsufficientFundsResponse( + wallet_id=wallet_id, + current_balance=float(wallet.balance), + current_balance_exact=str(wallet.balance), + required_amount=float(charge_amount), + required_amount_exact=str(charge_amount), + shortfall=float(charge_amount - remaining), + shortfall_exact=str(charge_amount - remaining), + top_up_url="/v1/billing/top-up/prepare", + message="Child wallet lifetime spend cap exceeded.", + ) + shortfall = charge_amount - wallet.balance return InsufficientFundsResponse( wallet_id=wallet_id, current_balance=float(wallet.balance), diff --git a/app/services/idempotency.py b/app/services/idempotency.py index 369c33ab..90e84d23 100644 --- a/app/services/idempotency.py +++ b/app/services/idempotency.py @@ -18,6 +18,7 @@ from app.db.database import get_session_factory from app.db.models import ( IdempotencyRecordModel, + LedgerEntryModel, McpDispatchAttemptModel, ReceiptModel, ) @@ -612,6 +613,99 @@ async def reconcile_stuck_records( record.status_code = status_code session.add(record) repaired += 1 + + # Third pass: a governed call that is NOT upstream MCP, whose + # debit committed but whose checkpoint never landed. + # + # ``money.charge()`` commits in its own transaction and + # ``mark_charged()`` runs in a separate one, so a crash between + # them leaves the wallet debited while the record still reads + # ``ledger_entry_id IS NULL``. Such a record matched *neither* + # pass above: the first requires ``operation_kind == + # "upstream_mcp"``, the second requires a non-NULL + # ``ledger_entry_id``. It therefore sat forever and, because + # nothing counted it, ``needs_review`` stayed 0 -- an orphaned + # debit that no operator was ever told about. Unlike the + # read-modify-write defects elsewhere in this codebase, this one + # does not depend on a row lock being a no-op: it is exposed on + # PostgreSQL too. + # + # The debit is recoverable because governed charges carry + # ``operation_key = record_id``. Adopting it restores the + # checkpoint the crash lost, which lets the pass above resolve + # the record into a real replay on the next run. + orphaned = ( + ( + await session.execute( + select(IdempotencyRecordModel) + .where( + cast( + ColumnElement[bool], + cast(Any, IdempotencyRecordModel.response_json).is_( + None + ), + ), + cast( + ColumnElement[bool], + cast( + Any, IdempotencyRecordModel.ledger_entry_id + ).is_(None), + ), + cast( + ColumnElement[bool], + or_( + cast( + ColumnElement[bool], + IdempotencyRecordModel.operation_kind + != "upstream_mcp", + ), + cast( + ColumnElement[bool], + cast( + Any, + IdempotencyRecordModel.operation_kind, + ).is_(None), + ), + ), + ), + cast( + ColumnElement[bool], + IdempotencyRecordModel.created_at < cutoff, + ), + ) + .with_for_update() + ) + ) + .scalars() + .all() + ) + for record in orphaned: + debit = ( + await session.execute( + select(LedgerEntryModel).where( + cast( + ColumnElement[bool], + LedgerEntryModel.wallet_id == record.wallet_id, + ), + cast( + ColumnElement[bool], + LedgerEntryModel.operation_key == record.record_id, + ), + ) + ) + ).scalar_one_or_none() + if debit is None: + # No debit ever committed, so there is nothing orphaned + # here. Leave the record alone: the effect-free cleanup + # above is deliberately scoped to the one identity whose + # ordering makes deletion provably safe. + continue + record.ledger_entry_id = debit.entry_id + session.add(record) + # Surface it. The money has moved and the caller has no + # replayable answer, which is precisely the state an + # operator needs to see rather than a silent zero. + needs_review += 1 return repaired, needs_review diff --git a/docs/PROOF_MATRIX.md b/docs/PROOF_MATRIX.md index a018cd73..8f0d63f3 100644 --- a/docs/PROOF_MATRIX.md +++ b/docs/PROOF_MATRIX.md @@ -159,7 +159,10 @@ or any application table that already holds rows, and it takes an advisory lock so two runs cannot overlap. Without the opt-in environment variables it skips, so it never interferes with `make test`. -CI runs the same six tests in the `postgres_permit_concurrency` job. That job +CI runs these in the `postgres_permit_concurrency` job, as two steps against +one PostgreSQL service: the **13** crash-consistency tests in +`tests/test_mcp_postgres_multiprocess.py`, then the **11** row-lock +concurrency tests in `tests/test_permit_postgres_concurrency.py`. That job installs dependencies with `pip` rather than `uv`, so its step is spelled inline; the Make target is the operator-facing equivalent, not a literal copy of the CI step. @@ -168,7 +171,7 @@ copy of the CI step. | Command | Enforces | |---|---| -| `make trust-coverage-gate` | 20 focused trust test files at an **80% coverage floor** across 18 named trust-plane control modules | +| `make trust-coverage-gate` | 24 focused trust test files at an **80% coverage floor** across 22 named trust-plane control modules | | `make trust-release-gate` | A 13-file trust suite (including the in-process adversarial pass over the five claims, `tests/test_adversarial_five_claims.py`), then the coverage gate, then the demo proof, then discovery-drift tests, then committed-OpenAPI parity, then simulation-inventory parity | CI runs `scripts/trust_release_gate.sh` as a dedicated required check diff --git a/docs/invention-inventory.md b/docs/invention-inventory.md index da14b395..02b31571 100644 --- a/docs/invention-inventory.md +++ b/docs/invention-inventory.md @@ -66,13 +66,23 @@ have been regenerated from `wc -l`; see the [appendix](#appendix-mechanism-to-fi **Evidence confidence: Verified** (after corrections 2, 10, 11) -Four operations commit as one crash-consistent unit: budget reservation under -row lock, permit scope validation, creation of a durable dispatch attempt in -`prepared`, and the debit ledger entry. Its docstring states the invariant -directly β€” *"every durable reservation has a row the reconciler can compensate, -and a failed transaction leaves neither"* +Four operations commit as one crash-consistent unit: the idempotency record, +the permit row and its budget reservation, permit scope validation, and +creation of a durable dispatch attempt in `prepared`. Its docstring states the +invariant directly β€” *"every durable reservation has a row the reconciler can +compensate, and a failed transaction leaves neither"* (`mcp_dispatch_attempts.py:360-366`). +**The debit is not in that transaction.** `money.charge()` commits on its own +and the idempotency record learns the entry id afterwards, through +`mark_charged()`. An earlier revision of this section listed the debit ledger +entry as the fourth member of the atomic unit; that was an over-claim, and the +correction matters because the gap between those two commits is a real crash +window rather than an absent one. What closes it is not atomicity but +recoverability: governed charges carry `operation_key = record_id`, so a debit +whose checkpoint was lost is still findable, and `reconcile_stuck_records` +adopts it and raises it for review instead of leaving it orphaned. + A second boundary, the `dispatched` checkpoint, is written immediately before the network send. Past that point the system classifies failure as `delivery_uncertain`: the charge is retained and the call is **never** diff --git a/tests/test_mcp_trust.py b/tests/test_mcp_trust.py index dc92e652..7b21a9f0 100644 --- a/tests/test_mcp_trust.py +++ b/tests/test_mcp_trust.py @@ -1327,3 +1327,111 @@ def broke_guard_tool() -> dict: assert wallet.lifetime_debits == lifetime_debits_before finally: registry.unregister_local("broke-guard-tool") + + +@pytest.mark.anyio +async def test_reconcile_surfaces_a_local_debit_orphaned_before_its_checkpoint( + client, clean_database +): + """A committed local debit whose checkpoint never landed must be surfaced. + + ``money.charge()`` commits in its own transaction and ``mark_charged()`` + runs in a separate one, so a crash between them leaves the wallet debited + while the idempotency record still reads ``ledger_entry_id IS NULL``. + + Such a record used to match *neither* reconciler pass: the effect-free + cleanup requires ``operation_kind == "upstream_mcp"`` and the charged-but- + stuck repair requires a non-NULL ``ledger_entry_id``. The record sat + forever and, because nothing counted it, ``needs_review`` stayed 0 β€” money + had moved and no operator was ever told. Unlike the read-modify-write + defects elsewhere in this codebase this one does not depend on a row lock + being a no-op; it is exposed on PostgreSQL too. + """ + provisioned = await provision_agent_wallet(client) + registry = get_service_registry() + + def orphan_debit_tool() -> dict: + return {"ok": True} + + registry.register_local( + service_id="orphan-debit-tool", + name="Orphan Debit Tool", + description="Governed trust reconciliation test tool", + category=ServiceCategory.AGENT_COMMS, + func=orphan_debit_tool, + credits_per_unit=2.0, + unit_name="call", + ) + + async def crash_before_checkpoint(self, *args, **kwargs): + raise RuntimeError("process died between charge and mark_charged") + + try: + await _orphan_debit_body(client, provisioned, idem=get_idempotency_service(), + crash=crash_before_checkpoint) + finally: + registry.unregister_local("orphan-debit-tool") + + +async def _orphan_debit_body(client, provisioned, *, idem, crash): + permit = await create_tool_permit( + client, + wallet_id=provisioned["agent_wallet_id"], + key_id=provisioned["key_id"], + tool_name="orphan-debit-tool", + idem_key="orphan-debit-permit-create-1", + ) + idempotency_key = "orphan-debit-invoke-1" + body = { + "jsonrpc": "2.0", + "id": "orphan-debit-call", + "method": "tools/call", + "params": { + "name": "orphan-debit-tool", + "arguments": {}, + "mcpContext": { + "wallet_id": provisioned["agent_wallet_id"], + "permit_id": permit["permit_id"], + "idempotency_key": idempotency_key, + }, + }, + } + with patch.object(type(idem), "mark_charged", crash): + await client.post( + "/mcp/messages", json=body, headers=provisioned["agent_headers"] + ) + + factory = get_session_factory() + async with factory() as session: + record = ( + await session.execute( + select(IdempotencyRecordModel).where( + IdempotencyRecordModel.wallet_id == provisioned["agent_wallet_id"], + IdempotencyRecordModel.idempotency_key == idempotency_key, + ) + ) + ).scalar_one_or_none() + + if record is None or record.ledger_entry_id is not None: + pytest.skip( + "this build does not leave the charge/checkpoint window observable" + ) + + await _age_idempotency_record(provisioned["agent_wallet_id"], idempotency_key) + + _repaired, needs_review = await idem.reconcile_stuck_records(idle_seconds=900) + + # The orphaned debit is surfaced rather than silently ignored. + assert needs_review >= 1 + + # And the checkpoint the crash lost has been adopted from the debit, which + # is findable because governed charges carry operation_key == record_id. + async with factory() as session: + healed = ( + await session.execute( + select(IdempotencyRecordModel).where( + IdempotencyRecordModel.record_id == record.record_id + ) + ) + ).scalar_one() + assert healed.ledger_entry_id is not None diff --git a/tests/test_wallet_ledger_integrity.py b/tests/test_wallet_ledger_integrity.py index eec6ac7c..aa0f33b4 100644 --- a/tests/test_wallet_ledger_integrity.py +++ b/tests/test_wallet_ledger_integrity.py @@ -226,3 +226,62 @@ async def _concurrent_reclaim() -> None: # The parent may gain exactly what the child held, never twice. assert await _balance(parent_id) == parent_before + child_before assert await _balance(child_id) == Decimal("0") + + +@pytest.mark.anyio +async def test_concurrent_charges_cannot_exceed_a_child_wallet_spend_cap( + client, clean_database, monkeypatch +): + """A delegated child may not be pushed past the cap its parent granted. + + ``lifetime_debits + charge_amount <= max_spend`` was checked against a + value read earlier and the increment happened several statements later, so + two concurrent charges both cleared a cap only one of them fits inside. + The child then spends more than the authority it was delegated β€” which is + the whole point of ``max_spend``, not an accounting detail. + + Raised by review on #305 after the balance guard landed: the balance and + the cap are two separate predicates, and fixing one does not carry the + other. + """ + money = get_agent_money() + provisioned = await provision_agent_wallet(client) + parent_id = provisioned["agent_wallet_id"] + await _set_balance(parent_id, Decimal("1000")) + child = await money.create_child_wallet( + parent_wallet_id=parent_id, + child_agent_id="cap-race-child", + budget_credits=Decimal("100"), + # Plenty of balance, but only enough cap for a single charge. + max_spend=Decimal("2"), + task_description="cap-race", + ) + child_id = child.wallet_id + + real_factory = get_session_factory() + state: dict = {} + + async def _concurrent_charge() -> None: + state["second"] = await money.charge( + wallet_id=child_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/cap-race-b", + ) + + monkeypatch.setattr( + money._billing_engine, + "_session_factory", + _interleaving_factory(real_factory, _concurrent_charge, state, fire_on=1), + ) + + state["first"] = await money.charge( + wallet_id=child_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/cap-race-a", + ) + + assert state.get("fired"), "the interleave never ran β€” the test proved nothing" + # Whatever mix of success and refusal, the delegated ceiling holds. + assert await _debits(child_id) <= Decimal("2") From 0b31bd94fa3b7db8e33ebbed35f186cf94bcd112 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 22:22:58 +0000 Subject: [PATCH 4/7] fix(review): restore the release state check; make the velocity reversal atomic Addresses CodeRabbit's review on #305. Each finding was verified against the code before acting; two were regressions of mine and are the important ones. 1. release_dispatch_budget_once lost its terminal-state check. This is a regression I introduced in the 3fca6b3 merge: resolving the conflict in favour of main's implementation dropped the "state != returned_error" pre-check that main's version actually kept, because the guard sat inside the replaced hunk. Without it, budget could be released for a prepared or dispatched attempt that then goes on to spend those credits, leaving the cap enforced against a reservation that no longer exists. Restored, with the budget_released_at fast path. Two negative-path tests added, per AGENTS.md; the non-terminal one fails DID NOT RAISE without the check. 2. reverse_velocity_record() was a read-modify-write, in the same function whose debit this branch had already converted to a guarded UPDATE. It is now one relative statement using clamped_decrement, and all six call sites await it. These counters are what the spend cap and anomaly auto-freeze are measured against, so a lost reversal over-counts spend and throttles a caller who never spent the money. 3. reserve_budget's PermitError escaped as a 500 on the AWI path. The reservation happens before the try block and PermitError is caught nowhere on that route, so an expired or revoked permit answered 500 -- telling an operator the service is broken when the permit did its job. Now mapped to 403 with the permit's own reason, closing the idempotency key first. 4. Two of my own regression tests could pass without exercising the thing they test. The crash-recovery test skipped when the record was absent or already charged, so a broken injection passed silently; it now asserts the premise. The conservation test was satisfiable by two refusals; it now asserts a debit was actually recorded first. This is the same failure mode the "interleave fired" guard was written to catch, one level up. 5. Smaller, all verified: the AWI docstring claimed every failure path releases the reservation, but ChargeCreditMismatchError deliberately retains it ("Wallet was debited: keep permit budget reserved") -- an inaccuracy I introduced while fixing a different one; the orphan-debit adoption now requires action == debit; a test dereferenced get_permit() without a None check. Not taken, with reasons stated on the PR: binding expiry to a server-side CURRENT_TIMESTAMP (dialect-dependent semantics against a naive-UTC column, and it would leave the sibling reservation path inconsistent), batching the orphan pass (bounded by the idle cutoff; an optimization, not a correctness issue), and a skipped-repair counter (observability, better as its own change). The new velocity test passes against the unfixed source as well and says so in its docstring: the only reversal branch this harness can reach refreshes the wallet first, which makes the stale read benign there. It is a forward-looking invariant guard, not a demonstration. Validation: 1788 passed / 32 skipped (app + b2a_sdk); ruff and mypy 2.3.1 clean over 173 files; doc references 279/243. --- app/services/awi_http_governance.py | 36 +++++++++-- app/services/billing_engine.py | 37 ++++++++--- app/services/idempotency.py | 10 +++ app/services/permits.py | 10 +++ tests/test_mcp_dispatch_reconciliation.py | 48 +++++++++++++- tests/test_mcp_trust.py | 12 ++-- tests/test_wallet_ledger_integrity.py | 77 ++++++++++++++++++++++- 7 files changed, 210 insertions(+), 20 deletions(-) diff --git a/app/services/awi_http_governance.py b/app/services/awi_http_governance.py index cb8f75ee..a88584db 100644 --- a/app/services/awi_http_governance.py +++ b/app/services/awi_http_governance.py @@ -29,7 +29,7 @@ IdempotencyReplay, get_idempotency_service, ) -from app.services.permits import get_permit_service +from app.services.permits import PermitError, get_permit_service from app.services.receipts import get_receipt_service logger = logging.getLogger(__name__) @@ -240,8 +240,11 @@ async def complete_awi_http_governed( """Reserve permit budget, charge wallet, write receipt, complete idempotency. The order is load-bearing and is stated here in the order the code runs. - The reservation is taken **first**, so every later failure path compensates - by releasing it β€” which is exactly what the error handlers below do. + The reservation is taken **first**, so a later failure compensates by + releasing it β€” with one deliberate exception: on ``ChargeCreditMismatchError`` + the wallet has already been debited, so the reservation is *retained* rather + than released. Releasing it there would free budget the caller has in fact + spent. An earlier revision of this line named the charge first; anyone reasoning about crash compensation from that reading would have had the direction of the required rollback backwards. @@ -254,7 +257,32 @@ async def complete_awi_http_governed( money = money or get_agent_money() permits = get_permit_service() idem = get_idempotency_service() - await permits.reserve_budget(ctx.permit_id, ctx.credits) + try: + await permits.reserve_budget(ctx.permit_id, ctx.credits) + except PermitError as exc: + # A permit that expired or was revoked between validation and this + # reservation is a *denial*, not a server fault. Nothing was reserved + # and nothing was charged, so the caller gets the same 403 shape the + # up-front validation would have produced. Without this the PermitError + # escaped uncaught and the route answered 500, which tells an operator + # the service is broken when in fact the permit did its job. + await abort_awi_http_governed( + ctx, + status_code=status.HTTP_403_FORBIDDEN, + error_payload={ + "error": exc.reason, + "message": exc.reason, + "tool": ctx.tool_name, + }, + ) + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail={ + "error": exc.reason, + "message": exc.reason, + "tool": ctx.tool_name, + }, + ) from exc unit_price = DEFAULT_PRICING[ServiceCategory.AGENT_COMMS][1] charge_units = ctx.credits / unit_price if unit_price else Decimal("1") diff --git a/app/services/billing_engine.py b/app/services/billing_engine.py index ad10dcf4..5a259c95 100644 --- a/app/services/billing_engine.py +++ b/app/services/billing_engine.py @@ -262,15 +262,34 @@ async def _apply_charge_to_locked_wallet( # wallet is absent, so there is nothing to reverse here. raise self._wallet_not_found_error(wallet_id) - def reverse_velocity_record() -> None: + async def reverse_velocity_record() -> None: # The velocity monitor committed these counters before this # transaction acquired the wallet lock. A rejected debit must # reverse exactly that increment. - wallet.hourly_spent = max(Decimal("0"), wallet.hourly_spent - charge_amount) - wallet.daily_spent = max(Decimal("0"), wallet.daily_spent - charge_amount) + # + # Relative, in one statement, for the same reason the increment in + # velocity_monitor is: reading a counter and writing back + # read-charge is a read-modify-write, and these counters are what + # the spend cap and the anomaly auto-freeze are measured against. + # A lost reversal here leaves the wallet permanently over-counted, + # which throttles a caller that never spent the money. + await session.execute( + sa_update(WalletModel) + .where(cast(ColumnElement[bool], WalletModel.wallet_id == wallet_id)) + .values( + hourly_spent=clamped_decrement( + WalletModel.hourly_spent, charge_amount + ), + daily_spent=clamped_decrement( + WalletModel.daily_spent, charge_amount + ), + ) + .execution_options(synchronize_session=False) + ) + await session.refresh(wallet) if wallet.status in _NON_SPENDABLE_WALLET_STATUSES: - reverse_velocity_record() + await reverse_velocity_record() return InsufficientFundsResponse( error=( "wallet_frozen" @@ -291,7 +310,7 @@ def reverse_velocity_record() -> None: try: await self._wallet_engine.ensure_wallet_not_expired(session, wallet) except WalletExpiredError: - reverse_velocity_record() + await reverse_velocity_record() return InsufficientFundsResponse( error="wallet_expired", wallet_id=wallet_id, @@ -310,7 +329,7 @@ def reverse_velocity_record() -> None: if new_debits > wallet.max_spend: remaining = wallet.max_spend - wallet.lifetime_debits shortfall = charge_amount - remaining - reverse_velocity_record() + await reverse_velocity_record() return InsufficientFundsResponse( wallet_id=wallet_id, current_balance=float(wallet.balance), @@ -327,7 +346,7 @@ def reverse_velocity_record() -> None: prior_daily_spent = wallet.daily_spent - charge_amount remaining = wallet.daily_limit - prior_daily_spent shortfall = charge_amount - remaining - reverse_velocity_record() + await reverse_velocity_record() return InsufficientFundsResponse( wallet_id=wallet_id, current_balance=float(wallet.balance), @@ -342,7 +361,7 @@ def reverse_velocity_record() -> None: if wallet.balance < charge_amount: shortfall = charge_amount - wallet.balance - reverse_velocity_record() + await reverse_velocity_record() return InsufficientFundsResponse( wallet_id=wallet_id, current_balance=float(wallet.balance), @@ -404,7 +423,7 @@ def reverse_velocity_record() -> None: # debited, so each case is reported exactly as failing the # corresponding check above would have reported it. await session.refresh(wallet) - reverse_velocity_record() + await reverse_velocity_record() max_spend = wallet.max_spend if ( wallet.wallet_type == WalletType.CHILD.value diff --git a/app/services/idempotency.py b/app/services/idempotency.py index 90e84d23..5d268009 100644 --- a/app/services/idempotency.py +++ b/app/services/idempotency.py @@ -22,6 +22,7 @@ McpDispatchAttemptModel, ReceiptModel, ) +from app.schemas.billing import LedgerAction from app.services.signing_keys import sha256_hex GOVERNED_MCP_IDEMPOTENCY_ENDPOINT = "/mcp/invoke" @@ -691,6 +692,15 @@ async def reconcile_stuck_records( ColumnElement[bool], LedgerEntryModel.operation_key == record.record_id, ), + # Adopt only an actual debit. The operation key + # is unique per wallet, but naming the action + # keeps a future credit sharing that key from + # being mistaken for the charge we are + # recovering. + cast( + ColumnElement[bool], + LedgerEntryModel.action == LedgerAction.DEBIT.value, + ), ) ) ).scalar_one_or_none() diff --git a/app/services/permits.py b/app/services/permits.py index 89e106e3..6047b2a3 100644 --- a/app/services/permits.py +++ b/app/services/permits.py @@ -896,6 +896,16 @@ async def _once() -> bool: attempt = await session.get(McpDispatchAttemptModel, attempt_id) if attempt is None: raise PermitError("dispatch_attempt_not_found") + # Only a terminal returned_error attempt has a reservation + # to give back. Releasing budget for a prepared or + # dispatched attempt frees credits that attempt may still + # go on to spend, so the cap would be enforced against a + # reservation that no longer exists. This pre-check is the + # contract; the guarded claim below is the once-only gate. + if attempt.state != "returned_error": + raise PermitError("dispatch_budget_release_state_invalid") + if attempt.budget_released_at is not None: + return False now = utc_now() # Claim the release atomically before touching the permit. # The read above is guarded by a row lock, but SQLAlchemy diff --git a/tests/test_mcp_dispatch_reconciliation.py b/tests/test_mcp_dispatch_reconciliation.py index 744b012e..cb933356 100644 --- a/tests/test_mcp_dispatch_reconciliation.py +++ b/tests/test_mcp_dispatch_reconciliation.py @@ -34,7 +34,7 @@ from app.services.mcp_dispatch_reconciliation import ( get_mcp_dispatch_reconciliation_service, ) -from app.services.permits import get_permit_service +from app.services.permits import PermitError, get_permit_service from app.services.receipts import get_receipt_service from tests.test_trust_helpers import create_tool_permit, provision_agent_wallet @@ -1088,7 +1088,9 @@ async def test_dispatch_budget_release_is_once_only_under_a_stale_read( error_code="upstream_returned_error", ) permits = get_permit_service() - spent_before = (await permits.get_permit(seed.permit_id)).spent_credits + seeded_permit = await permits.get_permit(seed.permit_id) + assert seeded_permit is not None + spent_before = seeded_permit.spent_credits assert spent_before == CREDITS import app.services.permits as permits_module @@ -1141,3 +1143,45 @@ async def __aexit__(self, *exc): permit = await permits.get_permit(seed.permit_id) assert permit is not None assert permit.spent_credits == spent_before - CREDITS + + +@pytest.mark.anyio +async def test_release_dispatch_budget_rejects_a_missing_attempt( + client: AsyncClient, + clean_database, +) -> None: + """An unknown attempt id must not move budget.""" + with pytest.raises(PermitError) as excinfo: + await get_permit_service().release_dispatch_budget_once("att-does-not-exist") + assert excinfo.value.reason == "dispatch_attempt_not_found" + + +@pytest.mark.anyio +async def test_release_dispatch_budget_rejects_a_non_terminal_attempt( + client: AsyncClient, + clean_database, +) -> None: + """Only a terminal ``returned_error`` attempt has a reservation to release. + + A ``dispatched`` attempt may still complete and consume its credits. Giving + its reservation back early would leave the permit enforcing a cap against a + reservation that no longer exists, so the same permit could fund the + in-flight call twice over. + """ + seed = await _seed_attempt( + client, + suffix="release-non-terminal", + state="dispatched", + ) + permits = get_permit_service() + before = await permits.get_permit(seed.permit_id) + assert before is not None + + with pytest.raises(PermitError) as excinfo: + await permits.release_dispatch_budget_once(seed.attempt_id) + assert excinfo.value.reason == "dispatch_budget_release_state_invalid" + + # And no budget moved on the way to the refusal. + after = await permits.get_permit(seed.permit_id) + assert after is not None + assert after.spent_credits == before.spent_credits diff --git a/tests/test_mcp_trust.py b/tests/test_mcp_trust.py index 7b21a9f0..63a42535 100644 --- a/tests/test_mcp_trust.py +++ b/tests/test_mcp_trust.py @@ -1412,10 +1412,14 @@ async def _orphan_debit_body(client, provisioned, *, idem, crash): ) ).scalar_one_or_none() - if record is None or record.ledger_entry_id is not None: - pytest.skip( - "this build does not leave the charge/checkpoint window observable" - ) + # Assert the premise rather than skipping on it. A skip here would let a + # broken injection pass silently: the whole point is that a debit committed + # while the checkpoint did not, and if that state was never produced the + # test proves nothing about the reconciler. + assert record is not None, "no idempotency record β€” the crash injection missed" + assert record.ledger_entry_id is None, ( + "the checkpoint landed, so the orphaned-debit window was never entered" + ) await _age_idempotency_record(provisioned["agent_wallet_id"], idempotency_key) diff --git a/tests/test_wallet_ledger_integrity.py b/tests/test_wallet_ledger_integrity.py index aa0f33b4..8a6b8414 100644 --- a/tests/test_wallet_ledger_integrity.py +++ b/tests/test_wallet_ledger_integrity.py @@ -160,9 +160,13 @@ async def _concurrent_charge() -> None: ) assert state.get("fired"), "the interleave never ran β€” the test proved nothing" + # Money must actually have moved, or conservation is trivially satisfied by + # two refusals and the test proves nothing. + debited = await _debits(wallet_id) + assert debited > Decimal("0"), "no debit was recorded β€” no money moved" # Whatever mix of success and refusal the two charges produced, the balance # must account for exactly the debits the ledger actually recorded. - assert await _balance(wallet_id) == Decimal("10") - await _debits(wallet_id) + assert await _balance(wallet_id) == Decimal("10") - debited # And it may never go negative. assert await _balance(wallet_id) >= Decimal("0") @@ -285,3 +289,74 @@ async def _concurrent_charge() -> None: assert state.get("fired"), "the interleave never ran β€” the test proved nothing" # Whatever mix of success and refusal, the delegated ceiling holds. assert await _debits(child_id) <= Decimal("2") + + +@pytest.mark.anyio +async def test_a_rejected_charge_does_not_erase_another_charges_velocity( + client, clean_database, monkeypatch +): + """Reversing one charge's velocity must not undo another's increment. + + When a charge is rejected after the velocity monitor already committed its + increment, `reverse_velocity_record()` backs that increment out. It did so + by reading `hourly_spent`/`daily_spent` and writing back read-charge β€” a + read-modify-write, in the same function whose debit this PR had already + converted to a guarded UPDATE. A charge committing in between is erased. + + That matters more than a skewed metric: these counters are what the spend + cap and the anomaly auto-freeze are measured against. Losing an increment + under-counts spend and quietly raises the ceiling; losing a *reversal* + over-counts it and throttles a caller who never spent the money. + + Honest limit on what this proves: unlike the other tests in this file, it + passes against the unfixed source as well. The only reversal branch this + interleave can reliably reach is the lost-race one, and that branch calls + ``session.refresh(wallet)`` first, which happens to make the stale read + benign there. The five branches that reverse *without* a refresh are the + exposed ones, and they need a rejection this harness cannot force on the + same wallet the concurrent charge must succeed on. So treat this as a + forward-looking invariant guard, not as a demonstration of the defect. + """ + provisioned = await provision_agent_wallet(client) + wallet_id = provisioned["agent_wallet_id"] + # Enough for the concurrent charge to succeed, nothing for the outer one: + # the outer charge is rejected and must reverse only its own increment. + await _set_balance(wallet_id, Decimal("1.5")) + + money = get_agent_money() + real_factory = get_session_factory() + state: dict = {} + + async def _concurrent_charge() -> None: + state["second"] = await money.charge( + wallet_id=wallet_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/velocity-b", + ) + + monkeypatch.setattr( + money._billing_engine, + "_session_factory", + _interleaving_factory(real_factory, _concurrent_charge, state, fire_on=1), + ) + + state["first"] = await money.charge( + wallet_id=wallet_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/velocity-a", + ) + + assert state.get("fired"), "the interleave never ran β€” the test proved nothing" + debited = await _debits(wallet_id) + assert debited > Decimal("0"), "no debit was recorded β€” no money moved" + + # The surviving charge's velocity increment must still be on the books. + factory = get_session_factory() + async with factory() as session: + wallet = await session.get(WalletModel, wallet_id) + assert wallet is not None + assert wallet.daily_spent >= debited, ( + "a rejected charge's reversal erased a concurrent charge's increment" + ) From 55f03d48626437c406871c7f1616b402b03588bc Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 22:33:17 +0000 Subject: [PATCH 5/7] fix(billing,awi): stop a mid-charge freeze debiting; don't freeze contention into a denial Second round of CodeRabbit review on #305. Both findings verified against the code; one is a bug I introduced in the previous commit. permit_write_contended was being stored as a terminal 403. reserve_budget raises it after retry exhaustion on a transient write conflict, and the handler I added one commit ago caught every PermitError and *completed* the idempotency record with a denial. That turns a momentary database conflict into a permanent stored answer: every retry of that key replays the 403 long after the contention cleared. It is now split -- transient contention abandons the uncharged record (so the key stays usable) and answers 503; only terminal denials complete as 403. Wallet spendability now travels with the guarded debit. The status check read a row loaded at the top of the transaction and the write happened several statements later, so a freeze committing in between was invisible and the debit went through. A freeze is the control that exists to stop spending at that moment, not at the next request. A zero-row result is classified so a freeze is reported as wallet_frozen rather than as an empty balance. Confirmed against the unfixed source by reverting the service, not the test: the frozen wallet is debited, balance_after 98.5, a real ledger entry minted. Not taken, with reasons on the PR: the daily-limit predicate (daily_spent is maintained by the velocity monitor in its own transaction and is already reversed on rejection; folding it into the debit predicate would double-count the interaction between the two and deserves its own change), and the amount < 0 filter on ledger adoption (moot -- that pass is removed here, see below). Also dropped this branch's orphaned-local-debit pass in favour of #306, which implements the same fix better: it runs *before* the stuck pass with an explicit flush, so the record resolves in the same run instead of the next, and it lets the stuck pass classify repaired vs needs_review instead of counting every adoption as review. Its test matrix is also stronger. No reason to carry two implementations of one reconciler pass into a conflict. Validation: 1788 passed / 32 skipped (app + b2a_sdk); ruff and mypy 2.3.1 clean over 173 files; doc references resolve. --- CHANGELOG.md | 12 --- app/services/awi_http_governance.py | 45 +++++++---- app/services/billing_engine.py | 29 +++++++ app/services/idempotency.py | 103 ------------------------ tests/test_mcp_trust.py | 110 -------------------------- tests/test_wallet_ledger_integrity.py | 48 +++++++++++ 6 files changed, 106 insertions(+), 241 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4bcc7c43..1fb787cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -317,18 +317,6 @@ full release gate; do not backfill a final `v1.2.0` tag. ### πŸ”’ Security -- **An orphaned local debit is no longer invisible.** `money.charge()` commits - in its own transaction and `mark_charged()` runs in a separate one, so a - crash between them leaves the wallet debited while the idempotency record - still reads `ledger_entry_id IS NULL`. Such a record matched **neither** - reconciler pass β€” the effect-free cleanup requires - `operation_kind == "upstream_mcp"`, the charged-but-stuck repair requires a - non-NULL `ledger_entry_id` β€” so it sat forever and `needs_review` stayed 0. - Money had moved and no operator was ever told. **Unlike the read-modify-write - fixes below, this one does not depend on a row lock being a no-op: it is - exposed on PostgreSQL too.** Governed charges carry - `operation_key = record_id`, so the debit is still findable; the reconciler - now adopts it and raises it for review. - **The child-wallet lifetime spend cap is enforced by the database.** `lifetime_debits + charge_amount <= max_spend` was checked against an earlier read while the increment happened later, so concurrent charges could push a diff --git a/app/services/awi_http_governance.py b/app/services/awi_http_governance.py index a88584db..9eee6dde 100644 --- a/app/services/awi_http_governance.py +++ b/app/services/awi_http_governance.py @@ -260,28 +260,41 @@ async def complete_awi_http_governed( try: await permits.reserve_budget(ctx.permit_id, ctx.credits) except PermitError as exc: - # A permit that expired or was revoked between validation and this - # reservation is a *denial*, not a server fault. Nothing was reserved - # and nothing was charged, so the caller gets the same 403 shape the - # up-front validation would have produced. Without this the PermitError - # escaped uncaught and the route answered 500, which tells an operator - # the service is broken when in fact the permit did its job. + detail = { + "error": exc.reason, + "message": exc.reason, + "tool": ctx.tool_name, + } + if exc.reason == "permit_write_contended": + # Transient: the guarded write lost to contention and exhausted its + # retries. Nothing was reserved and nothing was charged, so the key + # must stay usable -- *completing* the record here would freeze a + # momentary database conflict into a permanent stored denial that + # every retry of that idempotency key replays, long after the + # contention cleared. Release the key and answer 503 instead. + await idem.abandon( + wallet_id=ctx.wallet_id, + endpoint=ctx.endpoint, + idempotency_key=ctx.idempotency_key, + ) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=detail, + ) from exc + # Terminal: a permit that expired or was revoked between validation and + # this reservation is a *denial*, not a server fault. Nothing was + # reserved and nothing was charged, so the caller gets the same 403 + # shape the up-front validation would have produced. Without this the + # PermitError escaped uncaught and the route answered 500, which tells + # an operator the service is broken when in fact the permit did its job. await abort_awi_http_governed( ctx, status_code=status.HTTP_403_FORBIDDEN, - error_payload={ - "error": exc.reason, - "message": exc.reason, - "tool": ctx.tool_name, - }, + error_payload=detail, ) raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, - detail={ - "error": exc.reason, - "message": exc.reason, - "tool": ctx.tool_name, - }, + detail=detail, ) from exc unit_price = DEFAULT_PRICING[ServiceCategory.AGENT_COMMS][1] diff --git a/app/services/billing_engine.py b/app/services/billing_engine.py index 5a259c95..32dae440 100644 --- a/app/services/billing_engine.py +++ b/app/services/billing_engine.py @@ -409,6 +409,16 @@ async def reverse_velocity_record() -> None: ), ), ), + # Spendability travels with the debit too. The status check + # above reads a value loaded before this statement, so a freeze + # landing in between would otherwise still be debited -- the + # freeze exists precisely to stop that. + cast( + ColumnElement[bool], + cast(Any, WalletModel.status).notin_( + tuple(_NON_SPENDABLE_WALLET_STATUSES) + ), + ), ) .values( balance=WalletModel.balance - charge_amount, @@ -424,6 +434,25 @@ async def reverse_velocity_record() -> None: # corresponding check above would have reported it. await session.refresh(wallet) await reverse_velocity_record() + if wallet.status in _NON_SPENDABLE_WALLET_STATUSES: + # A freeze landed between the read and the write. Report it as + # the freeze it is, not as an empty balance. + return InsufficientFundsResponse( + error=( + "wallet_frozen" + if wallet.status == WalletStatus.FROZEN.value + else "insufficient_funds" + ), + wallet_id=wallet_id, + current_balance=float(wallet.balance), + current_balance_exact=str(wallet.balance), + required_amount=float(charge_amount), + required_amount_exact=str(charge_amount), + shortfall=float(charge_amount), + shortfall_exact=str(charge_amount), + top_up_url="", + message="Wallet is not spendable.", + ) max_spend = wallet.max_spend if ( wallet.wallet_type == WalletType.CHILD.value diff --git a/app/services/idempotency.py b/app/services/idempotency.py index 5d268009..9f776347 100644 --- a/app/services/idempotency.py +++ b/app/services/idempotency.py @@ -18,11 +18,9 @@ from app.db.database import get_session_factory from app.db.models import ( IdempotencyRecordModel, - LedgerEntryModel, McpDispatchAttemptModel, ReceiptModel, ) -from app.schemas.billing import LedgerAction from app.services.signing_keys import sha256_hex GOVERNED_MCP_IDEMPOTENCY_ENDPOINT = "/mcp/invoke" @@ -615,107 +613,6 @@ async def reconcile_stuck_records( session.add(record) repaired += 1 - # Third pass: a governed call that is NOT upstream MCP, whose - # debit committed but whose checkpoint never landed. - # - # ``money.charge()`` commits in its own transaction and - # ``mark_charged()`` runs in a separate one, so a crash between - # them leaves the wallet debited while the record still reads - # ``ledger_entry_id IS NULL``. Such a record matched *neither* - # pass above: the first requires ``operation_kind == - # "upstream_mcp"``, the second requires a non-NULL - # ``ledger_entry_id``. It therefore sat forever and, because - # nothing counted it, ``needs_review`` stayed 0 -- an orphaned - # debit that no operator was ever told about. Unlike the - # read-modify-write defects elsewhere in this codebase, this one - # does not depend on a row lock being a no-op: it is exposed on - # PostgreSQL too. - # - # The debit is recoverable because governed charges carry - # ``operation_key = record_id``. Adopting it restores the - # checkpoint the crash lost, which lets the pass above resolve - # the record into a real replay on the next run. - orphaned = ( - ( - await session.execute( - select(IdempotencyRecordModel) - .where( - cast( - ColumnElement[bool], - cast(Any, IdempotencyRecordModel.response_json).is_( - None - ), - ), - cast( - ColumnElement[bool], - cast( - Any, IdempotencyRecordModel.ledger_entry_id - ).is_(None), - ), - cast( - ColumnElement[bool], - or_( - cast( - ColumnElement[bool], - IdempotencyRecordModel.operation_kind - != "upstream_mcp", - ), - cast( - ColumnElement[bool], - cast( - Any, - IdempotencyRecordModel.operation_kind, - ).is_(None), - ), - ), - ), - cast( - ColumnElement[bool], - IdempotencyRecordModel.created_at < cutoff, - ), - ) - .with_for_update() - ) - ) - .scalars() - .all() - ) - for record in orphaned: - debit = ( - await session.execute( - select(LedgerEntryModel).where( - cast( - ColumnElement[bool], - LedgerEntryModel.wallet_id == record.wallet_id, - ), - cast( - ColumnElement[bool], - LedgerEntryModel.operation_key == record.record_id, - ), - # Adopt only an actual debit. The operation key - # is unique per wallet, but naming the action - # keeps a future credit sharing that key from - # being mistaken for the charge we are - # recovering. - cast( - ColumnElement[bool], - LedgerEntryModel.action == LedgerAction.DEBIT.value, - ), - ) - ) - ).scalar_one_or_none() - if debit is None: - # No debit ever committed, so there is nothing orphaned - # here. Leave the record alone: the effect-free cleanup - # above is deliberately scoped to the one identity whose - # ordering makes deletion provably safe. - continue - record.ledger_entry_id = debit.entry_id - session.add(record) - # Surface it. The money has moved and the caller has no - # replayable answer, which is precisely the state an - # operator needs to see rather than a silent zero. - needs_review += 1 return repaired, needs_review diff --git a/tests/test_mcp_trust.py b/tests/test_mcp_trust.py index 63a42535..e52d9bb9 100644 --- a/tests/test_mcp_trust.py +++ b/tests/test_mcp_trust.py @@ -1329,113 +1329,3 @@ def broke_guard_tool() -> dict: registry.unregister_local("broke-guard-tool") -@pytest.mark.anyio -async def test_reconcile_surfaces_a_local_debit_orphaned_before_its_checkpoint( - client, clean_database -): - """A committed local debit whose checkpoint never landed must be surfaced. - - ``money.charge()`` commits in its own transaction and ``mark_charged()`` - runs in a separate one, so a crash between them leaves the wallet debited - while the idempotency record still reads ``ledger_entry_id IS NULL``. - - Such a record used to match *neither* reconciler pass: the effect-free - cleanup requires ``operation_kind == "upstream_mcp"`` and the charged-but- - stuck repair requires a non-NULL ``ledger_entry_id``. The record sat - forever and, because nothing counted it, ``needs_review`` stayed 0 β€” money - had moved and no operator was ever told. Unlike the read-modify-write - defects elsewhere in this codebase this one does not depend on a row lock - being a no-op; it is exposed on PostgreSQL too. - """ - provisioned = await provision_agent_wallet(client) - registry = get_service_registry() - - def orphan_debit_tool() -> dict: - return {"ok": True} - - registry.register_local( - service_id="orphan-debit-tool", - name="Orphan Debit Tool", - description="Governed trust reconciliation test tool", - category=ServiceCategory.AGENT_COMMS, - func=orphan_debit_tool, - credits_per_unit=2.0, - unit_name="call", - ) - - async def crash_before_checkpoint(self, *args, **kwargs): - raise RuntimeError("process died between charge and mark_charged") - - try: - await _orphan_debit_body(client, provisioned, idem=get_idempotency_service(), - crash=crash_before_checkpoint) - finally: - registry.unregister_local("orphan-debit-tool") - - -async def _orphan_debit_body(client, provisioned, *, idem, crash): - permit = await create_tool_permit( - client, - wallet_id=provisioned["agent_wallet_id"], - key_id=provisioned["key_id"], - tool_name="orphan-debit-tool", - idem_key="orphan-debit-permit-create-1", - ) - idempotency_key = "orphan-debit-invoke-1" - body = { - "jsonrpc": "2.0", - "id": "orphan-debit-call", - "method": "tools/call", - "params": { - "name": "orphan-debit-tool", - "arguments": {}, - "mcpContext": { - "wallet_id": provisioned["agent_wallet_id"], - "permit_id": permit["permit_id"], - "idempotency_key": idempotency_key, - }, - }, - } - with patch.object(type(idem), "mark_charged", crash): - await client.post( - "/mcp/messages", json=body, headers=provisioned["agent_headers"] - ) - - factory = get_session_factory() - async with factory() as session: - record = ( - await session.execute( - select(IdempotencyRecordModel).where( - IdempotencyRecordModel.wallet_id == provisioned["agent_wallet_id"], - IdempotencyRecordModel.idempotency_key == idempotency_key, - ) - ) - ).scalar_one_or_none() - - # Assert the premise rather than skipping on it. A skip here would let a - # broken injection pass silently: the whole point is that a debit committed - # while the checkpoint did not, and if that state was never produced the - # test proves nothing about the reconciler. - assert record is not None, "no idempotency record β€” the crash injection missed" - assert record.ledger_entry_id is None, ( - "the checkpoint landed, so the orphaned-debit window was never entered" - ) - - await _age_idempotency_record(provisioned["agent_wallet_id"], idempotency_key) - - _repaired, needs_review = await idem.reconcile_stuck_records(idle_seconds=900) - - # The orphaned debit is surfaced rather than silently ignored. - assert needs_review >= 1 - - # And the checkpoint the crash lost has been adopted from the debit, which - # is findable because governed charges carry operation_key == record_id. - async with factory() as session: - healed = ( - await session.execute( - select(IdempotencyRecordModel).where( - IdempotencyRecordModel.record_id == record.record_id - ) - ) - ).scalar_one() - assert healed.ledger_entry_id is not None diff --git a/tests/test_wallet_ledger_integrity.py b/tests/test_wallet_ledger_integrity.py index 8a6b8414..61d02957 100644 --- a/tests/test_wallet_ledger_integrity.py +++ b/tests/test_wallet_ledger_integrity.py @@ -360,3 +360,51 @@ async def _concurrent_charge() -> None: assert wallet.daily_spent >= debited, ( "a rejected charge's reversal erased a concurrent charge's increment" ) + + +@pytest.mark.anyio +async def test_a_wallet_frozen_mid_charge_is_not_debited( + client, clean_database, monkeypatch +): + """A freeze landing after the read must still stop the debit. + + ``charge`` checks ``wallet.status`` against the row it loaded at the top of + the transaction and writes several statements later. A freeze committing in + between was invisible to that check, so the debit went through β€” and a + freeze is the control that exists precisely to stop spending *now*, not at + the next request. Spendability therefore travels with the debit. + """ + provisioned = await provision_agent_wallet(client) + wallet_id = provisioned["agent_wallet_id"] + await _set_balance(wallet_id, Decimal("100")) + + money = get_agent_money() + real_factory = get_session_factory() + state: dict = {} + + async def _freeze_the_wallet() -> None: + async with real_factory() as s: + async with s.begin(): + wallet = await s.get(WalletModel, wallet_id) + assert wallet is not None + wallet.status = "frozen" + s.add(wallet) + + monkeypatch.setattr( + money._billing_engine, + "_session_factory", + _interleaving_factory(real_factory, _freeze_the_wallet, state, fire_on=1), + ) + + result = await money.charge( + wallet_id=wallet_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/freeze-race", + ) + + assert state.get("fired"), "the interleave never ran β€” the test proved nothing" + # Refused, and nothing debited. + assert not hasattr(result, "entry_id"), f"a frozen wallet was debited: {result}" + assert await _debits(wallet_id) == Decimal("0") + assert await _balance(wallet_id) == Decimal("100") From 5525051df1cb968357739a2bc8eb59d92f2d6eb7 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 22:43:06 +0000 Subject: [PATCH 6/7] fix(billing): a freeze is not a balance deficit, in the race path too `charge_wallet`'s lost-the-race branch re-reads the wallet and reports the refusal as whichever check would have caught it. For the freeze case it reported `shortfall=charge_amount`, while the ordinary frozen path a few hundred lines above reports `shortfall=0`. A freeze blocks spending regardless of balance, so a caller told it is short by the full charge will top the wallet up and still be refused. Align the race branch with the non-race path -- same `shortfall`, `shortfall_exact`, and message -- and assert all three in the frozen-mid-charge race test, which previously only checked that no debit landed. --- app/services/billing_engine.py | 9 ++++++--- tests/test_wallet_ledger_integrity.py | 4 ++++ 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/app/services/billing_engine.py b/app/services/billing_engine.py index 32dae440..474e02ba 100644 --- a/app/services/billing_engine.py +++ b/app/services/billing_engine.py @@ -448,10 +448,13 @@ async def reverse_velocity_record() -> None: current_balance_exact=str(wallet.balance), required_amount=float(charge_amount), required_amount_exact=str(charge_amount), - shortfall=float(charge_amount), - shortfall_exact=str(charge_amount), + # A freeze blocks spending regardless of balance, so this + # is not a balance deficit -- matching the non-race frozen + # path above, which also reports zero. + shortfall=0.0, + shortfall_exact="0", top_up_url="", - message="Wallet is not spendable.", + message=f"Wallet is {wallet.status} and cannot be charged.", ) max_spend = wallet.max_spend if ( diff --git a/tests/test_wallet_ledger_integrity.py b/tests/test_wallet_ledger_integrity.py index 61d02957..54f40661 100644 --- a/tests/test_wallet_ledger_integrity.py +++ b/tests/test_wallet_ledger_integrity.py @@ -406,5 +406,9 @@ async def _freeze_the_wallet() -> None: assert state.get("fired"), "the interleave never ran β€” the test proved nothing" # Refused, and nothing debited. assert not hasattr(result, "entry_id"), f"a frozen wallet was debited: {result}" + # A freeze is not a balance deficit: the refusal must not report a shortfall. + assert getattr(result, "error", None) == "wallet_frozen" + assert result.shortfall == 0.0 + assert result.shortfall_exact == "0" assert await _debits(wallet_id) == Decimal("0") assert await _balance(wallet_id) == Decimal("100") From aa75f5b30cd16096a7372a2572062a481fe7f628 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 23:02:26 +0000 Subject: [PATCH 7/7] fix(permits,stripe): surface skipped budget repairs; timestamp the fiat clawback Two review findings, both verified against the code before acting. `reconcile_budgets` returns a count of *writes*. A permit whose guarded repair matched no row -- a concurrent reservation moved `spent_credits` between the receipt scan and the write -- is correctly left alone, but it was also invisible: a pass that skipped every permit and a pass that found nothing to repair both returned 0, and `app/main.py` logs only a nonzero result. A permit that never stops being skipped, the signature of a hot permit under sustained contention, surfaced nowhere. Count the skips and log them, keeping the return value a count of writes so the eight existing call sites are unaffected. The docstring now states which permits are not counted and that they are re-examined next pass. The Stripe refund clawback was the one balance-moving wallet UPDATE in this branch that did not set `updated_at`. `WalletModel.updated_at` has a default but no `onupdate`, and a Core UPDATE bypasses the ORM attribute write, so the most consequential move the wallet sees -- real fiat returned -- was the one leaving the row looking untouched. Both are covered by tests that fail against the unfixed source. The reconcile test records the service's logger rather than capturing through stdlib logging: tests/test_migrations.py runs Alembic in-process and migrations/env.py calls fileConfig, which disables every logger created before it, so handler- and caplog-based capture depends on test order. Production runs migrations in a separate process (scripts/docker_entrypoint.sh), so the operator log is live there. --- app/services/permits.py | 28 ++++++++++++- app/services/stripe_integration.py | 10 +++++ tests/test_permit_budget_integrity.py | 30 ++++++++++++++ tests/test_stripe_integration.py | 59 +++++++++++++++++++++++++++ 4 files changed, 126 insertions(+), 1 deletion(-) diff --git a/app/services/permits.py b/app/services/permits.py index 6047b2a3..4d1b4db4 100644 --- a/app/services/permits.py +++ b/app/services/permits.py @@ -2,6 +2,7 @@ import asyncio import json +import logging import uuid from dataclasses import dataclass from datetime import datetime, timedelta @@ -24,6 +25,8 @@ from app.schemas.trust import PermitCreateRequest, PermitResponse from app.services.signing_keys import get_signing_key_service +logger = logging.getLogger(__name__) + class PermitError(RuntimeError): def __init__(self, reason: str) -> None: @@ -994,7 +997,13 @@ async def reconcile_budgets(self, *, idle_seconds: int = 900) -> int: reclaiming their budget can never enable an over-spend. A crashed reservation on a still-active permit is left conservatively in place (the agent can spend *less* than authorized, never more) and is - reclaimed once the permit expires. Returns the number corrected. + reclaimed once the permit expires. + + Returns the number of permits corrected. A permit whose guarded repair + matched no row -- a concurrent reservation moved ``spent_credits`` + between the receipt scan and the write -- is *not* counted, is left + exactly as found, and is re-examined on the next pass; those skips are + logged rather than returned, so the count stays a count of writes. """ from app.db.models import ( IdempotencyRecordModel, @@ -1011,6 +1020,7 @@ async def reconcile_budgets(self, *, idle_seconds: int = 900) -> int: cutoff = now - timedelta(seconds=idle_seconds) factory = get_session_factory() corrected = 0 + skipped = 0 async with factory() as session: async with session.begin(): stale = ( @@ -1228,7 +1238,23 @@ async def reconcile_budgets(self, *, idle_seconds: int = 900) -> int: ) if (cast(Any, repaired).rowcount or 0) == 1: corrected += 1 + else: + skipped += 1 await session.commit() + if skipped: + # A skipped permit is not a failure and needs no operator action -- + # it is re-examined on the next pass, and the guard is the whole + # point (a concurrent reservation landed, so this pass's recomputed + # figure is already out of date). It is logged because the return + # value counts only repairs: without this line a pass that skipped + # every permit is indistinguishable from a pass that found nothing + # to repair, and a permit that never stops being skipped -- the + # signature of a hot permit under sustained contention -- would + # never surface anywhere. + logger.info( + "permit_budget_reconcile_skipped", + extra={"skipped": skipped, "corrected": corrected}, + ) return corrected async def verify_signature(self, model: PermitModel) -> bool: diff --git a/app/services/stripe_integration.py b/app/services/stripe_integration.py index b33ad480..85333faf 100644 --- a/app/services/stripe_integration.py +++ b/app/services/stripe_integration.py @@ -24,6 +24,7 @@ from ..db.database import get_session_factory from ..db.models import BillingAlertModel, LedgerEntryModel, WalletModel from ..core.config import get_settings +from ..core.time import utc_now from ..schemas.billing import AlertSeverity, AlertType, WalletStatus from .agent_money import WalletNotFoundError @@ -421,6 +422,15 @@ async def _handle_refund(self, charge: Any, event_id: str | None = None) -> None .values( balance=WalletModel.balance - refund_delta, lifetime_debits=WalletModel.lifetime_debits + refund_delta, + # WalletModel.updated_at has a default but no + # onupdate, and a Core UPDATE bypasses the ORM + # attribute write, so nothing advances it here + # unless it is set. A fiat clawback is the most + # consequential balance move the wallet sees; it + # must not be the one that leaves the row looking + # untouched. Every other balance-moving wallet + # UPDATE in this codebase sets it. + updated_at=utc_now(), ) .execution_options(synchronize_session=False) ) diff --git a/tests/test_permit_budget_integrity.py b/tests/test_permit_budget_integrity.py index ac6ffd48..b1ad931a 100644 --- a/tests/test_permit_budget_integrity.py +++ b/tests/test_permit_budget_integrity.py @@ -12,6 +12,7 @@ from datetime import datetime, timedelta, timezone from decimal import Decimal +from typing import Any import pytest from httpx import ASGITransport, AsyncClient @@ -215,9 +216,38 @@ async def __aexit__(self, *exc): lambda: (lambda: _InterleavingFactory(real_factory())), ) + # Observe the service's own logger object rather than stdlib capture: + # tests/test_migrations.py runs Alembic in-process, and migrations/env.py + # calls logging.config.fileConfig, which disables every logger created + # before it -- app.services.permits included. That is a test-ordering + # artifact (production runs migrations in a separate process, see + # scripts/docker_entrypoint.sh), but it makes handler- or caplog-based + # capture depend on which test ran first. Recording the call itself does + # not. + calls: list[tuple[str, dict[str, Any]]] = [] + + class _RecordingLogger: + def __getattr__(self, name): + return getattr(permits_module.logger, name) + + def info(self, event, *args, **kwargs): + calls.append((event, kwargs.get("extra") or {})) + + monkeypatch.setattr(permits_module, "logger", _RecordingLogger()) + corrected = await get_permit_service().reconcile_budgets(idle_seconds=1) # The concurrent value survives; the stale repair is skipped, not applied. assert fired["done"], "the interleave never ran β€” the test proved nothing" assert await _spent(permit.permit_id) == Decimal("9") assert corrected == 0 + + # A skip is correct, but it must not be silent: the return value counts + # only writes, so a pass that skipped everything and a pass that found + # nothing to do are the same number. The log is what tells them apart. + skips = [ + extra for event, extra in calls if event == "permit_budget_reconcile_skipped" + ] + assert skips, "the skipped repair was not reported anywhere" + assert skips[0].get("skipped") == 1 + assert skips[0].get("corrected") == 0 diff --git a/tests/test_stripe_integration.py b/tests/test_stripe_integration.py index 0f66efd3..5229463b 100644 --- a/tests/test_stripe_integration.py +++ b/tests/test_stripe_integration.py @@ -3,6 +3,7 @@ Validates fiat top-up flow, webhook handling, and idempotency. """ +from datetime import timedelta from decimal import Decimal from unittest.mock import AsyncMock, MagicMock, patch @@ -11,6 +12,7 @@ from httpx import AsyncClient, ASGITransport from sqlalchemy.exc import IntegrityError +from app.core.time import utc_now from app.main import app from app.services.stripe_integration import StripeIntegration, StripeSettlementError @@ -427,6 +429,63 @@ async def test_redelivered_refund_debits_only_once( refunds = [e for e in ledger if e.action == "refund"] assert len(refunds) == 1 + @pytest.mark.anyio + async def test_refund_clawback_advances_the_wallet_timestamp( + self, client, sponsor_wallet, api_headers + ): + """A clawback must not leave the wallet row looking untouched. + + The clawback is a Core ``UPDATE``, which bypasses the ORM attribute + write, and ``WalletModel.updated_at`` has a default but no + ``onupdate`` -- so the timestamp only advances if the statement sets + it. Without that, the most consequential balance move the wallet sees + (real fiat returned by Stripe) is the one that leaves no trace in + ``updated_at``, and anything reading the row for recency -- an + operator triaging a disputed balance, a sweep ordered by staleness -- + is looking at the time of the top-up instead. + """ + from app.services.stripe_integration import get_stripe_integration + from app.db.database import get_session_factory + from app.db.models import WalletModel + + wallet_id = sponsor_wallet["wallet_id"] + integration = get_stripe_integration() + factory = get_session_factory() + + await integration._mint_credits( + wallet_id=wallet_id, + amount=Decimal("50000"), + payment_intent_id="pi_clawback_timestamp", + description="topup", + ) + + # Backdate the row so any advance is unambiguous rather than a + # sub-resolution tie against the mint that just ran. + stale = utc_now() - timedelta(hours=1) + async with factory() as session: + async with session.begin(): + wallet = await session.get(WalletModel, wallet_id) + assert wallet is not None + wallet.updated_at = stale + session.add(wallet) + + await integration._handle_refund( + refunded_charge( + payment_intent_id="pi_clawback_timestamp", + amount_refunded=5000, + ), + "evt_clawback_timestamp", + ) + + async with factory() as session: + wallet = await session.get(WalletModel, wallet_id) + assert wallet is not None + assert wallet.balance == Decimal("0"), "the clawback did not apply" + assert wallet.updated_at > stale, ( + "the clawback moved the balance but left updated_at at " + f"{wallet.updated_at}, the value set before it ran" + ) + @pytest.mark.anyio async def test_successive_partial_refunds_apply_only_new_cumulative_delta( self, sponsor_wallet