feat(decisioning): ProposalManager v1 — Protocol + MockProposalManager forwarder + tenant binding

Foundation for the two-platform composition established in the product
architecture design doc (PR #502, docs/proposals/product-architecture.md).
Adopters can now wire a separate ProposalManager that handles
get_products / refine, while DecisioningPlatform handles create_media_buy
and lifecycle. The recipe (typed implementation_config) is the contract
between them.

v1 surface:

- ProposalManager Protocol (sync/async, capability-gated refine)
- ProposalCapabilities dataclass (sales_specialism + flags)
- Recipe Pydantic base with recipe_kind discriminator
- MockProposalManager v1 default forwarder (symmetric with Phase 2's
  upstream_for mock-mode dispatch on DecisioningPlatform)
- proposal_manager= kwarg on serve() / create_adcp_server_from_platform
- Dispatcher routing in PlatformHandler.get_products: routes to
  ProposalManager when wired (with refine/get_products selection by
  buying_mode + capability + method-presence); falls through to
  platform.get_products otherwise — backward-compat by construction

Out of scope (deferred to subsequent PRs, called out in module doc):

- Session cache for in-flight proposals
- finalize transition (buying_mode='refine' + action='finalize')
- expires_at enforcement
- capability_overlap declaration on Recipe + framework validation
- Recipe persistence through buy lifecycle (hydration in
  create_media_buy / update_media_buy / get_delivery)
- Per-tenant ProposalManager binding via PlatformRouter
- MIGRATION.md updates for v3 reference seller

Tests: 16 new in tests/test_proposal_manager.py covering Protocol
conformance, capability validation, MockProposalManager forwarding
(respx-mocked), dispatcher routing with/without proposal_manager,
adopter subclass dispatch, sync+async support, and refine routing
across all four (capability, method, buying_mode, fall-through) cases.
Full regression suite (3718 tests) green.

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

Author: bokelley

examples/hello_proposal_manager.py

@@ -0,0 +1,180 @@
+"""Hello-proposal-manager — the v1 two-platform composition demo.
+
+Wires a :class:`MockProposalManager` (proposal side) and a trivial
+:class:`DecisioningPlatform` (execution side) together. The
+:class:`MockProposalManager` forwards ``get_products`` requests to a
+running mock-server (``bin/adcp.js mock-server sales-non-guaranteed``);
+the platform handles ``create_media_buy`` directly without consulting
+the mock-server.
+
+This is the on-ramp shape for adopters whose proposal logic isn't
+ready yet but who already have an adapter for their upstream:
+
+* The mock-server provides product fixtures + stub recipes.
+* The platform's ``create_media_buy`` translates buyer requests into
+  upstream calls (here, it's a no-op stub).
+* As the adopter writes real proposal logic, they replace
+  :class:`MockProposalManager` with their own
+  :class:`ProposalManager` subclass — incrementally, one slice at a
+  time.
+
+**Prerequisite:** start the mock-server before running this example::
+
+    npx -y adcp-mock-server@latest sales-non-guaranteed --port 4500
+
+Then::
+
+    uv run python examples/hello_proposal_manager.py
+
+The server answers ``get_products`` by forwarding to the mock-server
+on port 4500; ``create_media_buy`` runs entirely in this process.
+Tail the mock-server log to see the proposal-side traffic; tail this
+process's log to see the execution-side traffic.
+
+See ``docs/proposals/product-architecture.md`` for the full design
+context — § "The two-platform composition" + § "Shape 3 — Mock-backed".
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from adcp.decisioning import (
+    DecisioningCapabilities,
+    DecisioningPlatform,
+    MockProposalManager,
+    RequestContext,
+    SalesPlatform,
+    SingletonAccounts,
+    serve,
+)
+from adcp.decisioning.capabilities import (
+    Account as CapabilitiesAccount,
+)
+from adcp.decisioning.capabilities import (
+    Adcp,
+    IdempotencySupported,
+    MediaBuy,
+)
+
+
+class HelloDecisioningPlatform(DecisioningPlatform, SalesPlatform):
+    """Trivial execution-side platform.
+
+    Handles ``create_media_buy`` / ``update_media_buy`` /
+    ``sync_creatives`` / ``get_media_buy_delivery`` with stub responses.
+    Does NOT implement ``get_products`` — the wired
+    :class:`MockProposalManager` handles that surface.
+
+    In production, this is where the adopter's adapter code lives —
+    translating wire ``CreateMediaBuyRequest`` payloads into upstream
+    calls (GAM, Kevel, Meta, etc.) and projecting the results back
+    onto the wire response shape.
+    """
+
+    capabilities = DecisioningCapabilities(
+        specialisms=["sales-non-guaranteed"],
+        adcp=Adcp(
+            major_versions=[3],
+            idempotency=IdempotencySupported(
+                supported=True,
+                replay_ttl_seconds=86400,
+            ),
+        ),
+        account=CapabilitiesAccount(supported_billing=["operator"]),
+        media_buy=MediaBuy(supported_pricing_models=["cpm"]),
+    )
+    accounts = SingletonAccounts(account_id="hello-proposal-manager-acct")
+
+    def create_media_buy(
+        self,
+        req: Any,
+        ctx: RequestContext[Any],
+    ) -> dict[str, Any]:
+        """Stub create_media_buy — returns a synthetic media_buy_id.
+
+        Production: this is where the adopter's adapter code calls the
+        upstream API (the recipe attached to the request's products
+        carries the upstream-specific config the adapter consumes).
+        """
+        del ctx
+        idem_key = getattr(req, "idempotency_key", "unknown")
+        return {
+            "media_buy_id": f"mb_demo_{idem_key}",
+            "status": "active",
+            "packages": [],
+        }
+
+    def update_media_buy(
+        self,
+        media_buy_id: str,
+        patch: Any,
+        ctx: RequestContext[Any],
+    ) -> dict[str, Any]:
+        del patch, ctx
+        return {"media_buy_id": media_buy_id, "status": "active", "packages": []}
+
+    def sync_creatives(
+        self,
+        req: Any,
+        ctx: RequestContext[Any],
+    ) -> dict[str, Any]:
+        del ctx
+        creatives = getattr(req, "creatives", None) or []
+        return {
+            "creatives": [
+                {
+                    "creative_id": (
+                        c.creative_id if hasattr(c, "creative_id") else c.get("creative_id")
+                    ),
+                    "approval_status": "approved",
+                }
+                for c in creatives
+            ],
+        }
+
+    def get_media_buy_delivery(
+        self,
+        req: Any,
+        ctx: RequestContext[Any],
+    ) -> dict[str, Any]:
+        del ctx
+        return {
+            "media_buy_deliveries": [
+                {
+                    "media_buy_id": getattr(req, "media_buy_id", "mb_unknown"),
+                    "totals": {"impressions": 0, "spend": 0.0},
+                },
+            ],
+        }
+
+
+if __name__ == "__main__":
+    # Mock-server URL — adopters in production point this at the
+    # appropriate `bin/adcp.js mock-server <specialism>` instance, or
+    # at a fixture-server in their own infra. Override via env var
+    # for CI / local dev.
+    mock_url = os.environ.get(
+        "ADCP_MOCK_PROPOSAL_URL",
+        "http://localhost:4500",
+    )
+
+    # The proposal manager — forwards get_products to the running
+    # mock-server. Adopters replace this with their own ProposalManager
+    # subclass as their proposal logic comes online.
+    proposal_manager = MockProposalManager(
+        mock_upstream_url=mock_url,
+        sales_specialism="sales-non-guaranteed",
+    )
+
+    # serve() composes the two platforms. The dispatcher routes
+    # get_products to the proposal manager and create_media_buy /
+    # update_media_buy / etc. to the platform.
+    serve(
+        HelloDecisioningPlatform(),
+        proposal_manager=proposal_manager,
+        name="hello-proposal-manager",
+        port=3001,
+        auto_emit_completion_webhooks=False,
+    )

src/adcp/decisioning/__init__.py

@@ -121,6 +121,13 @@ def create_media_buy(
     resolve_property_list,
     validate_property_list_config,
 )
+from adcp.decisioning.proposal_manager import (
+    MockProposalManager,
+    ProposalCapabilities,
+    ProposalManager,
+    SalesSpecialism,
+)
+from adcp.decisioning.recipe import Recipe
 from adcp.decisioning.registry import (
     ApiKeyCredential,
     BillingMode,
@@ -297,13 +304,16 @@ def __init__(self, *args: object, **kwargs: object) -> None:
     "MediaBuyNotFoundError",
     "MediaBuyStore",
     "MockAdServer",
+    "MockProposalManager",
     "NoAuth",
     "OAuthCredential",
     "PermissionDeniedError",
     "PgTaskRegistry",
     "PlatformRouter",
     "PostgresTaskRegistry",
     "Proposal",
+    "ProposalCapabilities",
+    "ProposalManager",
     "PropertyList",
     "PropertyListFetcher",
     "PropertyListReference",
@@ -315,11 +325,13 @@ def __init__(self, *args: object, **kwargs: object) -> None:
     "validate_property_list_config",
     "RateLimitedBuyerAgentRegistry",
     "RateLimitedError",
+    "Recipe",
     "RequestContext",
     "ResolveContext",
     "ResourceResolver",
     "SalesPlatform",
     "SalesResult",
+    "SalesSpecialism",
     "ServiceUnavailableError",
     "SignalsPlatform",
     "SingletonAccounts",

src/adcp/decisioning/handler.py

@@ -156,6 +156,7 @@
 
     from adcp.decisioning.platform import DecisioningPlatform
     from adcp.decisioning.property_list import PropertyListFetcher
+    from adcp.decisioning.proposal_manager import ProposalManager
     from adcp.decisioning.registry import BuyerAgent, BuyerAgentRegistry
     from adcp.decisioning.resolve import ResourceResolver
     from adcp.decisioning.state import StateReader
@@ -702,6 +703,7 @@ def __init__(
         buyer_agent_registry: BuyerAgentRegistry | None = None,
         config_store: ProductConfigStore | None = None,
         property_list_fetcher: PropertyListFetcher | None = None,
+        proposal_manager: ProposalManager | None = None,
     ) -> None:
         super().__init__()
         self._platform = platform
@@ -715,6 +717,7 @@ def __init__(
         self._buyer_agent_registry = buyer_agent_registry
         self._config_store = config_store
         self._property_list_fetcher = property_list_fetcher
+        self._proposal_manager = proposal_manager
 
         # Cache whether the platform's create_media_buy accepts 'configs'
         # so we only pay the inspect.signature cost at construction time.
@@ -1062,21 +1065,45 @@ async def get_products(  # type: ignore[override]
         params: GetProductsRequest,
         context: ToolContext | None = None,
     ) -> GetProductsResponse:
-        """Invoke the platform's ``get_products`` method and apply fields projection.
+        """Invoke ``get_products`` on the wired ProposalManager when present,
+        else fall through to the platform's ``get_products`` method.
+
+        Routing precedence:
+
+        1. If a :class:`ProposalManager` is wired AND
+           ``params.buying_mode == 'refine'`` AND the manager declares
+           :attr:`ProposalCapabilities.refine` AND implements
+           ``refine_products`` → route to ``proposal_manager.refine_products``.
+        2. If a :class:`ProposalManager` is wired (any buying_mode) →
+           route to ``proposal_manager.get_products``.
+        3. Otherwise (no proposal_manager wired) → fall through to
+           ``platform.get_products``. Existing adopters who haven't
+           wired a ProposalManager keep their current behaviour
+           unchanged — backward-compat by construction.
 
         When ``params.fields`` is set the framework drops unrequested product
-        fields after the platform method returns, always retaining the eight
+        fields after the method returns, always retaining the eight
         schema-required fields.  When ``params.fields`` is ``None`` the
         response passes through unchanged.
         """
         tool_ctx = context or ToolContext()
         account = await self._resolve_account(params.account, tool_ctx)
         ctx = self._build_ctx(tool_ctx, account)
+
+        target: Any
+        method_name: str
+        if self._proposal_manager is not None:
+            target = self._proposal_manager
+            method_name = self._select_proposal_method(params)
+        else:
+            target = self._platform
+            method_name = "get_products"
+
         response = cast(
             "GetProductsResponse",
             await _invoke_platform_method(
-                self._platform,
-                "get_products",
+                target,
+                method_name,
                 params,
                 ctx,
                 executor=self._executor,
@@ -1106,6 +1133,40 @@ async def get_products(  # type: ignore[override]
             response = _project_product_fields(response, params.fields)
         return response
 
+    def _select_proposal_method(self, params: GetProductsRequest) -> str:
+        """Choose between ``get_products`` and ``refine_products`` on the
+        wired ProposalManager.
+
+        Refine is dispatched only when all three conditions hold:
+
+        1. The request's ``buying_mode`` is ``'refine'``.
+        2. The manager's ``capabilities.refine`` flag is True.
+        3. The manager subclass implements ``refine_products``
+           (``hasattr`` covers the Protocol's "present-or-absent"
+           semantics).
+
+        Otherwise routes to ``get_products``. Adopters whose
+        ``get_products`` handler also handles refine internally keep
+        working without declaring the refine capability.
+        """
+        manager = self._proposal_manager
+        if manager is None:
+            return "get_products"
+        buying_mode = getattr(params, "buying_mode", None)
+        # ``buying_mode`` may be a string or a generated enum (the
+        # Pydantic model coerces). Normalize via ``getattr(.., 'value',
+        # buying_mode)`` so both shapes compare cleanly.
+        buying_mode_str = getattr(buying_mode, "value", buying_mode)
+        if buying_mode_str != "refine":
+            return "get_products"
+        caps = getattr(manager, "capabilities", None)
+        refine_supported = bool(getattr(caps, "refine", False))
+        if not refine_supported:
+            return "get_products"
+        if not hasattr(manager, "refine_products"):
+            return "get_products"
+        return "refine_products"
+
     async def create_media_buy(  # type: ignore[override]
         self,
         params: CreateMediaBuyRequest,