fix(idempotency): reconcile the local governed crash window between the debit and its link

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(

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")