From 237a0cc048fd834c949d6c271d12d0e3b08d3fce Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 23:39:41 +0000 Subject: [PATCH 1/5] fix(wallets): guard the provisioning and transfer debits against lost updates #305 converted the charge debit, the child reclaim, refunds, the velocity counters, and the Stripe clawback to guarded UPDATEs. Three read-modify-write debits in wallet_engine were not in that sweep and are still live: wallet_engine.py:285 sponsor.balance -= budget_credits (provisioning) wallet_engine.py:399 parent.balance -= budget_credits (delegation) wallet_engine.py:675 source.balance -= amount (transfer) Each reads a balance, checks it, and writes the decided value back under a SELECT ... FOR UPDATE that SQLAlchemy drops silently on SQLite. Two concurrent provisions hand out the same credits; a transfer credits the destination with credits the competing writer already rolled back off the source, minting them outright. All three now route through _apply_balance_delta, which restates balance, spendable status, and the child lifetime cap in the WHERE clause and reports a lost race as whichever check it actually failed. The cap predicate compares against the row's own max_spend rather than a value read into Python, the same way #305 shaped it in billing_engine, so a concurrently changed cap is honoured too. Tests tests/test_wallet_concurrency.py covers only these three paths; the ones #305 fixed are already covered by tests/test_wallet_ledger_integrity.py. All four are confirmed to fail against this base, and three of them fail by succeeding -- the operation completes when it should have been refused, including a delegation debiting a parent frozen mid-operation. The fourth pins the sponsor-provisioning guard: that path awaits nothing between its read and its write, so there is no seam to inject a competing commit into. The module skips on any non-SQLite engine, where the injected writer would block on a lock that genuinely works. tests/test_permit_postgres_concurrency gains three real-concurrency conservation tests -- six provisions against a budget funding four, five transfers out of one balance, three reclaims of one child -- verified against a live PostgreSQL 16. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01FsXYfTHDLiFNQ24R5JPN7H --- app/services/wallet_engine.py | 179 ++++++++++++-- tests/test_permit_postgres_concurrency.py | 137 +++++++++++ tests/test_wallet_concurrency.py | 280 ++++++++++++++++++++++ 3 files changed, 578 insertions(+), 18 deletions(-) create mode 100644 tests/test_wallet_concurrency.py diff --git a/app/services/wallet_engine.py b/app/services/wallet_engine.py index 9c30af5f..afc4c1f5 100644 --- a/app/services/wallet_engine.py +++ b/app/services/wallet_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.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.sql.elements import ColumnElement @@ -152,6 +152,93 @@ async def wallet_is_expired(self, wallet_id: str) -> bool: expires_at = await self._effective_expiry(session, wallet) return expires_at is not None and expires_at <= utc_now() + async def _apply_balance_delta( + self, + session: AsyncSession, + wallet: WalletModel, + *, + balance_delta: Decimal, + lifetime_debits_delta: Decimal = Decimal("0"), + lifetime_credits_delta: Decimal = Decimal("0"), + require_balance: Decimal | None = None, + require_child_cap_for: Decimal | None = None, + require_spendable: bool = False, + ) -> bool: + """Move a wallet balance with a guarded UPDATE, never a write-back. + + The provisioning and transfer paths reached this module's checks the + same way the charge debit did before #305: read the balance, decide, + write the decided value back. ``SELECT ... FOR UPDATE`` is a silent + no-op on SQLite, so the lock those checks lean on is not always taken, + and the decision is made against a value another writer may already + have moved. + + Restating each racing precondition in the WHERE clause makes the + database re-check it in the same statement as the write. Returns + whether the row was updated. The caller's ``wallet`` instance is + refreshed either way: ``synchronize_session=False`` bypasses the + identity map, and both the ``balance_after`` written onto ledger + entries and the errors raised on a lost race must reflect the row + rather than the stale read. + """ + guards: list[ColumnElement[bool]] = [ + cast(ColumnElement[bool], WalletModel.wallet_id == wallet.wallet_id) + ] + if require_balance is not None: + guards.append( + cast(ColumnElement[bool], WalletModel.balance >= require_balance) + ) + if require_child_cap_for is not None: + # Compared against the row's own max_spend rather than a value read + # into Python, so a cap changed concurrently is honoured too. + # Non-child wallets and children with no cap are unconstrained. + guards.append( + 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 + require_child_cap_for + <= cast(Any, WalletModel.max_spend), + ), + ), + ) + ) + if require_spendable: + guards.append( + cast( + ColumnElement[bool], + cast(Any, WalletModel.status).notin_( + tuple(_NON_SPENDABLE_WALLET_STATUSES) + ), + ) + ) + + values: dict[str, Any] = {"balance": WalletModel.balance + balance_delta} + if lifetime_debits_delta: + values["lifetime_debits"] = ( + WalletModel.lifetime_debits + lifetime_debits_delta + ) + if lifetime_credits_delta: + values["lifetime_credits"] = ( + WalletModel.lifetime_credits + lifetime_credits_delta + ) + values["updated_at"] = utc_now() + + applied = await session.execute( + sa_update(WalletModel) + .where(*guards) + .values(**values) + .execution_options(synchronize_session=False) + ) + await session.refresh(wallet) + return (cast(Any, applied).rowcount or 0) == 1 + async def _lock_wallets_in_order( self, session: AsyncSession, wallet_ids: list[str] ) -> dict[str, WalletModel]: @@ -274,17 +361,29 @@ async def create_agent_wallet( f"Sponsor wallet is {sponsor.status} and cannot provision " "agent wallets" ) - if sponsor.balance < budget_credits: + # Deduct from sponsor. The balance and status checks above are + # a fast path; the guarded UPDATE is what makes them hold, so + # two concurrent provisions cannot hand out the same credits + # and a freeze landing in between is not outrun. + if not await self._apply_balance_delta( + session, + sponsor, + balance_delta=-budget_credits, + lifetime_debits_delta=budget_credits, + require_balance=budget_credits, + require_spendable=True, + ): + if sponsor.status in _NON_SPENDABLE_WALLET_STATUSES: + raise ValueError( + f"Sponsor wallet is {sponsor.status} and cannot " + "provision agent wallets" + ) raise self._insufficient_funds_error( sponsor_wallet_id, sponsor.balance, budget_credits, ) - # Deduct from sponsor - sponsor.balance -= budget_credits - sponsor.lifetime_debits += budget_credits - # Create agent wallet agent_wallet_id = f"agt-{uuid.uuid4().hex[:12]}" agent_wallet = WalletModel( @@ -388,17 +487,37 @@ async def create_child_wallet( and parent.lifetime_debits + budget_credits > parent.max_spend ): raise ValueError("Child wallet lifetime spend cap exceeded") - if parent.balance < budget_credits: + + # Deduct from parent. Balance, spendable status, and the child + # lifetime cap all travel with the debit: two concurrent + # delegations must not both clear a limit only one of them + # fits inside. + if not await self._apply_balance_delta( + session, + parent, + balance_delta=-budget_credits, + lifetime_debits_delta=budget_credits, + require_balance=budget_credits, + require_spendable=True, + require_child_cap_for=budget_credits, + ): + if parent.status in _NON_SPENDABLE_WALLET_STATUSES: + raise ValueError( + f"Parent wallet is {parent.status} and cannot spawn " + "child wallets" + ) + if ( + parent.wallet_type == WalletType.CHILD.value + and parent.max_spend + and parent.lifetime_debits + budget_credits > parent.max_spend + ): + raise ValueError("Child wallet lifetime spend cap exceeded") raise self._insufficient_funds_error( parent_wallet_id, parent.balance, budget_credits, ) - # Deduct from parent - parent.balance -= budget_credits - parent.lifetime_debits += budget_credits - # Create child wallet child_wallet_id = f"chd-{uuid.uuid4().hex[:12]}" effective_ttl_seconds = ttl_seconds @@ -666,16 +785,40 @@ async def transfer( ): raise ValueError("Child wallet lifetime spend cap exceeded") - if source.balance < amount: + # Execute transfer. The debit is guarded on everything a + # concurrent writer could move, so the credit below only ever + # pays out credits this statement actually took from the + # source -- a lost update here would mint credits, landing + # them on the destination without removing them anywhere. + if not await self._apply_balance_delta( + session, + source, + balance_delta=-amount, + lifetime_debits_delta=amount, + require_balance=amount, + require_spendable=True, + require_child_cap_for=amount, + ): + if source.status in _NON_SPENDABLE_WALLET_STATUSES: + raise ValueError( + f"Source wallet is {source.status} and cannot " + "transfer credits" + ) + if ( + source.wallet_type == WalletType.CHILD.value + and source.max_spend + and source.lifetime_debits + amount > source.max_spend + ): + raise ValueError("Child wallet lifetime spend cap exceeded") raise self._insufficient_funds_error( from_wallet_id, source.balance, amount ) - - # Execute transfer - source.balance -= amount - source.lifetime_debits += amount - dest.balance += amount - dest.lifetime_credits += amount + await self._apply_balance_delta( + session, + dest, + balance_delta=amount, + lifetime_credits_delta=amount, + ) # Create ledger entries on both sides session.add( diff --git a/tests/test_permit_postgres_concurrency.py b/tests/test_permit_postgres_concurrency.py index e784622d..38cf4845 100644 --- a/tests/test_permit_postgres_concurrency.py +++ b/tests/test_permit_postgres_concurrency.py @@ -1405,3 +1405,140 @@ async def test_conflicting_payload_during_upstream_dispatch_rejects_in_postgres( if first_task is not None and not first_task.done(): await asyncio.gather(first_task, return_exceptions=True) get_service_registry().unregister_local(tool_name) + + +# --------------------------------------------------------------------------- +# Wallet balance conservation under real concurrency +# +# The guarded UPDATEs on the wallet paths make a lost update impossible in the +# statement rather than in the lock; SQLite proves that in +# tests/test_wallet_concurrency.py, where the lock is a no-op and the window +# can be injected into. Here the lock is real, so the interesting question is +# different: with genuinely concurrent operations on one balance, does the +# money add up? These would catch a regression that removed both the guard and +# the lock, which no single-engine test can -- and unlike the SQLite proofs +# they exercise the real engine the deployment runs on. +# --------------------------------------------------------------------------- + + +async def _seed_wallet_for_concurrency(credits: Decimal) -> str: + """Create a sponsor wallet holding *credits*, returning its id.""" + wallet = await get_agent_money().create_sponsor_wallet( + sponsor_name=f"pg-concurrency-{uuid.uuid4().hex[:8]}", + email="pg-concurrency@example.com", + initial_credits=credits, + require_kyc=False, + ) + return wallet.wallet_id + + +async def _wallet_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 + + +@pytest.mark.asyncio(loop_scope="session") +async def test_concurrent_provisioning_never_overdraws_in_postgres() -> None: + """Six concurrent provisions against a budget that funds four. + + Exactly four must succeed and the sponsor must land on zero -- never + negative, and never with credits stranded because a lost update erased a + sibling's debit. + """ + _require_opted_in_postgres() + sponsor_id = await _seed_wallet_for_concurrency(Decimal("400")) + + async def provision(index: int) -> object: + try: + return await get_agent_money().create_agent_wallet( + sponsor_wallet_id=sponsor_id, + agent_id=f"pg-conc-{index}-{uuid.uuid4().hex[:6]}", + budget_credits=Decimal("100"), + ) + except Exception as exc: # refusal is a valid outcome, not a failure + return exc + + results = await asyncio.gather(*(provision(i) for i in range(6))) + granted = [r for r in results if not isinstance(r, Exception)] + + assert len(granted) == 4 + assert await _wallet_balance(sponsor_id) == Decimal("0") + + +@pytest.mark.asyncio(loop_scope="session") +async def test_concurrent_transfers_conserve_credits_in_postgres() -> None: + """Concurrent transfers out of one wallet must conserve the total. + + Whatever mix of transfers is admitted, the credits that left the source + must equal the credits that arrived -- a transfer that pays out more than + it took would mint credits from nothing. + """ + _require_opted_in_postgres() + sponsor_id = await _seed_wallet_for_concurrency(Decimal("600")) + money = get_agent_money() + source = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_id, + agent_id=f"pg-src-{uuid.uuid4().hex[:6]}", + budget_credits=Decimal("300"), + ) + dest = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_id, + agent_id=f"pg-dst-{uuid.uuid4().hex[:6]}", + budget_credits=Decimal("0"), + ) + + async def move() -> object: + try: + return await money.transfer( + from_wallet_id=source.wallet_id, + to_wallet_id=dest.wallet_id, + amount=Decimal("100"), + ) + except Exception as exc: + return exc + + results = await asyncio.gather(*(move() for _ in range(5))) + moved = len([r for r in results if not isinstance(r, Exception)]) + + assert moved == 3 + source_balance = await _wallet_balance(source.wallet_id) + dest_balance = await _wallet_balance(dest.wallet_id) + assert source_balance == Decimal("300") - Decimal("100") * moved + assert dest_balance == Decimal("100") * moved + assert source_balance + dest_balance == Decimal("300") + + +@pytest.mark.asyncio(loop_scope="session") +async def test_concurrent_reclaims_credit_the_parent_once_in_postgres() -> None: + """Three concurrent reclaims of one child credit the parent exactly once.""" + _require_opted_in_postgres() + sponsor_id = await _seed_wallet_for_concurrency(Decimal("600")) + money = get_agent_money() + parent = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_id, + agent_id=f"pg-parent-{uuid.uuid4().hex[:6]}", + budget_credits=Decimal("400"), + ) + child = await money.create_child_wallet( + parent_wallet_id=parent.wallet_id, + child_agent_id=f"pg-child-{uuid.uuid4().hex[:6]}", + budget_credits=Decimal("150"), + max_spend=Decimal("150"), + ) + parent_before = await _wallet_balance(parent.wallet_id) + + async def reclaim() -> object: + try: + return await money.reclaim_child_wallet(child.wallet_id) + except Exception as exc: + return exc + + results = await asyncio.gather(*(reclaim() for _ in range(3))) + succeeded = [r for r in results if not isinstance(r, Exception)] + + assert len(succeeded) == 1 + assert await _wallet_balance(parent.wallet_id) == parent_before + Decimal("150") + assert await _wallet_balance(child.wallet_id) == Decimal("0") diff --git a/tests/test_wallet_concurrency.py b/tests/test_wallet_concurrency.py new file mode 100644 index 00000000..41ba5a21 --- /dev/null +++ b/tests/test_wallet_concurrency.py @@ -0,0 +1,280 @@ +"""Lost-update proofs for the wallet provisioning and transfer debits. + +#305 converted the charge debit, the child reclaim, refunds, and the velocity +counters to guarded updates. It left three: the sponsor provisioning debit, +the parent delegation debit, and the transfer debit, all still reading a +balance, deciding, and writing the decided value back. They rely on +``SELECT ... FOR UPDATE`` to keep a second writer out of that window, and +SQLAlchemy **silently drops** ``FOR UPDATE`` on SQLite, so on that engine the +window is wide open. + +Ordinary concurrency does not reproduce it: SQLite serializes the write +transactions, so two ``asyncio.gather``-ed operations simply take turns and +the second one reads fresh data. What the dropped lock actually allows is +narrower -- a competing writer committing *between* one operation's read and +its write -- so these tests inject exactly that, by wrapping a call the +operation already awaits inside that window and committing the competing +write from it. Nothing about the assertion depends on the injection: the +operation is left to decide on its own, and a decision made from the value it +read rather than the value the row holds is what fails. + +They run on SQLite, because that is where the defect is reachable: the +injection only works while the row lock is a no-op. Pointed at PostgreSQL the +injected writer would block on the lock the operation under test genuinely +holds, so they skip there. The PostgreSQL side of the same guarantee is +covered by real concurrent operations in +``tests/test_permit_postgres_concurrency.py``, and the paths #305 already +fixed are covered by ``tests/test_wallet_ledger_integrity.py``. +""" + +from __future__ import annotations + +import os +import uuid +from collections.abc import Awaitable, Callable +from decimal import Decimal + +import pytest +import pytest_asyncio +from sqlalchemy import update as sa_update + +from app.db.database import get_engine, get_session_factory +from app.db.models import WalletModel +from app.schemas.billing import WalletStatus +from app.services.agent_money import get_agent_money +from app.services.wallet_engine import WalletEngine + +_SKIP_REASON = ( + "the read-to-write window these tests inject into is only open where " + "SELECT ... FOR UPDATE is dropped; on an engine that honours it the " + "injected writer would block on the lock the operation already holds" +) + +# Checked before any database fixture runs: the PostgreSQL jobs point +# DATABASE_URL at a real server, and even reaching a fixture there would open +# connections this module has no use for. +_DATABASE_URL = os.environ.get("DATABASE_URL", "") +if _DATABASE_URL and not _DATABASE_URL.startswith("sqlite"): + pytest.skip(_SKIP_REASON, allow_module_level=True) + +pytestmark = pytest.mark.anyio + + +def _skip_unless_sqlite() -> None: + """Skip where ``FOR UPDATE`` is honoured and the injection would block.""" + engine = get_engine() + if engine is not None and engine.dialect.name != "sqlite": + pytest.skip(_SKIP_REASON) + + +async def _balance(wallet_id: str) -> Decimal: + """Read a wallet balance in its own transaction.""" + 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 _commit_debit(wallet_id: str, amount: Decimal) -> None: + """Commit a debit from outside the operation under test.""" + factory = get_session_factory() + async with factory() as session: + async with session.begin(): + await session.execute( + sa_update(WalletModel) + .where(WalletModel.wallet_id == wallet_id) + .values(balance=WalletModel.balance - amount) + ) + + +def _interfere_once( + monkeypatch: pytest.MonkeyPatch, + attr: str, + interference: Callable[[], Awaitable[None]], + *, + before: bool = False, +) -> None: + """Commit a competing write inside the operation's read-to-write window. + + ``attr`` names a ``WalletEngine`` coroutine the operation under test + already awaits after loading the wallet and before writing it. This is the + window a working ``SELECT ... FOR UPDATE`` would hold closed and that + SQLite leaves open; the wrapper fires once, so only the first operation in + a test is raced. Pass ``before`` when the named call *is* the write, so + the competing commit still lands inside the window rather than after it. + """ + original = getattr(WalletEngine, attr) + state = {"fired": False} + + async def wrapper(self, *args, **kwargs): + if before and not state["fired"]: + state["fired"] = True + await interference() + result = await original(self, *args, **kwargs) + if not before and not state["fired"]: + state["fired"] = True + await interference() + return result + + monkeypatch.setattr(WalletEngine, attr, wrapper) + + +@pytest_asyncio.fixture +async def sponsor_wallet(clean_database) -> str: + """A sponsor wallet with a known balance.""" + _skip_unless_sqlite() + wallet = await get_agent_money().create_sponsor_wallet( + sponsor_name=f"Concurrency Sponsor {uuid.uuid4().hex[:8]}", + email="concurrency@example.com", + initial_credits=Decimal("1000"), + require_kyc=False, + ) + return wallet.wallet_id + + +async def test_delegation_refuses_a_parent_balance_that_moved_under_it( + sponsor_wallet: str, monkeypatch: pytest.MonkeyPatch +) -> None: + """A child wallet must not be funded with credits the parent no longer has.""" + money = get_agent_money() + parent = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"agent-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("400"), + ) + _interfere_once( + monkeypatch, + "ensure_wallet_not_expired", + lambda: _commit_debit(parent.wallet_id, Decimal("350")), + ) + + with pytest.raises(Exception) as excinfo: + await money.create_child_wallet( + parent_wallet_id=parent.wallet_id, + child_agent_id=f"child-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("200"), + max_spend=Decimal("200"), + ) + + assert "insufficient" in str(excinfo.value).lower() + assert await _balance(parent.wallet_id) == Decimal("50") + + +async def test_transfer_refuses_a_source_balance_that_moved_under_it( + sponsor_wallet: str, monkeypatch: pytest.MonkeyPatch +) -> None: + """A transfer must never pay out credits it did not take from the source. + + A lost update here is worse than an overdraft: the destination is credited + from a debit that was rolled over by the competing writer, so the credits + exist on one side and not the other. + """ + money = get_agent_money() + source = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"src-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("300"), + ) + dest = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"dst-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("100"), + ) + _interfere_once( + monkeypatch, + "ensure_wallet_not_expired", + lambda: _commit_debit(source.wallet_id, Decimal("250")), + ) + + with pytest.raises(Exception) as excinfo: + await money.transfer( + from_wallet_id=source.wallet_id, + to_wallet_id=dest.wallet_id, + amount=Decimal("200"), + ) + + assert "insufficient" in str(excinfo.value).lower() + assert await _balance(source.wallet_id) == Decimal("50") + assert await _balance(dest.wallet_id) == Decimal("100") + + +async def test_sponsor_provisioning_refuses_a_balance_that_moved_under_it( + sponsor_wallet: str, monkeypatch: pytest.MonkeyPatch +) -> None: + """Two provisions must not hand out the same sponsor credits twice. + + Unlike the paths above, this one awaits nothing between its read and its + write, so the window can only be entered at the write itself. That means + this test pins the guard's contract rather than reproducing the original + defect against the pre-fix code, which had no such seam -- the defect is + the same shape as the delegation debit it sits beside. + """ + _interfere_once( + monkeypatch, + "_apply_balance_delta", + lambda: _commit_debit(sponsor_wallet, Decimal("900")), + before=True, + ) + + with pytest.raises(Exception) as excinfo: + await get_agent_money().create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"agent-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("500"), + ) + + assert "insufficient" in str(excinfo.value).lower() + # The competing debit stands; nothing was provisioned from credits that + # had already been spent. + assert await _balance(sponsor_wallet) == Decimal("100") + + +async def _commit_freeze(wallet_id: str) -> None: + """Freeze a wallet from outside the operation under test.""" + factory = get_session_factory() + async with factory() as session: + async with session.begin(): + await session.execute( + sa_update(WalletModel) + .where(WalletModel.wallet_id == wallet_id) + .values(status=WalletStatus.FROZEN.value) + ) + + +async def test_delegation_refuses_a_parent_frozen_under_it( + sponsor_wallet: str, monkeypatch: pytest.MonkeyPatch +) -> None: + """A freeze landing mid-operation must stop the delegation debit. + + The parent's status is checked in Python and then awaited past, so a + freeze committed in that window is invisible to the check. Spending from + a frozen wallet is the failure the freeze exists to prevent, which is why + the status rides in the UPDATE alongside the balance rather than being + trusted from the read. + """ + money = get_agent_money() + parent = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"agent-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("400"), + ) + _interfere_once( + monkeypatch, + "ensure_wallet_not_expired", + lambda: _commit_freeze(parent.wallet_id), + ) + + with pytest.raises(Exception) as excinfo: + await money.create_child_wallet( + parent_wallet_id=parent.wallet_id, + child_agent_id=f"child-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("200"), + max_spend=Decimal("200"), + ) + + assert "frozen" in str(excinfo.value).lower() + # Nothing was spent out of the frozen wallet. + assert await _balance(parent.wallet_id) == Decimal("400") + + From 3c2f5bef3d7d1ac03c1a906bbcc3085dc30ac32e Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 23:44:10 +0000 Subject: [PATCH 2/5] fix(wallets): abort a transfer whose credit matches no row; tighten the proofs Review follow-ups on the three guarded debits. The transfer's destination credit discarded the helper's result. The source debit is already applied in the same transaction, so a credit matching zero rows left transfer reporting success with the credits removed from the source and never added to the destination. Only wallet_id guards that write, so a miss means the row is gone; the transfer now aborts. The proofs asserted pytest.raises(Exception), which would accept an AttributeError from the injection wrapper or a driver fault as a refusal. They now name the concrete class -- InsufficientFundsError for the funds refusals, ValueError for the frozen-parent refusal -- and keep the message check as a secondary assertion. The PostgreSQL conservation tests likewise assert that every non-success is the expected refusal, so a pool timeout or a deadlock can no longer produce the right split for the wrong reason. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01FsXYfTHDLiFNQ24R5JPN7H --- app/services/wallet_engine.py | 10 ++++++++-- tests/test_permit_postgres_concurrency.py | 12 ++++++++++++ tests/test_wallet_concurrency.py | 10 +++++----- 3 files changed, 25 insertions(+), 7 deletions(-) diff --git a/app/services/wallet_engine.py b/app/services/wallet_engine.py index afc4c1f5..9e3df574 100644 --- a/app/services/wallet_engine.py +++ b/app/services/wallet_engine.py @@ -813,12 +813,18 @@ async def transfer( raise self._insufficient_funds_error( from_wallet_id, source.balance, amount ) - await self._apply_balance_delta( + # The debit is already applied in this transaction, so a credit + # that matches no row would leave the transfer reporting success + # with the credits on neither side. Only ``wallet_id`` guards + # this write, so a miss means the destination row is gone -- + # abort rather than commit half a transfer. + if not await self._apply_balance_delta( session, dest, balance_delta=amount, lifetime_credits_delta=amount, - ) + ): + raise self._wallet_not_found_error(to_wallet_id) # Create ledger entries on both sides session.add( diff --git a/tests/test_permit_postgres_concurrency.py b/tests/test_permit_postgres_concurrency.py index 38cf4845..6415973e 100644 --- a/tests/test_permit_postgres_concurrency.py +++ b/tests/test_permit_postgres_concurrency.py @@ -1463,8 +1463,12 @@ async def provision(index: int) -> object: results = await asyncio.gather(*(provision(i) for i in range(6))) granted = [r for r in results if not isinstance(r, Exception)] + refused = [r for r in results if isinstance(r, Exception)] assert len(granted) == 4 + # A pool timeout or a deadlock would otherwise count as a refusal and the + # split would come out right for the wrong reason. + assert all("insufficient" in str(exc).lower() for exc in refused), refused assert await _wallet_balance(sponsor_id) == Decimal("0") @@ -1502,8 +1506,10 @@ async def move() -> object: results = await asyncio.gather(*(move() for _ in range(5))) moved = len([r for r in results if not isinstance(r, Exception)]) + refused = [r for r in results if isinstance(r, Exception)] assert moved == 3 + assert all("insufficient" in str(exc).lower() for exc in refused), refused source_balance = await _wallet_balance(source.wallet_id) dest_balance = await _wallet_balance(dest.wallet_id) assert source_balance == Decimal("300") - Decimal("100") * moved @@ -1538,7 +1544,13 @@ async def reclaim() -> object: results = await asyncio.gather(*(reclaim() for _ in range(3))) succeeded = [r for r in results if not isinstance(r, Exception)] + refused = [r for r in results if isinstance(r, Exception)] assert len(succeeded) == 1 + # The losers must lose the claim, not fail for some unrelated reason. + assert all( + "already closed" in str(exc).lower() or "balance changed" in str(exc).lower() + for exc in refused + ), refused assert await _wallet_balance(parent.wallet_id) == parent_before + Decimal("150") assert await _wallet_balance(child.wallet_id) == Decimal("0") diff --git a/tests/test_wallet_concurrency.py b/tests/test_wallet_concurrency.py index 41ba5a21..483df46a 100644 --- a/tests/test_wallet_concurrency.py +++ b/tests/test_wallet_concurrency.py @@ -41,7 +41,7 @@ from app.db.database import get_engine, get_session_factory from app.db.models import WalletModel from app.schemas.billing import WalletStatus -from app.services.agent_money import get_agent_money +from app.services.agent_money import InsufficientFundsError, get_agent_money from app.services.wallet_engine import WalletEngine _SKIP_REASON = ( @@ -149,7 +149,7 @@ async def test_delegation_refuses_a_parent_balance_that_moved_under_it( lambda: _commit_debit(parent.wallet_id, Decimal("350")), ) - with pytest.raises(Exception) as excinfo: + with pytest.raises(InsufficientFundsError) as excinfo: await money.create_child_wallet( parent_wallet_id=parent.wallet_id, child_agent_id=f"child-{uuid.uuid4().hex[:8]}", @@ -187,7 +187,7 @@ async def test_transfer_refuses_a_source_balance_that_moved_under_it( lambda: _commit_debit(source.wallet_id, Decimal("250")), ) - with pytest.raises(Exception) as excinfo: + with pytest.raises(InsufficientFundsError) as excinfo: await money.transfer( from_wallet_id=source.wallet_id, to_wallet_id=dest.wallet_id, @@ -217,7 +217,7 @@ async def test_sponsor_provisioning_refuses_a_balance_that_moved_under_it( before=True, ) - with pytest.raises(Exception) as excinfo: + with pytest.raises(InsufficientFundsError) as excinfo: await get_agent_money().create_agent_wallet( sponsor_wallet_id=sponsor_wallet, agent_id=f"agent-{uuid.uuid4().hex[:8]}", @@ -265,7 +265,7 @@ async def test_delegation_refuses_a_parent_frozen_under_it( lambda: _commit_freeze(parent.wallet_id), ) - with pytest.raises(Exception) as excinfo: + with pytest.raises(ValueError) as excinfo: await money.create_child_wallet( parent_wallet_id=parent.wallet_id, child_agent_id=f"child-{uuid.uuid4().hex[:8]}", From 2bb8ed491a0fcab1b49c2b6acfd1aa2ff273f2c8 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 23:53:20 +0000 Subject: [PATCH 3/5] fix(wallets): report a vanished credit destination instead of failing to refresh _apply_balance_delta refreshed the wallet before inspecting rowcount. When the guarded UPDATE matched no row because the row was gone, the refresh raised InvalidRequestError ("Could not refresh instance") and that reached the caller in place of the miss the helper exists to report -- so a transfer to a deleted destination surfaced an opaque SQLAlchemy error rather than WalletNotFoundError. The rowcount is now inspected first. A matched row is refreshed as before; a miss still refreshes, so the caller's error message is built from committed values, but tolerates the row being absent. No money was ever at risk here: the exception aborted the transaction and the already-applied debit rolled back with it. The defect was the diagnosis, not the accounting, which is why the new test asserts both the exception type and that the source balance is untouched. Confirmed to fail against the unfixed helper with the exact InvalidRequestError. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01FsXYfTHDLiFNQ24R5JPN7H --- app/services/wallet_engine.py | 22 ++++++++++- tests/test_wallet_concurrency.py | 66 ++++++++++++++++++++++++++++++-- 2 files changed, 83 insertions(+), 5 deletions(-) diff --git a/app/services/wallet_engine.py b/app/services/wallet_engine.py index 9e3df574..24b1181c 100644 --- a/app/services/wallet_engine.py +++ b/app/services/wallet_engine.py @@ -14,6 +14,7 @@ from typing import Any, cast from sqlalchemy import or_, select, update as sa_update +from sqlalchemy.exc import InvalidRequestError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.sql.elements import ColumnElement @@ -236,8 +237,25 @@ async def _apply_balance_delta( .values(**values) .execution_options(synchronize_session=False) ) - await session.refresh(wallet) - return (cast(Any, applied).rowcount or 0) == 1 + matched = (cast(Any, applied).rowcount or 0) == 1 + if matched: + # Only a matched row can be refreshed. Refreshing before checking + # would raise on a row that no longer exists, which is the one case + # the caller most needs reported as a plain miss. + await session.refresh(wallet) + else: + # A miss usually means a guard failed and the row is still there, so + # refresh to give the caller the committed values its error message + # is built from. A miss can also mean the row is gone, and then + # there is nothing to refresh -- keep the stale instance rather than + # letting SQLAlchemy raise over the caller's own error. + # ObjectDeletedError subclasses InvalidRequestError, so this catches + # both the deleted-row and the unrefreshable-instance cases. + try: + await session.refresh(wallet) + except InvalidRequestError: + pass + return matched async def _lock_wallets_in_order( self, session: AsyncSession, wallet_ids: list[str] diff --git a/tests/test_wallet_concurrency.py b/tests/test_wallet_concurrency.py index 483df46a..3313bbd6 100644 --- a/tests/test_wallet_concurrency.py +++ b/tests/test_wallet_concurrency.py @@ -36,12 +36,16 @@ import pytest import pytest_asyncio -from sqlalchemy import update as sa_update +from sqlalchemy import delete as sa_delete, update as sa_update from app.db.database import get_engine, get_session_factory -from app.db.models import WalletModel +from app.db.models import LedgerEntryModel, WalletModel from app.schemas.billing import WalletStatus -from app.services.agent_money import InsufficientFundsError, get_agent_money +from app.services.agent_money import ( + InsufficientFundsError, + WalletNotFoundError, + get_agent_money, +) from app.services.wallet_engine import WalletEngine _SKIP_REASON = ( @@ -278,3 +282,59 @@ async def test_delegation_refuses_a_parent_frozen_under_it( assert await _balance(parent.wallet_id) == Decimal("400") + + +async def _delete_wallet(wallet_id: str) -> None: + """Remove a wallet row, and the ledger rows that reference it.""" + factory = get_session_factory() + async with factory() as session: + async with session.begin(): + await session.execute( + sa_delete(LedgerEntryModel).where( + LedgerEntryModel.wallet_id == wallet_id + ) + ) + await session.execute( + sa_delete(WalletModel).where(WalletModel.wallet_id == wallet_id) + ) + + +async def test_transfer_reports_a_destination_that_vanished( + sponsor_wallet: str, monkeypatch: pytest.MonkeyPatch +) -> None: + """A credit matching no row must surface as a plain not-found. + + The destination credit is guarded on ``wallet_id`` alone, so the only way + it matches nothing is that the row is gone. The debit is already applied by + then, so the transfer has to abort -- and it has to abort with the reason, + not with whatever SQLAlchemy raises first. Refreshing an instance whose row + no longer exists raises ``InvalidRequestError``, which would otherwise + reach the caller in place of the real answer. + """ + money = get_agent_money() + source = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"src-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("300"), + ) + dest = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"dst-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("100"), + ) + _interfere_once( + monkeypatch, + "ensure_wallet_not_expired", + lambda: _delete_wallet(dest.wallet_id), + ) + + with pytest.raises(WalletNotFoundError) as excinfo: + await money.transfer( + from_wallet_id=source.wallet_id, + to_wallet_id=dest.wallet_id, + amount=Decimal("50"), + ) + + assert dest.wallet_id in str(excinfo.value) + # The debit rolled back with the aborted transaction. + assert await _balance(source.wallet_id) == Decimal("300") From 4dee5b4a5dd258b08a4cb65dee9cfb17a4fbf405 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 00:03:19 +0000 Subject: [PATCH 4/5] fix(wallets): treat a zero lifetime cap as a cap; refuse non-positive budgets Two input-shape defects on the debits this PR guards, both found in review. A stored max_spend of Decimal("0") was read two different ways inside the same function. The guarded UPDATE compares lifetime_debits against the column, so a zero cap correctly refused the debit. The Python checks around it used truthiness, so zero read as "no cap configured": the pre-check waved the debit through, and the failure path skipped the cap branch and reported an empty balance instead. The result was a wallet holding 400 credits refusing a 10-credit delegation with "Insufficient funds: 400.00000000 < 10". All four checks (delegation and transfer, pre-check and failure path) now test `is not None`, so both halves agree with the SQL. Non-positive budget_credits is now refused before either provisioning debit. The guard a negative budget would meet is `balance >= budget_credits`, which it satisfies trivially, so the debit ran as a credit: the sponsor or parent gained and the new wallet opened holding a negative balance. A child in that state cannot be unwound by a later reclaim, which only credits back a positive reclaim_amount. Both request schemas already require gt=0, so this was never reachable over HTTP -- but the service must not corrupt state when called directly, and the two methods being hardened here are exactly the ones that would. All four new tests are confirmed to fail against the unfixed source: the two zero-cap cases report insufficient funds on a funded wallet, and the non-positive cases do not raise at all. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01FsXYfTHDLiFNQ24R5JPN7H --- app/services/wallet_engine.py | 20 ++++-- tests/test_wallet_concurrency.py | 120 +++++++++++++++++++++++++++++++ 2 files changed, 136 insertions(+), 4 deletions(-) diff --git a/app/services/wallet_engine.py b/app/services/wallet_engine.py index 24b1181c..3b13f903 100644 --- a/app/services/wallet_engine.py +++ b/app/services/wallet_engine.py @@ -353,6 +353,13 @@ async def create_agent_wallet( auto_refill_amount: Decimal = Decimal("1000.0"), ) -> WalletResponse: """Provision a pre-paid agent wallet from a sponsor's balance.""" + if budget_credits <= Decimal("0"): + # A negative budget inverts the debit into a credit: the guard + # ``balance >= budget_credits`` is trivially true for a negative + # amount, so the sponsor would be paid and the new wallet opened + # holding a negative balance. + raise ValueError("budget_credits must be positive") + async with self._session_factory()() as session: async with session.begin(): # Lock sponsor wallet for update @@ -465,6 +472,11 @@ async def create_child_wallet( ) -> WalletResponse: """Spawn a child sub-agent wallet from a parent agent's balance.""" _validate_child_wallet_ttl(ttl_seconds) + if budget_credits <= Decimal("0"): + # Same inversion as the sponsor path, and worse: a child opened + # with a negative balance cannot be unwound by a later reclaim, + # which only credits the parent back a positive reclaim_amount. + raise ValueError("budget_credits must be positive") async with self._session_factory()() as session: async with session.begin(): # Lock parent wallet @@ -501,7 +513,7 @@ async def create_child_wallet( ) if ( parent.wallet_type == WalletType.CHILD.value - and parent.max_spend + and parent.max_spend is not None and parent.lifetime_debits + budget_credits > parent.max_spend ): raise ValueError("Child wallet lifetime spend cap exceeded") @@ -526,7 +538,7 @@ async def create_child_wallet( ) if ( parent.wallet_type == WalletType.CHILD.value - and parent.max_spend + and parent.max_spend is not None and parent.lifetime_debits + budget_credits > parent.max_spend ): raise ValueError("Child wallet lifetime spend cap exceeded") @@ -798,7 +810,7 @@ async def transfer( await self.ensure_wallet_not_expired(session, source) if ( source.wallet_type == WalletType.CHILD.value - and source.max_spend + and source.max_spend is not None and source.lifetime_debits + amount > source.max_spend ): raise ValueError("Child wallet lifetime spend cap exceeded") @@ -824,7 +836,7 @@ async def transfer( ) if ( source.wallet_type == WalletType.CHILD.value - and source.max_spend + and source.max_spend is not None and source.lifetime_debits + amount > source.max_spend ): raise ValueError("Child wallet lifetime spend cap exceeded") diff --git a/tests/test_wallet_concurrency.py b/tests/test_wallet_concurrency.py index 3313bbd6..3844bec0 100644 --- a/tests/test_wallet_concurrency.py +++ b/tests/test_wallet_concurrency.py @@ -338,3 +338,123 @@ async def test_transfer_reports_a_destination_that_vanished( assert dest.wallet_id in str(excinfo.value) # The debit rolled back with the aborted transaction. assert await _balance(source.wallet_id) == Decimal("300") + + +async def _set_child_cap(wallet_id: str, max_spend: Decimal) -> None: + """Force a wallet into child shape with an exact lifetime cap. + + ``max_spend=0`` cannot be created through the API (the request schema + requires ``gt=0``), so a stored zero cap is reachable only from a direct + service call or a migration. It is still a value the column can hold, and + the guarded UPDATE treats it as a real cap, so the Python paths must agree. + """ + factory = get_session_factory() + async with factory() as session: + async with session.begin(): + await session.execute( + sa_update(WalletModel) + .where(WalletModel.wallet_id == wallet_id) + .values(wallet_type="child", max_spend=max_spend) + ) + + +async def test_zero_lifetime_cap_is_a_cap_not_an_absent_one( + sponsor_wallet: str, +) -> None: + """A stored cap of zero must refuse, and say it was the cap. + + The guarded UPDATE compares against the column, so a zero cap already + refuses the debit. The Python checks used truthiness, so zero read as "no + cap set" -- the pre-check waved the debit through and the failure path + reported an empty balance. Two halves of the same function disagreeing + about what zero means is the actual defect; the caller seeing the wrong + reason is the symptom. + """ + money = get_agent_money() + parent = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"agent-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("400"), + ) + await _set_child_cap(parent.wallet_id, Decimal("0")) + + with pytest.raises(ValueError) as excinfo: + await money.create_child_wallet( + parent_wallet_id=parent.wallet_id, + child_agent_id=f"child-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("10"), + max_spend=Decimal("10"), + ) + + assert "cap exceeded" in str(excinfo.value).lower() + # The balance was never the problem, and none of it moved. + assert await _balance(parent.wallet_id) == Decimal("400") + + +async def test_zero_lifetime_cap_refuses_a_transfer_as_a_cap( + sponsor_wallet: str, +) -> None: + """The transfer path must agree with the delegation path about zero.""" + money = get_agent_money() + source = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"src-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("300"), + ) + dest = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"dst-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("100"), + ) + await _set_child_cap(source.wallet_id, Decimal("0")) + + with pytest.raises(ValueError) as excinfo: + await money.transfer( + from_wallet_id=source.wallet_id, + to_wallet_id=dest.wallet_id, + amount=Decimal("10"), + ) + + assert "cap exceeded" in str(excinfo.value).lower() + assert await _balance(source.wallet_id) == Decimal("300") + assert await _balance(dest.wallet_id) == Decimal("100") + + +@pytest.mark.parametrize("budget", [Decimal("0"), Decimal("-50")]) +async def test_non_positive_provisioning_budget_is_refused( + sponsor_wallet: str, budget: Decimal +) -> None: + """A non-positive budget must be refused before any wallet is touched. + + The request schemas already require ``gt=0``, so this is unreachable over + HTTP -- but the guard it would otherwise meet is ``balance >= budget``, + which a negative budget satisfies trivially. The debit would then run as a + credit: the sponsor gains, and the new wallet opens holding a negative + balance that no later reclaim can unwind. + """ + money = get_agent_money() + before = await _balance(sponsor_wallet) + + with pytest.raises(ValueError, match="positive"): + await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"agent-{uuid.uuid4().hex[:8]}", + budget_credits=budget, + ) + + assert await _balance(sponsor_wallet) == before + + parent = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"parent-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("200"), + ) + with pytest.raises(ValueError, match="positive"): + await money.create_child_wallet( + parent_wallet_id=parent.wallet_id, + child_agent_id=f"child-{uuid.uuid4().hex[:8]}", + budget_credits=budget, + max_spend=Decimal("50"), + ) + + assert await _balance(parent.wallet_id) == Decimal("200") From 0058333e79a0161e83f1ca2aac34491bc6408d27 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 00:18:44 +0000 Subject: [PATCH 5/5] fix(wallets): allow a zero provisioning budget; only negative is refused The previous commit refused budget_credits <= 0. That was wrong, and CI caught it: tests/test_permit_postgres_concurrency.py seeds a destination wallet with budget_credits=0 and the guard rejected it. Zero is a legal, documented input. SelfProvisionRequest.budget_credits is declared ge=0 (app/routers/dev_keys.py:64), so a zero-budget self-provision is a supported request that opens an empty wallet for later top-up. Refusing it raised after the caller's sponsor wallet had already been committed -- answering 500 and leaving an orphaned sponsor behind. The hazard was only ever the negative case: the guard balance >= budget is trivially satisfied by a negative amount, so the debit runs as a credit. Zero debits nothing and opens an empty wallet, which is exactly what the caller asked for. Both guards now refuse only budget_credits < 0. Adds a test that a zero budget opens an empty wallet and leaves the sponsor balance untouched, so the self-provision contract is pinned rather than implied. Verified against a live PostgreSQL 16 this time, not just SQLite: both jobs that gate this PR pass locally -- test_permit_postgres_concurrency.py (14) and test_mcp_postgres_multiprocess.py (13). The regression reached CI because the previous run only exercised SQLite, where those files skip. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01FsXYfTHDLiFNQ24R5JPN7H --- app/services/wallet_engine.py | 13 +++++--- tests/test_wallet_concurrency.py | 52 ++++++++++++++++++++++++-------- 2 files changed, 49 insertions(+), 16 deletions(-) diff --git a/app/services/wallet_engine.py b/app/services/wallet_engine.py index 3b13f903..095015b1 100644 --- a/app/services/wallet_engine.py +++ b/app/services/wallet_engine.py @@ -353,12 +353,16 @@ async def create_agent_wallet( auto_refill_amount: Decimal = Decimal("1000.0"), ) -> WalletResponse: """Provision a pre-paid agent wallet from a sponsor's balance.""" - if budget_credits <= Decimal("0"): + if budget_credits < Decimal("0"): # A negative budget inverts the debit into a credit: the guard # ``balance >= budget_credits`` is trivially true for a negative # amount, so the sponsor would be paid and the new wallet opened # holding a negative balance. - raise ValueError("budget_credits must be positive") + # + # Zero is deliberately allowed. It debits nothing and opens an + # empty wallet to be topped up later, which is what + # SelfProvisionRequest.budget_credits (ge=0) offers callers. + raise ValueError("budget_credits cannot be negative") async with self._session_factory()() as session: async with session.begin(): @@ -472,11 +476,12 @@ async def create_child_wallet( ) -> WalletResponse: """Spawn a child sub-agent wallet from a parent agent's balance.""" _validate_child_wallet_ttl(ttl_seconds) - if budget_credits <= Decimal("0"): + if budget_credits < Decimal("0"): # Same inversion as the sponsor path, and worse: a child opened # with a negative balance cannot be unwound by a later reclaim, # which only credits the parent back a positive reclaim_amount. - raise ValueError("budget_credits must be positive") + # Zero is allowed here too, for the same reason. + raise ValueError("budget_credits cannot be negative") async with self._session_factory()() as session: async with session.begin(): # Lock parent wallet diff --git a/tests/test_wallet_concurrency.py b/tests/test_wallet_concurrency.py index 3844bec0..ee932fea 100644 --- a/tests/test_wallet_concurrency.py +++ b/tests/test_wallet_concurrency.py @@ -420,22 +420,27 @@ async def test_zero_lifetime_cap_refuses_a_transfer_as_a_cap( assert await _balance(dest.wallet_id) == Decimal("100") -@pytest.mark.parametrize("budget", [Decimal("0"), Decimal("-50")]) -async def test_non_positive_provisioning_budget_is_refused( - sponsor_wallet: str, budget: Decimal +async def test_negative_provisioning_budget_is_refused( + sponsor_wallet: str, ) -> None: - """A non-positive budget must be refused before any wallet is touched. - - The request schemas already require ``gt=0``, so this is unreachable over - HTTP -- but the guard it would otherwise meet is ``balance >= budget``, - which a negative budget satisfies trivially. The debit would then run as a - credit: the sponsor gains, and the new wallet opens holding a negative - balance that no later reclaim can unwind. + """A negative budget must be refused before any wallet is touched. + + The wallet-creation schemas require ``gt=0``, so this is unreachable over + those routes -- but the guard it would otherwise meet is + ``balance >= budget``, which a negative budget satisfies trivially. The + debit would then run as a credit: the sponsor gains, and the new wallet + opens holding a negative balance that no later reclaim can unwind. + + Zero is NOT refused, and must not be: ``SelfProvisionRequest`` accepts + ``budget_credits=0`` (``ge=0``) to open an empty wallet for later top-up, + and refusing it there would 500 the request after its sponsor wallet had + already been created. That is asserted separately below. """ + budget = Decimal("-50") money = get_agent_money() before = await _balance(sponsor_wallet) - with pytest.raises(ValueError, match="positive"): + with pytest.raises(ValueError, match="negative"): await money.create_agent_wallet( sponsor_wallet_id=sponsor_wallet, agent_id=f"agent-{uuid.uuid4().hex[:8]}", @@ -449,7 +454,7 @@ async def test_non_positive_provisioning_budget_is_refused( agent_id=f"parent-{uuid.uuid4().hex[:8]}", budget_credits=Decimal("200"), ) - with pytest.raises(ValueError, match="positive"): + with pytest.raises(ValueError, match="negative"): await money.create_child_wallet( parent_wallet_id=parent.wallet_id, child_agent_id=f"child-{uuid.uuid4().hex[:8]}", @@ -458,3 +463,26 @@ async def test_non_positive_provisioning_budget_is_refused( ) assert await _balance(parent.wallet_id) == Decimal("200") + + +async def test_zero_provisioning_budget_opens_an_empty_wallet( + sponsor_wallet: str, +) -> None: + """Zero is a legal budget: it debits nothing and opens an empty wallet. + + ``SelfProvisionRequest.budget_credits`` is declared ``ge=0``, so a + zero-budget provision is a supported request. Refusing it would raise + after the caller's sponsor wallet had already been committed, leaving an + orphan behind and answering 500 to a well-formed request. + """ + money = get_agent_money() + before = await _balance(sponsor_wallet) + + agent = await money.create_agent_wallet( + sponsor_wallet_id=sponsor_wallet, + agent_id=f"agent-{uuid.uuid4().hex[:8]}", + budget_credits=Decimal("0"), + ) + + assert await _balance(agent.wallet_id) == Decimal("0") + assert await _balance(sponsor_wallet) == before