diff --git a/examples/a2a_db_tasks.py b/examples/a2a_db_tasks.py
index f0a03505..deffa175 100644
--- a/examples/a2a_db_tasks.py
+++ b/examples/a2a_db_tasks.py
@@ -27,15 +27,17 @@
 **Security model — push-notification config store adds two threats
 tenant-scoping alone does NOT address:**
 
-1. **SSRF via unvalidated webhook URLs.** Clients supply
+1. **SSRF via webhook URLs.** Clients supply
    ``PushNotificationConfig.url`` when subscribing to task progress;
    a2a-sdk's push-notif sender POSTs the full task JSON to that URL
    with no built-in validation. An attacker can register
    ``url=http://169.254.169.254/…`` (cloud metadata),
    ``http://localhost:5432/`` (internal services), link-local IPs,
-   etc. The store persists URLs verbatim — URL validation is the
-   seller's responsibility. Reject non-https, reject RFC 1918 / IPv6
-   link-local, check against an egress allowlist before persisting.
+   etc. The store rejects URLs unless they use HTTPS and their canonical
+   hostname appears in ``allowed_destination_hosts``. Configure
+   only operator-owned, stable hosts. This storage-time gate does not replace
+   sender-side DNS resolution checks and IP-pinned connections on every send;
+   implement a custom ``PushNotificationSender`` for that production boundary.
 2. **Webhook secrets stored plaintext.**
    ``PushNotificationConfig.authentication.credentials`` and
    ``PushNotificationConfig.token`` are bearer tokens / shared
@@ -72,8 +74,12 @@
 
 Run::
 
-    uv run python examples/a2a_db_tasks.py
-    # or: python -m adcp.examples.a2a_db_tasks
+    A2A_PUSH_ALLOWED_HOSTS=callback.example \
+      uv run python examples/a2a_db_tasks.py
+
+Omit the environment variable when push callbacks are not needed. The empty
+default denies every push destination; any configured callback URL must match
+one of the comma-separated canonical hostnames exactly.
 """
 
 from __future__ import annotations
@@ -83,7 +89,7 @@
 import sqlite3
 import uuid
 import warnings
-from collections.abc import Callable
+from collections.abc import AsyncIterator, Callable
 from contextlib import asynccontextmanager
 from contextvars import ContextVar
 from pathlib import Path
@@ -106,6 +112,11 @@
 from google.protobuf.json_format import MessageToJson, Parse
 
 from adcp.server import ADCPHandler, serve
+from adcp.server.a2a_push_security import (
+    normalize_allowed_push_hosts,
+    scope_from_server_context,
+    validate_a2a_push_notification_url,
+)
 from adcp.server.responses import capabilities_response, products_response
 
 _ANONYMOUS_SCOPE = "__anonymous__"
@@ -115,6 +126,11 @@
 is part of every WHERE clause."""
 
 
+def _scope_from_server_context(context: ServerCallContext | None) -> str:
+    """Derive a verified principal scope from an a2a-sdk call context."""
+    return scope_from_server_context(context) or _ANONYMOUS_SCOPE
+
+
 # ----------------------------------------------------------------------
 # SQLite-backed TaskStore
 # ----------------------------------------------------------------------
@@ -171,17 +187,10 @@ def _scope_from_context(self, context: ServerCallContext | None) -> str:
         key on every read/write; anything you don't include here
         *cannot* be enforced by the store.
         """
-        user = getattr(context, "user", None) if context is not None else None
-        if user is None:
-            return _ANONYMOUS_SCOPE
-        user_name = getattr(user, "user_name", None)
-        is_authenticated = getattr(user, "is_authenticated", False)
-        if is_authenticated and isinstance(user_name, str) and user_name:
-            return user_name
-        return _ANONYMOUS_SCOPE
+        return _scope_from_server_context(context)
 
     @asynccontextmanager
-    async def _conn(self):
+    async def _conn(self) -> AsyncIterator[sqlite3.Connection]:
         # SQLite connections aren't safe across threads. Open a fresh
         # connection per operation and commit-on-success / rollback-on-error
         # so a port to psycopg / aiomysql doesn't silently leak partial
@@ -278,22 +287,19 @@ async def list(
 #    a single-user host but loses that guarantee across backups,
 #    Docker bind mounts with wrong umask, and DB migrations. Either
 #    encrypt those fields or move them to a secrets backend.
-# 3. Isolate by principal, not just by scope. Within a single auth
-#    scope (e.g. "tenant-acme") multiple principals may share access
-#    to the same task. The reference impl keys on ``(scope, task_id,
-#    config_id)`` and falls ``config_id`` back to ``task_id`` when
-#    the client omits it — two principals registering without a
-#    ``config_id`` overwrite each other silently. Either require an
-#    explicit ``config_id`` from the client, or widen the scope key to
-#    include the principal.
+# 3. Isolate by principal, not just by a coarse organization scope. The
+#    normal SDK path below uses the authenticated ``user_name`` directly.
+#    Adopters replacing it with a custom provider that groups principals
+#    must widen that key or authorize each config row explicitly.
 _current_push_config_scope: ContextVar[str | None] = ContextVar(
     "adcp_push_config_scope", default=None
 )
-"""Default ContextVar used by ``SqlitePushNotificationConfigStore`` when
-no ``scope_provider`` is supplied. HTTP auth middleware sets it per
-request; the store reads it on every op. Exposed at module level so
-a seller with their own auth middleware can pair it with this
-reference impl without subclassing."""
+"""Fallback ContextVar for context-less direct/background store calls.
+
+Normal a2a-sdk handler calls carry ``ServerCallContext`` and do not consult
+this value. Exposed so a custom sender can restore the owning scope when it
+later reads configs without a request context.
+"""
 
 
 def _default_push_config_scope_provider() -> str | None:
@@ -306,17 +312,13 @@ def _default_push_config_scope_provider() -> str | None:
 class SqlitePushNotificationConfigStore(PushNotificationConfigStore):
     """Durable A2A ``PushNotificationConfigStore`` backed by a single
     SQLite file, scoped by an authenticated principal resolved at
-    set/get/delete time via a ``scope_provider`` callable.
-
-    a2a-sdk's ``PushNotificationConfigStore`` ABC does **not** pass a
-    ``ServerCallContext`` to ``set_info`` / ``get_info`` /
-    ``delete_info`` (unlike the ``TaskStore`` ABC), so scoping has to
-    happen out-of-band. The canonical pattern is a ``ContextVar`` the
-    seller's HTTP auth middleware populates per request — the
-    ``_default_push_config_scope_provider()`` factory below reads the
-    module-level ``_current_push_config_scope``. Sellers who already
-    maintain their own ContextVar (or prefer thread-locals, Starlette
-    ``request.state``, etc.) inject a custom provider.
+    set/get/delete time from the a2a-sdk ``ServerCallContext``.
+
+    a2a-sdk 1.0 passes ``ServerCallContext`` to all three store methods;
+    normal handler calls therefore bind directly to the authenticated
+    ``user.user_name`` just like :class:`SqliteTaskStore`. A ContextVar
+    ``scope_provider`` remains as a fallback for background sender and direct
+    calls that genuinely lack a context.
 
     Example — wiring the default ContextVar from auth middleware::
 
@@ -353,7 +355,7 @@ async def dispatch(self, request, call_next):
             scope_provider=lambda: my_scope.get(default=None),
         )
 
-    **Fails closed on anonymous requests.** If the provider returns
+    **Fails loudly on anonymous fallback.** If a context-less call's provider returns
     ``None``, a ``UserWarning`` is emitted once per store instance and
     the store falls through to ``__anonymous__`` — unauthenticated
     requests end up sharing one giant scope. Operators should reject
@@ -362,8 +364,8 @@ async def dispatch(self, request, call_next):
     forgot to.
 
     **Background-task caveat — sender path.** a2a-sdk's push-notif
-    sender calls ``get_info()`` from a background ``asyncio.Task``
-    spawned by ``DefaultRequestHandler``. That task inherits the
+    sender may call ``get_info()`` from a background ``asyncio.Task``
+    without a ``ServerCallContext``. That task inherits the
     ContextVar snapshot captured at task-creation time; if the
     seller's auth middleware has already reset the ContextVar before
     the background task reads it, ``get_info()`` will return an empty
@@ -381,9 +383,11 @@ def __init__(
         db_path: str | Path = "a2a_push_configs.db",
         *,
         scope_provider: Callable[[], str | None] | None = None,
+        allowed_destination_hosts: frozenset[str] = frozenset(),
     ) -> None:
         self._db_path = str(db_path)
         self._scope_provider = scope_provider or _default_push_config_scope_provider
+        self._allowed_destination_hosts = normalize_allowed_push_hosts(allowed_destination_hosts)
         self._init_schema()
         self._warned_anonymous = False
 
@@ -409,8 +413,12 @@ def _init_schema(self) -> None:
             with contextlib.suppress(OSError):
                 os.chmod(self._db_path, 0o600)
 
-    def _scope(self) -> str:
-        scope = self._scope_provider()
+    def _scope(self, context: ServerCallContext | None) -> str:
+        # An explicit context is authoritative. Never let an unauthenticated
+        # request inherit an ambient tenant from a ContextVar/provider.
+        scope = (
+            scope_from_server_context(context) if context is not None else self._scope_provider()
+        )
         if not scope:
             if not self._warned_anonymous:
                 self._warned_anonymous = True
@@ -429,7 +437,7 @@ def _scope(self) -> str:
         return scope
 
     @asynccontextmanager
-    async def _conn(self):
+    async def _conn(self) -> AsyncIterator[sqlite3.Connection]:
         conn = sqlite3.connect(self._db_path)
         try:
             yield conn
@@ -447,7 +455,11 @@ async def set_info(
         notification_config: PushNotificationConfig,
         context: ServerCallContext | None = None,
     ) -> None:
-        scope = self._scope()
+        scope = self._scope(context)
+        validate_a2a_push_notification_url(
+            str(notification_config.url),
+            allowed_hosts=self._allowed_destination_hosts,
+        )
         # PushNotificationConfig.id is optional on the wire; when the
         # client didn't supply one we synthesise a UUID so two clients
         # registering on the same task without explicit ids don't
@@ -471,7 +483,7 @@ async def get_info(
         task_id: str,
         context: ServerCallContext | None = None,
     ) -> list[PushNotificationConfig]:
-        scope = self._scope()
+        scope = self._scope(context)
         async with self._conn() as conn:
             rows = conn.execute(
                 "SELECT config_json FROM a2a_push_configs WHERE scope = ? AND task_id = ?",
@@ -485,7 +497,7 @@ async def delete_info(
         context: ServerCallContext | None = None,
         config_id: str | None = None,
     ) -> None:
-        scope = self._scope()
+        scope = self._scope(context)
         async with self._conn() as conn:
             if config_id is None:
                 # a2a-sdk's ABC semantic: ``delete_info(task_id, None)``
@@ -514,7 +526,7 @@ async def delete_info(
 # ----------------------------------------------------------------------
 
 
-class DemoAgent(ADCPHandler):
+class DemoAgent(ADCPHandler[Any]):
     async def get_adcp_capabilities(self, params: Any, context: Any = None) -> dict[str, Any]:
         return capabilities_response(["media_buy"])
 
@@ -529,7 +541,13 @@ async def get_products(self, params: Any, context: Any = None) -> dict[str, Any]
 
 def main() -> None:
     task_store = SqliteTaskStore(db_path="a2a_tasks.db")
-    push_store = SqlitePushNotificationConfigStore(db_path="a2a_push_configs.db")
+    allowed_push_hosts = frozenset(
+        host for host in os.environ.get("A2A_PUSH_ALLOWED_HOSTS", "").split(",") if host
+    )
+    push_store = SqlitePushNotificationConfigStore(
+        db_path="a2a_push_configs.db",
+        allowed_destination_hosts=allowed_push_hosts,
+    )
     serve(
         DemoAgent(),
         name="a2a-db-tasks-demo",
diff --git a/examples/a2a_sqlalchemy_tasks.py b/examples/a2a_sqlalchemy_tasks.py
index 3af2fa29..ef979391 100644
--- a/examples/a2a_sqlalchemy_tasks.py
+++ b/examples/a2a_sqlalchemy_tasks.py
@@ -21,10 +21,11 @@
 etc.) and the wrapper just maps those that the protocol cares about.
 
 **Security model — same as the SQLite reference.** Tenant-scoped
-lookups via ``ServerCallContext.user.user_name``; SSRF-vulnerable
-``PushNotificationConfig.url`` MUST be validated by the seller before
-persistence (this example does NOT validate — see the SQLite reference
-docstring for the egress-allowlist pattern). Webhook secrets in
+lookups via ``ServerCallContext.user.user_name``; push-notification URLs
+must use HTTPS and match the store's explicit destination-host
+allowlist before persistence. Allowlist only operator-owned, stable hosts;
+storage-time validation does not replace sender-side DNS resolution checks
+and IP-pinned connections on every delivery. Webhook secrets in
 ``authentication.credentials`` / ``token`` should be envelope-encrypted
 or moved to a secrets backend in production; this example persists
 them plaintext for runnability.
@@ -44,17 +45,21 @@
 
 Run::
 
-    uv run python examples/a2a_sqlalchemy_tasks.py
+    A2A_PUSH_ALLOWED_HOSTS=callback.example \
+      uv run python examples/a2a_sqlalchemy_tasks.py
 
 Then connect any A2A client to ``http://localhost:3001/`` —
 ``message/send`` carries a ``configuration.push_notification_config``
-that lands in the ``a2a_push_configs`` SQLite table; ``tasks/get``
+whose URL host must match that comma-separated allowlist before it lands in
+the ``a2a_push_configs`` SQLite table; ``tasks/get``
 reads from ``a2a_tasks``. Tear down by deleting ``a2a_sqlalchemy.db``.
 """
 
 from __future__ import annotations
 
 import contextlib
+import os
+import uuid
 import warnings
 from contextvars import ContextVar
 from datetime import datetime, timezone
@@ -76,6 +81,12 @@
 from a2a.types import TaskPushNotificationConfig as PushNotificationConfig
 from google.protobuf.json_format import MessageToJson, Parse
 
+from adcp.server.a2a_push_security import (
+    normalize_allowed_push_hosts,
+    scope_from_server_context,
+    validate_a2a_push_notification_url,
+)
+
 try:
     from sqlalchemy import (
         Boolean,
@@ -184,18 +195,14 @@ def _scope_from_context(context: ServerCallContext | None) -> str:
     context never falls through to a "no filter" query that would leak
     other tenants' tasks.
     """
-    if context is None or context.user is None:
-        return _NO_AUTH_SCOPE
-    return context.user.user_name or _NO_AUTH_SCOPE
+    return scope_from_server_context(context) or _NO_AUTH_SCOPE
 
 
 # Adopter-side push-notif scope hook. The
-# ``PushNotificationConfigStore`` Protocol does NOT receive the
-# ``ServerCallContext`` (a2a-sdk caveat — see the SQLite reference's
-# docstring); adopters compose with their tenant-scoped ``TaskStore``
-# to derive scope by walking from ``task_id`` to the owning row. This
-# example uses a contextvar that the surrounding handler populates
-# from its own auth middleware.
+# a2a-sdk 1.0 passes ``ServerCallContext`` to the push-config store.
+# The ContextVar remains a compatibility fallback for direct calls and
+# older surrounding middleware, while normal handler calls derive scope
+# directly from the authenticated context.
 _push_config_scope: ContextVar[str | None] = ContextVar("_push_config_scope", default=None)
 
 
@@ -294,20 +301,27 @@ async def list(
 class SqlAlchemyPushNotificationConfigStore(PushNotificationConfigStore):
     """Tenant-scoped, SQLAlchemy-backed push-notification config store.
 
-    URL validation is the seller's responsibility. This example does
-    NOT validate ``config.push_notification_config.url`` before
-    persisting — production deployments MUST reject non-https,
-    RFC-1918, link-local IPv6, and the cloud metadata service URL
-    before this method runs. See the SQLite reference's module
-    docstring for the SSRF threat model.
+    The destination-host allowlist is intentionally empty by default, so
+    registrations fail closed until the operator configures trusted callback
+    hosts.
     """
 
-    def __init__(self, session_factory: sessionmaker[Session]) -> None:
+    def __init__(
+        self,
+        session_factory: sessionmaker[Session],
+        *,
+        allowed_destination_hosts: frozenset[str] = frozenset(),
+    ) -> None:
         self._session_factory = session_factory
+        self._allowed_destination_hosts = normalize_allowed_push_hosts(allowed_destination_hosts)
 
     @staticmethod
-    def _scope() -> str:
-        scope = _push_config_scope.get()
+    def _scope(context: ServerCallContext | None) -> str:
+        # An explicit context is authoritative. Never let an unauthenticated
+        # request inherit an ambient tenant from the ContextVar.
+        scope = (
+            scope_from_server_context(context) if context is not None else _push_config_scope.get()
+        )
         if scope is None:
             warnings.warn(
                 "PushNotificationConfigStore scope contextvar unset — "
@@ -324,12 +338,14 @@ async def set_info(
         self,
         task_id: str,
         notification_config: PushNotificationConfig,
+        context: ServerCallContext | None = None,
     ) -> None:
-        scope = self._scope()
-        config_id = (
-            notification_config.push_notification_config.id
-            or notification_config.push_notification_config.url
+        scope = self._scope(context)
+        validate_a2a_push_notification_url(
+            str(notification_config.url),
+            allowed_hosts=self._allowed_destination_hosts,
         )
+        config_id = notification_config.id or f"auto-{uuid.uuid4()}"
         with self._session_factory() as session:
             row = A2APushConfigRow(
                 scope=scope,
@@ -342,8 +358,12 @@ async def set_info(
             session.merge(row)
             session.commit()
 
-    async def get_info(self, task_id: str) -> list[PushNotificationConfig]:
-        scope = self._scope()
+    async def get_info(
+        self,
+        task_id: str,
+        context: ServerCallContext | None = None,
+    ) -> list[PushNotificationConfig]:
+        scope = self._scope(context)
         with self._session_factory() as session:
             rows = session.execute(
                 select(A2APushConfigRow).where(
@@ -354,8 +374,13 @@ async def get_info(self, task_id: str) -> list[PushNotificationConfig]:
             ).scalars()
             return [Parse(row.payload, PushNotificationConfig()) for row in rows]
 
-    async def delete_info(self, task_id: str, config_id: str | None = None) -> None:
-        scope = self._scope()
+    async def delete_info(
+        self,
+        task_id: str,
+        context: ServerCallContext | None = None,
+        config_id: str | None = None,
+    ) -> None:
+        scope = self._scope(context)
         with self._session_factory() as session:
             stmt = delete(A2APushConfigRow).where(
                 A2APushConfigRow.scope == scope,
@@ -404,7 +429,7 @@ def build_engine_and_sessions(
 # ----------------------------------------------------------------------
 
 
-class DemoAgent(ADCPHandler):
+class DemoAgent(ADCPHandler[Any]):
     async def get_adcp_capabilities(self, params: Any, context: Any = None) -> dict[str, Any]:
         return capabilities_response(["media_buy"])
 
@@ -420,7 +445,13 @@ async def get_products(self, params: Any, context: Any = None) -> dict[str, Any]
 def main() -> None:
     session_factory = build_engine_and_sessions()
     task_store = SqlAlchemyTaskStore(session_factory)
-    push_store = SqlAlchemyPushNotificationConfigStore(session_factory)
+    allowed_push_hosts = frozenset(
+        host for host in os.environ.get("A2A_PUSH_ALLOWED_HOSTS", "").split(",") if host
+    )
+    push_store = SqlAlchemyPushNotificationConfigStore(
+        session_factory,
+        allowed_destination_hosts=allowed_push_hosts,
+    )
     serve(
         DemoAgent(),
         name="a2a-sqlalchemy-demo",
diff --git a/src/adcp/audit_sink.py b/src/adcp/audit_sink.py
index 14e63928..7cb30662 100644
--- a/src/adcp/audit_sink.py
+++ b/src/adcp/audit_sink.py
@@ -226,6 +226,10 @@ class SlackAlertSink:
         operation/identity/error summary. Prevents accidental egress of
         financial fields (budgets, credit limits), PII (contact info),
         or buyer-supplied free text.
+    :param include_error_message: Include raw exception text in Slack alerts.
+        Defaults to ``False`` because exception messages can contain request
+        values, upstream response fragments, or credentials. Enable only when
+        Slack is inside the same trusted logging boundary.
     :param timeout_seconds: Per-call HTTP timeout. The middleware also
         applies its own ``sink_timeout_seconds`` ceiling; the tighter of
         the two governs.
@@ -243,6 +247,7 @@ def __init__(
         *,
         sensitive_operations: frozenset[str] | None = None,
         allowed_fields: frozenset[str] = frozenset(),
+        include_error_message: bool = False,
         timeout_seconds: float = 5.0,
         allow_private_destinations: bool = False,
         allowed_destination_ports: frozenset[int] | None = None,
@@ -255,6 +260,7 @@ def __init__(
         self._webhook_url = webhook_url
         self._sensitive_operations = sensitive_operations
         self._allowed_fields = allowed_fields
+        self._include_error_message = include_error_message
         self._timeout = timeout_seconds
         self._allow_private = allow_private_destinations
         self._allowed_ports = allowed_destination_ports
@@ -313,7 +319,7 @@ def _format(self, event: AuditEvent) -> str:
             parts.append(f"request_id={event.request_id}")
         if not event.success and event.error_type:
             parts.append(f"error={event.error_type}")
-            if event.error_message:
+            if self._include_error_message and event.error_message:
                 parts.append(f"msg={event.error_message}")
         if self._allowed_fields:
             filtered = {k: v for k, v in event.details.items() if k in self._allowed_fields}
@@ -329,6 +335,7 @@ def make_audit_middleware(
     sinks: Sequence[AuditSink],
     *,
     sink_timeout_seconds: float = 5.0,
+    include_error_message: bool = False,
 ) -> SkillMiddleware:
     """Compose one or more :class:`AuditSink` instances into a
     :data:`~adcp.server.SkillMiddleware`.
@@ -396,7 +403,7 @@ async def audit_middleware(
                     tenant_id=context.tenant_id,
                     request_id=context.request_id,
                     error_type=type(exc).__name__,
-                    error_message=str(exc)[:200],
+                    error_message=str(exc)[:200] if include_error_message else None,
                 ),
                 timeout_seconds=sink_timeout_seconds,
             )
diff --git a/src/adcp/decisioning/property_list.py b/src/adcp/decisioning/property_list.py
index be696e0b..b03ebeb9 100644
--- a/src/adcp/decisioning/property_list.py
+++ b/src/adcp/decisioning/property_list.py
@@ -21,7 +21,9 @@
 from __future__ import annotations
 
 import logging
+import re
 from typing import Any, Protocol, runtime_checkable
+from urllib.parse import urlsplit
 
 logger = logging.getLogger(__name__)
 
@@ -35,12 +37,10 @@ class PropertyListFetcher(Protocol):
     reference.  Adopters plug in their own HTTP client — the framework ships
     no hidden HTTP dependency.
 
-    Typical implementation::
+    Implementations MUST pin delivery to the validated IP, disable redirects
+    and environment proxies, and URL-encode ``list_id``. A safe httpx pattern::
 
         class MyFetcher:
-            def __init__(self, client: httpx.AsyncClient) -> None:
-                self._client = client
-
             async def fetch(
                 self,
                 agent_url: str,
@@ -48,13 +48,17 @@ async def fetch(
                 *,
                 auth_token: str | None = None,
             ) -> list[str]:
+                url = f"{agent_url.rstrip('/')}/property-lists/{quote(list_id, safe='')}"
+                transport = build_async_ip_pinned_transport(url)
                 headers = {"Authorization": f"Bearer {auth_token}"} if auth_token else {}
-                resp = await self._client.get(
-                    f"{agent_url}/property-lists/{list_id}",
-                    headers=headers,
-                )
-                resp.raise_for_status()
-                return resp.json()["property_ids"]
+                async with httpx.AsyncClient(
+                    transport=transport,
+                    follow_redirects=False,
+                    trust_env=False,
+                ) as client:
+                    resp = await client.get(url, headers=headers)
+                    resp.raise_for_status()
+                    return resp.json()["property_ids"]
 
     Wire the fetcher via::
 
@@ -95,33 +99,64 @@ async def resolve_property_list(
         ``auth_token`` is never included in the error details.
     """
     from adcp.decisioning.types import AdcpError
+    from adcp.webhooks import (
+        WebhookDestinationPolicy,
+        WebhookDestinationValidationError,
+        validate_webhook_destination_url,
+    )
 
     list_id: str = ref.list_id
     agent_url: str = str(ref.agent_url)
     auth_token: str | None = getattr(ref, "auth_token", None)
 
+    if not re.fullmatch(r"[A-Za-z0-9._~-]+", list_id):
+        raise AdcpError(
+            "INVALID_REQUEST",
+            message="Property list_id must be a single URL-safe path segment",
+            recovery="correctable",
+            details={"list_id": list_id},
+        )
+
     try:
-        ids = await fetcher.fetch(agent_url, list_id, auth_token=auth_token)
+        validation = validate_webhook_destination_url(
+            agent_url,
+            policy=WebhookDestinationPolicy.production(),
+            field="property_list.agent_url",
+        )
+    except WebhookDestinationValidationError as exc:
+        raise AdcpError(
+            "INVALID_REQUEST",
+            message="Property list agent_url failed destination policy",
+            recovery="correctable",
+            details={"reason": exc.reason},
+        ) from None
+
+    parsed = urlsplit(validation.original_url)
+    safe_origin = f"{parsed.scheme}://{parsed.hostname or ''}"
+    if parsed.port is not None:
+        safe_origin += f":{parsed.port}"
+
+    try:
+        ids = await fetcher.fetch(validation.original_url, list_id, auth_token=auth_token)
         return set(ids)
     except Exception as exc:
-        # Log the raw exception server-side; never include it in the wire
-        # error message — the exception repr may carry auth_token or other
-        # credential-shaped values from the upstream HTTP response.
+        # Exception text may carry auth_token or credential-shaped upstream
+        # values. Log only the class and deliberately omit the exception chain.
         logger.warning(
-            "[adcp.property_list] fetch failed for list_id=%r agent_url=%r: %s",
+            "[adcp.property_list] fetch failed for list_id=%r agent_origin=%r (%s)",
             list_id,
-            agent_url,
-            exc,
+            safe_origin,
+            type(exc).__name__,
         )
         raise AdcpError(
             "SERVICE_UNAVAILABLE",
             message=(
                 f"Property list fetch failed for list_id={list_id!r} "
-                f"from agent_url={agent_url!r}"
+                f"from agent_origin={safe_origin!r}"
             ),
             recovery="transient",
-            details={"list_id": list_id, "agent_url": agent_url},
-        ) from exc
+            details={"list_id": list_id, "agent_origin": safe_origin},
+        ) from None
 
 
 def filter_products_by_property_list(
@@ -177,9 +212,7 @@ def _product_matches(product: Any, allowed: set[str]) -> bool:
 
         if st == "by_id":
             raw_ids: list[Any] = list(getattr(pp, "property_ids", None) or [])
-            product_ids = {
-                (pid.root if hasattr(pid, "root") else str(pid)) for pid in raw_ids
-            }
+            product_ids = {(pid.root if hasattr(pid, "root") else str(pid)) for pid in raw_ids}
             if permissive:
                 if product_ids & allowed:
                     logger.debug(
@@ -262,9 +295,7 @@ async def maybe_apply_property_list_filter(
     products: list[Any] = list(getattr(response, "products", None) or [])
     filtered = filter_products_by_property_list(products, allowed)
 
-    return response.model_copy(
-        update={"products": filtered, "property_list_applied": True}
-    )
+    return response.model_copy(update={"products": filtered, "property_list_applied": True})
 
 
 def validate_property_list_config(
diff --git a/src/adcp/server/a2a_push_security.py b/src/adcp/server/a2a_push_security.py
new file mode 100644
index 00000000..b205bd45
--- /dev/null
+++ b/src/adcp/server/a2a_push_security.py
@@ -0,0 +1,84 @@
+"""Shared A2A push-notification destination and identity policy."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any
+
+from a2a.utils.errors import InvalidParamsError
+
+from adcp.signing._idna_canonicalize import canonicalize_host
+from adcp.webhooks import WebhookDestinationPolicy, WebhookDestinationValidationError
+from adcp.webhooks import validate_webhook_destination_url as validate_destination
+
+
+def normalize_allowed_push_hosts(hosts: Iterable[str]) -> frozenset[str]:
+    """Canonicalize an operator-managed set of stable destination hosts."""
+    try:
+        return frozenset(canonicalize_host(host.strip()) for host in hosts if host.strip())
+    except (UnicodeError, ValueError) as exc:
+        raise WebhookDestinationValidationError(
+            "allowed push-notification hostname is invalid",
+            reason="invalid_allowed_hostname",
+            field="allowed_destination_hosts",
+        ) from exc
+
+
+def scope_from_server_context(context: Any | None) -> str | None:
+    """Return a verified authenticated principal name, or ``None``."""
+    user = getattr(context, "user", None) if context is not None else None
+    if user is None or not getattr(user, "is_authenticated", False):
+        return None
+    user_name = getattr(user, "user_name", None)
+    return user_name if isinstance(user_name, str) and user_name else None
+
+
+def validate_push_notification_url(
+    url: str,
+    *,
+    allowed_hosts: frozenset[str],
+    allowed_ports: frozenset[int] | None = None,
+) -> None:
+    """Validate a callback through the SDK's shared SSRF classifier."""
+    canonical_allowed_hosts = normalize_allowed_push_hosts(allowed_hosts)
+    validation = validate_destination(
+        url,
+        policy=WebhookDestinationPolicy.production(
+            allowed_destination_ports=allowed_ports,
+        ),
+        field="push_notification_config.url",
+    )
+    if validation.hostname not in canonical_allowed_hosts:
+        raise WebhookDestinationValidationError(
+            "push notification URL hostname is not in allowed_destination_hosts",
+            reason="hostname_not_allowed",
+            field="push_notification_config.url",
+            url=url,
+            effective_url=validation.effective_url,
+            policy=validation.policy,
+        )
+
+
+def validate_a2a_push_notification_url(
+    url: str,
+    *,
+    allowed_hosts: frozenset[str],
+    allowed_ports: frozenset[int] | None = None,
+) -> None:
+    """Validate a callback and map policy failures to the A2A wire error."""
+    try:
+        validate_push_notification_url(
+            url,
+            allowed_hosts=allowed_hosts,
+            allowed_ports=allowed_ports,
+        )
+    except WebhookDestinationValidationError as exc:
+        raise InvalidParamsError(message=str(exc), data=exc.to_error()) from None
+
+
+__all__ = [
+    "normalize_allowed_push_hosts",
+    "scope_from_server_context",
+    "validate_a2a_push_notification_url",
+    "validate_push_notification_url",
+]
diff --git a/tests/test_a2a_push_security.py b/tests/test_a2a_push_security.py
new file mode 100644
index 00000000..79430fc3
--- /dev/null
+++ b/tests/test_a2a_push_security.py
@@ -0,0 +1,243 @@
+"""A2A push destination policy and SQLAlchemy example parity tests."""
+
+from __future__ import annotations
+
+import socket
+
+import pytest
+from a2a.auth.user import UnauthenticatedUser, User
+from a2a.server.context import ServerCallContext
+from a2a.types import TaskPushNotificationConfig
+from a2a.utils.errors import InvalidParamsError
+
+from adcp.server.a2a_push_security import (
+    normalize_allowed_push_hosts,
+    validate_push_notification_url,
+)
+
+
+class _AuthenticatedUser(User):
+    def __init__(self, user_name: str) -> None:
+        self._user_name = user_name
+
+    @property
+    def is_authenticated(self) -> bool:
+        return True
+
+    @property
+    def user_name(self) -> str:
+        return self._user_name
+
+
+@pytest.fixture(autouse=True)
+def _resolve_example_hosts(monkeypatch: pytest.MonkeyPatch):
+    original = socket.getaddrinfo
+
+    def resolve(host: str, port: object, *args: object, **kwargs: object):
+        if host.rstrip(".").endswith(".example") or host.rstrip(".") == "example.com":
+            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))]
+        return original(host, port, *args, **kwargs)
+
+    monkeypatch.setattr(socket, "getaddrinfo", resolve)
+
+
+def test_push_url_canonicalizes_idna_case_and_trailing_dot() -> None:
+    allowed = normalize_allowed_push_hosts(["BÜCHER.Example."])
+    assert allowed == frozenset({"xn--bcher-kva.example"})
+    validate_push_notification_url(
+        "https://xn--bcher-kva.EXAMPLE./callback",
+        allowed_hosts=frozenset({"BÜCHER.Example."}),
+    )
+
+
+@pytest.mark.parametrize(
+    ("url", "message"),
+    [
+        ("https://user@example.com/hook", "userinfo"),
+        ("https://user:password@example.com/hook", "userinfo"),
+        ("https://example.com:not-a-port/hook", "Port could not be cast"),
+        ("http://example.com/hook", "https"),
+    ],
+)
+def test_push_url_rejects_unsafe_authority_forms(url: str, message: str) -> None:
+    with pytest.raises(ValueError, match=message):
+        validate_push_notification_url(
+            url,
+            allowed_hosts=normalize_allowed_push_hosts(["example.com"]),
+        )
+
+
+@pytest.mark.parametrize("host", ["127.0.0.1", "10.0.0.1", "::1", "fe80::1"])
+def test_push_url_rejects_private_ipv4_and_ipv6_literals(host: str) -> None:
+    rendered_host = f"[{host}]" if ":" in host else host
+    with pytest.raises(ValueError, match="blocked|private|SSRF"):
+        validate_push_notification_url(
+            f"https://{rendered_host}/hook",
+            allowed_hosts=normalize_allowed_push_hosts([host]),
+        )
+
+
+@pytest.mark.parametrize(
+    "host",
+    [
+        "192.88.99.1",
+        "192.31.196.1",
+        "192.52.193.1",
+        "192.175.48.1",
+        "2001:20::1",
+        "64:ff9b::7f00:1",
+    ],
+)
+def test_push_url_rejects_globally_classified_reserved_literals(host: str) -> None:
+    """Shared SSRF policy covers ranges ``ipaddress.is_global`` misses."""
+    rendered_host = f"[{host}]" if ":" in host else host
+    with pytest.raises(ValueError, match="blocked|reserved|SSRF"):
+        validate_push_notification_url(
+            f"https://{rendered_host}/hook",
+            allowed_hosts=normalize_allowed_push_hosts([host]),
+        )
+
+
+def test_push_url_allows_nonstandard_tls_port_by_default() -> None:
+    validate_push_notification_url(
+        "https://example.com:9443/hook",
+        allowed_hosts=frozenset({"example.com"}),
+    )
+
+
+@pytest.mark.asyncio
+async def test_sqlite_push_store_uses_a2a_context_for_tenant_isolation(tmp_path) -> None:
+    import examples.a2a_db_tasks as example
+
+    store = example.SqlitePushNotificationConfigStore(
+        tmp_path / "push.db",
+        allowed_destination_hosts=frozenset({"callback.example"}),
+    )
+    tenant_a = ServerCallContext(user=_AuthenticatedUser("tenant-a"))
+    tenant_b = ServerCallContext(user=_AuthenticatedUser("tenant-b"))
+    config = TaskPushNotificationConfig(
+        id="cfg-1",
+        task_id="task-shared",
+        url="https://callback.example/hook",
+    )
+
+    await store.set_info("task-shared", config, tenant_a)
+    assert [item.id for item in await store.get_info("task-shared", tenant_a)] == ["cfg-1"]
+    assert await store.get_info("task-shared", tenant_b) == []
+
+    await store.delete_info("task-shared", tenant_b)
+    assert [item.id for item in await store.get_info("task-shared", tenant_a)] == ["cfg-1"]
+
+
+@pytest.mark.asyncio
+async def test_sqlite_explicit_unauthenticated_context_cannot_inherit_ambient_scope(
+    tmp_path,
+) -> None:
+    import examples.a2a_db_tasks as example
+
+    store = example.SqlitePushNotificationConfigStore(
+        tmp_path / "push.db",
+        scope_provider=lambda: "tenant-a",
+        allowed_destination_hosts=frozenset({"callback.example"}),
+    )
+    tenant_a = ServerCallContext(user=_AuthenticatedUser("tenant-a"))
+    unauthenticated = ServerCallContext(user=UnauthenticatedUser())
+    config = TaskPushNotificationConfig(
+        id="cfg-1",
+        task_id="task-shared",
+        url="https://callback.example/hook",
+    )
+
+    await store.set_info("task-shared", config, tenant_a)
+    assert await store.get_info("task-shared", unauthenticated) == []
+
+
+@pytest.mark.asyncio
+async def test_sqlalchemy_push_store_matches_a2a_v1_set_get_delete_contract() -> None:
+    import examples.a2a_sqlalchemy_tasks as example
+
+    session_factory = example.build_engine_and_sessions(database_url="sqlite:///:memory:")
+    store = example.SqlAlchemyPushNotificationConfigStore(
+        session_factory,
+        allowed_destination_hosts=frozenset({"callback.example"}),
+    )
+    tenant_a = ServerCallContext(user=_AuthenticatedUser("tenant-a"))
+    tenant_b = ServerCallContext(user=_AuthenticatedUser("tenant-b"))
+    first = TaskPushNotificationConfig(
+        id="cfg-1",
+        task_id="task-1",
+        url="https://callback.example/first",
+    )
+    second = TaskPushNotificationConfig(
+        id="cfg-2",
+        task_id="task-1",
+        url="https://callback.example/second",
+    )
+    await store.set_info("task-1", first, tenant_a)
+    await store.set_info("task-1", second, tenant_a)
+
+    stored = await store.get_info("task-1", tenant_a)
+    assert {config.id for config in stored} == {"cfg-1", "cfg-2"}
+    assert {config.url for config in stored} == {
+        "https://callback.example/first",
+        "https://callback.example/second",
+    }
+    assert await store.get_info("task-1", tenant_b) == []
+
+    await store.delete_info("task-1", tenant_b)
+    assert len(await store.get_info("task-1", tenant_a)) == 2
+
+    await store.delete_info("task-1", tenant_a, "cfg-1")
+    remaining = await store.get_info("task-1", tenant_a)
+    assert [config.id for config in remaining] == ["cfg-2"]
+
+    await store.delete_info("task-1", tenant_a)
+    assert await store.get_info("task-1", tenant_a) == []
+
+
+@pytest.mark.asyncio
+async def test_sqlalchemy_explicit_unauthenticated_context_cannot_inherit_ambient_scope() -> None:
+    import examples.a2a_sqlalchemy_tasks as example
+
+    session_factory = example.build_engine_and_sessions(database_url="sqlite:///:memory:")
+    store = example.SqlAlchemyPushNotificationConfigStore(
+        session_factory,
+        allowed_destination_hosts=frozenset({"callback.example"}),
+    )
+    tenant_a = ServerCallContext(user=_AuthenticatedUser("tenant-a"))
+    unauthenticated = ServerCallContext(user=UnauthenticatedUser())
+    config = TaskPushNotificationConfig(
+        id="cfg-1",
+        task_id="task-shared",
+        url="https://callback.example/hook",
+    )
+    token = example._push_config_scope.set("tenant-a")
+    try:
+        await store.set_info("task-shared", config, tenant_a)
+        assert [item.id for item in await store.get_info("task-shared", None)] == ["cfg-1"]
+        assert await store.get_info("task-shared", unauthenticated) == []
+    finally:
+        example._push_config_scope.reset(token)
+
+
+@pytest.mark.asyncio
+async def test_sqlalchemy_push_store_defaults_to_deny_all_destinations() -> None:
+    import examples.a2a_sqlalchemy_tasks as example
+
+    session_factory = example.build_engine_and_sessions(database_url="sqlite:///:memory:")
+    store = example.SqlAlchemyPushNotificationConfigStore(session_factory)
+    scope_token = example._push_config_scope.set("tenant-a")
+    try:
+        with pytest.raises(InvalidParamsError, match="allowed_destination_hosts") as exc:
+            await store.set_info(
+                "task-1",
+                TaskPushNotificationConfig(
+                    task_id="task-1",
+                    url="https://callback.example/hook",
+                ),
+                ServerCallContext(user=UnauthenticatedUser()),
+            )
+        assert exc.value.data is not None
+        assert exc.value.data["code"] == "INVALID_REQUEST"
+    finally:
+        example._push_config_scope.reset(scope_token)
diff --git a/tests/test_a2a_server.py b/tests/test_a2a_server.py
index a2f7a419..392de4e7 100644
--- a/tests/test_a2a_server.py
+++ b/tests/test_a2a_server.py
@@ -4,6 +4,7 @@
 
 import contextlib
 import json
+import socket
 import sys
 from typing import Any
 
@@ -1106,7 +1107,10 @@ async def test_sqlite_push_config_store_isolates_scopes_by_contextvar():
 
     with tempfile.TemporaryDirectory() as tmp:
         db = Path(tmp) / "push.db"
-        store = mod.SqlitePushNotificationConfigStore(db_path=db)
+        store = mod.SqlitePushNotificationConfigStore(
+            db_path=db,
+            allowed_destination_hosts=frozenset({"callback.tenant-a.example"}),
+        )
         scope_var = mod._current_push_config_scope
 
         cfg = PushNotificationConfig(id="cfg-1", url="https://callback.tenant-a.example/webhook")
@@ -1138,13 +1142,51 @@ async def test_sqlite_push_config_store_isolates_scopes_by_contextvar():
         tok_a2 = scope_var.set("tenant-a")
         try:
             still_a = await store.get_info("task-shared")
-            assert len(still_a) == 1, (
-                "SqlitePushNotificationConfigStore cross-scope delete " "removed tenant A's config."
-            )
+            assert (
+                len(still_a) == 1
+            ), "SqlitePushNotificationConfigStore cross-scope delete removed tenant A's config."
         finally:
             scope_var.reset(tok_a2)
 
 
+async def test_sqlite_push_config_store_rejects_untrusted_destinations():
+    """Push callback URLs fail closed before attacker-controlled storage."""
+    import importlib.util
+    import tempfile
+    from pathlib import Path
+
+    from a2a.types import TaskPushNotificationConfig as PushNotificationConfig
+    from a2a.utils.errors import InvalidParamsError
+
+    example_path = Path(__file__).parent.parent / "examples" / "a2a_db_tasks.py"
+    spec = importlib.util.spec_from_file_location("_a2a_db_tasks_ex_ssrf", example_path)
+    assert spec is not None and spec.loader is not None
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+
+    with tempfile.TemporaryDirectory() as tmp:
+        store = mod.SqlitePushNotificationConfigStore(
+            db_path=Path(tmp) / "push.db",
+            scope_provider=lambda: "tenant-a",
+        )
+        with pytest.raises(InvalidParamsError, match="allowed_destination_hosts"):
+            await store.set_info(
+                "task-1",
+                PushNotificationConfig(url="https://attacker.example/hook"),
+            )
+
+        private_store = mod.SqlitePushNotificationConfigStore(
+            db_path=Path(tmp) / "private.db",
+            scope_provider=lambda: "tenant-a",
+            allowed_destination_hosts=frozenset({"127.0.0.1"}),
+        )
+        with pytest.raises(InvalidParamsError, match="reserved range"):
+            await private_store.set_info(
+                "task-1",
+                PushNotificationConfig(url="https://127.0.0.1/hook"),
+            )
+
+
 @pytest.mark.skipif(
     sys.version_info < (3, 11),
     reason="a2a-sdk starlette integration requires Python 3.11+",
@@ -1197,6 +1239,58 @@ async def test_custom_push_config_store_receives_sets_from_handler():
     )
 
 
+@pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="a2a-sdk starlette integration requires Python 3.11+",
+)
+async def test_push_config_destination_policy_reaches_jsonrpc_error_envelope(tmp_path):
+    """Production A2A dispatch rejects unsafe callbacks as invalid params."""
+    import httpx
+
+    import examples.a2a_db_tasks as example
+
+    task_store = _RecordingTaskStore()
+    await task_store.save(
+        Task(id="task-1", context_id="ctx-1", status=pb.TaskStatus(state="working")),
+        _empty_call_context(),
+    )
+    push_store = example.SqlitePushNotificationConfigStore(
+        tmp_path / "push.db",
+        allowed_destination_hosts=frozenset({"127.0.0.1"}),
+    )
+    app = create_a2a_server(
+        _TestHandler(),
+        name="push-policy-wire",
+        task_store=task_store,
+        push_config_store=push_store,
+    )
+
+    transport = httpx.ASGITransport(app=app, raise_app_exceptions=True)
+    async with httpx.AsyncClient(
+        transport=transport,
+        base_url="http://testserver",
+    ) as client:
+        response = await client.post(
+            "/",
+            headers={"A2A-Version": "1.0"},
+            json={
+                "jsonrpc": "2.0",
+                "id": "push-1",
+                "method": "CreateTaskPushNotificationConfig",
+                "params": {
+                    "taskId": "task-1",
+                    "url": "https://127.0.0.1/hook",
+                },
+            },
+        )
+
+    assert response.status_code == 200
+    error = response.json()["error"]
+    assert error["code"] == -32602
+    assert "push_notification_config.url" in str(error)
+    assert await push_store.get_info("task-1", _empty_call_context()) == []
+
+
 async def test_sqlite_push_config_store_warns_once_on_anonymous_scope():
     """Reference impl must fail LOUD when the scope_provider returns
     None — silent fall-through to the anonymous bucket is the
@@ -1220,7 +1314,11 @@ async def test_sqlite_push_config_store_warns_once_on_anonymous_scope():
         db = Path(tmp) / "anon.db"
         # Force the anonymous path by supplying a provider that always
         # returns None.
-        store = mod.SqlitePushNotificationConfigStore(db_path=db, scope_provider=lambda: None)
+        store = mod.SqlitePushNotificationConfigStore(
+            db_path=db,
+            scope_provider=lambda: None,
+            allowed_destination_hosts=frozenset({"x.example"}),
+        )
 
         cfg = PushNotificationConfig(url="https://x.example/hook")
         with _warnings.catch_warnings(record=True) as caught:
@@ -1255,7 +1353,11 @@ async def test_sqlite_push_config_store_synthesises_config_id_when_omitted():
 
     with tempfile.TemporaryDirectory() as tmp:
         db = Path(tmp) / "uuid.db"
-        store = mod.SqlitePushNotificationConfigStore(db_path=db, scope_provider=lambda: "tenant-a")
+        store = mod.SqlitePushNotificationConfigStore(
+            db_path=db,
+            scope_provider=lambda: "tenant-a",
+            allowed_destination_hosts=frozenset({"first.example", "second.example"}),
+        )
 
         await store.set_info(
             "shared-task",
@@ -1686,3 +1788,15 @@ def composed(ctx: RequestContext) -> tuple[str | None, dict[str, Any]]:
     event = await queue.dequeue_event()
     assert isinstance(event, Task)
     assert event.status.state == pb.TaskState.TASK_STATE_COMPLETED
+
+
+@pytest.fixture(autouse=True)
+def _resolve_a2a_example_hosts(monkeypatch: pytest.MonkeyPatch):
+    original = socket.getaddrinfo
+
+    def resolve(host: str, port: object, *args: object, **kwargs: object):
+        if host.rstrip(".").endswith(".example"):
+            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))]
+        return original(host, port, *args, **kwargs)
+
+    monkeypatch.setattr(socket, "getaddrinfo", resolve)
diff --git a/tests/test_audit_sink.py b/tests/test_audit_sink.py
index 2cf5699e..d7133473 100644
--- a/tests/test_audit_sink.py
+++ b/tests/test_audit_sink.py
@@ -334,6 +334,35 @@ def _handler(request: httpx.Request) -> httpx.Response:
     assert "ops@buyer.example" not in text
 
 
+@pytest.mark.asyncio
+async def test_slack_alert_sink_omits_exception_message_by_default() -> None:
+    captured: dict[str, Any] = {}
+
+    def _handler(request: httpx.Request) -> httpx.Response:
+        captured["body"] = json.loads(request.content.decode())
+        return httpx.Response(200, text="ok")
+
+    sink = SlackAlertSink("https://hooks.slack.com/services/T/B/x")
+    event = AuditEvent(
+        operation="create_media_buy",
+        success=False,
+        occurred_at=datetime.now(UTC),
+        error_type="RuntimeError",
+        error_message="Authorization=Bearer secret-token",
+    )
+
+    with patch(
+        "adcp.signing.ip_pinned_transport.build_async_ip_pinned_transport",
+        return_value=httpx.MockTransport(_handler),
+    ):
+        await sink.record(event)
+
+    text = captured["body"]["text"]
+    assert "RuntimeError" in text
+    assert "secret-token" not in text
+    assert "Authorization" not in text
+
+
 @pytest.mark.asyncio
 async def test_slack_alert_sink_emits_only_allowlisted_details() -> None:
     captured: dict[str, Any] = {}
@@ -398,7 +427,7 @@ def _handler(request: httpx.Request) -> httpx.Response:
 @pytest.mark.asyncio
 async def test_middleware_records_on_success() -> None:
     sink = _RecordingSink()
-    middleware = make_audit_middleware([sink])
+    middleware = make_audit_middleware([sink], include_error_message=True)
 
     async def handler() -> dict[str, str]:
         return {"ok": "yes"}
@@ -420,7 +449,7 @@ async def handler() -> dict[str, str]:
 @pytest.mark.asyncio
 async def test_middleware_records_failure_and_reraises() -> None:
     sink = _RecordingSink()
-    middleware = make_audit_middleware([sink])
+    middleware = make_audit_middleware([sink], include_error_message=True)
 
     class _BoomError(RuntimeError):
         pass
@@ -441,7 +470,7 @@ async def handler() -> Any:
 @pytest.mark.asyncio
 async def test_middleware_truncates_long_error_messages() -> None:
     sink = _RecordingSink()
-    middleware = make_audit_middleware([sink])
+    middleware = make_audit_middleware([sink], include_error_message=True)
 
     async def handler() -> Any:
         raise RuntimeError("x" * 500)
diff --git a/tests/test_decisioning_property_list.py b/tests/test_decisioning_property_list.py
index 53a56ccc..34256df7 100644
--- a/tests/test_decisioning_property_list.py
+++ b/tests/test_decisioning_property_list.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import socket
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock
 
@@ -17,6 +18,19 @@
 from adcp.decisioning.types import AdcpError
 
 
+@pytest.fixture(autouse=True)
+def _resolve_example_hosts(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Give documentation-only example hosts a public test address."""
+    original = socket.getaddrinfo
+
+    def fake_getaddrinfo(host: str, port: int, *args: Any, **kwargs: Any) -> Any:
+        if host.endswith(".example.com"):
+            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", port))]
+        return original(host, port, *args, **kwargs)
+
+    monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo)
+
+
 # ---------------------------------------------------------------------------
 # Helpers — minimal wire-shape stubs
 # ---------------------------------------------------------------------------
@@ -139,7 +153,8 @@ def test_by_id_strict_partial_match_excluded(self) -> None:
             property_targeting_allowed=False,
         )
         result = filter_products_by_property_list(
-            [product], allowed_property_ids={"home"}  # missing "sports"
+            [product],
+            allowed_property_ids={"home"},  # missing "sports"
         )
         assert result == []
 
@@ -160,9 +175,7 @@ def test_by_id_permissive_any_intersection_sufficient(self) -> None:
             [_make_pp_by_id(["home", "sports"])],
             property_targeting_allowed=True,
         )
-        result = filter_products_by_property_list(
-            [product], allowed_property_ids={"sports"}
-        )
+        result = filter_products_by_property_list([product], allowed_property_ids={"sports"})
         assert result == [product]
 
     def test_by_id_permissive_no_intersection_excluded(self) -> None:
@@ -201,9 +214,7 @@ def test_mixed_by_id_and_by_tag_respects_by_id(self) -> None:
             [_make_pp_by_tag(["ctv"]), _make_pp_by_id(["home"])],
             property_targeting_allowed=True,
         )
-        result = filter_products_by_property_list(
-            [product], allowed_property_ids={"home"}
-        )
+        result = filter_products_by_property_list([product], allowed_property_ids={"home"})
         assert result == [product]
 
     def test_multiple_products_filtered_correctly(self) -> None:
@@ -252,9 +263,7 @@ def test_by_id_empty_property_ids_excluded_permissive(self) -> None:
             [_make_pp_by_id([])],
             property_targeting_allowed=True,
         )
-        result = filter_products_by_property_list(
-            [product], allowed_property_ids={"home"}
-        )
+        result = filter_products_by_property_list([product], allowed_property_ids={"home"})
         assert result == []
 
     def test_property_targeting_allowed_none_treated_as_false(self) -> None:
@@ -279,9 +288,7 @@ class TestResolvePropertyList:
     async def test_returns_set_from_fetcher(self) -> None:
         fetcher = AsyncMock(spec=PropertyListFetcher)
         fetcher.fetch = AsyncMock(return_value=["home", "sports", "news"])
-        ref = _make_property_list_ref(
-            agent_url="https://agent.example.com", list_id="list_1"
-        )
+        ref = _make_property_list_ref(agent_url="https://agent.example.com", list_id="list_1")
 
         result = await resolve_property_list(ref, fetcher=fetcher)
 
@@ -329,13 +336,28 @@ async def test_error_details_do_not_include_auth_token(self) -> None:
             await resolve_property_list(ref, fetcher=fetcher)
 
         err = exc_info.value
-        # details should include list_id and agent_url but NOT auth_token
+        # details include only a sanitized origin, never the full URL or token.
         assert err.details is not None
         assert "list_id" in err.details
-        assert "agent_url" in err.details
+        assert "agent_origin" in err.details
         assert "auth_token" not in err.details
         assert "secret_bearer_token" not in str(err.details)
 
+    @pytest.mark.asyncio
+    async def test_fetch_failure_log_omits_exception_text(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        secret = "secret_bearer_token"
+        fetcher = AsyncMock(spec=PropertyListFetcher)
+        fetcher.fetch = AsyncMock(side_effect=RuntimeError(f"Authorization=Bearer {secret}"))
+        ref = _make_property_list_ref(auth_token=secret)
+
+        with caplog.at_level("WARNING"), pytest.raises(AdcpError):
+            await resolve_property_list(ref, fetcher=fetcher)
+
+        assert "RuntimeError" in caplog.text
+        assert secret not in caplog.text
+
 
 # ---------------------------------------------------------------------------
 # validate_property_list_config
@@ -471,9 +493,7 @@ async def test_model_copy_used_not_in_place_mutation(self) -> None:
         """Response is updated via model_copy, not in-place mutation."""
         p = _make_product("p1", [_make_pp_all()])
         original_response = _make_response([p])
-        original_response.model_copy = MagicMock(
-            side_effect=original_response.model_copy
-        )
+        original_response.model_copy = MagicMock(side_effect=original_response.model_copy)
         params = MagicMock()
         params.property_list = _make_property_list_ref()