diff --git a/app/services/wallet_engine.py b/app/services/wallet_engine.py index 9c30af5f..095015b1 100644 --- a/app/services/wallet_engine.py +++ b/app/services/wallet_engine.py @@ -13,7 +13,8 @@ from decimal import Decimal from typing import Any, cast -from sqlalchemy import select, update as sa_update +from sqlalchemy import or_, select, update as sa_update +from sqlalchemy.exc import InvalidRequestError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.sql.elements import ColumnElement @@ -152,6 +153,110 @@ 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) + ) + 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] ) -> dict[str, WalletModel]: @@ -248,6 +353,17 @@ 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. + # + # 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(): # Lock sponsor wallet for update @@ -274,17 +390,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( @@ -348,6 +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"): + # 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. + # 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 @@ -384,21 +518,41 @@ 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") - 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 is not None + 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 @@ -661,21 +815,51 @@ 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") - 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 is not None + 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 + # 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 e784622d..6415973e 100644 --- a/tests/test_permit_postgres_concurrency.py +++ b/tests/test_permit_postgres_concurrency.py @@ -1405,3 +1405,152 @@ 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)] + 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") + + +@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)]) + 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 + 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)] + 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 new file mode 100644 index 00000000..ee932fea --- /dev/null +++ b/tests/test_wallet_concurrency.py @@ -0,0 +1,488 @@ +"""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 delete as sa_delete, update as sa_update + +from app.db.database import get_engine, get_session_factory +from app.db.models import LedgerEntryModel, WalletModel +from app.schemas.billing import WalletStatus +from app.services.agent_money import ( + InsufficientFundsError, + WalletNotFoundError, + 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(InsufficientFundsError) 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(InsufficientFundsError) 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(InsufficientFundsError) 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(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("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") + + + + +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") + + +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") + + +async def test_negative_provisioning_budget_is_refused( + sponsor_wallet: str, +) -> None: + """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="negative"): + 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="negative"): + 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") + + +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