feat: publish deal.created from every booking path

.gitignore

@@ -77,6 +77,7 @@ cython_debug/
 
 # Database files
 *.db
+data/audit_fallback.jsonl
 *.sqlite
 *.sqlite3
 

CHANGELOG.md

@@ -19,6 +19,10 @@ All notable changes to the IAB Tech Lab Seller Agent are documented here.
   wire-status filter; unserializable rows reported in `skipped`), and
   `GET /api/v1/deals/export` now reads stored deals instead of an index
   nothing wrote.
+- `deal.created` is now published from every booking path (quote,
+  template, curated, bulk, migration) with a `source` field; the event
+  is audit-class, so a bus failure falls back to the audit log and the
+  booking still succeeds.
 
 ### Changed
 

docs/event-bus/overview.md

@@ -27,7 +27,7 @@ All event types are defined in `src/ad_seller/events/models.py` as the `EventTyp
 
 | Event Type | Value | Description |
 |-----------|-------|-------------|
-| `DEAL_CREATED` | `deal.created` | A deal was generated from an accepted proposal |
+| `DEAL_CREATED` | `deal.created` | A deal was persisted by a booking path; `payload.source` is one of `quote`, `template`, `curated`, `bulk`, `migration` |
 | `DEAL_REGISTERED` | `deal.registered` | The deal was registered in the system |
 | `DEAL_SYNCED` | `deal.synced` | The deal was synced to the ad server |
 

src/ad_seller/services/deal_service.py

@@ -189,6 +189,37 @@ def match_agentic_audience(ref: dict[str, Any]) -> dict[str, Any]:
 # =============================================================================
 
 
+async def _emit_deal_created(deal_data: dict[str, Any], source: str) -> None:
+    """Publish ``deal.created`` for a deal just persisted by a booking path.
+
+    Call it after the path's last write (quote marked booked, replaced deal
+    deprecated), so a listener reading storage sees the finished state.
+
+    ``source`` names the path (``quote``, ``template``, ``curated``,
+    ``bulk``, ``migration``) so consumers of the event feed can tell how
+    the deal came to exist.
+    ``deal.created`` is audit-class (``AUDIT_EVENT_TYPES``): on bus failure
+    the event is written to the audit fallback file and the booking still
+    succeeds; if that write also fails the error propagates after the deal
+    is already persisted.
+    """
+    from ..events.helpers import emit_event
+    from ..events.models import EventType
+
+    product = deal_data.get("product") or {}
+    await emit_event(
+        event_type=EventType.DEAL_CREATED,
+        deal_id=deal_data.get("deal_id", ""),
+        payload={
+            "source": source,
+            "deal_type": deal_data.get("deal_type"),
+            "status": deal_data.get("status"),
+            "product_id": product.get("product_id") or deal_data.get("product_id"),
+            "quote_id": deal_data.get("quote_id"),
+        },
+    )
+
+
 async def _resolve_accepted_negotiation(storage: Any, quote_id: str) -> Optional[dict[str, Any]]:
     """Return the ACCEPTED negotiation about ``quote_id``, or None.
 
@@ -393,6 +424,7 @@ async def book_deal(request: Any) -> dict[str, Any]:
     # The snapshot fields land on the persisted record so
     # `honor_audience_plan_snapshot()` can read them at fulfillment time.
     await storage.set_deal(deal_id, deal_data)
+    await _emit_deal_created(deal_data, source="quote")
 
     return deal_data
 
@@ -667,6 +699,7 @@ async def create_deal_from_template(
 
     storage = await get_storage()
     await storage.set_deal(deal_id, deal_data)
+    await _emit_deal_created(deal_data, source="template")
 
     return deal_data
 
@@ -809,6 +842,7 @@ def _result(index: int, action: str, success: bool, deal_id=None, error=None):
                 # Mark quote as booked
                 quote["status"] = QuoteStatus.BOOKED.value
                 await storage.set_quote(op.quote_id, quote)
+                await _emit_deal_created(deal_data, source="bulk")
 
                 results.append(_result(i, "create", True, deal_id=deal_id))
 
@@ -1426,6 +1460,7 @@ async def create_curated_deal(request: Any, catalog: dict[str, Any]) -> dict[str
 
     storage = await get_storage()
     await storage.set_deal(deal_id, deal_data)
+    await _emit_deal_created(deal_data, source="curated")
 
     return {
         "deal_id": deal_id,
@@ -1523,6 +1558,7 @@ async def migrate_deal(deal_id: str, request: Any) -> dict[str, Any]:
     old_deal["deprecated_reason"] = request.reason or "Replaced by migration"
     old_deal["replacement_deal_id"] = new_deal_id
     await storage.set_deal(deal_id, old_deal)
+    await _emit_deal_created(new_deal, source="migration")
 
     return {
         "new_deal_id": new_deal_id,

tests/unit/test_audience_plan_validation.py

@@ -19,7 +19,7 @@
 import logging
 import sys
 from types import ModuleType
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -37,6 +37,7 @@
 import httpx  # noqa: E402
 from httpx import ASGITransport  # noqa: E402
 
+from ad_seller.events.bus import InMemoryEventBus  # noqa: E402
 from ad_seller.flows.proposal_handling_flow import (  # noqa: E402
     ProposalHandlingFlow,
     ProposalState,
@@ -495,7 +496,11 @@ def http_client():
         "expires_at": (datetime.utcnow() + timedelta(hours=24)).isoformat() + "Z",
         "created_at": datetime.utcnow().isoformat() + "Z",
     }
-    yield c, storage, quote_id
+    # deal.created is audit-class: without a bus, the booking falls back to
+    # writing data/audit_fallback.jsonl. Publish to an in-memory bus instead
+    # so the suite leaves no file behind (_FakeStorage has no ``set``).
+    with patch("ad_seller.events.bus.get_event_bus", AsyncMock(return_value=InMemoryEventBus())):
+        yield c, storage, quote_id
     app.dependency_overrides.clear()