feat(webhooks): support tenant-scoped signing senders

docs/handler-authoring.md

@@ -1446,6 +1446,73 @@ copy of the callback token is cleared in that transaction. Workers use expiring
 leases and exact retries; the 1–7 day horizon begins on the first attempt and
 must exactly match the advertised value.
 
+Multi-tenant sellers can resolve a different signing identity for each trusted
+server-side tenant scope. Use `sender_resolver=` on the outbox and pair it with
+`webhook_signing_scope_resolver=` on the registry:
+
+```python
+from adcp.decisioning import (
+    PgTaskRegistry,
+    PgTaskWebhookOutbox,
+    ScopePermanentlyUnknown,
+    ScopeTransientlyUnavailable,
+    WebhookSenderResolution,
+)
+
+class TenantWebhookSenders:
+    async def resolve(self, signing_scope_id: str):
+        credential = await internal_key_store.active_for_scope(signing_scope_id)
+        if credential is None:
+            raise ScopePermanentlyUnknown
+        if credential.rotation_in_progress:
+            raise ScopeTransientlyUnavailable
+        return WebhookSenderResolution(
+            sender=credential.webhook_sender,  # cached, lifecycle-managed sender
+            # Read from the same trusted record used by request-scoped capabilities.
+            advertised_algorithms=frozenset(credential.webhook_signing_algorithms),
+        )
+
+def trusted_signing_scope(context):
+    # Use only internal tenant/platform metadata populated by the server.
+    return context.account.metadata.webhook_signing_scope_id
+
+outbox = PgTaskWebhookOutbox(
+    pool=pool,
+    sender_resolver=TenantWebhookSenders(),
+    encryption_key=task_webhook_encryption_key,
+    delivery_retry_horizon_seconds=86_400,
+)
+registry = PgTaskRegistry(
+    pool=pool,
+    task_webhook_outbox=outbox,
+    webhook_signing_scope_resolver=trusted_signing_scope,
+)
+```
+
+The scope is encrypted at task issuance, persisted on the durable outbox row,
+and authenticated as envelope AAD. The worker resolves a fresh sender on every
+attempt, so key rotation changes the signature without changing the stored body
+or idempotency key. `ScopeTransientlyUnavailable` releases the row for retry;
+`ScopePermanentlyUnknown` quarantines it for operator reconciliation. Every
+resolved sender is revalidated as an RFC 9421 sender using an SDK-owned,
+IP-pinned transport with private destinations disabled. Its actual key
+algorithm must also appear in the trusted scope's advertised algorithm set;
+a mismatch is quarantined before any request is sent.
+
+The resolver owns sender lifecycle. Return cached senders whose clients are
+closed during application shutdown; do not allocate a new `WebhookSender` (and
+therefore a new connection pool) on every delivery attempt.
+
+Never derive the signing scope from `push_notification_config`, the request's
+buyer-supplied `context`, or an unqualified buyer account id. It must be an
+opaque identifier obtained from trusted internal tenant/platform metadata.
+Pass exactly one of `sender=` or `sender_resolver=`; the fixed-sender path and
+existing `NULL signing_scope_id` rows remain backward compatible while that
+fixed-sender mode is retained. Before switching an existing deployment to
+resolver mode, drain or reconcile its pre-migration `NULL` rows: the worker
+cannot safely infer a tenant key for them and will quarantine them rather than
+guess a signing identity.
+
 Production adopters may set `auto_emit_task_webhooks=False` only when an external
 durable outbox owns publication, retries, immutable body/key retention, and
 reconciliation. Set `webhook_signing_managed_externally=True` in the corresponding

src/adcp/__init__.py

@@ -724,13 +724,17 @@ def _resolve_version() -> str:
         "LegacyHmacFallback",
         "MemoryBackend",
         "PreparedWebhook",
+        "ScopePermanentlyUnknown",
+        "ScopeTransientlyUnavailable",
         "WebhookChallengeError",
         "WebhookChallengeResult",
         "WebhookDedupStore",
         "WebhookDestinationPolicy",
         "WebhookReceiver",
         "WebhookReceiverConfig",
         "WebhookSender",
+        "WebhookSenderResolution",
+        "WebhookSenderResolver",
         "WebhookVerifyOptions",
         "challenge_webhook_destination",
         "create_a2a_webhook_payload",
@@ -968,6 +972,10 @@ def get_adcp_version() -> str:
     "WebhookReceiver",
     "WebhookReceiverConfig",
     "WebhookSender",
+    "WebhookSenderResolution",
+    "WebhookSenderResolver",
+    "ScopePermanentlyUnknown",
+    "ScopeTransientlyUnavailable",
     "WebhookVerifyOptions",
     "WebhookDedupStore",
     "MemoryBackend",
@@ -2140,13 +2148,17 @@ def get_adcp_version() -> str:
         LegacyHmacFallback,
         MemoryBackend,
         PreparedWebhook,
+        ScopePermanentlyUnknown,
+        ScopeTransientlyUnavailable,
         WebhookChallengeError,
         WebhookChallengeResult,
         WebhookDedupStore,
         WebhookDestinationPolicy,
         WebhookReceiver,
         WebhookReceiverConfig,
         WebhookSender,
+        WebhookSenderResolution,
+        WebhookSenderResolver,
         WebhookVerifyOptions,
         challenge_webhook_destination,
         create_a2a_webhook_payload,

src/adcp/decisioning/__init__.py

@@ -273,6 +273,12 @@ def create_media_buy(
     validate_capabilities_response_shape,
     validate_capabilities_response_shape_async,
 )
+from adcp.webhook_sender import (
+    ScopePermanentlyUnknown,
+    ScopeTransientlyUnavailable,
+    WebhookSenderResolution,
+    WebhookSenderResolver,
+)
 
 # Conditional import: PgTaskRegistry needs the [pg] extra. Always expose
 # the name — when psycopg isn't installed we fall through to a stub class whose
@@ -289,6 +295,7 @@ def create_media_buy(
         PgTaskRegistry,
         PgTaskWebhookOutbox,
         PostgresTaskRegistry,
+        WebhookSigningScopeResolver,
     )
 except ImportError:  # pragma: no cover — exercised by the [pg] extra tests
     from typing import ClassVar as _ClassVar
@@ -341,6 +348,8 @@ def __init__(self, *args: object, **kwargs: object) -> None:
                 "(Poetry: `poetry add 'adcp[pg]'`)."
             )
 
+    from adcp.decisioning.pg.task_registry import WebhookSigningScopeResolver
+
 
 __all__ = [
     "Account",
@@ -446,6 +455,8 @@ def __init__(self, *args: object, **kwargs: object) -> None:
     "SalesResult",
     "SalesSpecialism",
     "ServiceUnavailableError",
+    "ScopePermanentlyUnknown",
+    "ScopeTransientlyUnavailable",
     "SignalsPlatform",
     "SingletonAccounts",
     "SELF_SERVE_UPDATE_ACTION_MODES",
@@ -458,6 +469,9 @@ def __init__(self, *args: object, **kwargs: object) -> None:
     "TaskHandoffContext",
     "TaskRegistry",
     "TaskState",
+    "WebhookSenderResolver",
+    "WebhookSenderResolution",
+    "WebhookSigningScopeResolver",
     "TranslationMap",
     "UNKNOWN_UPDATE_ACTION",
     "UnsupportedFeatureError",

src/adcp/decisioning/dispatch.py

@@ -2092,10 +2092,15 @@ async def _project_handoff(
         and getattr(registry, "task_webhook_outbox", None) is not None
     ):
         push_url, push_token = push_target
+        signing_scope_id: str | None = None
+        signing_scope_resolver = getattr(registry, "resolve_webhook_signing_scope", None)
+        if signing_scope_resolver is not None:
+            signing_scope_id = await signing_scope_resolver(ctx)
         issue_kwargs.update(
             webhook_url=push_url,
             webhook_operation_id=_extract_push_operation_id(request_params),
             webhook_token=push_token,
+            webhook_signing_scope_id=signing_scope_id,
         )
     task_id = await registry.issue(**issue_kwargs)
 
@@ -2364,10 +2369,15 @@ async def _project_workflow_handoff(
         and getattr(registry, "task_webhook_outbox", None) is not None
     ):
         push_url, push_token = push_target
+        signing_scope_id: str | None = None
+        signing_scope_resolver = getattr(registry, "resolve_webhook_signing_scope", None)
+        if signing_scope_resolver is not None:
+            signing_scope_id = await signing_scope_resolver(ctx)
         issue_kwargs.update(
             webhook_url=push_url,
             webhook_operation_id=_extract_push_operation_id(request_params),
             webhook_token=push_token,
+            webhook_signing_scope_id=signing_scope_id,
         )
     task_id = await registry.issue(**issue_kwargs)
     handoff_ctx = TaskHandoffContext(id=task_id, _registry=registry)

src/adcp/decisioning/pg/__init__.py

@@ -36,7 +36,11 @@
     PgBuyerAgentRegistry,
 )
 from adcp.decisioning.pg.proposal_store import PgProposalStore
-from adcp.decisioning.pg.task_registry import PgTaskRegistry, PostgresTaskRegistry
+from adcp.decisioning.pg.task_registry import (
+    PgTaskRegistry,
+    PostgresTaskRegistry,
+    WebhookSigningScopeResolver,
+)
 from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox
 
 __all__ = [
@@ -47,4 +51,5 @@
     "PgTaskRegistry",
     "PgTaskWebhookOutbox",
     "PostgresTaskRegistry",
+    "WebhookSigningScopeResolver",
 ]

src/adcp/decisioning/pg/task_registry.py

@@ -62,17 +62,20 @@ async def main():
 
 from __future__ import annotations
 
+import inspect
 import json
 import re
 import time
 import uuid
-from typing import TYPE_CHECKING, Any, ClassVar
+from collections.abc import Awaitable, Callable
+from typing import TYPE_CHECKING, Any, ClassVar, TypeAlias
 
 from adcp.decisioning.account_projection import strip_credentials_from_wire_result
 
 if TYPE_CHECKING:
     from psycopg_pool import AsyncConnectionPool
 
+    from adcp.decisioning.context import RequestContext
     from adcp.decisioning.pg.task_webhook_outbox import PgTaskWebhookOutbox
 
 try:
@@ -95,6 +98,8 @@ async def main():
 # the only protection against SQL injection or Unicode homoglyph substitution.
 _SAFE_IDENTIFIER_RE = re.compile(r"^[a-z_][a-z0-9_]{0,62}$")
 
+WebhookSigningScopeResolver: TypeAlias = Callable[["RequestContext[Any]"], str | Awaitable[str]]
+
 
 class PgTaskRegistry:
     """PostgreSQL-backed :class:`~adcp.decisioning.TaskRegistry` — v6.1.
@@ -131,6 +136,7 @@ def __init__(
         *,
         pool: AsyncConnectionPool,
         task_webhook_outbox: PgTaskWebhookOutbox | None = None,
+        webhook_signing_scope_resolver: WebhookSigningScopeResolver | None = None,
         _table: str = _DEFAULT_TABLE,
     ) -> None:
         if not PG_AVAILABLE:
@@ -141,9 +147,18 @@ def __init__(
             raise ValueError(
                 "PgTaskRegistry and PgTaskWebhookOutbox must use the same connection pool"
             )
+        uses_sender_resolver = (
+            task_webhook_outbox is not None and task_webhook_outbox._sender_resolver is not None
+        )
+        if uses_sender_resolver != (webhook_signing_scope_resolver is not None):
+            raise ValueError(
+                "webhook_signing_scope_resolver is required exactly when the "
+                "PgTaskWebhookOutbox uses sender_resolver"
+            )
         self._pool = pool
         self._table = _table
         self.task_webhook_outbox = task_webhook_outbox
+        self._webhook_signing_scope_resolver = webhook_signing_scope_resolver
         self.atomic_task_webhook_outbox = task_webhook_outbox is not None
 
         # Pre-format queries at construction so the hot path avoids f-strings per call.
@@ -247,6 +262,7 @@ async def issue(
         webhook_url: str | None = None,
         webhook_operation_id: str | None = None,
         webhook_token: str | None = None,
+        webhook_signing_scope_id: str | None = None,
         **_extra: Any,
     ) -> str:
         """Allocate a task_id, persist a ``submitted`` row, return the id.
@@ -269,6 +285,8 @@ async def issue(
             raise ValueError("webhook_url must be non-empty when supplied")
         if webhook_operation_id is not None and not webhook_operation_id:
             raise ValueError("webhook_operation_id must be non-empty when supplied")
+        if webhook_url is None and webhook_signing_scope_id is not None:
+            raise ValueError("webhook_signing_scope_id requires webhook_url")
         outbox = self.task_webhook_outbox
         if webhook_url is not None:
             if outbox is None:
@@ -287,6 +305,7 @@ async def issue(
                 url=webhook_url,
                 operation_id=webhook_operation_id,
                 token=webhook_token,
+                signing_scope_id=webhook_signing_scope_id,
             )
         now = time.time()
         async with self._pool.connection() as conn:
@@ -305,6 +324,46 @@ async def issue(
             )
         return task_id
 
+    async def resolve_webhook_signing_scope(
+        self,
+        context: RequestContext[Any],
+    ) -> str | None:
+        """Derive an opaque signing scope from trusted framework context.
+
+        The callback is operator wiring and receives the hydrated
+        :class:`RequestContext`. It must use internal tenant/platform metadata,
+        never buyer request ``context``, ``push_notification_config``, or an
+        unqualified buyer account id.
+        """
+        resolver = self._webhook_signing_scope_resolver
+        if resolver is None:
+            return None
+        from adcp.decisioning.types import AdcpError
+
+        try:
+            value: object = resolver(context)
+            if inspect.isawaitable(value):
+                value = await value
+        except Exception:
+            raise AdcpError(
+                "INTERNAL_ERROR",
+                message="Webhook signing scope resolution failed",
+                recovery="terminal",
+            ) from None
+        if not isinstance(value, str):
+            raise AdcpError(
+                "INTERNAL_ERROR",
+                message="Webhook signing scope resolver returned an invalid value",
+                recovery="terminal",
+            )
+        # Reuse the outbox's bounded opaque-ID validation before any task row
+        # is issued. This is trusted server state, but it still crosses a DB
+        # and authenticated-envelope boundary.
+        if self.task_webhook_outbox is None:
+            raise RuntimeError("signing scope resolver requires a task webhook outbox")
+        self.task_webhook_outbox._validate_signing_scope_id(value)
+        return value
+
     async def update_progress(
         self,
         task_id: str,
@@ -475,12 +534,14 @@ async def _enqueue_terminal_if_registered(
             )
         if registration_nonce is None:
             raise RuntimeError(f"Task {task_id!r} has incomplete webhook registration")
-        url, operation_id, token = self.task_webhook_outbox.open_registration(
-            account_id=account_id,
-            task_id=task_id,
-            task_type=task_type,
-            encrypted_registration=bytes(encrypted_registration),
-            nonce=bytes(registration_nonce),
+        url, operation_id, token, signing_scope_id = (
+            self.task_webhook_outbox._open_registration_with_scope(
+                account_id=account_id,
+                task_id=task_id,
+                task_type=task_type,
+                encrypted_registration=bytes(encrypted_registration),
+                nonce=bytes(registration_nonce),
+            )
         )
         await self.task_webhook_outbox.enqueue_terminal(
             conn,
@@ -492,6 +553,7 @@ async def _enqueue_terminal_if_registered(
             url=url,
             operation_id=operation_id,
             token=token,
+            signing_scope_id=signing_scope_id,
         )
         # The encrypted outbox envelope now owns the callback registration.
         # Clear the task-row copy in this same transaction.
@@ -515,4 +577,9 @@ async def discard(self, task_id: str) -> None:
 PostgresTaskRegistry = PgTaskRegistry
 
 
-__all__ = ["PG_AVAILABLE", "PgTaskRegistry", "PostgresTaskRegistry"]
+__all__ = [
+    "PG_AVAILABLE",
+    "PgTaskRegistry",
+    "PostgresTaskRegistry",
+    "WebhookSigningScopeResolver",
+]