feat(examples): MediaBuy.invoice_recipient first-class column (3.1 readiness)

examples/v3_reference_seller/README.md

@@ -179,7 +179,8 @@ Adopters typically *don't* change:
 
 This seller is **3.0-compliant on the wire** — every field it sends
 matches the AdCP 3.0 schemas. The schema and architecture is
-**3.1-ready** (`billing_entity` + `reporting_bucket` columns,
-typed `BillingMode`, write-only bank-details projection). Sellers
-running this code today serve 3.0 buyers; the same code serves
-3.1 buyers when the spec lands.
+**3.1-ready** (`billing_entity` + `reporting_bucket` columns on
+`Account`; `invoice_recipient` column on `MediaBuy`; typed
+`BillingMode`; write-only bank-details projection via
+`BusinessEntityResponse`). Sellers running this code today serve
+3.0 buyers; the same code serves 3.1 buyers when the spec lands.

examples/v3_reference_seller/src/models.py

@@ -19,7 +19,9 @@
   details — projection-guarded) and ``reporting_bucket`` (offline
   delivery target).
 * :class:`MediaBuy` — terminal artifact of ``create_media_buy``.
-  Idempotency-keyed for replay safety.
+  Idempotency-keyed for replay safety. Carries the 3.1-ready column
+  ``invoice_recipient`` (per-buy billing entity override —
+  projection-guarded; ``bank`` is stripped on response).
 
 Admin API and protocol-side audit log live in separate tables
 (:mod:`audit` ships :class:`AuditEvent`).
@@ -327,6 +329,14 @@ class MediaBuy(Base):
     start_time: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     end_time: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
 
+    #: Per-buy invoice recipient override — stored as a raw BusinessEntity
+    #: JSON blob. Adopters MUST project through BusinessEntityResponse
+    #: before serializing on any response payload (strips write-only
+    #: bank details). NULL when the buyer did not supply invoice_recipient.
+    #: NOTE: adding this column to an existing DB requires a migration:
+    #:   ALTER TABLE media_buys ADD COLUMN invoice_recipient JSONB;
+    invoice_recipient: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
+
     request_snapshot: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
     response_snapshot: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
 

examples/v3_reference_seller/src/platform.py

@@ -43,6 +43,8 @@
     CreateMediaBuySuccessResponse,
     GetMediaBuyDeliveryRequest,
     GetMediaBuyDeliveryResponse,
+    GetMediaBuysRequest,
+    GetMediaBuysResponse,
     GetProductsRequest,
     GetProductsResponse,
     Product,
@@ -51,6 +53,7 @@
     UpdateMediaBuyRequest,
     UpdateMediaBuySuccessResponse,
 )
+from adcp.types.projections import BusinessEntityResponse
 
 if TYPE_CHECKING:
     from sqlalchemy.ext.asyncio import async_sessionmaker
@@ -220,6 +223,9 @@ async def create_media_buy(
         budget_amount = req.total_budget.amount if req.total_budget else None
         budget_currency = req.total_budget.currency if req.total_budget else None
         start_dt = _project_start_time(req.start_time)
+        invoice_recipient_json = (
+            req.invoice_recipient.model_dump(mode="json") if req.invoice_recipient else None
+        )
         row = MediaBuyRow(
             tenant_id=ctx.account.metadata["tenant_id"],
             account_id=ctx.account.id,
@@ -231,6 +237,7 @@ async def create_media_buy(
             currency=budget_currency,
             start_time=start_dt,
             end_time=req.end_time,
+            invoice_recipient=invoice_recipient_json,
             request_snapshot=req.model_dump(mode="json"),
         )
         async with self._sessionmaker() as session, session.begin():
@@ -241,10 +248,12 @@ async def create_media_buy(
             ctx.account.id,
             ctx.buyer_agent.agent_url,
         )
+        ir_response = _project_invoice_recipient(invoice_recipient_json)
         return CreateMediaBuySuccessResponse(
             media_buy_id=media_buy_id,
             packages=[],
             status="active",
+            invoice_recipient=ir_response,
         )
 
     # ----- update_media_buy ------------------------------------------------
@@ -283,12 +292,84 @@ async def update_media_buy(
                 row.status = "paused"
             elif patch.paused is False and row.status == "paused":
                 row.status = "active"
+            if "invoice_recipient" in patch.model_fields_set:
+                # Allow explicit null to clear an existing override.
+                row.invoice_recipient = (
+                    patch.invoice_recipient.model_dump(mode="json")
+                    if patch.invoice_recipient is not None
+                    else None
+                )
             row.updated_at = datetime.now(timezone.utc)
+        ir_response = _project_invoice_recipient(row.invoice_recipient)
         return UpdateMediaBuySuccessResponse(
             media_buy_id=row.media_buy_id,
             status=row.status,  # type: ignore[arg-type]
             packages=[],
+            invoice_recipient=ir_response,
+        )
+
+    # ----- get_media_buys --------------------------------------------------
+
+    async def get_media_buys(
+        self, req: GetMediaBuysRequest, ctx: RequestContext
+    ) -> GetMediaBuysResponse:
+        """Return media buys for the resolved account, optionally filtered
+        by IDs or status. Populates ``invoice_recipient`` from the
+        first-class column, with ``bank`` stripped via
+        :class:`~adcp.types.projections.BusinessEntityResponse`."""
+        if ctx.account is None:
+            raise AdcpError(
+                "INTERNAL_ERROR",
+                message="Dispatch should have populated account.",
+                recovery="terminal",
+            )
+        tenant_id = ctx.account.metadata["tenant_id"]
+        stmt = select(MediaBuyRow).where(
+            MediaBuyRow.tenant_id == tenant_id,
+            MediaBuyRow.account_id == ctx.account.id,
         )
+        if req.media_buy_ids:
+            stmt = stmt.where(MediaBuyRow.media_buy_id.in_(list(req.media_buy_ids)))
+        else:
+            # Default status filter: active only (mirrors spec default).
+            status_filter = req.status_filter
+            if status_filter is None:
+                stmt = stmt.where(MediaBuyRow.status == "active")
+            elif hasattr(status_filter, "root"):
+                # StatusFilter is a RootModel wrapping list[MediaBuyStatus].
+                stmt = stmt.where(
+                    MediaBuyRow.status.in_([s.value for s in status_filter.root])
+                )
+            else:
+                # Single MediaBuyStatus enum value — use .value to get "paused" not repr.
+                stmt = stmt.where(MediaBuyRow.status == status_filter.value)
+        async with self._sessionmaker() as session:
+            result = await session.execute(stmt)
+            rows = list(result.scalars())
+
+        # Build items as plain dicts so GetMediaBuysResponse.model_validate handles
+        # inner type construction — avoids importing generated_poc classes directly.
+        items: list[dict[str, Any]] = []
+        for row in rows:
+            if row.currency is None or row.total_budget is None:
+                # Skip rows where required response fields were not captured at
+                # create time. Adopters who always supply total_budget will never
+                # hit this; reference seller stores NULL when buyer omits it.
+                continue
+            ir = _project_invoice_recipient(row.invoice_recipient)
+            item: dict[str, Any] = {
+                "media_buy_id": row.media_buy_id,
+                "status": row.status,
+                "currency": row.currency,
+                "total_budget": row.total_budget,
+                "packages": [],
+                "created_at": row.created_at,
+                "updated_at": row.updated_at,
+            }
+            if ir is not None:
+                item["invoice_recipient"] = ir.model_dump(mode="json")
+            items.append(item)
+        return GetMediaBuysResponse.model_validate({"media_buys": items})
 
     # ----- sync_creatives --------------------------------------------------
 
@@ -312,6 +393,20 @@ async def get_media_buy_delivery(
         return GetMediaBuyDeliveryResponse(media_buys=[])
 
 
+def _project_invoice_recipient(data: dict[str, Any] | None) -> BusinessEntityResponse | None:
+    """Project a raw ``invoice_recipient`` JSON blob to its response shape.
+
+    Mirrors :func:`adcp.types.projections.to_account_response` for the
+    per-buy billing override: strips write-only ``bank`` details before
+    constructing :class:`~adcp.types.projections.BusinessEntityResponse`.
+    """
+    if not data:
+        return None
+    payload = dict(data)
+    payload.pop("bank", None)
+    return BusinessEntityResponse.model_validate(payload)
+
+
 def _project_start_time(value: Any) -> datetime:
     """Project :class:`StartTiming` (root: ``'asap'`` | :class:`AwareDatetime`)
     into a flat timezone-aware datetime for SQL storage.

examples/v3_reference_seller/tests/test_smoke.py

@@ -96,6 +96,44 @@ def scalar_one_or_none(self):
     assert result is None
 
 
+def test_media_buy_has_invoice_recipient_column() -> None:
+    """MediaBuy ORM has a first-class invoice_recipient column (3.1-ready)."""
+    from src.models import MediaBuy
+
+    col_names = {c.key for c in MediaBuy.__table__.columns}
+    assert "invoice_recipient" in col_names
+
+
+def test_invoice_recipient_round_trips_without_bank() -> None:
+    """invoice_recipient stored as JSON and reconstructed via
+    _project_invoice_recipient strips write-only bank details.
+
+    Mirrors the platform path: req.invoice_recipient.model_dump() may
+    include bank (stored verbatim in the DB column), then on the
+    response edge _project_invoice_recipient pops bank before
+    constructing BusinessEntityResponse.
+    """
+    from adcp.types.projections import BusinessEntityResponse
+
+    # Simulate a DB row that was stored WITH bank details included.
+    stored_in_db = {
+        "legal_name": "Acme Billing LLC",
+        "tax_id": "12-3456789",
+        "bank": {
+            "account_holder": "Acme",
+            "iban": "DE89370400440532013000",
+        },
+    }
+    # _project_invoice_recipient logic: pop bank, then validate.
+    payload = dict(stored_in_db)
+    payload.pop("bank", None)
+    projected = BusinessEntityResponse.model_validate(payload)
+    serialized = projected.model_dump(mode="json", exclude_none=True)
+
+    assert serialized.get("legal_name") == "Acme Billing LLC"
+    assert "bank" not in serialized, "bank is write-only and must be stripped from responses"
+
+
 @pytest.mark.asyncio
 async def test_buyer_registry_returns_none_without_tenant() -> None:
     """Without a tenant context (ContextVar unset), the registry