diff --git a/app/services/idempotency.py b/app/services/idempotency.py index 369c33ab..ff1252b8 100644 --- a/app/services/idempotency.py +++ b/app/services/idempotency.py @@ -18,6 +18,7 @@ from app.db.database import get_session_factory from app.db.models import ( IdempotencyRecordModel, + LedgerEntryModel, McpDispatchAttemptModel, ReceiptModel, ) @@ -504,6 +505,101 @@ async def reconcile_stuck_records( await session.delete(record) repaired += 1 + # Local governed calls fall through both passes above: the + # unstarted pass is scoped to operation_kind="upstream_mcp", + # and the stuck pass below requires a linked ledger_entry_id. + # A crash between the committed debit and the write that + # attaches that debit to the record therefore left an + # unresolvable row -- response_json IS NULL and + # ledger_entry_id IS NULL -- that no branch selected, so it + # never repaired and never counted toward needs_review. The + # charge carries operation_key=record_id (see the governed + # money.charge call in app/routers/mcp.py), so the orphaned + # debit is recoverable by that key: link it here and the stuck + # pass below resolves it exactly like any other crashed + # finalization. Engine-independent -- no row-lock behaviour + # involved. + orphaned_local = ( + ( + await session.execute( + select(IdempotencyRecordModel) + .where( + cast( + ColumnElement[bool], + IdempotencyRecordModel.endpoint + == GOVERNED_MCP_IDEMPOTENCY_ENDPOINT, + ), + cast( + ColumnElement[bool], + IdempotencyRecordModel.operation_kind == "local", + ), + cast( + ColumnElement[bool], + cast(Any, IdempotencyRecordModel.response_json).is_( + None + ), + ), + cast( + ColumnElement[bool], + cast( + Any, IdempotencyRecordModel.ledger_entry_id + ).is_(None), + ), + cast( + ColumnElement[bool], + IdempotencyRecordModel.created_at < cutoff, + ), + ) + .with_for_update() + ) + ) + .scalars() + .all() + ) + for record in orphaned_local: + debit = ( + await session.execute( + select(LedgerEntryModel).where( + cast( + ColumnElement[bool], + LedgerEntryModel.wallet_id == record.wallet_id, + ), + cast( + ColumnElement[bool], + LedgerEntryModel.operation_key == record.record_id, + ), + # Compensation proof means money left the + # wallet. The wallet/operation-key uniqueness + # constraint already makes a same-key credit + # impossible alongside the debit, but stating + # the direction here means this pass can never + # adopt a credit as evidence of a charge. + cast( + ColumnElement[bool], + LedgerEntryModel.amount < 0, + ), + ) + ) + ).scalar_one_or_none() + if debit is None: + # Nothing provably moved, and a local identity must NOT + # be expired on that basis: local tools run their side + # effect before the debit exists, so deleting the record + # would let a retry execute it a second time. That + # invariant is asserted by + # test_stale_local_identity_is_not_deleted_without + # _compensation_proof. Leave it exactly as found. + continue + # A committed debit is compensation proof: money moved under + # this identity. Link it, and the stuck pass below resolves + # it from the receipt or counts it for manual review. + record.ledger_entry_id = debit.entry_id + session.add(record) + # The session does not autoflush, so the links just written are + # still pending; flush them so the stuck pass below selects the + # rows this pass just repaired into scope. + await session.flush() + stuck = ( ( await session.execute( diff --git a/tests/test_mcp_trust.py b/tests/test_mcp_trust.py index dc92e652..0f5fdb3a 100644 --- a/tests/test_mcp_trust.py +++ b/tests/test_mcp_trust.py @@ -10,7 +10,12 @@ from app.core.config import get_settings from app.db.database import get_session_factory -from app.db.models import IdempotencyRecordModel, WalletModel +from app.db.models import ( + IdempotencyRecordModel, + LedgerEntryModel, + ReceiptModel, + WalletModel, +) from app.main import app from app.schemas.billing import ServiceCategory from app.services.idempotency import ( @@ -1327,3 +1332,203 @@ def broke_guard_tool() -> dict: assert wallet.lifetime_debits == lifetime_debits_before finally: registry.unregister_local("broke-guard-tool") + + +async def _simulate_crash_before_ledger_link( + wallet_id: str, idempotency_key: str, *, drop_receipt: bool +) -> str: + """Reproduce the crash window between the committed debit and the link. + + A governed local call charges with ``operation_key=record_id`` and only + afterwards writes ``ledger_entry_id`` onto the idempotency record. A crash + in that gap leaves the debit committed and the record unlinked, which is + the state this helper recreates from a call that actually completed. + Returns the record id. + """ + factory = get_session_factory() + async with factory() as session: + record = ( + await session.execute( + select(IdempotencyRecordModel).where( + IdempotencyRecordModel.wallet_id == wallet_id, + IdempotencyRecordModel.endpoint + == GOVERNED_MCP_IDEMPOTENCY_ENDPOINT, + IdempotencyRecordModel.idempotency_key == idempotency_key, + ) + ) + ).scalar_one() + record_id = record.record_id + # The debit stays committed and still carries operation_key=record_id. + record.response_json = None + record.response_reference = None + record.ledger_entry_id = None + record.created_at = datetime.now(timezone.utc) - timedelta(hours=1) + session.add(record) + if drop_receipt: + for receipt in ( + ( + await session.execute( + select(ReceiptModel).where( + ReceiptModel.idempotency_record_id == record_id + ) + ) + ) + .scalars() + .all() + ): + await session.delete(receipt) + await session.commit() + return record_id + + +async def _invoke_local_governed_tool(client, provisioned, *, suffix: str) -> str: + """Drive one successful governed local tool call; returns its idem key. + + The caller is responsible for unregistering ``crash-before-link-tool-*``; + leaving it registered leaks into unrelated tool-listing tests. + """ + registry = get_service_registry() + tool = f"crash-before-link-tool-{suffix}" + + def _tool() -> dict: + return {"ok": True} + + registry.register_local( + service_id=tool, + name="Crash Before Link Tool", + description="Governed trust reconciliation test tool", + category=ServiceCategory.AGENT_COMMS, + func=_tool, + credits_per_unit=2.0, + unit_name="call", + ) + permit = await create_tool_permit( + client, + wallet_id=provisioned["agent_wallet_id"], + key_id=provisioned["key_id"], + tool_name=tool, + idem_key=f"crash-before-link-permit-{suffix}", + ) + idempotency_key = f"crash-before-link-invoke-{suffix}" + resp = await client.post( + "/mcp/messages", + json={ + "jsonrpc": "2.0", + "id": f"crash-before-link-{suffix}", + "method": "tools/call", + "params": { + "name": tool, + "arguments": {}, + "mcpContext": { + "wallet_id": provisioned["agent_wallet_id"], + "permit_id": permit["permit_id"], + "idempotency_key": idempotency_key, + }, + }, + }, + headers=provisioned["agent_headers"], + ) + assert resp.status_code == 200 + return idempotency_key + + +@pytest.mark.anyio +async def test_reconcile_links_orphaned_local_debit_and_repairs_it( + client, clean_database +): + """A local governed crash before the ledger link must not be invisible. + + The unstarted pass is scoped to operation_kind="upstream_mcp" and the + stuck pass requires ledger_entry_id IS NOT NULL, so a local record with a + committed debit and no link matched neither: it stuck forever and + needs_review stayed 0. The debit carries operation_key=record_id, so the + reconciler can adopt it and repair from the receipt. + """ + provisioned = await provision_agent_wallet(client) + try: + idempotency_key = await _invoke_local_governed_tool( + client, provisioned, suffix="repair" + ) + await _simulate_crash_before_ledger_link( + provisioned["agent_wallet_id"], idempotency_key, drop_receipt=False + ) + + idem = get_idempotency_service() + repaired, needs_review = await idem.reconcile_stuck_records(idle_seconds=900) + assert (repaired, needs_review) == (1, 0) + finally: + get_service_registry().unregister_local("crash-before-link-tool-repair") + + +@pytest.mark.anyio +async def test_reconcile_flags_orphaned_local_debit_without_a_receipt( + client, clean_database +): + """The same crash window with no receipt yet must be surfaced, not lost. + + This is the case that most needs an operator: real money moved and there + is no evidence row to rebuild the response from. It previously counted + zero on both sides, so nothing ever revealed it. + """ + provisioned = await provision_agent_wallet(client) + try: + idempotency_key = await _invoke_local_governed_tool( + client, provisioned, suffix="review" + ) + await _simulate_crash_before_ledger_link( + provisioned["agent_wallet_id"], idempotency_key, drop_receipt=True + ) + + idem = get_idempotency_service() + repaired, needs_review = await idem.reconcile_stuck_records(idle_seconds=900) + assert (repaired, needs_review) == (0, 1) + finally: + get_service_registry().unregister_local("crash-before-link-tool-review") + + +@pytest.mark.anyio +async def test_reconcile_refuses_a_credit_as_compensation_proof( + client, clean_database +): + """Only a debit proves the local charge happened. + + The linking pass adopts a ledger entry as evidence that money moved under + this identity. A credit carrying the same operation key proves the + opposite, so it must not be linked -- the record stays exactly as found + and reconciliation reports nothing, the same as when no entry exists. + """ + provisioned = await provision_agent_wallet(client) + try: + idempotency_key = await _invoke_local_governed_tool( + client, provisioned, suffix="credit" + ) + record_id = await _simulate_crash_before_ledger_link( + provisioned["agent_wallet_id"], idempotency_key, drop_receipt=False + ) + + factory = get_session_factory() + async with factory() as session: + entry = ( + await session.execute( + select(LedgerEntryModel).where( + LedgerEntryModel.operation_key == record_id + ) + ) + ).scalar_one() + # Same key, opposite direction: a refund-shaped row, not a charge. + entry.amount = abs(entry.amount) + entry.action = "credit" + session.add(entry) + await session.commit() + + idem = get_idempotency_service() + repaired, needs_review = await idem.reconcile_stuck_records(idle_seconds=900) + assert (repaired, needs_review) == (0, 0) + + async with factory() as session: + record = await session.get(IdempotencyRecordModel, record_id) + assert record is not None + assert record.ledger_entry_id is None + assert record.response_json is None + finally: + get_service_registry().unregister_local("crash-before-link-tool-credit")