feat(decisioning): framework-wide async-completion webhooks + #930 review fixes

Deliver the spec-required terminal completion / failure webhook on the
async (handoff) path of every spec-eligible verb when the buyer
registered push_notification_config (adcp#5389). Previously the SDK
emitted only on the sync path; create_media_buy / get_products /
get_signals and all other async ops left a registered push URL silent
and the buyer polling tasks/get.

Seam (approach a): emit centrally from the background completion path in
dispatch._project_handoff — the single seam every async task flows
through (_invoke_platform_method for all verbs, plus proposal finalize).
The terminal webhook fires EXACTLY ONCE after registry.complete /
registry.fail, with the buyer's operation_id echoed verbatim and the
registry task_id included. The sync auto-emit gate already skips the
{task_id, status} submitted projection, so the two paths never
double-deliver. Webhook emission lives in the decisioning runtime
(webhook_emit.emit_terminal_completion_webhook), reusing the sync path's
WebhookSender / supervisor and send_mcp payload builder; the pluggable
TaskRegistry stays webhook-agnostic. Gated by auto_emit_completion_webhooks
so manual-emit adopters are unaffected; no change to the sync defaults.

- webhook_emit.py: emit_terminal_completion_webhook (self-isolating,
  logged-and-swallowed) + operation_id extraction from push config.
- dispatch.py: thread webhook_target / webhook_auto_emit through
  _invoke_platform_method -> _project_handoff; fire on both the
  registry.complete (completed) and _fail (failed) terminal arms.
- handler.py: _handoff_webhook_kwargs() threaded into every spec-eligible
  shim (create/update_media_buy, sync_creatives, get/activate_signal,
  get_products, sync_audiences/catalogs, brand/rights, property_list).
- webhook_supervisor{,_pg}.py: add operation_id to send_mcp; Pg persists
  it on the queue row (CREATE column + ADD COLUMN IF NOT EXISTS backfill)
  and replays it from the worker.

Review fixes on #930:
- MUST-FIX docstrings: corrected the now-true claims — async terminal
  completion delivers via push webhook when configured, always via
  tasks/get polling (handler / serve / specialism docstrings).
- MUST-FIX dead guard arm (c): removed the unreachable post-dispatch
  assert_account_resolved_for_async call; documented that
  compose_caller_identity owns the no-push-handoff + unresolved-account
  case (fails closed terminally at _build_ctx before any task is minted).
  Added a test asserting no registry row is issued.
- SHOULD-FIX side-effect leak (b): reject_wholesale_handoff_before_launch
  wired as a pre_handoff_reject callback so a wholesale handoff is
  rejected BEFORE the registry row / background task / draft / webhook --
  no side effects for a buyer told 'rejected'. Test asserts an empty
  registry after rejection.
- SHOULD-FIX persist-draft: regression test pinning that fields= /
  pagination= projections shape only the wire response, never the
  persisted draft (full product pricing retained).
- SHOULD-FIX wire-validator: tightened the discovery submitted test to
  assert variant == 'submitted' (no skip escape hatch).
- Added async-completion webhook tests on get_products, get_signals,
  create_media_buy (one completed webhook; operation_id echoed, task_id,
  result in payload), the failure path (one failed webhook), and the
  no-push no-webhook case. Updated the create_media_buy handoff test to
  assert the conformant exactly-once behavior.

No ADCP_VERSION bump, no schema re-download, no public-API export change.

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

Author: bokelley

src/adcp/decisioning/discovery_guards.py

@@ -114,30 +114,60 @@ def assert_discovery_push_consistent(req: Any, *, mode_field: str) -> None:
         )
 
 
+def _wholesale_handoff_error(mode_field: str) -> AdcpError:
+    """Build the guard (b) rejection. Shared by the pre-dispatch callback
+    and the post-dispatch belt-and-braces check so both emit an identical
+    diagnostic."""
+    return AdcpError(
+        "INVALID_REQUEST",
+        message=(
+            f"{mode_field}='wholesale' returned a task handoff, but "
+            "wholesale discovery MUST complete synchronously. Return the "
+            "catalog directly; signal partial completion via the "
+            "response's incomplete[] field rather than handing off to a "
+            "background task. The 'brief' mode is the async-capable path."
+        ),
+        field=mode_field,
+        recovery="correctable",
+    )
+
+
+def reject_wholesale_handoff_before_launch(mode_field: str) -> None:
+    """Guard (b) pre-dispatch arm: reject a wholesale handoff BEFORE the
+    framework mints a registry row or launches the background task.
+
+    Wired as the ``pre_handoff_reject`` callback on
+    :func:`adcp.decisioning.dispatch._invoke_platform_method` only when the
+    resolved mode is wholesale. The dispatcher calls it the instant it
+    detects the adapter returned a :class:`TaskHandoff` — before
+    ``_project_handoff`` issues a task or starts background work — so a
+    rejected wholesale handoff leaves NO task row, NO persisted draft, NO
+    background coroutine, and NO completion webhook. Without this, the
+    post-dispatch check below would fire only after those side effects had
+    already happened.
+
+    :raises AdcpError: ``INVALID_REQUEST`` / ``correctable``,
+        ``field=<mode_field>``.
+    """
+    raise _wholesale_handoff_error(mode_field)
+
+
 def reject_wholesale_handoff(result: Any, *, mode: str | None, mode_field: str) -> None:
-    """Guard (b): post-dispatch reject of an adopter handoff on a wholesale call.
+    """Guard (b) post-dispatch arm: belt-and-braces reject of an adopter
+    handoff on a wholesale call.
 
+    Defense-in-depth for :func:`reject_wholesale_handoff_before_launch`.
     Called after ``_invoke_platform_method`` with the (already projected)
-    result. When the resolved mode is wholesale and the result is the
+    result; covers any dispatch path that did not wire the pre-launch
+    callback. When the resolved mode is wholesale and the result is the
     framework's submitted projection, the adopter's wholesale code path
     handed off — a contract violation. Reject (``field=<mode_field>``).
 
     :param mode: The request's coerced mode string.
     :raises AdcpError: ``INVALID_REQUEST`` / ``correctable``.
     """
     if mode == "wholesale" and _is_submitted_projection(result):
-        raise AdcpError(
-            "INVALID_REQUEST",
-            message=(
-                f"{mode_field}='wholesale' returned a task handoff, but "
-                "wholesale discovery MUST complete synchronously. Return the "
-                "catalog directly; signal partial completion via the "
-                "response's incomplete[] field rather than handing off to a "
-                "background task. The 'brief' mode is the async-capable path."
-            ),
-            field=mode_field,
-            recovery="correctable",
-        )
+        raise _wholesale_handoff_error(mode_field)
 
 
 def assert_account_resolved_for_async(
@@ -156,13 +186,21 @@ def assert_account_resolved_for_async(
     :func:`adcp.decisioning.dispatch.compose_caller_identity`'s contract
     across derived / implicit / explicit account modes.
 
-    Call this both pre-dispatch (with ``result=None``) when the buyer
-    supplied ``push_notification_config`` — so the rejection fires as a
-    clean ``INVALID_REQUEST`` before any platform / registry interaction —
-    and post-dispatch (passing the projected ``result``) so an adopter
-    handoff against an unresolved account is also caught. The framework's
-    registry independently rejects an empty ``account_id`` at issue-time;
-    this guard surfaces the buyer-facing ``field='account'`` diagnostic.
+    Used pre-dispatch (with ``result=None``) when the buyer supplied
+    ``push_notification_config`` — so the rejection fires as a clean
+    ``INVALID_REQUEST`` / ``correctable`` with the buyer-facing
+    ``field='account'`` diagnostic BEFORE any platform / registry
+    interaction.
+
+    The no-push handoff case (the adopter hands off against an unresolved
+    account without a push config) is NOT routed through this guard:
+    :func:`adcp.decisioning.dispatch.compose_caller_identity` already fails
+    closed at ``_build_ctx`` — which runs BEFORE the platform method — with
+    a terminal ``INVALID_REQUEST``, so no task row is ever minted against an
+    unresolved account. A post-dispatch call here passing the projected
+    ``result`` would be unreachable for an unresolved account (control never
+    reaches it because ``_build_ctx`` raised first) and a no-op for a
+    resolved one, so the shims do not make that call.
 
     :raises AdcpError: ``INVALID_REQUEST`` / ``correctable`` with
         ``field='account'``.
@@ -245,4 +283,5 @@ def reject_hand_rolled_submitted(result: Any) -> None:
     "assert_discovery_push_consistent",
     "reject_hand_rolled_submitted",
     "reject_wholesale_handoff",
+    "reject_wholesale_handoff_before_launch",
 ]

src/adcp/decisioning/dispatch.py

@@ -66,6 +66,7 @@
     is_task_handoff,
     is_workflow_handoff,
 )
+from adcp.decisioning.webhook_emit import emit_terminal_completion_webhook
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
@@ -77,6 +78,10 @@
     from adcp.decisioning.registry import BuyerAgent
     from adcp.decisioning.types import Account
     from adcp.server.base import ToolContext
+    from adcp.webhook_sender import WebhookSender
+    from adcp.webhook_supervisor import WebhookDeliverySupervisor
+
+    WebhookDeliveryTarget = WebhookSender | WebhookDeliverySupervisor
 
 logger = logging.getLogger(__name__)
 
@@ -1305,6 +1310,9 @@ async def _invoke_platform_method(
     extra_kwargs: dict[str, Any] | None = None,
     on_complete: Callable[[Any], Awaitable[None]] | None = None,
     on_failure: Callable[[BaseException], Awaitable[None]] | None = None,
+    webhook_target: WebhookDeliveryTarget | None = None,
+    webhook_auto_emit: bool = True,
+    pre_handoff_reject: Callable[[], None] | None = None,
 ) -> Any:
     """Invoke a platform method, projecting hybrid returns.
 
@@ -1351,6 +1359,30 @@ async def _invoke_platform_method(
         Symmetric with ``on_complete``. Used by v1.5 create_media_buy
         to release the consumption reservation so the buyer can retry.
         Hook errors are logged but never block exception propagation.
+
+    :param webhook_target: Forwarded to :func:`_project_handoff` so the
+        background completion path can deliver the terminal completion /
+        failure webhook when the buyer registered
+        ``push_notification_config``. The handler wires its
+        ``webhook_sender`` / ``webhook_supervisor``; only the handoff
+        (async) arm uses it — the sync arm's auto-emit is a separate
+        call in the handler shim.
+
+    :param webhook_auto_emit: Forwarded to :func:`_project_handoff`;
+        mirrors the handler's ``auto_emit_completion_webhooks`` so an
+        adopter emitting webhooks manually never gets a framework
+        double-delivery on the handoff path.
+
+    :param pre_handoff_reject: Optional zero-arg callback invoked when
+        the adapter returned a :class:`TaskHandoff`, BEFORE
+        :func:`_project_handoff` mints a registry row or launches the
+        background task. Raising from it (e.g. an :class:`AdcpError`)
+        rejects the handoff with NO side effects — no task row, no
+        background work, no completion webhook. The discovery
+        wholesale-is-synchronous guard uses this so an adopter handing
+        off on a ``wholesale`` request is rejected cleanly instead of
+        leaking a task the buyer was told was rejected. Runs only on the
+        ``TaskHandoff`` arm; sync / workflow-handoff returns ignore it.
     """
     # pydantic is a required dep; import here (not at module level) to mirror
     # the lazy-import discipline used throughout this module.
@@ -1513,6 +1545,13 @@ async def _invoke_platform_method(
         raise wrapped from exc
 
     if is_task_handoff(result):
+        # Reject before any side effect (registry row, background task,
+        # completion webhook) is created. The wholesale discovery guard
+        # uses this so an adopter handing off on a synchronous-only
+        # wholesale request never leaks a task the buyer is told is
+        # rejected.
+        if pre_handoff_reject is not None:
+            pre_handoff_reject()
         return await _project_handoff(
             result,
             ctx,
@@ -1522,6 +1561,8 @@ async def _invoke_platform_method(
             on_complete=on_complete,
             on_failure=on_failure,
             request_params=params,
+            webhook_target=webhook_target,
+            webhook_auto_emit=webhook_auto_emit,
         )
     if is_workflow_handoff(result):
         return await _project_workflow_handoff(
@@ -1588,6 +1629,8 @@ async def _project_handoff(
     on_complete: Callable[[Any], Awaitable[None]] | None = None,
     on_failure: Callable[[BaseException], Awaitable[None]] | None = None,
     request_params: BaseModel | None = None,
+    webhook_target: WebhookDeliveryTarget | None = None,
+    webhook_auto_emit: bool = True,
 ) -> dict[str, Any]:
     """Promote a TaskHandoff to a background task.
 
@@ -1647,7 +1690,26 @@ async def _project_handoff(
         (``registry.fail``) paths — closes #563. Mirrors the sync
         AdcpError path's context-passthrough (PR #560). When ``None``,
         no echo happens (e.g. test fixtures invoking the handoff
-        helper directly).
+        helper directly). Also the source of the buyer's
+        ``push_notification_config`` (url / token / operation_id) for
+        the terminal-completion webhook.
+
+    :param webhook_target: The wired :class:`~adcp.webhook_sender.WebhookSender`
+        or :class:`~adcp.webhook_supervisor.WebhookDeliverySupervisor`. When
+        the buyer supplied ``push_notification_config`` on
+        ``request_params``, the background completion path emits the
+        terminal completion / failure webhook to that URL EXACTLY ONCE —
+        on success after ``registry.complete``, on failure after
+        ``registry.fail``. This is the async-path half of the spec
+        webhook contract (adcp#5389): a ``Submitted`` task carrying a
+        push config MUST deliver at least the terminal notification.
+        ``None`` (and the no-push case) skips delivery — the buyer polls
+        ``tasks/get`` instead. The framework's polling path is unchanged.
+
+    :param webhook_auto_emit: Mirrors the handler's
+        ``auto_emit_completion_webhooks`` flag. When ``False`` the
+        adopter emits webhooks manually inside their handler; the
+        framework skips the terminal emission so it never double-delivers.
 
     The handoff fn is extracted via the type-identity dispatch in
     :func:`adcp.decisioning.types.is_task_handoff`. Subclassed
@@ -1699,7 +1761,23 @@ async def _fail(exc: AdcpError) -> None:
                     "still recorded in the registry",
                     task_id,
                 )
-        await registry.fail(task_id, exc.to_wire())
+        error_wire = exc.to_wire()
+        await registry.fail(task_id, error_wire)
+        # Terminal failure webhook (spec MUST when push config present).
+        # Fired AFTER registry.fail so the buyer's tasks/get poll and the
+        # push notification observe the same terminal state. The error
+        # wire dict is the payload `result`; the helper is self-isolating
+        # (logged-and-swallowed) so a delivery failure never re-raises into
+        # the background task.
+        await emit_terminal_completion_webhook(
+            target=webhook_target,
+            enabled=webhook_auto_emit,
+            method_name=method_name,
+            params=request_params,
+            status="failed",
+            task_id=task_id,
+            result=error_wire,
+        )
 
     async def _run() -> None:
         try:
@@ -1792,6 +1870,22 @@ async def _run() -> None:
         # surfaces it under the top-level ``context`` key; nothing to
         # do here on the result path.
         await registry.complete(task_id, persisted)
+        # Terminal completion webhook (spec MUST when push config present).
+        # Fired AFTER registry.complete so the buyer's tasks/get poll and
+        # the push notification observe the same terminal artifact. EXACTLY
+        # ONCE — the sync auto-emit gate skips the {task_id, status}
+        # submitted projection, so the handoff path owns the completion
+        # notification end-to-end. The helper is self-isolating
+        # (logged-and-swallowed); a delivery failure never re-raises here.
+        await emit_terminal_completion_webhook(
+            target=webhook_target,
+            enabled=webhook_auto_emit,
+            method_name=method_name,
+            params=request_params,
+            status="completed",
+            task_id=task_id,
+            result=persisted,
+        )
 
     # ``asyncio.create_task`` only weak-refs the resulting Task — under
     # GC pressure or with no outer awaiter, the task can be collected