feat(decisioning): ProductConfigStore lookup helper for create_media_buy (#498)

* feat(decisioning): ProductConfigStore lookup helper for create_media_buy

Adds a pluggable ProductConfigStore Protocol that the framework calls
before invoking SalesPlatform.create_media_buy, injecting seller-side
per-product implementation configs as a configs= kwarg. Adopters who
don't wire a store or don't declare configs in their signature see no
change (fully backward-compatible). Closes #497.

https://claude.ai/code/session_01WS8EChCkSo6diBNgbXBMM1

* fix(decisioning): stable product_id ordering, extra_kwargs TypeError diagnostic, test warning scope

- dict.fromkeys() instead of set comprehension in handler.py for stable
  insertion-order deduplication of product_ids before store lookup
- targeted TypeError diagnostic branch in dispatch.py so extra_kwargs
  signature drift surfaces as INVALID_REQUEST rather than INTERNAL_ERROR
- move warnings.catch_warnings() to wrap _make_handler() construction in
  test_no_store_wired_adopter_gets_empty_configs (warning fires at init,
  not at call time)

https://claude.ai/code/session_01WS8EChCkSo6diBNgbXBMM1

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: bokelley

src/adcp/decisioning/__init__.py

@@ -96,6 +96,7 @@ def create_media_buy(
     ValidationError,
 )
 from adcp.decisioning.helpers import ref_account_id
+from adcp.decisioning.implementation_config import ProductConfigStore
 from adcp.decisioning.media_buy_store import (
     MediaBuyStore,
     create_media_buy_store,
@@ -298,6 +299,7 @@ def __init__(self, *args: object, **kwargs: object) -> None:
     "Proposal",
     "PropertyList",
     "PropertyListReference",
+    "ProductConfigStore",
     "PropertyListsPlatform",
     "RateLimitedBuyerAgentRegistry",
     "RateLimitedError",

src/adcp/decisioning/dispatch.py

@@ -1040,6 +1040,7 @@ async def _invoke_platform_method(
     executor: ThreadPoolExecutor,
     registry: TaskRegistry,
     arg_projector: dict[str, Any] | None = None,
+    extra_kwargs: dict[str, Any] | None = None,
 ) -> Any:
     """Invoke a platform method, projecting hybrid returns.
 
@@ -1064,15 +1065,22 @@ async def _invoke_platform_method(
     :param arg_projector: Optional kwargs dict for tools whose Python
         method signature differs from the wire shape (D1
         arg-projection, e.g. ``update_media_buy(media_buy_id, patch,
-        ctx)``). Codegen-emitted shims pass this for those tools;
-        most tools call with ``None``.
+        ctx)``). Replaces the default positional ``(params, ctx)``
+        call entirely. Codegen-emitted shims pass this for those
+        tools; most tools call with ``None``.
+    :param extra_kwargs: Optional additional kwargs appended to the
+        normal ``(params, ctx)`` call, used when the framework injects
+        framework-computed values (e.g. ``configs=`` from
+        ``ProductConfigStore``). Ignored when ``arg_projector`` is set.
     """
     method = getattr(platform, method_name)
 
     try:
         if asyncio.iscoroutinefunction(method):
             if arg_projector is not None:
                 result = await method(**arg_projector, ctx=ctx)
+            elif extra_kwargs:
+                result = await method(params, ctx, **extra_kwargs)
             else:
                 result = await method(params, ctx)
         else:
@@ -1084,6 +1092,11 @@ async def _invoke_platform_method(
                     executor,
                     functools.partial(ctx_snapshot.run, method, **projected_kwargs),
                 )
+            elif extra_kwargs:
+                result = await loop.run_in_executor(
+                    executor,
+                    functools.partial(ctx_snapshot.run, method, params, ctx, **extra_kwargs),
+                )
             else:
                 result = await loop.run_in_executor(
                     executor,
@@ -1094,8 +1107,7 @@ async def _invoke_platform_method(
         # outer middleware projects to the wire envelope.
         raise
     except TypeError as exc:
-        # Most likely an arg_projector signature-drift bug — adopter
-        # renamed update_media_buy's `patch` kwarg → `update`, etc.
+        # Most likely an arg_projector or extra_kwargs signature-drift bug.
         # Bare INTERNAL_ERROR would hide the cause; project to
         # INVALID_REQUEST with a hint pointing at the adopter's
         # method signature so they fix it without a server-log dive.
@@ -1120,6 +1132,23 @@ async def _invoke_platform_method(
                 ),
                 recovery="terminal",
             ) from exc
+        if extra_kwargs is not None:
+            logger.exception(
+                "TypeError invoking platform.%s — likely extra_kwargs "
+                "signature drift (injected kwargs %s vs adopter signature)",
+                method_name,
+                sorted(extra_kwargs.keys()),
+            )
+            raise AdcpError(
+                "INVALID_REQUEST",
+                message=(
+                    f"Platform method {method_name!r} rejected framework-injected "
+                    f"kwargs {sorted(extra_kwargs.keys())!r}. Declare the matching "
+                    "parameter(s) in your platform method signature, or remove them "
+                    "if you don't need them."
+                ),
+                recovery="terminal",
+            ) from exc
         # Non-projected TypeError — fall through to generic wrap.
         logger.exception(
             "Unhandled exception in platform.%s — wrapping to INTERNAL_ERROR",

src/adcp/decisioning/handler.py

@@ -32,6 +32,9 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
+import logging
+import warnings
 from typing import TYPE_CHECKING, Any, ClassVar, cast
 
 from adcp.decisioning._get_products_helpers import _project_product_fields
@@ -40,9 +43,12 @@
     _build_request_context,
     _invoke_platform_method,
 )
+from adcp.decisioning.implementation_config import ProductConfigStore
 from adcp.decisioning.webhook_emit import maybe_emit_sync_completion
 from adcp.server.base import ADCPHandler, ToolContext
 
+logger = logging.getLogger(__name__)
+
 # Pydantic Request/Response types are imported at module scope (NOT
 # under TYPE_CHECKING) so that ``typing.get_type_hints(method)`` can
 # resolve every shim's ``params`` annotation at runtime. The dispatcher
@@ -575,6 +581,18 @@ def _project_sync_audiences(result: Any) -> Any:
     return result
 
 
+def _method_accepts_configs(platform: Any, method_name: str) -> bool:
+    """Return True when the platform's ``method_name`` declares a ``configs`` parameter."""
+    method = getattr(platform, method_name, None)
+    if method is None:
+        return False
+    try:
+        sig = inspect.signature(method)
+        return "configs" in sig.parameters
+    except (ValueError, TypeError):
+        return False
+
+
 class PlatformHandler(ADCPHandler[ToolContext]):
     """ADCPHandler subclass that routes wire requests to a
     :class:`DecisioningPlatform` via :func:`_invoke_platform_method`.
@@ -676,6 +694,7 @@ def __init__(
         webhook_supervisor: WebhookDeliverySupervisor | None = None,
         auto_emit_completion_webhooks: bool = True,
         buyer_agent_registry: BuyerAgentRegistry | None = None,
+        config_store: ProductConfigStore | None = None,
     ) -> None:
         super().__init__()
         self._platform = platform
@@ -687,6 +706,23 @@ def __init__(
         self._webhook_supervisor = webhook_supervisor
         self._auto_emit_completion_webhooks = auto_emit_completion_webhooks
         self._buyer_agent_registry = buyer_agent_registry
+        self._config_store = config_store
+
+        # Cache whether the platform's create_media_buy accepts 'configs'
+        # so we only pay the inspect.signature cost at construction time.
+        self._create_media_buy_accepts_configs = _method_accepts_configs(
+            platform, "create_media_buy"
+        )
+        if config_store is None and self._create_media_buy_accepts_configs:
+            warnings.warn(
+                "create_media_buy declares a 'configs' parameter but no "
+                "ProductConfigStore was wired — the framework will inject "
+                "configs={} (empty dict) on every call. Wire a store via "
+                "config_store= in create_adcp_server_from_platform to enable "
+                "automatic implementation_config lookup.",
+                UserWarning,
+                stacklevel=2,
+            )
 
     # ----- account resolution helper -----
 
@@ -1048,16 +1084,50 @@ async def create_media_buy(  # type: ignore[override]
         params: CreateMediaBuyRequest,
         context: ToolContext | None = None,
     ) -> CreateMediaBuySuccessResponse:
+        from adcp.decisioning.types import AdcpError
+
         tool_ctx = context or ToolContext()
         account = await self._resolve_account(params.account, tool_ctx)
         ctx = self._build_ctx(tool_ctx, account)
+
+        configs: dict[str, Any] = {}
+        if self._config_store is not None:
+            # proposal_id flows have packages=None — skip lookup, inject {}
+            if params.packages:
+                product_ids = list(dict.fromkeys(p.product_id for p in params.packages))
+                try:
+                    configs = await self._config_store.lookup_implementation_configs(
+                        product_ids, ctx
+                    )
+                except AdcpError:
+                    raise
+                except Exception as exc:
+                    logger.exception(
+                        "[adcp.decisioning] ProductConfigStore.lookup_implementation_configs "
+                        "raised for create_media_buy — translating to SERVICE_UNAVAILABLE"
+                    )
+                    raise AdcpError(
+                        "SERVICE_UNAVAILABLE",
+                        message=(
+                            "implementation_config lookup failed for "
+                            f"{len(product_ids)} product(s). Retry the request; "
+                            "if the problem persists contact the seller."
+                        ),
+                        recovery="transient",
+                        details={"caused_by": {"type": type(exc).__name__}},
+                    ) from exc
+
+        extra: dict[str, Any] | None = (
+            {"configs": configs} if self._create_media_buy_accepts_configs else None
+        )
         result = await _invoke_platform_method(
             self._platform,
             "create_media_buy",
             params,
             ctx,
             executor=self._executor,
             registry=self._registry,
+            extra_kwargs=extra,
         )
         self._maybe_auto_emit_sync_completion("create_media_buy", params, result)
         return cast("CreateMediaBuySuccessResponse", result)

src/adcp/decisioning/implementation_config.py

@@ -0,0 +1,68 @@
+"""ProductConfigStore — pluggable implementation_config lookup for create_media_buy.
+
+Seller adapters (GAM, Kevel, Xandr, Prebid Server, …) attach seller-side
+configuration to each ``Product`` under ``Product.ext.implementation_config``
+by convention.  Every adopter that implements ``create_media_buy`` writes the
+same boilerplate: pull the packages off the request, collect the product_ids,
+look up the per-product config from a store, hand it to the adapter alongside
+the wire request.
+
+This module lifts that loop into the framework.  The adopter wires a
+``ProductConfigStore`` once at server construction; the framework calls
+``lookup_implementation_configs`` before invoking the platform method and
+injects the result as ``configs=`` kwarg.  Adopters who declare ``configs`` in
+their ``create_media_buy`` signature receive the resolved dict automatically.
+Adopters who don't wire a store receive ``configs={}`` (current behaviour,
+fully backward-compatible).
+
+Reference pattern: ``adcp.decisioning.webhook_emit`` — same "framework
+intercepts at the handler seam before the platform method fires" design.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from adcp.decisioning.context import RequestContext
+
+
+@runtime_checkable
+class ProductConfigStore(Protocol):
+    """Adopter-supplied lookup for seller-side product configuration.
+
+    The framework calls ``lookup_implementation_configs`` once per
+    ``create_media_buy`` request, before invoking the platform method.
+    Adopters return a dict keyed by ``product_id``; the framework injects
+    it as ``configs=`` on the platform method call.
+
+    If the store returns a partial dict (some product_ids missing), the
+    adopter handles it — the framework does not fabricate entries for missing
+    ids.  If the store raises, the framework surfaces
+    ``SERVICE_UNAVAILABLE`` with ``recovery='transient'`` to the buyer.
+
+    The ``ctx`` parameter carries the resolved ``RequestContext``, including
+    ``ctx.account`` for multi-tenant stores that need tenant isolation.
+    """
+
+    async def lookup_implementation_configs(
+        self,
+        product_ids: list[str],
+        ctx: RequestContext[Any],
+    ) -> dict[str, dict[str, Any]]:
+        """Return seller-side config keyed by product_id.
+
+        :param product_ids: Deduplicated list of product ids from
+            ``CreateMediaBuyRequest.packages[*].product_id``.  Empty when
+            the request uses ``proposal_id`` without an explicit
+            ``packages`` array.
+        :param ctx: Resolved request context — use ``ctx.account`` for
+            tenant scoping.
+        :returns: Dict mapping each product_id to its config dict.  Missing
+            keys mean "no config for this product"; the framework passes the
+            partial dict to the adopter unchanged.
+        """
+        ...
+
+
+__all__ = ["ProductConfigStore"]

src/adcp/decisioning/serve.py

@@ -40,6 +40,7 @@
 from adcp.decisioning.types import AdcpError
 
 if TYPE_CHECKING:
+    from adcp.decisioning.implementation_config import ProductConfigStore
     from adcp.decisioning.platform import DecisioningPlatform
     from adcp.decisioning.registry import BuyerAgentRegistry
     from adcp.decisioning.resolve import ResourceResolver
@@ -83,6 +84,7 @@ def create_adcp_server_from_platform(
     webhook_supervisor: WebhookDeliverySupervisor | None = None,
     auto_emit_completion_webhooks: bool = True,
     buyer_agent_registry: BuyerAgentRegistry | None = None,
+    config_store: ProductConfigStore | None = None,
 ) -> tuple[PlatformHandler, ThreadPoolExecutor, TaskRegistry]:
     """Build the :class:`PlatformHandler` + supporting wiring from a
     :class:`DecisioningPlatform`.
@@ -272,6 +274,7 @@ def create_adcp_server_from_platform(
         webhook_supervisor=webhook_supervisor,
         auto_emit_completion_webhooks=auto_emit_completion_webhooks,
         buyer_agent_registry=buyer_agent_registry,
+        config_store=config_store,
     )
 
     # F12 boot-time fail-fast (Emma sales-direct P0 root cause): if
@@ -322,6 +325,7 @@ def serve(
     webhook_supervisor: WebhookDeliverySupervisor | None = None,
     auto_emit_completion_webhooks: bool = True,
     buyer_agent_registry: BuyerAgentRegistry | None = None,
+    config_store: ProductConfigStore | None = None,
     advertise_all: bool = False,
     mock_ad_server: Any | None = None,
     enable_debug_endpoints: bool = False,
@@ -399,6 +403,7 @@ def serve(
         webhook_supervisor=webhook_supervisor,
         auto_emit_completion_webhooks=auto_emit_completion_webhooks,
         buyer_agent_registry=buyer_agent_registry,
+        config_store=config_store,
     )
 
     # Phase 1 sandbox-authority — wire the comply controller's account

src/adcp/decisioning/specialisms/sales.py

@@ -136,6 +136,21 @@ def create_media_buy(
         * ``media_buy_id`` field present → sync success
         * ``task_id`` + ``status='submitted'`` → poll ``tasks_get`` or
           receive webhook
+
+        **Framework injection:** when a :class:`~adcp.decisioning.ProductConfigStore`
+        is wired via ``config_store=`` in
+        :func:`adcp.decisioning.serve.create_adcp_server_from_platform`,
+        the framework calls the store before invoking this method and
+        injects the result as ``configs: dict[str, dict[str, Any]]``.
+        Declare ``configs`` in your method signature to receive it::
+
+            def create_media_buy(self, req, ctx, configs=None):
+                configs = configs or {}
+                line_item_id = configs.get(pkg.product_id, {}).get("line_item_id")
+
+        When the store is not wired, or when ``req.packages`` is
+        ``None`` (proposal-id flow), ``configs`` arrives as an empty
+        dict ``{}``.
         """
         ...