diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e9e278f..1fb787cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -317,6 +317,66 @@ full release gate; do not backfill a final `v1.2.0` tag. ### πŸ”’ Security +- **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 + 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 + 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/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/awi_http_governance.py b/app/services/awi_http_governance.py index c18d6ef4..9eee6dde 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__) @@ -237,7 +237,18 @@ 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 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. + """ if ctx.replay_response is not None: return ctx.replay_response @@ -246,7 +257,45 @@ 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: + 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=detail, + ) + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=detail, + ) 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 74b0fa74..474e02ba 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 or_, 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, @@ -261,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" @@ -290,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, @@ -309,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), @@ -326,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), @@ -341,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), @@ -353,9 +373,123 @@ 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), + # 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), + ), + ), + ), + # 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, + 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: 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) + 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), + # 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=f"Wallet is {wallet.status} and cannot be charged.", + ) + 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), + 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 +742,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 +768,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 + ) + await session.execute( + sa_update(WalletModel) + .where( + cast( + ColumnElement[bool], WalletModel.wallet_id == wallet_id + ) ) - wallet.updated_at = utc_now() + .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/idempotency.py b/app/services/idempotency.py index ff1252b8..9db6671c 100644 --- a/app/services/idempotency.py +++ b/app/services/idempotency.py @@ -708,6 +708,7 @@ async def reconcile_stuck_records( record.status_code = status_code session.add(record) repaired += 1 + return repaired, needs_review diff --git a/app/services/permits.py b/app/services/permits.py index 7630f5dd..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: @@ -67,6 +70,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,11 +762,12 @@ 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. - # The expiry term matters for the same reason it does there: - # an expired permit keeps status="active" in storage, so - # without it this second reservation path would still spend - # against a permit that has crossed expires_at. + # *and the expiry* are enforced by the WHERE clause, not a + # read-then-write. The expiry term matters for the same + # reason it does in authorize_and_reserve: an expired permit + # keeps status="active" in storage because the sweeper flips + # it lazily, so without this term a second reservation path + # still spends against a permit that has crossed expires_at. reserved = await session.execute( sa_update(PermitModel) .where( @@ -777,9 +799,14 @@ 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("permit_not_active") - if exists.expires_at <= now: + raise PermitError(f"permit_{exists.status}") + if to_naive_utc(exists.expires_at) <= now: raise PermitError("permit_expired") raise PermitError("permit_budget_exceeded") @@ -869,13 +896,15 @@ 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") + # 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: @@ -968,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, @@ -985,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 = ( @@ -1171,12 +1207,54 @@ 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 + 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/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..85333faf 100644 --- a/app/services/stripe_integration.py +++ b/app/services/stripe_integration.py @@ -13,16 +13,18 @@ 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 from ..core.config import get_settings +from ..core.time import utc_now from ..schemas.billing import AlertSeverity, AlertType, WalletStatus from .agent_money import WalletNotFoundError @@ -402,8 +404,39 @@ 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, + # 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) + ) + # 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/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_dispatch_reconciliation.py b/tests/test_mcp_dispatch_reconciliation.py index 7c8d9c59..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 @@ -1058,3 +1058,130 @@ async def test_concurrent_budget_release_decrements_exactly_once( assert await permits.release_dispatch_budget_once(seed.attempt_id) is False final = await permits.get_permit(seed.permit_id) assert final is not None and final.spent_credits == after.spent_credits + + +@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() + 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 + + 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 + + +@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_permit_budget_integrity.py b/tests/test_permit_budget_integrity.py new file mode 100644 index 00000000..b1ad931a --- /dev/null +++ b/tests/test_permit_budget_integrity.py @@ -0,0 +1,253 @@ +"""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 +from typing import Any + +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())), + ) + + # 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 diff --git a/tests/test_wallet_ledger_integrity.py b/tests/test_wallet_ledger_integrity.py new file mode 100644 index 00000000..54f40661 --- /dev/null +++ b/tests/test_wallet_ledger_integrity.py @@ -0,0 +1,414 @@ +"""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" + # 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") - debited + # 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") + + +@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") + + +@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" + ) + + +@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}" + # 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")