From 3971a981a82d4c31bb7b9348fb47b532aac24ed0 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 21:03:14 +0000 Subject: [PATCH 1/2] fix(idempotency): reconcile the local governed crash window between the debit and its link MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The engine-independent finding from #302's 'Not addressed here' list. A local governed call charges with operation_key= and only afterwards attaches that ledger entry to its idempotency record. A crash in that gap left response_json IS NULL and ledger_entry_id IS NULL, which matched neither reconciler pass: - the unstarted pass is scoped to operation_kind == 'upstream_mcp' - the stuck pass requires ledger_entry_id IS NOT NULL So the record stuck forever, replays kept raising in-progress, and needs_review stayed 0 — real money had moved and nothing surfaced it. Adds a pass that runs before the stuck pass and adopts the orphan by the key the charge already carries: if a ledger entry exists with operation_key == record_id, link it, and the existing stuck logic then repairs the record from its receipt or counts it for manual review. No new repair semantics. Deliberately NOT deleting the no-debit case. Local tools run their side effect before the debit exists, so expiring the identity would let a retry execute it a second time — the invariant asserted by test_stale_local_identity_is_not_deleted_without_compensation_proof, which caught exactly that mistake in an earlier revision of this change. Without a committed debit as compensation proof the record is left untouched. The session is created with autoflush=False, so the newly written links are flushed explicitly before the stuck pass selects them. Tests (both confirmed to fail against the unfixed source): - debit + receipt orphaned -> (repaired, needs_review) == (1, 0) - debit, receipt missing -> (repaired, needs_review) == (0, 1) Unfixed, both report (0, 0) — the record is invisible to reconciliation. Tests: 1709 passed, 32 skipped; production_trust 6 passed; ruff clean; mypy clean across 172 source files; demo and red-team trust-plane gates --assert clean. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01FsXYfTHDLiFNQ24R5JPN7H --- app/services/idempotency.py | 86 ++++++++++++++++++++ tests/test_mcp_trust.py | 154 +++++++++++++++++++++++++++++++++++- 2 files changed, 239 insertions(+), 1 deletion(-) diff --git a/app/services/idempotency.py b/app/services/idempotency.py index 369c33ab..c62c6188 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,91 @@ 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, + ), + ) + ) + ).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..a3e7e775 100644 --- a/tests/test_mcp_trust.py +++ b/tests/test_mcp_trust.py @@ -10,7 +10,7 @@ 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, ReceiptModel, WalletModel from app.main import app from app.schemas.billing import ServiceCategory from app.services.idempotency import ( @@ -1327,3 +1327,155 @@ 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") From 9abb9cf86da25698b5220120228513b9a131030b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 22:08:28 +0000 Subject: [PATCH 2/2] fix(idempotency): accept only a debit as compensation proof when linking The linking pass adopted any ledger entry carrying the record's operation_key. The wallet/operation-key uniqueness constraint already makes a same-key credit impossible alongside the debit, so this was not reachable today -- but the pass's whole premise is "money left this wallet", and that direction should be stated in the query rather than inferred from a constraint elsewhere. Adds a negative test that plants a same-key credit and asserts reconciliation reports (0, 0) and leaves ledger_entry_id and response_json untouched. Confirmed to fail against the unguarded query (reports (1, 0): the credit is adopted as proof of a charge). Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01FsXYfTHDLiFNQ24R5JPN7H --- app/services/idempotency.py | 10 +++++++ tests/test_mcp_trust.py | 55 ++++++++++++++++++++++++++++++++++++- 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/app/services/idempotency.py b/app/services/idempotency.py index c62c6188..ff1252b8 100644 --- a/app/services/idempotency.py +++ b/app/services/idempotency.py @@ -568,6 +568,16 @@ async def reconcile_stuck_records( 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() diff --git a/tests/test_mcp_trust.py b/tests/test_mcp_trust.py index a3e7e775..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, ReceiptModel, 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 ( @@ -1479,3 +1484,51 @@ async def test_reconcile_flags_orphaned_local_debit_without_a_receipt( 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")