fix(v3-reference-seller): expert-review fix-pack + serve(asgi_middleware=)

The 4-expert review of #357 surfaced multiple convergent blockers
in the v3 reference seller MVP:

* `app.py` `build_app()` imported `_build_mcp_and_a2a_app` from the
  wrong module → ImportError at adopter call time
* `main()` never wired `SubdomainTenantMiddleware` → multi-tenant
  claim was false; `current_tenant()` would return None
* `platform.create_media_buy` extracted `total_budget` / `currency` /
  `start_time` / `end_time` against a wrong-shape API surface
  (TotalBudget object, StartTiming union) — every persisted media
  buy stored corrupted data or NULL
* `media_buy_id = f"mb_{uuid[:16]}"` collides at scale → IntegrityError

Bundled fix:

* Framework: add `asgi_middleware=Sequence[(cls, kwargs)]` kwarg to
  `adcp.server.serve()` (and forward through
  `adcp.decisioning.serve()`) so adopters layer Starlette-style
  ASGI middleware without dropping to `_build_mcp_and_a2a_app`.
* Example: `main()` now passes `asgi_middleware=[(SubdomainTenant
  Middleware, {"router": router})]`. Dead `build_app()` helper
  removed.
* `platform.py`: project `req.total_budget.amount/.currency` and
  unwrap `StartTiming.root` (`'asap'` → now()); `media_buy_id` is a
  full UUID (the (tenant_id, idempotency_key) unique constraint
  already guards replay).
* `models.py` + `buyer_registry.py` + `seed.py`: `billing_capabilities`,
  `default_terms`, `allowed_brands` switched from JSON-encoded
  String columns to native JSON columns (matches the rest of the
  schema).
* `audit.py`: docstring corrected — `record()` is best-effort but
  awaited inline (not fire-and-forget).
* `tenant_router.py`: cache renamed from "LRU-ish" to bounded FIFO
  to match actual eviction behavior.
* `README.md`: drop inaccurate claims (HTTP-Sig wired,
  project_account_for_response wired inline). Add dev-only callout
  for docker-compose trust auth + seed bearer token.

Test: `tests/test_serve_asgi_middleware.py` covers the new kwarg's
no-op + outermost-first composition contract.

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

Author: bokelley

examples/v3_reference_seller/README.md

@@ -10,13 +10,14 @@ ships into one runnable binary:
 | Component | Module | Source |
 |---|---|---|
 | Tier 2 commercial-identity gate | `src/buyer_registry.py` | `adcp.decisioning.BuyerAgentRegistry` |
-| HTTP-Sig verifier → AuthInfo | (use `AuthInfo.from_verified_signer`) | `adcp.decisioning.AuthInfo` |
-| Subdomain tenant routing | `src/tenant_router.py` | `adcp.server.SubdomainTenantMiddleware` |
-| Account v3 projection (bank guard) | inline in platform | `adcp.types.project_account_for_response` |
+| Subdomain tenant routing | `src/tenant_router.py` + `src/app.py` | `adcp.server.SubdomainTenantMiddleware` |
+| Account v3 storage (bank-details column) | `src/models.py` | `Account.billing_entity` JSON column |
 | Audit trail | `src/audit.py` | `adcp.audit_sink.AuditSink` |
+| MCP + A2A on one binary | `src/app.py` | `serve(transport="both", asgi_middleware=...)` |
 | Durable HITL tasks (optional) | swap to `PgTaskRegistry` | `adcp.decisioning.pg.PgTaskRegistry` |
 | Durable webhook delivery (optional) | swap to `PgWebhookDeliverySupervisor` | `adcp.webhook_supervisor_pg` |
-| MCP + A2A on one binary | `src/app.py` | `serve(transport="both")` |
+| HTTP-Sig verifier → AuthInfo (TODO) | adopter middleware | `adcp.decisioning.AuthInfo.from_verified_signer` |
+| Account v3 projection on read (TODO) | adopter wires in `sync_accounts` | `adcp.types.project_account_for_response` |
 
 ## Run it
 
@@ -36,6 +37,15 @@ DATABASE_URL=postgresql+asyncpg://postgres@localhost/adcp \
 
 The server binds `0.0.0.0:3001` and serves both transports.
 
+> ⚠️ **Local-dev only.** `docker-compose.yml` uses
+> `POSTGRES_HOST_AUTH_METHOD=trust` and exposes 5432 on
+> `0.0.0.0`. Do not run this compose file on a host reachable from
+> an untrusted network. `seed.py` plants a literal dev bearer
+> token (`dev-bearer-token-acme-1`) — do not run `seed.py` against
+> a production `DATABASE_URL`. Production deployments point
+> `DATABASE_URL` at managed Postgres with scram-sha-256 + a real
+> password, and seed via the admin API (not this script).
+
 ## What's wired
 
 ### Schema (`src/models.py`)
@@ -50,9 +60,13 @@ Four tables — the spine of a multi-tenant v3 seller:
   rejects suspended (transient) and blocked (terminal) agents with
   structured errors.
 - `accounts` — buyer-side accounts under recognized agents. Carries
-  the spec 3.1-ready `billing_entity` (write-only bank details
-  guarded via `project_account_for_response`) and `reporting_bucket`
-  (offline reporting target).
+  the spec 3.1-ready `billing_entity` (write-only bank details on
+  responses) and `reporting_bucket` (offline reporting target). The
+  reference seller does not implement `sync_accounts`, so the
+  bank-details projection is a column-level architectural seam, not
+  an enforced runtime guard — adopters who add `sync_accounts`
+  MUST project through `adcp.types.project_account_for_response`
+  before returning the row.
 - `media_buys` — terminal artifact of `create_media_buy`,
   idempotency-keyed for replay safety.
 
@@ -83,10 +97,11 @@ alerting compose with `adcp.audit_sink.SlackAlertSink` via
 ### Platform (`src/platform.py`)
 
 `V3ReferenceSeller` implements `sales-non-guaranteed` — the five
-required Sales methods. Every method body reads
-`ctx.buyer_agent` (the resolved Tier 2 record) and `ctx.account`
-(the resolved account); both are populated by the framework's
-dispatch gate before the method runs.
+required Sales methods (`get_products`, `create_media_buy`,
+`update_media_buy`, `sync_creatives`, `get_media_buy_delivery`).
+Every method body reads `ctx.buyer_agent` (the resolved Tier 2
+record) and `ctx.account` (the resolved account); both are
+populated by the framework's dispatch gate before the method runs.
 
 This file is the bulk of what an adopter customizes. Everything
 else is boilerplate the seller wires once.

examples/v3_reference_seller/seed.py

@@ -19,7 +19,6 @@
 from __future__ import annotations
 
 import asyncio
-import json
 import os
 
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
@@ -55,7 +54,7 @@ async def main() -> None:
                         agent_url="https://signed-buyer.example/",
                         display_name="Signed Buyer",
                         status="active",
-                        billing_capabilities=json.dumps(["operator", "agent"]),
+                        billing_capabilities=["operator", "agent"],
                         api_key_id=None,
                     ),
                     BuyerAgent(
@@ -64,7 +63,7 @@ async def main() -> None:
                         agent_url="https://bearer-buyer.example/",
                         display_name="Bearer Buyer",
                         status="active",
-                        billing_capabilities=json.dumps(["operator"]),
+                        billing_capabilities=["operator"],
                         api_key_id="dev-bearer-token-acme-1",
                     ),
                     BuyerAgent(
@@ -73,7 +72,7 @@ async def main() -> None:
                         agent_url="https://suspended.example/",
                         display_name="Suspended Buyer",
                         status="suspended",
-                        billing_capabilities=json.dumps(["operator"]),
+                        billing_capabilities=["operator"],
                     ),
                 ]
             )

examples/v3_reference_seller/src/app.py

@@ -12,8 +12,9 @@
    * :class:`DbAuditSink` for compliance trail
    * :class:`V3ReferenceSeller` (the platform impl)
 
-4. ``adcp.decisioning.serve(transport="both", ...)`` — single
-   binary serving MCP at ``/mcp`` and A2A at ``/``.
+4. ``adcp.decisioning.serve(transport="both", asgi_middleware=[...])``
+   — single binary serving MCP at ``/mcp`` and A2A at ``/`` with
+   :class:`SubdomainTenantMiddleware` layered on the outer HTTP app.
 
 Adopters fork this file and replace the platform impl, the seller-
 specific column populators, and the seed fixtures. Everything else
@@ -38,6 +39,7 @@
 
 from adcp.decisioning import serve
 from adcp.server import (
+    SubdomainTenantMiddleware,
     ToolContext,
     current_tenant,
 )
@@ -54,15 +56,14 @@
 logger = logging.getLogger(__name__)
 
 
-def _build_context_factory(router: SqlSubdomainTenantRouter):
+def _build_context_factory():
     """``context_factory`` that pins :attr:`ToolContext.tenant_id`
     from the resolved tenant.
 
     The middleware sets ``current_tenant()`` on the contextvar before
     dispatch; this factory reads it and writes ``tenant_id`` so the
     framework's idempotency middleware scopes correctly.
     """
-    del router  # router is consumed via current_tenant(); kept for symmetry
 
     def build(meta: RequestMetadata) -> ToolContext:
         tenant = current_tenant()
@@ -114,84 +115,28 @@ def main() -> None:
     )
     logger.info("Audit sink wired: %s. Tenant router cache: 256 hosts.", type(audit_sink).__name__)
 
-    # The framework's serve() drives uvicorn. We layer the tenant
-    # middleware on the parent Starlette app via a build hook —
-    # for the MVP we wire it inline by passing a custom ``app``
-    # builder. For the Tier-2 wiring we use the ``buyer_agent_registry``
-    # and ``context_factory`` kwargs the framework already supports.
     serve(
         platform=platform,
         name="v3-reference-seller",
         port=port,
-        transport="both",
-        buyer_agent_registry=buyer_registry,
-        context_factory=_build_context_factory(router),
-        # NOTE: SubdomainTenantMiddleware is added to the unified
-        # Starlette app via ``add_middleware``. The current public
-        # surface of ``serve()`` doesn't expose a hook for adopter
-        # ASGI middleware on the parent — adopters who want it today
-        # call ``_build_mcp_and_a2a_app`` directly and run uvicorn
-        # themselves. Filed as a follow-up DX issue: the helper
-        # below shows the pattern.
         host="0.0.0.0",
-    )
-
-
-# Adopter-side helper for wiring SubdomainTenantMiddleware directly.
-# Until ``serve()`` accepts an ``asgi_middleware=`` kwarg, callers who
-# need tenant routing build the unified app themselves and run
-# uvicorn outside the framework's ``serve()`` wrapper.
-def build_app(
-    platform: V3ReferenceSeller,
-    *,
-    router: SqlSubdomainTenantRouter,
-    buyer_registry,  # type: ignore[no-untyped-def]
-    audit_sink,  # type: ignore[no-untyped-def]
-):
-    """Construct the unified MCP+A2A app with the tenant middleware.
-
-    Returns an ASGI app callable; the caller runs uvicorn against it.
-    """
-    from concurrent.futures import ThreadPoolExecutor
-
-    from adcp.decisioning.handler import PlatformHandler
-    from adcp.decisioning.serve import _build_mcp_and_a2a_app
-    from adcp.decisioning.task_registry import InMemoryTaskRegistry
-
-    executor = ThreadPoolExecutor(max_workers=8, thread_name_prefix="v3-ref-")
-    handler = PlatformHandler(
-        platform,
-        executor=executor,
-        registry=InMemoryTaskRegistry(),
+        transport="both",
         buyer_agent_registry=buyer_registry,
+        context_factory=_build_context_factory(),
+        # SubdomainTenantMiddleware reads the request Host header,
+        # resolves it via the SQL router, and sets the
+        # ``current_tenant()`` contextvar before the handler runs.
+        # The buyer_registry, AccountStore, and audit sink all read
+        # that contextvar — this is the only multi-tenant wiring
+        # point.
+        asgi_middleware=[
+            (SubdomainTenantMiddleware, {"router": router}),
+        ],
     )
-    app = _build_mcp_and_a2a_app(
-        handler,
-        name="v3-reference-seller",
-        port=3001,
-        host="0.0.0.0",
-        instructions=None,
-        test_controller=None,
-        context_factory=_build_context_factory(router),
-    )
-    # Wrap with the tenant middleware. ASGI middleware composes
-    # outermost-first: requests pass through the tenant resolver
-    # before reaching the dispatcher.
-    from adcp.server.tenant_router import SubdomainTenantMiddleware as _SubdomainTenantMiddleware
-
-    class _Wrapped:
-        def __init__(self, app):
-            self._app = app
-            self._mw = _SubdomainTenantMiddleware(app, router=router)
-
-        async def __call__(self, scope, receive, send):
-            await self._mw(scope, receive, send)
-
-    return _Wrapped(app)
 
 
 if __name__ == "__main__":
     main()
 
 
-__all__ = ["build_app", "main"]
+__all__ = ["main"]

examples/v3_reference_seller/src/audit.py

@@ -5,14 +5,17 @@
 queries ("who suspended buyer X yesterday?").
 
 Adopters with Slack / PagerDuty alerting compose this sink with
-:class:`adcp.audit_sink.SlackAlertSink` via a
-``CompositeAuditSink`` — both fire on every event; Slack is
-fire-and-forget, the DB sink is the durable record.
-
-The sink is **fire-and-forget** by contract: failures inside
-:meth:`record` are bounded and swallowed by the framework's audit
-middleware. Adopters needing transactional audit (refusing the
-request if audit fails) implement their own pre-dispatch middleware.
+:class:`adcp.audit_sink.SlackAlertSink` via a ``CompositeAuditSink``
+so each sink fires on every event — Slack for the alert, this DB
+sink for the durable record.
+
+Failure semantics: ``record()`` is **best-effort** — it ``await`` s
+the DB write inline, then swallows any exception so the audit step
+never fails the request. The audit step is therefore in the latency
+path of every skill call; adopters with high-volume audit traffic
+either decouple via an ``asyncio.Queue`` worker or batch writes.
+Adopters needing transactional audit (refusing the request if audit
+fails) implement their own pre-dispatch middleware.
 """
 
 from __future__ import annotations

examples/v3_reference_seller/src/buyer_registry.py

@@ -16,7 +16,6 @@
 
 from __future__ import annotations
 
-import json
 import logging
 from typing import TYPE_CHECKING
 
@@ -106,10 +105,10 @@ async def resolve_by_credential(
 
 def _row_to_agent(row: BuyerAgentRow) -> BuyerAgent:
     """Project ORM row → framework typed :class:`BuyerAgent`."""
-    capabilities = frozenset(json.loads(row.billing_capabilities))
+    capabilities = frozenset(row.billing_capabilities or ())
     terms: BuyerAgentDefaultTerms | None = None
-    if row.default_terms_json:
-        d = json.loads(row.default_terms_json)
+    if row.default_terms:
+        d = row.default_terms
         terms = BuyerAgentDefaultTerms(
             rate_card=d.get("rate_card"),
             payment_terms=d.get("payment_terms"),
@@ -118,7 +117,7 @@ def _row_to_agent(row: BuyerAgentRow) -> BuyerAgent:
         )
     brands: frozenset[str] | None = None
     if row.allowed_brands:
-        brands = frozenset(json.loads(row.allowed_brands))
+        brands = frozenset(row.allowed_brands)
     return BuyerAgent(
         agent_url=row.agent_url,
         display_name=row.display_name,

examples/v3_reference_seller/src/models.py

@@ -27,7 +27,6 @@
 
 from __future__ import annotations
 
-import json
 import uuid
 from datetime import datetime, timezone
 from typing import Any
@@ -131,8 +130,9 @@ class BuyerAgent(Base):
     :class:`buyer_registry.TenantScopedBuyerAgentRegistry` on top.
 
     ``billing_capabilities`` is the framework-enforced
-    ``frozenset[BillingMode]`` from the spec — adopters store as a
-    JSON array and project at the seam.
+    ``frozenset[BillingMode]`` from the spec — stored as a native
+    JSON array column and projected to ``frozenset`` at the
+    registry seam.
     """
 
     __tablename__ = "buyer_agents"
@@ -156,24 +156,24 @@ class BuyerAgent(Base):
     #: (terminal) before the platform method runs.
     status: Mapped[str] = mapped_column(String(32), nullable=False, default="active")
 
-    #: JSON-encoded list of permitted ``BillingMode`` values. Default
-    #: ``["operator"]`` is the pre-trust passthrough posture — agents
-    #: under this row get operator-billed accounts only.
-    billing_capabilities: Mapped[str] = mapped_column(
-        String(255), nullable=False, default=lambda: json.dumps(["operator"])
+    #: List of permitted ``BillingMode`` values. Default ``["operator"]``
+    #: is the pre-trust passthrough posture — agents under this row
+    #: get operator-billed accounts only.
+    billing_capabilities: Mapped[list[str]] = mapped_column(
+        JSON, nullable=False, default=lambda: ["operator"]
     )
 
     #: Pre-trust beta API key. Adopters running signing-only auth
     #: leave NULL.
     api_key_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
 
     #: Default account terms (rate card, payment terms, credit limit,
-    #: billing entity FK). JSON for portability.
-    default_terms_json: Mapped[str | None] = mapped_column(String, nullable=True)
+    #: billing entity FK).
+    default_terms: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
 
     #: Pre-RFC allowlist of brand domains. Layered with the (Tier 3)
     #: ``BrandAuthorizationResolver`` once spec #3690 lands.
-    allowed_brands: Mapped[str | None] = mapped_column(String, nullable=True)
+    allowed_brands: Mapped[list[str] | None] = mapped_column(JSON, nullable=True)
 
     #: Adopter passthrough — internal ids, contract refs, etc.
     ext: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)