diff --git a/gateway/app/core/config.py b/gateway/app/core/config.py index 4aaa057f..8a9780ee 100644 --- a/gateway/app/core/config.py +++ b/gateway/app/core/config.py @@ -307,6 +307,33 @@ def demo_allowed_tools_list(self) -> list[str]: ENABLE_STANDARD_MCP_ENDPOINT: bool = True STANDARD_MCP_PERMIT_TTL_SECONDS: int = 120 + # Unkeyed tools/call retry safety (SEL-484 / G1, MCP issue #3394). + # A mutating call without a client idempotency key gets a deterministic + # key derived from wallet, API key, verified enterprise principal, tool and + # canonical arguments. The window slides from the first call (a retry that + # crosses a bucket edge still joins it), so a lost-response retry within + # this many seconds replays the original receipt instead of dispatching + # again; a refunded pre-dispatch failure is retried, not replayed. + # Read-only tools get no derived key, so unkeyed read-only calls are + # refused with idempotency_key_required (SEL-498); there is never a + # random per-call key. Identical deliberate unkeyed calls inside one + # window collapse to one dispatch; callers needing repeats must send + # distinct keys. 0 disables derived keys, so every unkeyed call is + # refused. The effective window is capped at the auto-permit TTL + # (STANDARD_MCP_PERMIT_TTL_SECONDS, default 120; lower for demo): a + # joined slot reuses its auto permit. With defaults the window is 120 s, + # and a later identical retry runs and bills again. + MCP_UNKEYED_RETRY_WINDOW_SECONDS: int = 300 + + @field_validator("MCP_UNKEYED_RETRY_WINDOW_SECONDS") + @classmethod + def _validate_unkeyed_retry_window(cls, value: int) -> int: + if value < 0: + raise ValueError( + "MCP_UNKEYED_RETRY_WINDOW_SECONDS must be >= 0 (0 disables)" + ) + return value + # --- Public MCP endpoint (POST /mcp/public) --- # Unauthenticated, read-only Streamable HTTP surface for MCP clients # that cannot hold an API key (public plugin runtimes such as ChatGPT diff --git a/gateway/app/core/reason_codes.py b/gateway/app/core/reason_codes.py index d36678a9..687f70cd 100644 --- a/gateway/app/core/reason_codes.py +++ b/gateway/app/core/reason_codes.py @@ -115,6 +115,7 @@ "idempotency_key_control_characters", "idempotency_key_not_a_string", "idempotency_key_not_utf8", + "idempotency_key_required", "idempotency_key_required_for_human_approval", "idempotency_key_reused", "idempotency_key_too_long", diff --git a/gateway/app/routers/mcp.py b/gateway/app/routers/mcp.py index 49a9727c..8589d2b1 100644 --- a/gateway/app/routers/mcp.py +++ b/gateway/app/routers/mcp.py @@ -939,6 +939,27 @@ def _verified_enterprise_principal( return parse_enterprise_token(enterprise_bearer_token) +async def _assert_replay_iga_access(*, auth: AuthContext, tool_name: str) -> None: + """Run the enterprise IGA gate before serving any stored result. + + A replay must not hand one human principal another's result (or a result + it is no longer entitled to). The check spends no use (``consume=False``): + the original call already did. + """ + try: + principal = _verified_enterprise_principal(auth.enterprise_bearer_token) + if principal is None: + return + decision = await enforce_tool_call(principal, tool_name, consume=False) + except IGAError as exc: + raise ToolPermissionDenied(exc.reason) from exc + if not decision.allowed: + raise ToolPermissionDenied( + decision.reason, + details={"reason_code": decision.reason, **decision.details}, + ) + + async def _execute_registered_tool( *, tool_name: str, @@ -1366,6 +1387,9 @@ async def _execute_registered_tool_inner( key_id=auth.key_id, replay=replay, ) + # Same IGA gate as the registered-tool replay path: a revoked + # principal must not read stored results via a removed tool. + await _assert_replay_iga_access(auth=auth, tool_name=tool_name) await _raise_replayed_error(replay) return replay.response_json reason = f"Tool not found: {tool_name}" @@ -1638,6 +1662,7 @@ async def _execute_registered_tool_inner( key_id=auth.key_id, replay=replay, ) + await _assert_replay_iga_access(auth=auth, tool_name=tool_name) await _raise_replayed_error(replay) return replay.response_json @@ -4558,19 +4583,43 @@ async def _maybe_mint_demo_permit(request: GovernedRequest) -> str | None: if registry.get_action_binding(record) is not None: raise ToolPermissionDenied("action_permit_required") - from .mcp_standard import _mint_auto_permit, _permit_error + from .mcp_standard import ( + _mint_auto_permit, + _permit_error, + _resolve_unkeyed_retry, + ) try: - permit_id = await _mint_auto_permit( - auth=request.auth, - tool_name=request.tool_name, - record=record, - client_key=request.idempotency_key, - requires_human_approval=await wallet_human_approval_required( - request.auth.wallet_id - ), - map_mcp_errors=False, - ) + requires_approval = await wallet_human_approval_required(request.auth.wallet_id) + resolved = None + if not request.idempotency_key: + # Deterministic retry slot for unkeyed calls so a retry resolves + # to the same auto permit and replays instead of dispatching + # twice (SEL-484). + resolved = await _resolve_unkeyed_retry( + auth=request.auth, + tool_name=request.tool_name, + record=record, + arguments=request.arguments, + requires_human_approval=requires_approval, + map_mcp_errors=False, + ) + if resolved is not None: + request.idempotency_key, permit_id = resolved + elif not request.idempotency_key: + # No deterministic retry slot and no client key: refuse rather + # than mint a random per-attempt key that would let a retry + # dispatch and bill twice (SEL-498 / #208). + raise ValueError("idempotency_key_required") + else: + permit_id = await _mint_auto_permit( + auth=request.auth, + tool_name=request.tool_name, + record=record, + client_key=request.idempotency_key, + requires_human_approval=requires_approval, + map_mcp_errors=False, + ) except PermitWriteContendedError: raise except PermitError as exc: @@ -4586,7 +4635,6 @@ async def _maybe_mint_demo_permit(request: GovernedRequest) -> str | None: except IdempotencyConflictError as exc: raise ValueError("idempotency_key_reused") from exc request.permit_id = permit_id - request.idempotency_key = request.idempotency_key or f"mcp-auto-{uuid.uuid4().hex}" return permit_id diff --git a/gateway/app/routers/mcp_standard.py b/gateway/app/routers/mcp_standard.py index 31f8fe1d..7d46ce06 100644 --- a/gateway/app/routers/mcp_standard.py +++ b/gateway/app/routers/mcp_standard.py @@ -61,7 +61,7 @@ import hashlib import json import logging -import uuid +import time from datetime import timedelta from decimal import Decimal, InvalidOperation from typing import Any @@ -84,6 +84,14 @@ from app.services.billing_engine import LedgerWriteContendedError from app.core.config import get_settings from app.core.time import utc_now +from app.core.oidc_iga import IGAError +from app.services.mcp_retry_key import ( + bucket_for, + derive_retry_identity, + is_no_effect_failure, + principal_binding, + slot_key, +) from app.routers.mcp import ( _MAX_JSON_NESTING_DEPTH, GovernedToolError, @@ -99,11 +107,13 @@ _loads_strict_json, _terminal_record_contended_data, _value_error_jsonrpc_code, + _verified_enterprise_principal, ) from app.schemas.trust import PermitCreateRequest from app.services.agent_money import get_agent_money from app.services.idempotency import ( IdempotencyConflictError, + IdempotencyFence, IdempotencyInProgressError, InvalidIdempotencyKeyError, get_idempotency_service, @@ -148,6 +158,20 @@ ) +_IDEMPOTENCY_KEY_REQUIRED_DATA: dict[str, Any] = { + "error": "idempotency_key_required", + "reason_code": "idempotency_key_required", + "remediation": { + "type": "retry_with_idempotency_key", + "detail": ( + "Send an Idempotency-Key header (or params._meta" + '["io.agentmiddleware/idempotency_key"]) and reuse the same key ' + "when retrying the same call." + ), + }, +} + + def _mcp_error(code: int, message: str, data: dict[str, Any] | None = None) -> McpError: return McpError(mcp_types.ErrorData(code=code, message=message, data=data)) @@ -270,6 +294,213 @@ def _client_idempotency_key( raise _mcp_error(-32602, str(exc), exc.as_error_data()) from exc +def _auto_permit_mint_key(client_key: str) -> str: + return "smcp-" + hashlib.sha256(client_key.encode("utf-8")).hexdigest()[:48] + + +_UNKEYED_RETRY_LOCK_ENDPOINT = "/mcp#unkeyed-retry-lock" +_UNKEYED_RETRY_LOCK_WAIT_SECONDS = 5.0 +# Far above the worst-case hold (a few DB round trips plus one permit mint). +_UNKEYED_RETRY_LOCK_STALE_SECONDS = 120 +# A pre-dispatch (no side effect) failure rolls the call forward to a fresh +# attempt slot; bounded so a permanently failing upstream eventually replays +# its failure instead of being re-run forever. +_UNKEYED_RETRY_MAX_ATTEMPTS = 3 + + +def _unkeyed_clock() -> float: + """Epoch seconds for window bucketing (patched in tests).""" + return time.time() + + +def _auto_permit_ttl_seconds( + *, tenant: str | None, requires_human_approval: bool +) -> int: + """TTL in seconds of the auto permit ``_mint_auto_permit`` would mint. + + An approval-gated permit must stay valid for the whole decision window + plus the normal TTL as execution margin: permit expiry is validated + before the approval is polled, so a shorter-lived permit would strand a + decision made late in the window as a terminal permit_expired denial. + Demo permits are further capped by the demo TTL limits. + """ + settings = get_settings() + ttl_seconds = settings.STANDARD_MCP_PERMIT_TTL_SECONDS + if requires_human_approval: + ttl_seconds += approval_window_seconds() + if tenant == "demo": + ttl_seconds = min(ttl_seconds, settings.DEMO_AUTO_PERMIT_TTL_SECONDS) + if settings.DEMO_MAX_PERMIT_TTL_MINUTES is not None: + ttl_seconds = min(ttl_seconds, settings.DEMO_MAX_PERMIT_TTL_MINUTES * 60) + return ttl_seconds + + +def _effective_unkeyed_retry_window( + *, tenant: str | None = None, requires_human_approval: bool = False +) -> int: + """Retry window in seconds, capped at the TTL of the auto permit minted. + + A joined slot reuses the permit minted for the first attempt, so a window + longer than that permit's lifetime would resolve a late retry to an + expired permit (a denial instead of a replay or a fresh mint). The cap is + the TTL the mint will actually use, including the demo limits. 0 disables. + """ + window = get_settings().MCP_UNKEYED_RETRY_WINDOW_SECONDS + if window <= 0: + return 0 + ttl_seconds = _auto_permit_ttl_seconds( + tenant=tenant, requires_human_approval=requires_human_approval + ) + return min(window, max(ttl_seconds, 0)) + + +async def _acquire_unkeyed_retry_lock(wallet_id: str, lock_key: str) -> str: + """Serialize slot selection for one call identity; return the lock row id. + + The lock is an uncompleted idempotency row: the unique insert is the + mutex, ``abandon`` releases it. A row left by a crashed holder is + released once it is older than the stale threshold, so a holder that + merely stalls can lose the lock: it must re-check ownership with + its record id (the mint-row insert is fenced on it) before acting and release only its own + row (``expected_record_id``). + """ + idem = get_idempotency_service() + deadline = time.monotonic() + _UNKEYED_RETRY_LOCK_WAIT_SECONDS + while True: + try: + begun = await idem.begin_with_record( + wallet_id=wallet_id, + endpoint=_UNKEYED_RETRY_LOCK_ENDPOINT, + idempotency_key=lock_key, + request_payload={"kind": "unkeyed-retry-lock"}, + ) + if begun.replay is not None: + # Never expected for a lock row (it is never completed). + raise IdempotencyInProgressError("idempotency_in_progress") + return begun.record_id + except IdempotencyInProgressError: + held = await idem.get_record( + wallet_id=wallet_id, + endpoint=_UNKEYED_RETRY_LOCK_ENDPOINT, + idempotency_key=lock_key, + ) + if held is not None and held.created_at < utc_now() - timedelta( + seconds=_UNKEYED_RETRY_LOCK_STALE_SECONDS + ): + await idem.abandon( + wallet_id=wallet_id, + endpoint=_UNKEYED_RETRY_LOCK_ENDPOINT, + idempotency_key=lock_key, + expected_record_id=held.record_id, + ) + continue + if time.monotonic() >= deadline: + raise + await anyio.sleep(0.02) + + +async def _resolve_unkeyed_retry( + *, + auth: AuthContext, + tool_name: str, + record: dict[str, Any], + arguments: dict[str, Any], + requires_human_approval: bool, + map_mcp_errors: bool = True, +) -> tuple[str, str] | None: + """Pick the retry slot for an unkeyed mutating call and mint its permit. + + Returns ``(idempotency_key, permit_id)``, or None when no slot applies and + the caller must refuse the unkeyed call (SEL-498): read-only tools (a replay would serve a + stale read), the window disabled, or an enterprise bearer that fails + verification (the governed pipeline denies that call regardless). + + Under a per-identity lock: + * a live slot from the previous window (first minted less than one window + ago) is joined, so a retry or a concurrent twin straddling the bucket + boundary resolves to the same key: the window slides from the first + call instead of resetting at fixed boundaries; + * an attempt whose stored result is a refunded pre-dispatch failure (no + side effects) rolls forward to the next attempt slot, so the retry runs + again instead of replaying that failure; + * the auto permit is minted before the lock is released, so any later + caller sees the chosen slot. + """ + window = _effective_unkeyed_retry_window( + tenant=auth.tenant, requires_human_approval=requires_human_approval + ) + wallet_id = auth.wallet_id + if window <= 0 or not wallet_id or record.get("read_only"): + return None + try: + principal = _verified_enterprise_principal(auth.enterprise_bearer_token) + except IGAError: + return None + identity = derive_retry_identity( + wallet_id=wallet_id, + key_id=auth.key_id, + principal=principal_binding(principal), + tool_name=tool_name, + arguments=arguments, + ) + idem = get_idempotency_service() + lock_key = "ulock-" + identity[:48] + try: + lock_record_id = await _acquire_unkeyed_retry_lock(wallet_id, lock_key) + except IdempotencyInProgressError as exc: + if map_mcp_errors: + raise _mcp_error(-32005, "idempotency_in_progress") from exc + raise + try: + bucket = bucket_for(_unkeyed_clock(), window) + chosen = bucket + previous = await idem.get_record( + wallet_id=wallet_id, + endpoint=_AUTO_PERMIT_ENDPOINT, + idempotency_key=_auto_permit_mint_key( + slot_key(identity, window_seconds=window, bucket=bucket - 1, attempt=0) + ), + ) + if previous is not None and previous.created_at > utc_now() - timedelta( + seconds=window + ): + chosen = bucket - 1 + key = slot_key(identity, window_seconds=window, bucket=chosen, attempt=0) + for attempt in range(_UNKEYED_RETRY_MAX_ATTEMPTS): + key = slot_key( + identity, window_seconds=window, bucket=chosen, attempt=attempt + ) + stored = await idem.get_governed_mcp_record( + wallet_id=wallet_id, idempotency_key=key + ) + if stored is None or not is_no_effect_failure(stored.response_json): + break + # The mint-row insert is fenced on this lock row: it is written only + # if we still own the lock, in the same transaction. A holder that + # stalled past the stale threshold and was taken over gets -32005 + # and never mints or dispatches. + permit_id = await _mint_auto_permit( + auth=auth, + tool_name=tool_name, + record=record, + client_key=key, + lock_fence=(_UNKEYED_RETRY_LOCK_ENDPOINT, lock_key, lock_record_id), + requires_human_approval=requires_human_approval, + map_mcp_errors=map_mcp_errors, + ) + return key, permit_id + finally: + # Shielded so a cancelled request still releases its lock row + # instead of blocking the identity until the stale threshold. + with anyio.CancelScope(shield=True): + await idem.abandon( + wallet_id=wallet_id, + endpoint=_UNKEYED_RETRY_LOCK_ENDPOINT, + idempotency_key=lock_key, + expected_record_id=lock_record_id, + ) + + async def _mint_auto_permit( *, auth: AuthContext, @@ -278,6 +509,7 @@ async def _mint_auto_permit( client_key: str | None, requires_human_approval: bool = False, map_mcp_errors: bool = True, + lock_fence: IdempotencyFence | None = None, ) -> str: """Mint the bounded single-tool permit backing this call. @@ -290,13 +522,9 @@ async def _mint_auto_permit( if not wallet_id: # narrowed by the caller; fail closed regardless raise _mcp_error(-32003, "wallet_scoped_key_required") settings = get_settings() - # An approval-gated permit must stay valid for the whole decision window - # plus the normal TTL as execution margin: permit expiry is validated - # before the approval is polled, so a shorter-lived permit would strand a - # decision made late in the window as a terminal permit_expired denial. - ttl_seconds = settings.STANDARD_MCP_PERMIT_TTL_SECONDS - if requires_human_approval: - ttl_seconds += approval_window_seconds() + ttl_seconds = _auto_permit_ttl_seconds( + tenant=auth.tenant, requires_human_approval=requires_human_approval + ) budget = _tool_call_budget(record) if auth.tenant == "demo": demo_tools = settings.demo_allowed_tools_list @@ -322,9 +550,6 @@ async def _mint_auto_permit( if map_mcp_errors: raise _permit_error(error, tool_name=tool_name, record=record) raise error - ttl_seconds = min(ttl_seconds, settings.DEMO_AUTO_PERMIT_TTL_SECONDS) - if settings.DEMO_MAX_PERMIT_TTL_MINUTES is not None: - ttl_seconds = min(ttl_seconds, settings.DEMO_MAX_PERMIT_TTL_MINUTES * 60) permit_request = PermitCreateRequest( issuer_wallet_id=wallet_id, subject_wallet_id=wallet_id, @@ -347,7 +572,7 @@ async def _mint_auto_permit( return permit.permit_id idem = get_idempotency_service() - mint_key = "smcp-" + hashlib.sha256(client_key.encode("utf-8")).hexdigest()[:48] + mint_key = _auto_permit_mint_key(client_key) mint_payload: dict[str, Any] = { "kind": "standard-mcp-auto-permit", "tool": tool_name, @@ -360,11 +585,12 @@ async def _mint_auto_permit( if requires_human_approval: mint_payload["requires_human_approval"] = True try: - mint_replay = await idem.begin( + mint_begun = await idem.begin_with_record( wallet_id=wallet_id, endpoint=_AUTO_PERMIT_ENDPOINT, idempotency_key=mint_key, request_payload=mint_payload, + fence=lock_fence, ) except IdempotencyConflictError as exc: if map_mcp_errors: @@ -374,20 +600,26 @@ async def _mint_auto_permit( if map_mcp_errors: raise _mcp_error(-32005, "idempotency_in_progress") from exc raise + mint_replay = mint_begun.replay if mint_replay and mint_replay.response_json: return mint_replay.response_json["permit_id"] + # Any mint failure (PermitError, PermitWriteContendedError, another + # exception, or cancellation) releases only this attempt's row, so a + # retry can mint again instead of waiting on an orphaned in-progress row. try: permit = await get_permit_service().create_permit( permit_request, subject_key_id=auth.key_id, auth=auth ) - except PermitError as exc: - await idem.abandon( - wallet_id=wallet_id, - endpoint=_AUTO_PERMIT_ENDPOINT, - idempotency_key=mint_key, - ) - if map_mcp_errors: + except BaseException as exc: + with anyio.CancelScope(shield=True): + await idem.abandon( + wallet_id=wallet_id, + endpoint=_AUTO_PERMIT_ENDPOINT, + idempotency_key=mint_key, + expected_record_id=mint_begun.record_id, + ) + if isinstance(exc, PermitError) and map_mcp_errors: raise _permit_error(exc, tool_name=tool_name, record=record) from exc raise await idem.complete( @@ -498,14 +730,38 @@ async def _governed_tools_call( ) auto_minted = permit_id is None - idempotency_key = client_idempotency_key or f"mcp-auto-{uuid.uuid4().hex}" - permit_id = permit_id or await _mint_auto_permit( - auth=auth, - tool_name=tool_name, - record=record, - client_key=client_idempotency_key, - requires_human_approval=requires_human_approval, - ) + # Unkeyed mutating calls get a deterministic retry slot so a lost-response + # retry resolves to the same auto permit and replays the original receipt + # (SEL-484). Keyed calls are untouched. + resolved = None + if client_idempotency_key is None and permit_id is None: + resolved = await _resolve_unkeyed_retry( + auth=auth, + tool_name=tool_name, + record=record, + arguments=arguments, + requires_human_approval=requires_human_approval, + ) + if resolved is not None: + idempotency_key, permit_id = resolved + else: + if not client_idempotency_key: + # No client key and no deterministic retry slot (read-only tool, + # window 0, or an unverifiable enterprise bearer). A random + # per-attempt key would let a lost-response retry dispatch and + # bill twice, so refuse before any permit, dispatch, or charge + # (SEL-498 / #208). + raise _mcp_error( + -32003, "idempotency_key_required", _IDEMPOTENCY_KEY_REQUIRED_DATA + ) + idempotency_key = client_idempotency_key + permit_id = permit_id or await _mint_auto_permit( + auth=auth, + tool_name=tool_name, + record=record, + client_key=client_idempotency_key, + requires_human_approval=requires_human_approval, + ) call_params = { "name": tool_name, diff --git a/gateway/app/services/dogfood_tool.py b/gateway/app/services/dogfood_tool.py index a240a0c5..87ecedf8 100644 --- a/gateway/app/services/dogfood_tool.py +++ b/gateway/app/services/dogfood_tool.py @@ -161,6 +161,7 @@ def register_dogfood_second_tool() -> None: credits_per_unit=1.0, unit_name="call", require_permit=True, + read_only=True, ) _second_registered = True logger.info("Registered second dogfood MCP tool: %s", DOGFOOD_SECOND_TOOL_ID) diff --git a/gateway/app/services/idempotency.py b/gateway/app/services/idempotency.py index 5ad85ae6..40595936 100644 --- a/gateway/app/services/idempotency.py +++ b/gateway/app/services/idempotency.py @@ -10,7 +10,7 @@ from datetime import timedelta from typing import Any, cast -from sqlalchemy import and_, or_, select +from sqlalchemy import and_, or_, select, update from sqlalchemy.exc import IntegrityError, OperationalError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql.elements import ColumnElement @@ -124,6 +124,17 @@ class IdempotencyInProgressError(RuntimeError): """Raised when an idempotency key is already executing without a result.""" +class IdempotencyFenceLostError(IdempotencyInProgressError): + """The fencing record named by ``fence`` is gone or now belongs to another + holder, so the new record was not written. Callers treat it as in progress: + the current holder owns the work.""" + + +# (endpoint, idempotency_key, record_id) of a lock row the writer must still +# own, in the same wallet, for a new record to be inserted. +IdempotencyFence = tuple[str, str, str] + + class IdempotencyCreationDisabledError(RuntimeError): """Raised when only existing requests may be replayed.""" @@ -488,7 +499,20 @@ async def begin_with_record( wait_timeout_seconds: float = 0.0, poll_interval_seconds: float = 0.05, allow_new_record: bool = True, + fence: IdempotencyFence | None = None, ) -> IdempotencyBegin: + """Claim (or replay) one idempotency record. + + ``fence`` makes creating a NEW record conditional on the caller still + owning a lock row: inside the same transaction as the insert, the + fence row is claimed with a conditional no-op ``UPDATE`` that matches + only our open lock row (rowcount 1). The write takes the PostgreSQL + row lock and the SQLite database write lock until commit, so a + concurrent stale-lock takeover cannot delete it in between on either + backend; SQLite ignores ``FOR UPDATE``, so it is not relied on. A missing or + re-issued fence row raises :class:`IdempotencyFenceLostError` and + nothing is written. Replaying an existing record is not fenced. + """ if wait_timeout_seconds < 0 or poll_interval_seconds <= 0: raise ValueError("idempotency_wait_invalid") request_hash = sha256_hex(request_payload) @@ -525,6 +549,41 @@ async def begin_with_record( if not allow_new_record: raise IdempotencyCreationDisabledError("idempotency_creation_disabled") + if fence is not None: + fence_endpoint, fence_key, fence_record_id = fence + # Claim the fence with a conditional write, not SELECT ... + # FOR UPDATE (SQLite ignores FOR UPDATE). The no-op UPDATE + # matches only while our lock row is still ours and open; it + # takes the PostgreSQL row lock and the SQLite write lock in + # this transaction, so a stale-lock takeover (DELETE) cannot + # land before this insert commits. rowcount 0 => lost. + try: + claimed = await session.execute( + update(IdempotencyRecordModel) + .where( + *_idempotency_predicates( + wallet_id, fence_endpoint, fence_key + ), + col(IdempotencyRecordModel.record_id) == fence_record_id, + col(IdempotencyRecordModel.response_json).is_(None), + ) + .values(request_hash=IdempotencyRecordModel.request_hash) + .execution_options(synchronize_session=False) + ) + except OperationalError as exc: + # SQLite lock timeout while another writer (e.g. a + # takeover) holds the database: the current holder owns + # the work, so this is retryable in-progress, not a 500. + if "database is locked" not in str(exc): + raise + await session.rollback() + raise IdempotencyInProgressError( + "idempotency_in_progress" + ) from None + if cast(Any, claimed).rowcount != 1: + await session.rollback() + raise IdempotencyFenceLostError("idempotency_in_progress") + record = IdempotencyRecordModel( record_id=f"idm-{uuid.uuid4().hex[:16]}", wallet_id=wallet_id, @@ -620,6 +679,7 @@ async def begin( request_payload: dict[str, Any], allow_new_record: bool = True, compatible_request_payload: dict[str, Any] | None = None, + fence: IdempotencyFence | None = None, ) -> IdempotencyReplay | None: """Replay an explicitly supplied historical shape without writing that shape.""" begun = await self.begin_with_compatible_record( @@ -629,6 +689,7 @@ async def begin( request_payload=request_payload, allow_new_record=allow_new_record, compatible_request_payload=compatible_request_payload, + fence=fence, ) return begun.replay @@ -641,6 +702,7 @@ async def begin_with_compatible_record( request_payload: dict[str, Any], allow_new_record: bool = True, compatible_request_payload: dict[str, Any] | None = None, + fence: IdempotencyFence | None = None, ) -> IdempotencyBegin: """Keep the granted identity while permitting historical-shape replay.""" try: @@ -650,6 +712,7 @@ async def begin_with_compatible_record( idempotency_key=idempotency_key, request_payload=request_payload, allow_new_record=allow_new_record, + fence=fence, ) except IdempotencyConflictError as conflict: if compatible_request_payload is None: diff --git a/gateway/app/services/mcp_phase9_tools.py b/gateway/app/services/mcp_phase9_tools.py index 7d9b58dd..33e9aab9 100644 --- a/gateway/app/services/mcp_phase9_tools.py +++ b/gateway/app/services/mcp_phase9_tools.py @@ -339,6 +339,7 @@ def register_phase9_tools() -> None: credits_per_unit=tool["credits_per_unit"], unit_name=tool["unit_name"], require_permit=bool(tool.get("require_permit", False)), + read_only=bool(tool.get("read_only", False)), ) logger.info(f"Registered Phase 9 MCP tool: {tool['service_id']}") diff --git a/gateway/app/services/mcp_retry_key.py b/gateway/app/services/mcp_retry_key.py new file mode 100644 index 00000000..10d24589 --- /dev/null +++ b/gateway/app/services/mcp_retry_key.py @@ -0,0 +1,109 @@ +"""Deterministic idempotency keys for unkeyed MCP ``tools/call`` requests. + +Without a client key, the gateway used to mint ``mcp-auto-`` per +attempt, so a retry after a lost response got a new key and a new auto +permit and the tool ran twice (SEL-484; MCP issue #3394: "generating a fresh +key on every invocation does not correlate separate retry attempts"). + +The key binds the caller (wallet, API key, and the verified enterprise/IGA +principal when one rode in), the tool, and the canonical JSON of the +arguments. The JSON-RPC request id is deliberately excluded: retries arrive +with a fresh id. Keyed calls never reach this module. + +The time dimension is a *slot*: ``floor(now / window)`` plus an attempt +counter. Which slot a call joins is decided by +``app.routers.mcp_standard._resolve_unkeyed_retry`` under a per-identity +lock, so the previous window's live slot is reused across a boundary +(sliding window from the first call) and an attempt that provably failed +before dispatch rolls forward to a fresh attempt instead of replaying. +""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +DERIVED_KEY_PREFIX = "mcp-retry-" + + +def canonical_arguments(arguments: Any) -> str: + """Stable JSON for arguments: sorted keys, no whitespace variance.""" + return json.dumps( + arguments if arguments is not None else {}, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + default=str, + ) + + +def principal_binding(principal: Any) -> str: + """Stable string for a verified enterprise principal ('' when absent).""" + if principal is None: + return "" + return "\x1e".join( + [ + str(getattr(principal, "provider", "")), + str(getattr(principal, "issuer", "")), + str(getattr(principal, "subject", "")), + ] + ) + + +def derive_retry_identity( + *, + wallet_id: str, + key_id: str | None, + principal: str, + tool_name: str, + arguments: Any, +) -> str: + """Time-independent identity of one logical unkeyed call.""" + material = "\x1f".join( + [ + "v2", + wallet_id, + key_id or "", + principal, + tool_name, + canonical_arguments(arguments), + ] + ) + return hashlib.sha256(material.encode("utf-8")).hexdigest() + + +def slot_key(identity: str, *, window_seconds: int, bucket: int, attempt: int) -> str: + """Idempotency key for one (window bucket, attempt) slot of an identity.""" + material = f"{identity}\x1f{window_seconds}\x1f{bucket}\x1f{attempt}" + return DERIVED_KEY_PREFIX + hashlib.sha256(material.encode("utf-8")).hexdigest() + + +def bucket_for(now: float, window_seconds: int) -> int: + return int(now // window_seconds) + + +def is_no_effect_failure(response_json: Any) -> bool: + """True when a stored governed result is a refunded pre-dispatch failure. + + Only the pre-dispatch path stores ``upstream_result`` as the gateway's own + ``{"error": "failed_refunded", "error_code": ...}`` terminal payload with a + ``failed_refunded`` receipt: nothing reached the upstream, so a retry must + run again rather than replay the failure. + """ + if isinstance(response_json, str): + try: + response_json = json.loads(response_json) + except ValueError: + return False + if not isinstance(response_json, dict): + return False + upstream = response_json.get("upstream_result") + receipt = response_json.get("receipt") + return ( + isinstance(upstream, dict) + and upstream.get("error") == "failed_refunded" + and "error_code" in upstream + and isinstance(receipt, dict) + and receipt.get("outcome") == "failed_refunded" + ) diff --git a/gateway/app/services/service_registry.py b/gateway/app/services/service_registry.py index 84ebe56a..e64254fc 100644 --- a/gateway/app/services/service_registry.py +++ b/gateway/app/services/service_registry.py @@ -257,6 +257,7 @@ def register_local( unit_name: str = "call", owner_wallet_id: str | None = None, require_permit: bool = False, + read_only: bool = False, ) -> dict: """ Register a service locally (in-memory) for immediate use. @@ -267,6 +268,13 @@ def register_local( When ``require_permit`` is True, governed MCP invocation always requires a signed permit + idempotency key — even if legacy unpermitted MCP is otherwise allowed. + + ``read_only`` declares the tool free of side effects. It gets no + deterministic retry key (a replay would serve a stale read), and it is + still metered, so an unkeyed standard ``/mcp`` call to it is refused + with ``idempotency_key_required`` (SEL-498). The + default is False: an undeclared tool is treated as mutating, which + fails safe (deduped, never double-dispatched). """ input_schema = pydantic_to_mcp_schema(input_model) output_schema = pydantic_to_mcp_schema(output_model) @@ -294,6 +302,7 @@ def register_local( "output_schema": output_schema, "owner_wallet_id": owner_wallet_id, "require_permit": bool(require_permit), + "read_only": bool(read_only), "is_active": True, "is_local": True, "external": False, @@ -324,8 +333,15 @@ def register_upstream( upstream_origin: str, credits_per_unit_exact: str | None = None, action_binding: ActionToolBinding | None = None, + read_only: bool = False, ) -> dict[str, Any]: - """Register one runtime-backed remote MCP tool without persisting secrets.""" + """Register one runtime-backed remote MCP tool without persisting secrets. + + ``read_only`` is an operator declaration (never taken from the + upstream's own ``readOnlyHint``, which a remote server could set to + dodge retry dedupe). It exempts the tool from the deterministic + unkeyed retry key; the default treats the tool as mutating. + """ existing = self._local_registry.get(service_id) if existing and existing.get("execution_backend") != "upstream_mcp": raise ValueError(f"Service ID already registered: {service_id}") @@ -342,6 +358,7 @@ def register_upstream( "output_schema": output_schema, "owner_wallet_id": None, "require_permit": True, + "read_only": bool(read_only), "is_active": True, "is_local": False, "external": True, diff --git a/gateway/docs/concepts/request-keys.md b/gateway/docs/concepts/request-keys.md index ec688965..53aa4f55 100644 --- a/gateway/docs/concepts/request-keys.md +++ b/gateway/docs/concepts/request-keys.md @@ -28,10 +28,24 @@ earlier during auto-permit minting with JSON-RPC code `-32003` and the same message. Billing charge uses HTTP 409 with `detail.error` set to the code; permit creation uses HTTP 409 with string `detail`. -Standard `/mcp` makes a fresh `mcp-auto-` key for each call when the caller -supplies none. That fresh key cannot identify a retry, so an unkeyed call is -not retry-safe. The legacy `/mcp/messages` demo auto-permit path also assigns -a fresh `mcp-auto-` key when none is supplied. The SDK's `invoke_tool` +When the caller supplies no key to a mutating tool, standard `/mcp` and the +legacy `/mcp/messages` demo auto-permit path derive an `mcp-retry-` key from +wallet, API key, verified enterprise principal, tool, and canonical arguments. +An identical retry within the effective retry window of the first call +replays its receipt. The effective window is the smaller of +`MCP_UNKEYED_RETRY_WINDOW_SECONDS` (default 300) and the auto-permit TTL +(`STANDARD_MCP_PERMIT_TTL_SECONDS`, default 120; demo tenants are capped +lower by `DEMO_AUTO_PERMIT_TTL_SECONDS`), so with default settings it is +120 seconds. A retry after the effective window is a new call: it runs and +bills again. Send your own key if retries can arrive later than that; the window slides from that first call, +so a retry across a clock bucket edge still replays. If the stored result is +a refunded pre-dispatch failure (nothing reached the tool), the retry runs +again instead. Tools registered `read_only` get no derived key. +Identical deliberate calls inside one window collapse into one, so send your +own key when that matters. The gateway never mints a random per-call key: +an unkeyed call that gets no derived key (a `read_only` tool, a window of +`0`, or an enterprise bearer that fails verification) is refused with +`-32003 idempotency_key_required` before any permit, dispatch, or charge. The SDK's `invoke_tool` requires a key; its `"auto"` option derives one from arguments and is unsuitable when distinct calls can have the same arguments. diff --git a/gateway/docs/denial-details.md b/gateway/docs/denial-details.md index 5f77664a..0b459de9 100644 --- a/gateway/docs/denial-details.md +++ b/gateway/docs/denial-details.md @@ -122,6 +122,7 @@ marked API-only occur before a signed denial receipt can be created. | `key_tool_not_allowed` | The API key's own tool allowlist excludes the tool. | Use a key that allows the tool. | | `demo_tool_not_allowed` | Demo-tenant wallets may call only the demo allowlist. | Use an allowed demo tool or a non-demo wallet. | | `demo_auto_permit_budget_exceeded` | The tool costs more than the demo auto-permit credit cap. | Use a cheaper tool or a non-demo wallet. | +| `idempotency_key_required` | An unkeyed MCP `tools/call` got no derived retry key (a `read_only` tool, `MCP_UNKEYED_RETRY_WINDOW_SECONDS=0`, or an enterprise bearer that fails verification). Refused before any permit, dispatch, or charge instead of minting a random key. | Send an `Idempotency-Key` and reuse it when retrying the same call. | | `idempotency_key_required_for_human_approval` | The wallet policy requires approval, and without a client idempotency key each retry would page a human again. | Send an `Idempotency-Key` and retry the same call with the same key to poll the decision. | | `idempotency_key_not_a_string` | API-only: a supplied request key is not a string. | Send a string key. | | `idempotency_key_blank` | API-only: the key is empty or only whitespace. | Send a non-blank key. | diff --git a/gateway/docs/failure-semantics.md b/gateway/docs/failure-semantics.md index 5a366e39..46ae34cd 100644 --- a/gateway/docs/failure-semantics.md +++ b/gateway/docs/failure-semantics.md @@ -244,11 +244,46 @@ asked for: every retry carrying the same malformed key would become a new charged operation (`test_mcp_idempotency_key_validation`). Only a genuinely **absent** key — no header, no body entry, or a JSON `null` -entry — is treated as an un-keyed call. On the standard MCP endpoint that means -a generated key, and every un-keyed call, retries included, is a new charged -operation. Clients that retry must always send their own persistent key; a -governed `/mcp/messages` call without one is refused with -`idempotency_key_required`. +entry — is treated as an un-keyed call. On the standard MCP endpoint (and the +`/mcp/messages` demo auto-permit path) an un-keyed call to a mutating tool +gets a deterministic `mcp-retry-` key from the caller's wallet and API key, +the verified enterprise principal (provider, issuer, subject) when an IGA +bearer is present, the tool, and the canonical JSON of the arguments. The +JSON-RPC id is not part of it, because retries carry fresh ids. + +- **Sliding window.** Slot selection and the auto-permit mint run under a + per-identity lock row. A slot first minted less than + the effective window ago is joined (the smaller of + `MCP_UNKEYED_RETRY_WINDOW_SECONDS`, default 300, and the auto-permit TTL, + default 120 and lower for demo tenants; so 120 s by default, after which an + identical unkeyed retry runs and bills again) even if the + clock bucket has rolled over, so a retry or concurrent twin straddling the + edge resolves to one dispatch. A concurrent duplicate either replays the + same receipt or gets the retryable `idempotency_in_progress`. A lock older + than 120 s is treated as abandoned and can be taken over. The auto-permit + mint row is written only if the writer still owns the lock (checked in the + same transaction as the insert), so a stalled former holder never mints or + dispatches, and it releases only its own lock row. +- **IGA before replay.** A replay of a stored governed result re-runs the IGA + check (without spending a use) first, and the principal is part of the key, + so one human never receives another's stored result. +- **Mutating tools only.** Tools registered `read_only` (for example + `partner.notes.count`) get no derived key, so an unkeyed call to one is + refused (below). Undeclared tools are treated as mutating. +- **No-effect failures retry.** If the stored result is a refunded + pre-dispatch failure, the retry moves to the next attempt slot and runs + again (up to three attempts per window) instead of replaying the failure. + +Two deliberately identical un-keyed calls inside one window still collapse +into one; clients that need repeats must send their own key. The gateway never mints a +random per-call key: an unkeyed call with no derived key (a `read_only` tool, +the window set to `0`, or an enterprise bearer that fails verification) is +refused with JSON-RPC `-32003` `idempotency_key_required` (HTTP 400 on +`/mcp/tools/{service_id}/invoke`) before any permit, dispatch, or charge, so +a retry cannot run or bill twice. A +governed `/mcp/messages` call without a key is still refused with +`idempotency_key_required`, and a wallet that requires human approval still +needs an explicit key. The governed AWI HTTP routes — `POST /v1/awi/execute`, `/v1/awi/passkey/challenge`, `/v1/awi/passkey/verify`, `/v1/awi/dom/sync`, diff --git a/gateway/tests/test_mcp_idempotency_key_validation.py b/gateway/tests/test_mcp_idempotency_key_validation.py index 728ef389..a184cf77 100644 --- a/gateway/tests/test_mcp_idempotency_key_validation.py +++ b/gateway/tests/test_mcp_idempotency_key_validation.py @@ -384,14 +384,14 @@ async def test_standard_distinct_valid_keys_stay_distinct_operations( @pytest.mark.parametrize("meta", [_ABSENT, None], ids=["no-entry", "json-null"]) -async def test_standard_unkeyed_calls_are_documented_as_new_operations( +async def test_standard_unkeyed_retry_is_one_operation( client, standard_mcp_enabled, clean_database, counted_tool, meta ): - """No key anywhere (or a JSON null entry) means each call is a new charge. + """No key anywhere (or a JSON null entry) gets a derived retry key. - This pins the documented contract for un-keyed clients so it cannot drift - silently in either direction: absent must not be refused like malformed, - and it must not be quietly deduplicated either. + SEL-484: an identical un-keyed call inside the retry window replays the + first receipt instead of charging, dispatching, or minting again. Absent + must still not be refused like malformed. """ provisioned = await provision_agent_wallet(client) headers = {**provisioned["agent_headers"], **MCP_HEADERS} @@ -403,11 +403,11 @@ async def test_standard_unkeyed_calls_are_documented_as_new_operations( ]["receipt_id"] for _ in range(2) ] - assert len(set(receipts)) == 2 + assert len(set(receipts)) == 1 snapshot = await _snapshot(client, provisioned, counted_tool) - assert snapshot["effects"] == 2 - assert snapshot["ledger_entries"] == base["ledger_entries"] + 2 - assert snapshot["permits"] == base["permits"] + 2 + assert snapshot["effects"] == 1 + assert snapshot["ledger_entries"] == base["ledger_entries"] + 1 + assert snapshot["permits"] == base["permits"] + 1 # --------------------------------------------------------------------------- # diff --git a/gateway/tests/test_mcp_replay_iga.py b/gateway/tests/test_mcp_replay_iga.py new file mode 100644 index 00000000..937897d4 --- /dev/null +++ b/gateway/tests/test_mcp_replay_iga.py @@ -0,0 +1,59 @@ +# ruff: noqa: F811 (pytest fixtures imported from test_iga_policy) +"""SEL-484 / Codex blocking 2: the IGA gate runs before any stored-result replay. + +Reuses the signing-key, issuer and bundle fixtures from test_iga_policy; no +recorded data. +""" + +from __future__ import annotations + +import pytest + +from app.core.auth import AuthContext +from app.routers.mcp import ToolPermissionDenied, _assert_replay_iga_access +from tests.test_iga_policy import ( # noqa: F401 (fixtures) + TOOL, + _make_bundle, + _make_wallet, + _mint, + _okta_issuers, + _reset_iga_counters, + iga_config, + rsa_key, +) + + +def _auth(wallet_id: str, token: str | None) -> AuthContext: + return AuthContext( + source="api_key", + raw_key="", + key_id="k", + wallet_id=wallet_id, + enterprise_bearer_token=token, + ) + + +async def test_replay_denied_for_principal_without_role( + iga_config, clean_database, rsa_key +): + wallet_id = await _make_wallet() + policy_id = await _make_bundle(wallet_id, allowed_tools=[TOOL]) + iga_config(_okta_issuers(rsa_key), {"payments-ops": {"policy_id": policy_id}}) + + allowed = _mint(rsa_key, extra={"groups": ["payments-ops"]}) + await _assert_replay_iga_access(auth=_auth(wallet_id, allowed), tool_name=TOOL) + + intruder = _mint(rsa_key, sub="intruder", extra={"groups": ["random-team"]}) + with pytest.raises(ToolPermissionDenied): + await _assert_replay_iga_access(auth=_auth(wallet_id, intruder), tool_name=TOOL) + + +async def test_replay_denied_for_unverifiable_bearer(iga_config, rsa_key): + iga_config(_okta_issuers(rsa_key), {}) + expired = _mint(rsa_key, exp_delta=-60) + with pytest.raises(ToolPermissionDenied): + await _assert_replay_iga_access(auth=_auth("w", expired), tool_name=TOOL) + + +async def test_replay_without_enterprise_bearer_is_unchanged(): + await _assert_replay_iga_access(auth=_auth("w", None), tool_name=TOOL) diff --git a/gateway/tests/test_mcp_retry_key.py b/gateway/tests/test_mcp_retry_key.py new file mode 100644 index 00000000..e0b3ad33 --- /dev/null +++ b/gateway/tests/test_mcp_retry_key.py @@ -0,0 +1,88 @@ +"""Data-free unit tests for deterministic unkeyed MCP retry keys (SEL-484).""" + +import json +from uuid import uuid4 + +from app.services.mcp_retry_key import ( + DERIVED_KEY_PREFIX, + bucket_for, + derive_retry_identity, + is_no_effect_failure, + principal_binding, + slot_key, +) + +BASE = dict( + wallet_id=f"wallet-{uuid4()}", + key_id=f"key-{uuid4()}", + principal="", + tool_name=f"tool-{uuid4()}", +) + + +def _identity(**overrides): + return derive_retry_identity(**{**BASE, "arguments": {"a": 1, "b": 2}, **overrides}) + + +def _slot(identity=None, bucket=3, attempt=0, window=300): + return slot_key( + identity or _identity(), window_seconds=window, bucket=bucket, attempt=attempt + ) + + +def test_deterministic_and_prefixed(): + assert _slot() == _slot() + assert _slot().startswith(DERIVED_KEY_PREFIX) + + +def test_argument_order_and_empty_arguments(): + assert _identity(arguments={"b": 2, "a": 1}) == _identity() + assert _identity(arguments=None) == _identity(arguments={}) + + +def test_each_binding_changes_identity(): + base = _identity() + assert _identity(arguments={"a": 1, "b": 3}) != base + assert _identity(tool_name="other") != base + assert _identity(wallet_id="w2") != base + assert _identity(key_id="k2") != base + assert _identity(principal="okta\x1eiss\x1esub") != base + + +def test_principal_binding_separates_humans(): + class P: + def __init__(self, subject): + self.provider, self.issuer, self.subject = "okta", "iss", subject + + assert principal_binding(None) == "" + assert principal_binding(P("a")) == principal_binding(P("a")) + assert principal_binding(P("a")) != principal_binding(P("b")) + assert _identity(principal=principal_binding(P("a"))) != _identity( + principal=principal_binding(P("b")) + ) + + +def test_slots_differ_by_bucket_attempt_and_window(): + base = _slot() + assert _slot(bucket=4) != base + assert _slot(attempt=1) != base + assert _slot(window=60) != base + + +def test_bucket_for(): + assert bucket_for(299.999, 300) == 0 + assert bucket_for(300.001, 300) == 1 + + +def test_no_effect_failure_detection(): + refunded = { + "error": "upstream_unreachable", + "upstream_result": {"error": "failed_refunded", "error_code": "x"}, + "receipt": {"outcome": "failed_refunded"}, + } + assert is_no_effect_failure(refunded) + assert is_no_effect_failure(json.dumps(refunded)) + assert not is_no_effect_failure({"result": {"ok": True}, "receipt": {}}) + assert not is_no_effect_failure({**refunded, "receipt": {"outcome": "succeeded"}}) + assert not is_no_effect_failure("not json") + assert not is_no_effect_failure(None) diff --git a/gateway/tests/test_mcp_router_breaks.py b/gateway/tests/test_mcp_router_breaks.py index eedf303b..f49dee86 100644 --- a/gateway/tests/test_mcp_router_breaks.py +++ b/gateway/tests/test_mcp_router_breaks.py @@ -212,7 +212,13 @@ async def test_tool_call_without_session_does_not_skip_auth_or_crash( forged = await client.post( "/mcp", - json={**call, "id": 8}, + # Distinct arguments: an identical un-keyed call would replay the + # first receipt under the derived retry key (SEL-484). + json={ + **call, + "id": 8, + "params": {"name": TOOL, "arguments": {"text": "run-2"}}, + }, headers={**wallet_headers, "Mcp-Session-Id": "forged-session"}, ) assert forged.status_code < 500, forged.text @@ -220,7 +226,7 @@ async def test_tool_call_without_session_does_not_skip_auth_or_crash( forged.json()["result"]["receipt"]["wallet_id"] == (provisioned["agent_wallet_id"]) ) - assert effect_tool == [{"text": "run"}, {"text": "run"}] + assert effect_tool == [{"text": "run"}, {"text": "run-2"}] for bad_id in (None, True, [], {}): body: dict[str, Any] = { @@ -232,7 +238,7 @@ async def test_tool_call_without_session_does_not_skip_auth_or_crash( body["id"] = bad_id refused = await client.post("/mcp", json=body, headers=wallet_headers) assert refused.status_code < 500, refused.text - assert effect_tool == [{"text": "run"}, {"text": "run"}] + assert effect_tool == [{"text": "run"}, {"text": "run-2"}] @pytest.mark.anyio diff --git a/gateway/tests/test_mcp_upstream_governed.py b/gateway/tests/test_mcp_upstream_governed.py index 7aa8a573..b6d6152a 100644 --- a/gateway/tests/test_mcp_upstream_governed.py +++ b/gateway/tests/test_mcp_upstream_governed.py @@ -3279,3 +3279,28 @@ async def test_simulated_runtime_category_still_uses_the_runtime_mode_result( ] finally: get_service_registry().unregister_local(tool_name) + + +@pytest.mark.parametrize("read_only", [False, True]) +def test_register_upstream_read_only_is_operator_declared(read_only): + """SEL-484: remote tools default to mutating; an operator may opt out.""" + tool_name = f"upstream-ro-{read_only}" + registry = get_service_registry() + kwargs = {"read_only": True} if read_only else {} + try: + registry.register_upstream( + service_id=tool_name, + name=tool_name, + description="read_only flag test", + category=ServiceCategory.AGENT_COMMS, + executor=object(), + input_schema={"type": "object"}, + output_schema=None, + credits_per_unit=1.0, + upstream_tool_name="partner.echo", + upstream_origin="https://partner.example", + **kwargs, + ) + assert registry.get_local(tool_name)["read_only"] is read_only + finally: + registry.unregister_local(tool_name) diff --git a/gateway/tests/test_reason_code_registry.py b/gateway/tests/test_reason_code_registry.py index 0e4d454c..a2fa38a7 100644 --- a/gateway/tests/test_reason_code_registry.py +++ b/gateway/tests/test_reason_code_registry.py @@ -68,6 +68,7 @@ "idempotency_key_control_characters", "idempotency_key_not_a_string", "idempotency_key_not_utf8", + "idempotency_key_required", "idempotency_key_required_for_human_approval", "idempotency_key_reused", "idempotency_key_too_long", diff --git a/gateway/tests/test_standard_mcp_endpoint.py b/gateway/tests/test_standard_mcp_endpoint.py index 843b798c..3b3b6439 100644 --- a/gateway/tests/test_standard_mcp_endpoint.py +++ b/gateway/tests/test_standard_mcp_endpoint.py @@ -10,7 +10,12 @@ import hashlib +from decimal import Decimal + import pytest + +from tests.conftest import running_on_sqlite +from uuid import uuid4 from httpx import ASGITransport, AsyncClient from mcp.types import LATEST_PROTOCOL_VERSION from b2a_sdk.receipt_verifier import key_set_from_document, verify_bundle @@ -712,6 +717,929 @@ async def test_malformed_json_is_still_the_sdks_parse_error( assert resp.json()["error"]["code"] == -32700 +@pytest.mark.anyio +async def test_unkeyed_tools_call_retry_replays_without_second_dispatch( + client, standard_mcp_enabled, clean_database +): + """SEL-484: an unkeyed retry (fresh JSON-RPC id) must not dispatch twice.""" + provisioned = await provision_agent_wallet(client) + wallet_id = provisioned["agent_wallet_id"] + invocations: list[str] = [] + + def _echo(message="ok", n=0): + invocations.append(message) + return {"message": message} + + registry = get_service_registry() + registry.register_local( + service_id="standard-echo-retry", + name="Standard Echo Retry", + description="Unkeyed retry test tool", + category=ServiceCategory.AGENT_COMMS, + func=_echo, + credits_per_unit=2.0, + unit_name="call", + ) + headers = {**provisioned["agent_headers"], **MCP_HEADERS} + try: + first = await client.post( + "/mcp", + json=_rpc( + "tools/call", + request_id=101, + params={ + "name": "standard-echo-retry", + "arguments": {"message": "hi", "n": 1}, + }, + ), + headers=headers, + ) + assert first.status_code == 200 + receipt = first.json()["result"]["receipt"] + + # Retry: new JSON-RPC id, same arguments in a different key order. + retry = await client.post( + "/mcp", + json=_rpc( + "tools/call", + request_id=102, + params={ + "name": "standard-echo-retry", + "arguments": {"n": 1, "message": "hi"}, + }, + ), + headers=headers, + ) + assert retry.status_code == 200 + assert retry.json()["result"]["receipt"]["receipt_id"] == receipt["receipt_id"] + assert invocations == ["hi"] + + # Different arguments are a distinct call and dispatch. + other = await client.post( + "/mcp", + json=_rpc( + "tools/call", + request_id=103, + params={ + "name": "standard-echo-retry", + "arguments": {"message": "bye", "n": 1}, + }, + ), + headers=headers, + ) + assert other.status_code == 200 + assert other.json()["result"]["receipt"]["receipt_id"] != receipt["receipt_id"] + assert invocations == ["hi", "bye"] + + ledger = await client.get( + f"/v1/billing/ledger/{wallet_id}", headers=provisioned["agent_headers"] + ) + debits = [ + e + for e in ledger.json()["entries"] + if "standard-echo-retry" in e["description"] + ] + assert len(debits) == 2 + finally: + registry.unregister_local("standard-echo-retry") + + +async def _ledger_debits(client, provisioned, tool): + ledger = await client.get( + f"/v1/billing/ledger/{provisioned['agent_wallet_id']}", + headers=provisioned["agent_headers"], + ) + assert ledger.status_code == 200, ledger.text + return [e for e in ledger.json()["entries"] if tool in e["description"]] + + +def _assert_idempotency_key_required(resp): + assert resp.status_code == 200, resp.text + error = resp.json()["error"] + assert error["code"] == -32003 + assert error["message"] == "idempotency_key_required" + assert error["data"]["reason_code"] == "idempotency_key_required" + assert error["data"]["remediation"]["type"] == "retry_with_idempotency_key" + + +@pytest.mark.anyio +async def test_unkeyed_call_with_window_zero_is_refused_without_dispatch( + client, standard_mcp_enabled, clean_database, monkeypatch +): + """SEL-498: no retry slot and no key is refused, never a random key.""" + from app.core.config import get_settings + + monkeypatch.setattr(get_settings(), "MCP_UNKEYED_RETRY_WINDOW_SECONDS", 0) + provisioned = await provision_agent_wallet(client) + invocations: list[str] = [] + _register_counting_tool("standard-echo-legacy", invocations) + headers = {**provisioned["agent_headers"], **MCP_HEADERS} + try: + for rid in (201, 202): + resp = await _call(client, headers, "standard-echo-legacy", rid) + _assert_idempotency_key_required(resp) + assert invocations == [] + assert await _ledger_debits(client, provisioned, "standard-echo-legacy") == [] + + # The remediation works: a keyed call and its keyed retry dispatch + # and bill exactly once. + keyed = {**headers, "Idempotency-Key": "sel-498-window-zero"} + first = await _call(client, keyed, "standard-echo-legacy", 203) + retry = await _call(client, keyed, "standard-echo-legacy", 204) + assert first.status_code == retry.status_code == 200 + assert ( + retry.json()["result"]["receipt"]["receipt_id"] + == first.json()["result"]["receipt"]["receipt_id"] + ) + assert invocations == ["hi"] + assert ( + len(await _ledger_debits(client, provisioned, "standard-echo-legacy")) == 1 + ) + finally: + get_service_registry().unregister_local("standard-echo-legacy") + + +def _register_counting_tool(service_id: str, invocations: list, **extra): + def _echo(message="ok", n=0): + invocations.append(message) + return {"message": message} + + get_service_registry().register_local( + service_id=service_id, + name=service_id, + description="SEL-484 unkeyed retry test tool", + category=ServiceCategory.AGENT_COMMS, + func=_echo, + credits_per_unit=2.0, + unit_name="call", + **extra, + ) + + +async def _call(client, headers, tool, request_id, arguments=None): + return await client.post( + "/mcp", + json=_rpc( + "tools/call", + request_id=request_id, + params={"name": tool, "arguments": arguments or {"message": "hi"}}, + ), + headers=headers, + ) + + +@pytest.mark.anyio +async def test_unkeyed_retry_straddling_window_boundary_dispatches_once( + client, standard_mcp_enabled, clean_database, monkeypatch +): + """Codex blocking 1: a retry across a fixed bucket edge must still replay.""" + import time as _time + + from app.routers import mcp_standard + + window = mcp_standard._effective_unkeyed_retry_window() + edge = (int(_time.time() // window) + 1) * window + provisioned = await provision_agent_wallet(client) + headers = {**provisioned["agent_headers"], **MCP_HEADERS} + invocations: list[str] = [] + _register_counting_tool("echo-boundary", invocations) + try: + monkeypatch.setattr(mcp_standard, "_unkeyed_clock", lambda: edge - 0.001) + first = await _call(client, headers, "echo-boundary", 1) + monkeypatch.setattr(mcp_standard, "_unkeyed_clock", lambda: edge + 0.001) + retry = await _call(client, headers, "echo-boundary", 2) + assert first.status_code == retry.status_code == 200 + assert ( + retry.json()["result"]["receipt"]["receipt_id"] + == first.json()["result"]["receipt"]["receipt_id"] + ) + assert invocations == ["hi"] + finally: + get_service_registry().unregister_local("echo-boundary") + + +@pytest.mark.anyio +async def test_unkeyed_read_only_tool_is_refused_without_dispatch( + client, standard_mcp_enabled, clean_database +): + """Read-only tools get no deterministic slot, so unkeyed calls are refused.""" + provisioned = await provision_agent_wallet(client) + headers = {**provisioned["agent_headers"], **MCP_HEADERS} + invocations: list[str] = [] + _register_counting_tool("echo-read-only", invocations, read_only=True) + try: + for rid in (1, 2): + resp = await _call(client, headers, "echo-read-only", rid) + _assert_idempotency_key_required(resp) + assert invocations == [] + assert await _ledger_debits(client, provisioned, "echo-read-only") == [] + + # Distinct keys are distinct reads; each dispatches (no stale replay). + for rid, key in ((3, "sel-498-read-a"), (4, "sel-498-read-b")): + resp = await _call( + client, {**headers, "Idempotency-Key": key}, "echo-read-only", rid + ) + assert resp.status_code == 200 + assert "result" in resp.json(), resp.text + assert invocations == ["hi", "hi"] + # Lead's exemption test (#233): read-only calls ARE metered. Each keyed + # read debits the wallet at the tool's per-call price, so an unkeyed + # read could bill twice on retry and must stay refused. + debits = await _ledger_debits(client, provisioned, "echo-read-only") + assert len(debits) == 2 + assert all(Decimal(str(e["amount"])) != 0 for e in debits) + finally: + get_service_registry().unregister_local("echo-read-only") + + +@pytest.mark.anyio +async def test_demo_permit_mint_refuses_unkeyed_call_without_slot(monkeypatch): + """mcp.py demo path: no slot and no key raises before any permit is minted.""" + from types import SimpleNamespace + + from app.core.config import get_settings + from app.routers import mcp as mcp_router + from app.routers import mcp_standard + + minted: list[str] = [] + + async def _no_slot(**_kwargs): + return None + + async def _mint(**_kwargs): + minted.append("permit") + return "permit-should-not-exist" + + async def _no_approval(_wallet_id): + return False + + monkeypatch.setattr(mcp_standard, "_resolve_unkeyed_retry", _no_slot) + monkeypatch.setattr(mcp_standard, "_mint_auto_permit", _mint) + monkeypatch.setattr(mcp_router, "wallet_human_approval_required", _no_approval) + monkeypatch.setattr(get_settings(), "DEMO_ALLOWED_TOOLS", "partner.echo") + monkeypatch.setattr(mcp_router, "_ensure_local_mcp_tools_registered", lambda: None) + registry = get_service_registry() + existed = registry.get_local("partner.echo") is not None + if not existed: + _register_counting_tool("partner.echo", []) + request = SimpleNamespace( + auth=SimpleNamespace(tenant="demo", wallet_id="w-demo", allowed_tools=None), + wallet_id="w-demo", + permit_id=None, + tool_name="partner.echo", + arguments={"message": "hi"}, + idempotency_key=None, + ) + try: + with pytest.raises(ValueError, match="^idempotency_key_required$"): + await mcp_router._maybe_mint_demo_permit(request) + assert minted == [] + assert request.idempotency_key is None + finally: + if not existed: + registry.unregister_local("partner.echo") + + +def test_dogfood_notes_count_is_declared_read_only(monkeypatch): + from app.services import dogfood_tool + + monkeypatch.setattr(dogfood_tool, "_second_registered", False) + registry = get_service_registry() + existed = registry.get_local(dogfood_tool.DOGFOOD_SECOND_TOOL_ID) is not None + dogfood_tool.register_dogfood_second_tool() + try: + record = registry.get_local(dogfood_tool.DOGFOOD_SECOND_TOOL_ID) + assert record is not None and record["read_only"] is True + finally: + if not existed: + registry.unregister_local(dogfood_tool.DOGFOOD_SECOND_TOOL_ID) + + +@pytest.mark.anyio +async def test_unkeyed_retry_after_no_effect_failure_runs_again( + client, standard_mcp_enabled, clean_database +): + """Codex: a real refunded pre-dispatch failure is retried, not replayed. + + The upstream executor genuinely fails before dispatch (nothing reaches + the tool, the charge is refunded); the identical unkeyed retry then runs + once and a further retry replays that success. + """ + from tests.test_mcp_upstream_governed import ( + FakeUpstreamExecutor, + _register_upstream, + ) + + tool = f"echo-no-effect-{uuid4().hex[:8]}" + provisioned = await provision_agent_wallet(client) + headers = {**provisioned["agent_headers"], **MCP_HEADERS} + executor = FakeUpstreamExecutor("pre_dispatch_failure") + _register_upstream(tool, executor) + + async def _entries(): + ledger = await client.get( + f"/v1/billing/ledger/{provisioned['agent_wallet_id']}", + headers=provisioned["agent_headers"], + ) + assert ledger.status_code == 200, ledger.text + return {e["entry_id"]: e for e in ledger.json()["entries"]} + + def _net(entries): + return sum((Decimal(str(e["amount"])) for e in entries), Decimal("0")) + + try: + before = await _entries() + first = await _call(client, headers, tool, 1) + assert first.status_code == 200, first.text + assert executor.dispatch_count == 0 + assert len(executor.calls) == 1 + after_failure = await _entries() + failure_entries = [e for k, e in after_failure.items() if k not in before] + # Charged then refunded: no net charge for the pre-dispatch failure. + assert failure_entries, "expected charge+refund ledger rows" + assert _net(failure_entries) == 0, failure_entries + + executor.mode = "success" + retry = await _call(client, headers, tool, 2) + assert retry.status_code == 200, retry.text + after_success = await _entries() + success_entries = [ + e for k, e in after_success.items() if k not in after_failure + ] + debits = [e for e in success_entries if Decimal(str(e["amount"])) < 0] + assert len(debits) == 1, success_entries + assert _net(success_entries) == Decimal(str(debits[0]["amount"])) < 0 + assert "result" in retry.json(), retry.text + assert executor.dispatch_count == 1 + again = await _call(client, headers, tool, 3) + assert ( + again.json()["result"]["receipt"]["receipt_id"] + == retry.json()["result"]["receipt"]["receipt_id"] + ) + assert executor.dispatch_count == 1 + # The replay adds no ledger entry. + assert await _entries() == after_success + finally: + get_service_registry().unregister_local(tool) + + +@pytest.mark.anyio +async def test_simultaneous_identical_unkeyed_calls_dispatch_once( + client, standard_mcp_enabled, clean_database +): + """Codex non-blocking 2: two concurrent identical calls are one operation.""" + import asyncio + + provisioned = await provision_agent_wallet(client) + headers = {**provisioned["agent_headers"], **MCP_HEADERS} + invocations: list[str] = [] + _register_counting_tool("echo-concurrent", invocations) + try: + a, b = await asyncio.gather( + _call(client, headers, "echo-concurrent", 1), + _call(client, headers, "echo-concurrent", 2), + ) + assert invocations == ["hi"] + receipts = set() + for resp in (a, b): + body = resp.json() + if "result" in body: + receipts.add(body["result"]["receipt"]["receipt_id"]) + else: + assert body["error"]["code"] == -32005, resp.text + assert len(receipts) == 1 + finally: + get_service_registry().unregister_local("echo-concurrent") + + +async def _backdate_lock(wallet_id: str, lock_key: str) -> None: + """Age a held lock row past the stale threshold via its real created_at.""" + from datetime import timedelta + + from sqlalchemy import update + + from app.db.database import get_session_factory + from app.db.models import IdempotencyRecordModel + from app.routers import mcp_standard + from app.core.time import utc_now + + factory = get_session_factory() + async with factory() as session: + await session.execute( + update(IdempotencyRecordModel) + .where( + IdempotencyRecordModel.wallet_id == wallet_id, + IdempotencyRecordModel.endpoint + == mcp_standard._UNKEYED_RETRY_LOCK_ENDPOINT, + IdempotencyRecordModel.idempotency_key == lock_key, + ) + .values( + created_at=utc_now() + - timedelta(seconds=mcp_standard._UNKEYED_RETRY_LOCK_STALE_SECONDS + 5) + ) + ) + await session.commit() + + +def _stall_after_acquire(monkeypatch, on_stall): + """Holder A acquires the lock for real, then stalls (runs ``on_stall``) + before minting. Only the first acquire stalls; later ones are untouched. + Neither the ownership check nor the mint is patched.""" + from app.routers import mcp_standard + + real_acquire = mcp_standard._acquire_unkeyed_retry_lock + state = {"stalled": False} + + async def _acquire(wallet_id, lock_key): + record_id = await real_acquire(wallet_id, lock_key) + if not state["stalled"]: + state["stalled"] = True + await on_stall(wallet_id, lock_key, record_id) + return record_id + + monkeypatch.setattr(mcp_standard, "_acquire_unkeyed_retry_lock", _acquire) + + +@pytest.mark.anyio +async def test_stale_takeover_via_created_at_fences_stalled_holder_mint( + client, standard_mcp_enabled, clean_database, monkeypatch +): + """Codex blocker: no window between the ownership check and the mint. + A stalls holding the lock; its row ages past the stale threshold (real + created_at); worker B reclaims it through the real acquire path and runs + the call end-to-end. A then resumes: its fenced mint-row insert must fail + atomically or replay B's outcome, so the tool runs exactly once.""" + from app.routers import mcp_standard + from app.services.idempotency import get_idempotency_service + + provisioned = await provision_agent_wallet(client) + wallet_id = provisioned["agent_wallet_id"] + headers = {**provisioned["agent_headers"], **MCP_HEADERS} + invocations: list[str] = [] + _register_counting_tool("echo-stale", invocations) + idem = get_idempotency_service() + seen: dict[str, object] = {} + + async def _while_a_stalls(wid, lock_key, record_id): + await _backdate_lock(wid, lock_key) + seen["b"] = await _call(client, headers, "echo-stale", 2) + seen["key"] = lock_key + + _stall_after_acquire(monkeypatch, _while_a_stalls) + try: + stalled = await _call(client, headers, "echo-stale", 1) + b = seen["b"] + assert "result" in b.json(), b.text + assert invocations == ["hi"] + # A resumed after B finished: B's mint row already exists, so A's + # mint replays it (not fenced) and A returns B's receipt — never a + # second dispatch. The fence-lost path is the next test. + receipt_key = "io.agentmiddleware/receipt" + assert ( + stalled.json()["result"]["_meta"][receipt_key]["receipt_id"] + == b.json()["result"]["_meta"][receipt_key]["receipt_id"] + ), stalled.text + # B released its own lock; A's fenced release deleted nothing extra. + assert ( + await idem.get_record( + wallet_id=wallet_id, + endpoint=mcp_standard._UNKEYED_RETRY_LOCK_ENDPOINT, + idempotency_key=seen["key"], + ) + ) is None + finally: + get_service_registry().unregister_local("echo-stale") + + +@pytest.mark.anyio +async def test_stale_takeover_via_created_at_keeps_new_holders_lock( + client, standard_mcp_enabled, clean_database, monkeypatch +): + """A stalled holder reclaimed via the real stale path neither mints nor + dispatches, and its release leaves the new holder's lock row intact.""" + from app.routers import mcp_standard + from app.services.idempotency import get_idempotency_service + + provisioned = await provision_agent_wallet(client) + wallet_id = provisioned["agent_wallet_id"] + headers = {**provisioned["agent_headers"], **MCP_HEADERS} + invocations: list[str] = [] + _register_counting_tool("echo-takeover", invocations) + idem = get_idempotency_service() + taken: dict[str, str] = {} + real_acquire = mcp_standard._acquire_unkeyed_retry_lock + + async def _while_a_stalls(wid, lock_key, record_id): + await _backdate_lock(wid, lock_key) + # B goes through the unpatched acquire: stale created_at -> reclaim. + taken["id"] = await real_acquire(wid, lock_key) + taken["key"] = lock_key + assert taken["id"] != record_id + + _stall_after_acquire(monkeypatch, _while_a_stalls) + try: + stalled = await _call(client, headers, "echo-takeover", 1) + assert stalled.json()["error"]["code"] == -32005, stalled.text + assert invocations == [] + held = await idem.get_record( + wallet_id=wallet_id, + endpoint=mcp_standard._UNKEYED_RETRY_LOCK_ENDPOINT, + idempotency_key=taken["key"], + ) + assert held is not None and held.record_id == taken["id"] + + await idem.abandon( + wallet_id=wallet_id, + endpoint=mcp_standard._UNKEYED_RETRY_LOCK_ENDPOINT, + idempotency_key=taken["key"], + expected_record_id=taken["id"], + ) + retry = await _call(client, headers, "echo-takeover", 2) + assert "result" in retry.json(), retry.text + assert invocations == ["hi"] + finally: + get_service_registry().unregister_local("echo-takeover") + + +@pytest.mark.anyio +async def test_fenced_begin_writes_nothing_when_fence_row_reissued( + client, clean_database +): + """The mint-row insert and the lock ownership check are one transaction: + a fence naming a lost (missing or re-issued) lock row writes no record.""" + from app.services.idempotency import ( + IdempotencyFenceLostError, + get_idempotency_service, + ) + + wallet = (await provision_agent_wallet(client))["agent_wallet_id"] + idem = get_idempotency_service() + lock = await idem.begin_with_record( + wallet_id=wallet, + endpoint="/mcp#unkeyed-retry-lock", + idempotency_key="ulock-x", + request_payload={"kind": "unkeyed-retry-lock"}, + ) + for fence in ( + ("/mcp#unkeyed-retry-lock", "ulock-x", "idm-not-the-holder"), + ("/mcp#unkeyed-retry-lock", "ulock-gone", lock.record_id), + ): + with pytest.raises(IdempotencyFenceLostError): + await idem.begin_with_record( + wallet_id=wallet, + endpoint="/mcp#auto-permit", + idempotency_key="smcp-x", + request_payload={"k": 1}, + fence=fence, + ) + assert ( + await idem.get_record( + wallet_id=wallet, endpoint="/mcp#auto-permit", idempotency_key="smcp-x" + ) + ) is None + owned = await idem.begin_with_record( + wallet_id=wallet, + endpoint="/mcp#auto-permit", + idempotency_key="smcp-x", + request_payload={"k": 1}, + fence=("/mcp#unkeyed-retry-lock", "ulock-x", lock.record_id), + ) + assert owned.replay is None + + +def _sqlite_db_path() -> str: + from app.db.database import get_engine + + engine = get_engine() + assert engine is not None + return engine.sync_engine.url.database + + +def _external_takeover(lock_record_id: str) -> int: + """Delete the holder's lock row from a separate SQLite connection, the + way a stale-lock takeover would. Short timeout: if the holder has the + write lock this raises ``database is locked``; it is never swallowed.""" + import sqlite3 + + conn = sqlite3.connect(_sqlite_db_path(), timeout=0.2) + try: + cur = conn.execute( + "DELETE FROM idempotency_records WHERE record_id = ?", + (lock_record_id,), + ) + conn.commit() + return cur.rowcount + finally: + conn.close() + + +async def _fresh_lock(idem, wallet): + lock_key = f"ulock-{uuid4()}" + lock = await idem.begin_with_record( + wallet_id=wallet, + endpoint="/mcp#unkeyed-retry-lock", + idempotency_key=lock_key, + request_payload={"kind": "unkeyed-retry-lock"}, + ) + return lock_key, lock + + +@pytest.mark.anyio +@pytest.mark.skipif(not running_on_sqlite(), reason="SQLite lock semantics") +async def test_fence_claim_blocks_takeover_until_mint_commits(client, clean_database): + """Codex blocker 1 (deterministic): a takeover forced after the holder's + fence step and before its mint insert must lose, and the loser writes + nothing. Under the old SELECT ... FOR UPDATE (ignored by SQLite) the + takeover deletes the lock row and the mint still commits: both win.""" + import sqlite3 + + from sqlalchemy import event + + from app.db.database import get_engine + from app.services.idempotency import get_idempotency_service + + wallet = (await provision_agent_wallet(client))["agent_wallet_id"] + idem = get_idempotency_service() + lock_key, lock = await _fresh_lock(idem, wallet) + mint_key = f"smcp-{uuid4()}" + takeover: dict = {} + + def _before_insert(conn, cursor, statement, parameters, context, many): + if ( + statement.lstrip().upper().startswith("INSERT INTO IDEMPOTENCY_RECORDS") + and mint_key in str(parameters) + and not takeover + ): + try: + takeover["deleted"] = _external_takeover(lock.record_id) + except sqlite3.OperationalError as exc: + takeover["error"] = str(exc) + + sync_engine = get_engine().sync_engine + event.listen(sync_engine, "before_cursor_execute", _before_insert) + try: + minted = await idem.begin_with_record( + wallet_id=wallet, + endpoint="/mcp#auto-permit", + idempotency_key=mint_key, + request_payload={"k": 1}, + fence=("/mcp#unkeyed-retry-lock", lock_key, lock.record_id), + ) + finally: + event.remove(sync_engine, "before_cursor_execute", _before_insert) + assert minted.replay is None + # The takeover ran and lost: it hit the holder's write lock, deleted + # nothing, and the holder's lock row survives. + assert takeover == {"error": "database is locked"}, takeover + held = await idem.get_record( + wallet_id=wallet, + endpoint="/mcp#unkeyed-retry-lock", + idempotency_key=lock_key, + ) + assert held is not None and held.record_id == lock.record_id + + +@pytest.mark.anyio +@pytest.mark.skipif(not running_on_sqlite(), reason="SQLite lock semantics") +async def test_fence_lost_before_claim_writes_nothing(client, clean_database): + """Takeover forced between the holder's read and its fence claim: the + claim matches no row, the holder raises fence-lost and writes nothing.""" + from sqlalchemy import event + + from app.db.database import get_engine + from app.services.idempotency import ( + IdempotencyFenceLostError, + get_idempotency_service, + ) + + wallet = (await provision_agent_wallet(client))["agent_wallet_id"] + idem = get_idempotency_service() + lock_key, lock = await _fresh_lock(idem, wallet) + mint_key = f"smcp-{uuid4()}" + takeover: dict = {} + + def _before_claim(conn, cursor, statement, parameters, context, many): + upper = statement.lstrip().upper() + if ( + upper.startswith(("UPDATE IDEMPOTENCY_RECORDS", "SELECT")) + and lock_key in str(parameters) + and not takeover + ): + takeover["deleted"] = _external_takeover(lock.record_id) + + sync_engine = get_engine().sync_engine + event.listen(sync_engine, "before_cursor_execute", _before_claim) + try: + with pytest.raises(IdempotencyFenceLostError): + await idem.begin_with_record( + wallet_id=wallet, + endpoint="/mcp#auto-permit", + idempotency_key=mint_key, + request_payload={"k": 1}, + fence=("/mcp#unkeyed-retry-lock", lock_key, lock.record_id), + ) + finally: + event.remove(sync_engine, "before_cursor_execute", _before_claim) + assert takeover == {"deleted": 1} + assert ( + await idem.get_record( + wallet_id=wallet, endpoint="/mcp#auto-permit", idempotency_key=mint_key + ) + ) is None + + +@pytest.mark.anyio +@pytest.mark.skipif(not running_on_sqlite(), reason="SQLite lock semantics") +async def test_fence_claim_lock_timeout_is_retryable_in_progress( + client, clean_database +): + """Codex: a SQLite lock timeout on the fence claim maps to the retryable + idempotency_in_progress, not an internal error, and writes nothing.""" + import sqlite3 + + from sqlalchemy import event + + from app.db.database import get_engine + from app.services.idempotency import ( + IdempotencyFenceLostError, + IdempotencyInProgressError, + get_idempotency_service, + ) + + wallet = (await provision_agent_wallet(client))["agent_wallet_id"] + idem = get_idempotency_service() + lock_key, lock = await _fresh_lock(idem, wallet) + mint_key = f"smcp-{uuid4()}" + blocker = sqlite3.connect(_sqlite_db_path(), timeout=5) + blocker.isolation_level = None + + def _short_busy(conn, cursor, statement, parameters, context, many): + if statement.lstrip().upper().startswith("UPDATE IDEMPOTENCY_RECORDS"): + cursor.execute("PRAGMA busy_timeout=100") + + engine = get_engine() + event.listen(engine.sync_engine, "before_cursor_execute", _short_busy) + blocker.execute("BEGIN IMMEDIATE") + try: + with pytest.raises(IdempotencyInProgressError) as caught: + await idem.begin_with_record( + wallet_id=wallet, + endpoint="/mcp#auto-permit", + idempotency_key=mint_key, + request_payload={"k": 1}, + fence=("/mcp#unkeyed-retry-lock", lock_key, lock.record_id), + ) + assert not isinstance(caught.value, IdempotencyFenceLostError) + assert str(caught.value) == "idempotency_in_progress" + finally: + blocker.execute("ROLLBACK") + blocker.close() + event.remove(engine.sync_engine, "before_cursor_execute", _short_busy) + await engine.dispose() # drop the connection left at busy_timeout=100 + assert ( + await idem.get_record( + wallet_id=wallet, endpoint="/mcp#auto-permit", idempotency_key=mint_key + ) + ) is None + + +def test_unkeyed_retry_window_rejects_negative_and_keeps_zero(): + """CodeRabbit: negative windows are invalid; 0 stays the opt-out.""" + from pydantic import ValidationError + + from app.core.config import Settings + + with pytest.raises(ValidationError): + Settings(MCP_UNKEYED_RETRY_WINDOW_SECONDS=-1) + assert ( + Settings(MCP_UNKEYED_RETRY_WINDOW_SECONDS=0).MCP_UNKEYED_RETRY_WINDOW_SECONDS + == 0 + ) + assert Settings().MCP_UNKEYED_RETRY_WINDOW_SECONDS == 300 + + +def test_unregistered_tool_replay_runs_iga_check(): + """Codex should-fix: the replay served after a tool is removed must run + the same IGA gate as the registered-tool replay path.""" + import inspect + + from app.routers import mcp + + src = inspect.getsource(mcp) + end = src.index('reason = f"Tool not found') + branch = src[src.rindex("if not service:", 0, end) : end] + assert "_assert_replay_iga_access(auth=auth, tool_name=tool_name)" in branch + assert branch.index("_assert_replay_iga_access") < branch.index( + "_raise_replayed_error" + ) + + +def test_default_unkeyed_retry_window_is_permit_ttl_and_documented(): + """Codex blocker 2: with defaults the window is the 120 s permit TTL, + not the 300 s setting, and the docs say so (a later retry bills again).""" + from pathlib import Path + + from app.core.config import Settings, get_settings + from app.routers import mcp_standard + + defaults = Settings() + settings = get_settings() + assert defaults.MCP_UNKEYED_RETRY_WINDOW_SECONDS == 300 + assert defaults.STANDARD_MCP_PERMIT_TTL_SECONDS == 120 + assert settings.MCP_UNKEYED_RETRY_WINDOW_SECONDS == 300 + assert mcp_standard._effective_unkeyed_retry_window() == min( + 300, settings.STANDARD_MCP_PERMIT_TTL_SECONDS + ) + docs = Path(__file__).resolve().parents[1] / "docs" + for name in ("concepts/request-keys.md", "failure-semantics.md"): + text = " ".join((docs / name).read_text().split()) + assert "STANDARD_MCP_PERMIT_TTL_SECONDS" in text or "auto-permit TTL" in text + assert "120" in text + assert "bills again" in text + + +def test_effective_unkeyed_retry_window_capped_at_permit_ttl(monkeypatch): + """Bugbot: a joined slot must never outlive the permit it reuses.""" + from app.core.config import get_settings + from app.routers import mcp_standard + + settings = get_settings() + monkeypatch.setattr(settings, "STANDARD_MCP_PERMIT_TTL_SECONDS", 120) + monkeypatch.setattr(settings, "MCP_UNKEYED_RETRY_WINDOW_SECONDS", 300) + assert mcp_standard._effective_unkeyed_retry_window() == 120 + monkeypatch.setattr(settings, "MCP_UNKEYED_RETRY_WINDOW_SECONDS", 60) + assert mcp_standard._effective_unkeyed_retry_window() == 60 + monkeypatch.setattr(settings, "MCP_UNKEYED_RETRY_WINDOW_SECONDS", 0) + assert mcp_standard._effective_unkeyed_retry_window() == 0 + + +@pytest.mark.anyio +async def test_unkeyed_mint_failure_releases_row_for_retry( + client, standard_mcp_enabled, clean_database, monkeypatch +): + """CodeRabbit: a non-PermitError mint failure must not strand the mint row.""" + from app.services import permits as permits_module + + provisioned = await provision_agent_wallet(client) + headers = {**provisioned["agent_headers"], **MCP_HEADERS} + invocations: list[str] = [] + _register_counting_tool("echo-mint-fail", invocations) + service = permits_module.get_permit_service() + real_create = service.create_permit + calls = {"n": 0} + + async def _flaky_create(*args, **kwargs): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("transient permit store failure") + return await real_create(*args, **kwargs) + + monkeypatch.setattr(service, "create_permit", _flaky_create) + try: + # The failure surfaces either as a raised error or an error + # response depending on the transport; either way nothing dispatched. + raised: BaseException | None = None + try: + first = await _call(client, headers, "echo-mint-fail", 1) + assert "result" not in first.json() or first.json()["result"].get( + "isError" + ), first.text + except BaseException as exc: # noqa: BLE001 - transport-dependent + raised = exc + if raised is not None: + assert "transient permit store failure" in repr(raised) + assert invocations == [] + retry = await _call(client, headers, "echo-mint-fail", 2) + assert retry.status_code == 200, retry.text + assert "result" in retry.json(), retry.text + assert invocations == ["hi"] + finally: + get_service_registry().unregister_local("echo-mint-fail") + + +def test_effective_unkeyed_retry_window_capped_at_demo_permit_ttl(monkeypatch): + """Bugbot: demo auto permits are shorter-lived; the window must follow.""" + from app.core.config import get_settings + from app.routers import mcp_standard + + settings = get_settings() + monkeypatch.setattr(settings, "STANDARD_MCP_PERMIT_TTL_SECONDS", 120) + monkeypatch.setattr(settings, "MCP_UNKEYED_RETRY_WINDOW_SECONDS", 300) + monkeypatch.setattr(settings, "DEMO_AUTO_PERMIT_TTL_SECONDS", 45) + monkeypatch.setattr(settings, "DEMO_MAX_PERMIT_TTL_MINUTES", None) + assert mcp_standard._effective_unkeyed_retry_window(tenant="demo") == 45 + assert mcp_standard._effective_unkeyed_retry_window(tenant="acme") == 120 + monkeypatch.setattr(settings, "DEMO_AUTO_PERMIT_TTL_SECONDS", 200) + monkeypatch.setattr(settings, "DEMO_MAX_PERMIT_TTL_MINUTES", 1) + assert mcp_standard._effective_unkeyed_retry_window(tenant="demo") == 60 + # The window always equals the TTL the mint itself uses. + assert mcp_standard._effective_unkeyed_retry_window( + tenant="demo" + ) == mcp_standard._auto_permit_ttl_seconds( + tenant="demo", requires_human_approval=False + ) + + @pytest.mark.anyio async def test_tools_call_explicit_permit_out_of_scope_is_refused_without_charge( client, standard_mcp_enabled, clean_database