feat(decisioning): F12 — auto-emit completion webhook on sync mutating responses (#331)

* feat(decisioning): F12 — auto-emit completion webhook on sync mutating responses

Sync ``create_media_buy``, ``update_media_buy``, ``sync_creatives``
responses now auto-fire a completion webhook when the buyer supplied
``push_notification_config.url``. Previously only the HITL
TaskHandoff path emitted, so sync responses left buyers polling.

Mirrors the JS-side ``emitSyncCompletionWebhook`` implementation
(commits ``8dc427f9`` and ``7a887dfa`` on
``src/lib/server/decisioning/runtime/from-platform.ts``). Wire-format
is identical: ``task_type``, ``status: 'completed'``, ``result``
field carrying the projected sync response, echoed ``token`` via
``X-AdCP-Push-Token`` header. ``task_id`` is synthesized as
``f"sync-{uuid4()}"`` since sync responses don't allocate a registry
task; buyers correlate via the resource ids embedded in ``result``.

New module ``adcp.decisioning.webhook_emit``:

* ``SPEC_WEBHOOK_TASK_TYPES`` — closed 20-value set mirroring the
  on-disk spec enum at ``schemas/cache/enums/task-type.json``. The
  ``test_spec_webhook_task_types_matches_schema_cache`` test pins
  the constant so out-of-band drift surfaces in CI.
* ``maybe_emit_sync_completion`` — fire-and-forget gate. Skips when
  disabled, no sender wired, no push URL on the request, or the
  tool isn't in the spec enum (logged warning so adopters notice
  they extended the surface beyond spec).
* ``_BACKGROUND_WEBHOOK_TASKS`` — module-level strong-ref pin so the
  asyncio loop's weak-ref behavior doesn't garbage-collect in-flight
  emissions mid-flight. Mirrors the same pattern in
  ``dispatch._BACKGROUND_HANDOFF_TASKS``.

**Fire-and-forget posture (DoS defense).** Webhook delivery runs in a
background asyncio task; the sync response returns inline
immediately. A buyer-supplied slowloris webhook URL must not be able
to hold the seller's request worker for the full retry budget — the
JS round-2 fix at ``7a887dfa`` documented this DoS vector and Python
preserves the same posture from the start.

**TaskHandoff path doesn't double-fire.** The
``_maybe_auto_emit_sync_completion`` helper detects the projected
Submitted envelope (``status == 'submitted'`` shape) and skips
delivery. The HITL path's registry completion emits its own webhook
on terminal state.

Configuration on ``create_adcp_server_from_platform`` and ``serve``:

* ``webhook_sender: WebhookSender | None = None`` — BYO emitter.
  ``None`` silently disables auto-emit.
* ``auto_emit_completion_webhooks: bool = True`` — default-on.
  Adopters who emit webhooks manually inside their handlers pass
  ``False`` to avoid duplicate delivery.

21 new tests cover: drift-guard against the on-disk schema cache,
URL+token extraction (incl. dict-params test fixtures), gate skips
(disabled, no sender, no URL, tool outside spec enum, no running
loop), happy-path delivery via ``WebhookSender.send_mcp``, token
echo via ``X-AdCP-Push-Token`` header, delivery-failure swallow,
sync-success fires, TaskHandoff doesn't double-fire, opt-out
suppresses, default-on, no-sender silent, sync_creatives fires too.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(decisioning,webhooks): F12 round-2 — token via payload + exception isolation

Two P0 expert-review findings on PR #331:

P0-1 (cross-language wire divergence): the buyer's
``push_notification_config.token`` was being echoed via
``X-AdCP-Push-Token`` HTTP header. Per
``schemas/cache/core/push_notification_config.json`` ("Echoed back in
webhook payload to validate request authenticity") AND the JS
reference impl (``buildTaskWebhookPayload`` in
``src/lib/server/decisioning/runtime/from-platform.ts``), the token
belongs on ``payload.token``. Buyers validating against the spec read
``body.token``, not custom headers — header echo would silently fail
their auth check.

Fix: extend ``create_mcp_webhook_payload`` and
``WebhookSender.send_mcp`` to accept ``token`` and write it onto the
payload. Update F12's ``_emit_sync_completion_webhook`` to pass
``token=`` through instead of building ``extra_headers``.
Cross-language wire-parity restored.

P0-2 (exception isolation): ``maybe_emit_sync_completion`` runs
AFTER the platform method's successful return. ANY exception in the
gate body — extraction quirk on a weird ``params`` shape,
``loop.create_task`` failure — would propagate to the handler shim
and lose the buyer's sync response.

Fix: wrap the entire gate body in ``try/except Exception``;
logged-and-swallowed. Last-line defense ensures the post-success
path can never poison the buyer's response.

P1 fixes folded in:

* Submitted-shape detection tightened to the EXACT 2-key dict
  ``{"task_id", "status"}`` (not the loose
  ``status == "submitted"`` predicate). An adopter who legitimately
  returns a sync ``{"status": "submitted", ...}`` with extra metadata
  (queue acceptance) now correctly gets the auto-emit fired.
* No-running-loop branch bumped from ``logger.debug`` to
  ``logger.warning`` — production code landing here is mis-wired and
  should be visible.

Round-2 tests added:

* ``test_handler_returns_before_webhook_delivers`` — pins the
  non-blocking invariant (sync response returns before webhook
  delivery completes).
* ``test_concurrent_emissions_dont_corrupt_strong_ref_set`` — 100
  concurrent emissions exercising the
  ``_BACKGROUND_WEBHOOK_TASKS`` add/discard pattern.
* ``test_handler_does_not_skip_loose_submitted_shape`` — pins the
  tightened submitted-shape detection.
* ``test_gate_swallows_unexpected_exceptions`` — pins the
  exception-isolation invariant via a sender that raises on
  attribute access.

25 F12 tests pass total (up from 21); 2208 total tests pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bokelley

src/adcp/decisioning/handler.py

@@ -39,6 +39,7 @@
     _build_request_context,
     _invoke_platform_method,
 )
+from adcp.decisioning.webhook_emit import maybe_emit_sync_completion
 from adcp.server.base import ADCPHandler, ToolContext
 
 if TYPE_CHECKING:
@@ -70,6 +71,7 @@
         UpdateMediaBuyRequest,
         UpdateMediaBuySuccessResponse,
     )
+    from adcp.webhook_sender import WebhookSender
 
 
 # ---------------------------------------------------------------------------
@@ -141,13 +143,17 @@ def __init__(
         registry: TaskRegistry,
         state_reader: StateReader | None = None,
         resource_resolver: ResourceResolver | None = None,
+        webhook_sender: WebhookSender | None = None,
+        auto_emit_completion_webhooks: bool = True,
     ) -> None:
         super().__init__()
         self._platform = platform
         self._executor = executor
         self._registry = registry
         self._state_reader = state_reader
         self._resource_resolver = resource_resolver
+        self._webhook_sender = webhook_sender
+        self._auto_emit_completion_webhooks = auto_emit_completion_webhooks
 
     # ----- account resolution helper -----
 
@@ -211,6 +217,43 @@ def _extract_auth_info(ctx: ToolContext) -> AuthInfo | None:
             )
         return None
 
+    def _maybe_auto_emit_sync_completion(
+        self,
+        method_name: str,
+        params: Any,
+        result: Any,
+    ) -> None:
+        """Fire the F12 sync-completion webhook if applicable.
+
+        Skips TaskHandoff projections — those go through the registry
+        completion path which emits its own webhook on terminal state.
+        The auto-emit fires on the sync-success arm only, mirroring the
+        JS-side ``routeIfHandoff`` logic at
+        ``src/lib/server/decisioning/runtime/from-platform.ts``.
+
+        TaskHandoff projection returns the exact 2-key dict ``{"task_id":
+        ..., "status": "submitted"}`` from ``_project_handoff``; we
+        match the full key set rather than the loose ``status ==
+        "submitted"`` predicate so an adopter who legitimately returns a
+        sync ``{"status": "submitted", ...}`` (e.g., synchronous queue
+        acceptance with extra metadata) still gets the auto-emit.
+        """
+        if (
+            isinstance(result, dict)
+            and set(result.keys()) == {"task_id", "status"}
+            and result.get("status") == "submitted"
+        ):
+            # TaskHandoff projection — registry completion path emits
+            # its own webhook on terminal state.
+            return
+        maybe_emit_sync_completion(
+            sender=self._webhook_sender,
+            enabled=self._auto_emit_completion_webhooks,
+            method_name=method_name,
+            params=params,
+            result=result,
+        )
+
     def _build_ctx(
         self,
         tool_ctx: ToolContext,
@@ -260,17 +303,16 @@ async def create_media_buy(  # type: ignore[override]
         tool_ctx = context or ToolContext()
         account = await self._resolve_account(params.account, tool_ctx)
         ctx = self._build_ctx(tool_ctx, account)
-        return cast(
-            "CreateMediaBuySuccessResponse",
-            await _invoke_platform_method(
-                self._platform,
-                "create_media_buy",
-                params,
-                ctx,
-                executor=self._executor,
-                registry=self._registry,
-            ),
+        result = await _invoke_platform_method(
+            self._platform,
+            "create_media_buy",
+            params,
+            ctx,
+            executor=self._executor,
+            registry=self._registry,
         )
+        self._maybe_auto_emit_sync_completion("create_media_buy", params, result)
+        return cast("CreateMediaBuySuccessResponse", result)
 
     async def update_media_buy(  # type: ignore[override]
         self,
@@ -285,18 +327,17 @@ async def update_media_buy(  # type: ignore[override]
         tool_ctx = context or ToolContext()
         account = await self._resolve_account(params.account, tool_ctx)
         ctx = self._build_ctx(tool_ctx, account)
-        return cast(
-            "UpdateMediaBuySuccessResponse",
-            await _invoke_platform_method(
-                self._platform,
-                "update_media_buy",
-                params,
-                ctx,
-                executor=self._executor,
-                registry=self._registry,
-                arg_projector={"media_buy_id": params.media_buy_id, "patch": params},
-            ),
+        result = await _invoke_platform_method(
+            self._platform,
+            "update_media_buy",
+            params,
+            ctx,
+            executor=self._executor,
+            registry=self._registry,
+            arg_projector={"media_buy_id": params.media_buy_id, "patch": params},
         )
+        self._maybe_auto_emit_sync_completion("update_media_buy", params, result)
+        return cast("UpdateMediaBuySuccessResponse", result)
 
     async def sync_creatives(  # type: ignore[override]
         self,
@@ -306,17 +347,16 @@ async def sync_creatives(  # type: ignore[override]
         tool_ctx = context or ToolContext()
         account = await self._resolve_account(params.account, tool_ctx)
         ctx = self._build_ctx(tool_ctx, account)
-        return cast(
-            "SyncCreativesSuccessResponse",
-            await _invoke_platform_method(
-                self._platform,
-                "sync_creatives",
-                params,
-                ctx,
-                executor=self._executor,
-                registry=self._registry,
-            ),
+        result = await _invoke_platform_method(
+            self._platform,
+            "sync_creatives",
+            params,
+            ctx,
+            executor=self._executor,
+            registry=self._registry,
         )
+        self._maybe_auto_emit_sync_completion("sync_creatives", params, result)
+        return cast("SyncCreativesSuccessResponse", result)
 
     async def get_media_buy_delivery(  # type: ignore[override]
         self,

src/adcp/decisioning/serve.py

@@ -43,6 +43,7 @@
     from adcp.decisioning.resolve import ResourceResolver
     from adcp.decisioning.state import StateReader
     from adcp.decisioning.task_registry import TaskRegistry
+    from adcp.webhook_sender import WebhookSender
 
 
 def _is_production_env() -> bool:
@@ -75,6 +76,8 @@ def create_adcp_server_from_platform(
     registry: TaskRegistry | None = None,
     state_reader: StateReader | None = None,
     resource_resolver: ResourceResolver | None = None,
+    webhook_sender: WebhookSender | None = None,
+    auto_emit_completion_webhooks: bool = True,
 ) -> tuple[PlatformHandler, ThreadPoolExecutor, TaskRegistry]:
     """Build the :class:`PlatformHandler` + supporting wiring from a
     :class:`DecisioningPlatform`.
@@ -117,6 +120,24 @@ def create_adcp_server_from_platform(
         (D15 — async framework-mediated fetches). Default is the
         v6.0 stub (raises ``NotImplementedError`` with a pointer to
         v6.1).
+    :param webhook_sender: Bring-your-own
+        :class:`adcp.webhook_sender.WebhookSender` for sync-completion
+        and HITL-completion webhook delivery. Default ``None`` — when
+        unset, sync-completion auto-emit is a silent no-op (no URL to
+        deliver to, framework can't synthesize a sender). Adopters
+        wiring webhook delivery pass a configured sender (with their
+        signing key, IP-pinned transport, etc.).
+    :param auto_emit_completion_webhooks: F12 feature gate. When
+        ``True`` (default), the framework auto-fires a completion
+        webhook on the sync-success arm of mutating tools whenever the
+        request supplied ``push_notification_config.url`` AND the tool
+        is in :data:`adcp.decisioning.webhook_emit.SPEC_WEBHOOK_TASK_TYPES`.
+        Buyers passing the URL expect notification regardless of
+        whether the seller routed sync vs HITL. Set ``False`` for
+        adopters who emit webhooks manually inside their handlers
+        (avoid duplicate delivery; idempotency-key dedup at the
+        receiver would handle it but explicit suppression matches the
+        v5 manual-emit posture for adopters mid-migration).
 
     :raises ValueError: when ``executor`` and ``thread_pool_size`` are
         both supplied (D5 mutually-exclusive validation).
@@ -213,6 +234,8 @@ def create_adcp_server_from_platform(
         registry=registry,
         state_reader=state_reader,
         resource_resolver=resource_resolver,
+        webhook_sender=webhook_sender,
+        auto_emit_completion_webhooks=auto_emit_completion_webhooks,
     )
     return handler, executor, registry
 
@@ -226,6 +249,8 @@ def serve(
     registry: TaskRegistry | None = None,
     state_reader: StateReader | None = None,
     resource_resolver: ResourceResolver | None = None,
+    webhook_sender: WebhookSender | None = None,
+    auto_emit_completion_webhooks: bool = True,
     advertise_all: bool = False,
     **serve_kwargs: Any,
 ) -> None:
@@ -246,6 +271,14 @@ def serve(
         :class:`InMemoryTaskRegistry` (gated for production).
     :param state_reader: Custom :class:`StateReader` impl (D15).
     :param resource_resolver: Custom :class:`ResourceResolver` impl (D15).
+    :param webhook_sender: BYO :class:`adcp.webhook_sender.WebhookSender`
+        for completion webhook delivery (sync auto-emit + HITL terminal).
+        ``None`` disables auto-emit silently.
+    :param auto_emit_completion_webhooks: F12 — auto-fire a completion
+        webhook on the sync-success arm of mutating tools when the
+        request supplied ``push_notification_config.url``. Default
+        ``True``. Set ``False`` for adopters who emit webhooks
+        manually inside their handlers.
     :param advertise_all: Forwarded to :func:`adcp.server.serve`. When
         ``True``, ``tools/list`` advertises every method on the
         handler regardless of override status. Default ``False`` —
@@ -267,6 +300,8 @@ def serve(
         registry=registry,
         state_reader=state_reader,
         resource_resolver=resource_resolver,
+        webhook_sender=webhook_sender,
+        auto_emit_completion_webhooks=auto_emit_completion_webhooks,
     )
 
     server_name = name or type(platform).__name__