fix(signing): harden trust and request atomicity

src/adcp/canonical_formats/references.py

@@ -42,6 +42,7 @@
 DEFAULT_REFERENCE_TIMEOUT_SECONDS = 5.0
 DEFAULT_REFERENCE_BODY_LIMIT_BYTES = 1024 * 1024
 DEFAULT_MAX_SCHEMA_REFS = 256
+DEFAULT_MAX_SCHEMA_IDS = 32
 DEFAULT_MAX_REF_DEPTH = 8
 DEFAULT_MAX_SCHEMA_KEYWORDS = 10_000
 DEFAULT_MAX_SCHEMA_DEPTH = 128
@@ -157,6 +158,7 @@ def __init__(
         timeout: float = DEFAULT_REFERENCE_TIMEOUT_SECONDS,
         max_body_bytes: int = DEFAULT_REFERENCE_BODY_LIMIT_BYTES,
         max_schema_refs: int = DEFAULT_MAX_SCHEMA_REFS,
+        max_schema_ids: int = DEFAULT_MAX_SCHEMA_IDS,
         max_ref_depth: int = DEFAULT_MAX_REF_DEPTH,
         max_schema_keywords: int = DEFAULT_MAX_SCHEMA_KEYWORDS,
         max_schema_depth: int = DEFAULT_MAX_SCHEMA_DEPTH,
@@ -165,6 +167,7 @@ def __init__(
         self._timeout = timeout
         self._max_body_bytes = max_body_bytes
         self._max_schema_refs = max_schema_refs
+        self._max_schema_ids = max_schema_ids
         self._max_ref_depth = max_ref_depth
         self._max_schema_keywords = max_schema_keywords
         self._max_schema_depth = max_schema_depth
@@ -318,6 +321,7 @@ def _validate_schema(
             document,
             base_uri=reference.uri,
             max_refs=self._max_schema_refs,
+            max_ids=self._max_schema_ids,
             max_ref_depth=self._max_ref_depth,
             max_keywords=self._max_schema_keywords,
             max_depth=self._max_schema_depth,
@@ -534,11 +538,12 @@ def _validate_schema_refs(
     *,
     base_uri: str,
     max_refs: int,
+    max_ids: int,
     max_ref_depth: int,
     max_keywords: int,
     max_depth: int,
 ) -> str | None:
-    state = _SchemaRefState(max_refs=max_refs, max_keywords=max_keywords)
+    state = _SchemaRefState(max_refs=max_refs, max_ids=max_ids, max_keywords=max_keywords)
     return _walk_schema_refs(
         document,
         base_uri=base_uri,
@@ -552,8 +557,10 @@ def _validate_schema_refs(
 @dataclass
 class _SchemaRefState:
     max_refs: int
+    max_ids: int
     max_keywords: int
     refs: int = 0
+    ids: int = 0
     keywords: int = 0
 
 
@@ -577,6 +584,9 @@ def _walk_schema_refs(
         if raw_id is not None:
             if not isinstance(raw_id, str):
                 return "$id must be a string"
+            state.ids += 1
+            if state.ids > state.max_ids:
+                return "format_schema exceeds $id count bound"
             id_error, resolved_id = _validate_schema_id_value(raw_id, base_uri=base_uri)
             if id_error is not None:
                 return id_error
@@ -711,5 +721,6 @@ def _trusted_aao_catalog_origin(parts: SplitResult) -> bool:
     "CanonicalReferenceStatus",
     "DEFAULT_REFERENCE_BODY_LIMIT_BYTES",
     "DEFAULT_REFERENCE_TIMEOUT_SECONDS",
+    "DEFAULT_MAX_SCHEMA_IDS",
     "parse_canonical_reference",
 ]

src/adcp/server/idempotency/backends.py

@@ -4,12 +4,11 @@
 
 1. Retrieve a cached response by ``(principal_id, idempotency_key)``, honoring
    the seller's replay TTL.
-2. Atomically commit ``(payload_hash, response)`` on a fresh key. Atomicity
-   with the handler's business writes is the backend's choice —
-   :class:`MemoryBackend` makes no such guarantee; :class:`PgBackend` shares
-   a connection pool so adopters with the same Postgres can compose their
-   handler's transaction with the cache write (v1 commits in a separate
-   pool connection — co-tx wiring is a v1.1 affordance).
+2. Hold an execution lock across lookup, handler execution, and cache commit.
+3. Atomically insert webhook-dedup markers with first-writer-wins semantics.
+
+The lock prevents concurrent duplicate execution. Atomicity with unrelated
+business writes remains the adopter's responsibility.
 
 Backends expose async methods. The in-process :class:`MemoryBackend` is
 synchronous under the hood but wrapped in ``async`` signatures so the store
@@ -22,8 +21,11 @@
 import json
 import re
 import time
+import weakref
 from abc import ABC, abstractmethod
-from collections.abc import Callable
+from collections.abc import AsyncIterator, Callable
+from contextlib import asynccontextmanager
+from contextvars import ContextVar
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import Any
@@ -83,8 +85,8 @@ class IdempotencyBackend(ABC):
     """Abstract storage backend contract.
 
     All methods are async. Implementations MUST be safe to call concurrently
-    from multiple asyncio tasks — :class:`IdempotencyStore` does not serialize
-    access on the caller's behalf.
+    from multiple asyncio tasks. ``hold`` must coordinate all processes that
+    share the backend namespace.
     """
 
     @abstractmethod
@@ -109,6 +111,30 @@ async def put(
         or expired, so an overwrite in that window is a legitimate retry of
         the write itself."""
 
+    def hold(self, scope_key: str, key: str) -> Any:
+        """Return an async context manager holding the key's execution lock.
+
+        Custom backends must implement this operation to be usable by
+        :class:`IdempotencyStore`. It is concrete only to keep existing
+        backend subclasses importable while adopters migrate.
+        """
+        raise NotImplementedError(
+            f"{type(self).__name__} must implement atomic hold(scope_key, key)"
+        )
+
+    async def put_if_absent(self, scope_key: str, key: str, entry: CachedResponse) -> bool:
+        """Atomically insert a fresh/expired slot; return whether it won.
+
+        The default composes ``hold`` with the legacy ``get``/``put`` API,
+        allowing custom backends to implement one locking primitive. Backends
+        with a native conditional insert should override this method.
+        """
+        async with self.hold(scope_key, key):
+            if await self.get(scope_key, key) is not None:
+                return False
+            await self.put(scope_key, key, entry)
+            return True
+
     @abstractmethod
     async def delete_expired(self, now_epoch: float | None = None) -> int:
         """Best-effort sweep of expired entries. Returns the count removed.
@@ -139,6 +165,9 @@ class MemoryBackend(IdempotencyBackend):
     def __init__(self, *, clock: Callable[[], float] = time.time) -> None:
         self._store: dict[tuple[str, str], CachedResponse] = {}
         self._lock = asyncio.Lock()
+        self._key_locks: weakref.WeakValueDictionary[tuple[str, str], asyncio.Lock] = (
+            weakref.WeakValueDictionary()
+        )
         self._clock = clock
 
     async def get(self, scope_key: str, key: str) -> CachedResponse | None:
@@ -162,6 +191,28 @@ async def put(
         async with self._lock:
             self._store[(scope_key, key)] = entry
 
+    @asynccontextmanager
+    async def hold(self, scope_key: str, key: str) -> AsyncIterator[None]:
+        """Serialize one idempotent handler execution in this process."""
+        slot = (scope_key, key)
+        async with self._lock:
+            key_lock = self._key_locks.get(slot)
+            if key_lock is None:
+                key_lock = asyncio.Lock()
+                self._key_locks[slot] = key_lock
+        async with key_lock:
+            yield
+
+    async def put_if_absent(self, scope_key: str, key: str, entry: CachedResponse) -> bool:
+        """Atomically claim a missing or expired slot."""
+        slot = (scope_key, key)
+        async with self._lock:
+            existing = self._store.get(slot)
+            if existing is not None and existing.expires_at_epoch > self._clock():
+                return False
+            self._store[slot] = entry
+            return True
+
     async def delete_expired(self, now_epoch: float | None = None) -> int:
         cutoff = now_epoch if now_epoch is not None else self._clock()
         async with self._lock:
@@ -198,15 +249,17 @@ class PgBackend(IdempotencyBackend):
         from adcp.server.idempotency import IdempotencyStore, PgBackend
 
         pool = AsyncConnectionPool("postgresql://...", min_size=2, max_size=10)
-        backend = PgBackend(pool=pool)
+        lock_pool = AsyncConnectionPool("postgresql://...", min_size=2, max_size=10)
+        backend = PgBackend(pool=pool, lock_pool=lock_pool)
         await backend.create_schema()  # idempotent; safe to call on every boot
 
         store = IdempotencyStore(backend=backend, ttl_seconds=86400)
 
-    **Atomicity caveat (v1).** ``put`` commits on a fresh pool connection —
-    the cache write is NOT in the same transaction as the handler's
-    business writes. A crash between handler success and cache commit
-    leaves the slot empty; the next retry re-executes the handler.
+    **Atomicity caveat.** The backend holds a Postgres advisory lock and writes
+    the cache in its transaction, but the cache write is NOT automatically in
+    the same transaction as the handler's unrelated business writes. A crash
+    after an external side effect but before cache commit can still leave the
+    slot empty; the next retry re-executes the handler.
     Idempotent handlers absorb this without harm. **Handlers with
     non-idempotent side effects** (e.g., ``INSERT INTO media_buys``
     without a unique constraint on the buyer's idempotency_key) need
@@ -261,6 +314,10 @@ class PgBackend(IdempotencyBackend):
     :param pool: ``psycopg_pool.AsyncConnectionPool`` owned by the caller.
         Each operation acquires a short-lived connection. We don't open,
         own, or close the pool.
+    :param lock_pool: A distinct caller-owned pool reserved for advisory-lock
+        transactions. It MUST NOT be the business/cache ``pool``: ``hold``
+        keeps one connection checked out while adopter code runs, and sharing
+        that pool with handler SQL can deadlock under saturation.
     :param table_name: Override the default table name. Useful for
         multi-tenant schema scoping. Default ``adcp_idempotency``.
 
@@ -274,12 +331,19 @@ def __init__(
         self,
         *,
         pool: Any,  # psycopg_pool.AsyncConnectionPool — Any avoids runtime psycopg import
+        lock_pool: Any,
         table_name: str = DEFAULT_IDEMPOTENCY_TABLE,
     ) -> None:
         if not _PG_AVAILABLE:
             raise ImportError(_PG_INSTALL_HINT)
+        if lock_pool is pool:
+            raise ValueError("lock_pool must be distinct from pool to prevent handler deadlocks")
         self._pool = pool
+        self._lock_pool = lock_pool
         self._table = _safe_identifier(table_name)
+        self._active_connection: ContextVar[tuple[Any, asyncio.Task[Any] | None] | None] = (
+            ContextVar(f"adcp_idempotency_connection_{id(self)}", default=None)
+        )
 
         # Pre-format SQL once. Validated identifier so f-string interpolation
         # is byte-safe; values always go through %s parameterization. Same
@@ -309,6 +373,16 @@ def __init__(
             f"WHERE {t}.expires_at <= now()"
         )
         self._sql_delete_expired = f"DELETE FROM {t} WHERE expires_at <= %s"  # noqa: S608
+        self._sql_lock = "SELECT pg_advisory_xact_lock(hashtextextended(%s, 6217))"
+        self._sql_put_if_absent = (
+            f"INSERT INTO {t} "  # noqa: S608
+            f"(scope_key, key, payload_hash, response, expires_at) "
+            f"VALUES (%s, %s, %s, %s::jsonb, %s) "
+            f"ON CONFLICT (scope_key, key) DO UPDATE SET "
+            f"  payload_hash = EXCLUDED.payload_hash, response = EXCLUDED.response, "
+            f"  expires_at = EXCLUDED.expires_at "
+            f"WHERE {t}.expires_at <= now() RETURNING 1"
+        )
 
     async def create_schema(self) -> None:
         """Bootstrap the table + index. Idempotent.
@@ -341,17 +415,25 @@ async def get(self, scope_key: str, key: str) -> CachedResponse | None:
         sweeps them. ``get`` self-filters via ``expires_at > now()`` so a
         stale row never replays.
         """
+        active = self._active_connection.get()
+        if active is not None and active[1] is asyncio.current_task():
+            return await self._get_on_connection(active[0], scope_key, key)
         async with self._pool.connection() as conn:
-            cur = await conn.execute(self._sql_get, (scope_key, key))
-            row = await cur.fetchone()
-            if row is None:
-                return None
-            payload_hash, response, expires_at = row
-            return CachedResponse(
-                payload_hash=payload_hash,
-                response=response if isinstance(response, dict) else json.loads(response),
-                expires_at_epoch=_to_epoch(expires_at),
-            )
+            return await self._get_on_connection(conn, scope_key, key)
+
+    async def _get_on_connection(
+        self, conn: Any, scope_key: str, key: str
+    ) -> CachedResponse | None:
+        cur = await conn.execute(self._sql_get, (scope_key, key))
+        row = await cur.fetchone()
+        if row is None:
+            return None
+        payload_hash, response, expires_at = row
+        return CachedResponse(
+            payload_hash=payload_hash,
+            response=response if isinstance(response, dict) else json.loads(response),
+            expires_at_epoch=_to_epoch(expires_at),
+        )
 
     async def put(
         self,
@@ -366,9 +448,42 @@ async def put(
         that window is a legitimate retry of the write itself.
         """
         expires_at_dt = datetime.fromtimestamp(entry.expires_at_epoch, tz=timezone.utc)
+        params = (
+            scope_key,
+            key,
+            entry.payload_hash,
+            json.dumps(entry.response),
+            expires_at_dt,
+        )
+        active = self._active_connection.get()
+        if active is not None and active[1] is asyncio.current_task():
+            await active[0].execute(self._sql_put, params)
+            return
+        async with self._pool.connection() as conn:
+            await conn.execute(self._sql_put, params)
+
+    @asynccontextmanager
+    async def hold(self, scope_key: str, key: str) -> AsyncIterator[None]:
+        """Hold a cross-process transaction advisory lock for this key.
+
+        The same pooled connection remains checked out while the handler runs;
+        nested ``get``/``put`` calls reuse it via a context-local binding.
+        """
+        lock_identity = json.dumps([scope_key, key], separators=(",", ":"))
+        async with self._lock_pool.connection() as conn, conn.transaction():
+            await conn.execute(self._sql_lock, (lock_identity,))
+            token = self._active_connection.set((conn, asyncio.current_task()))
+            try:
+                yield
+            finally:
+                self._active_connection.reset(token)
+
+    async def put_if_absent(self, scope_key: str, key: str, entry: CachedResponse) -> bool:
+        """Atomically insert a webhook dedup marker, including stale replace."""
+        expires_at_dt = datetime.fromtimestamp(entry.expires_at_epoch, tz=timezone.utc)
         async with self._pool.connection() as conn:
-            await conn.execute(
-                self._sql_put,
+            cur = await conn.execute(
+                self._sql_put_if_absent,
                 (
                     scope_key,
                     key,
@@ -377,6 +492,7 @@ async def put(
                     expires_at_dt,
                 ),
             )
+            return await cur.fetchone() is not None
 
     async def delete_expired(self, now_epoch: float | None = None) -> int:
         """Best-effort sweep of expired entries. Returns rows removed."""

src/adcp/server/idempotency/lazy.py

@@ -13,7 +13,8 @@
 
     async def _resolve() -> IdempotencyBackend:
         pool = await app.get_pg_pool()
-        backend = PgBackend(pool=pool)
+        lock_pool = await app.get_idempotency_lock_pool()
+        backend = PgBackend(pool=pool, lock_pool=lock_pool)
         await backend.create_schema()
         return backend
 
@@ -33,7 +34,8 @@ async def _resolve() -> IdempotencyBackend:
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Awaitable, Callable
+from collections.abc import AsyncIterator, Awaitable, Callable
+from contextlib import asynccontextmanager
 
 from adcp.server.idempotency.backends import CachedResponse, IdempotencyBackend
 
@@ -108,6 +110,14 @@ async def get(self, scope_key: str, key: str) -> CachedResponse | None:
     async def put(self, scope_key: str, key: str, entry: CachedResponse) -> None:
         await (await self._resolve()).put(scope_key, key, entry)
 
+    @asynccontextmanager
+    async def hold(self, scope_key: str, key: str) -> AsyncIterator[None]:
+        async with (await self._resolve()).hold(scope_key, key):
+            yield
+
+    async def put_if_absent(self, scope_key: str, key: str, entry: CachedResponse) -> bool:
+        return await (await self._resolve()).put_if_absent(scope_key, key, entry)
+
     async def delete_expired(self, now_epoch: float | None = None) -> int:
         return await (await self._resolve()).delete_expired(now_epoch)