From 3e063b0f0acc43b59c2257c0bcefbdb1e26ebb41 Mon Sep 17 00:00:00 2001 From: numaras Date: Fri, 11 Sep 2026 19:41:31 -0300 Subject: [PATCH 1/9] feat: publish deal.created from every booking path Only the approval flow and the negotiation service emitted events. A deal booked over REST (quote -> book, from-template, or curated) left GET /events empty, so a client polling the feed never learned a deal existed. Add _emit_deal_created() and call it after each booking path persists its deal. The payload carries source (quote | template | curated), deal_type, status, product_id and quote_id. Fail-open like every non-audit event: a bus failure never fails the booking. Tests authenticate the quote booking call, since #77 requires a verified buyer on POST /api/v1/deals. --- src/ad_seller/services/deal_service.py | 28 +++++ tests/unit/test_booking_events.py | 137 +++++++++++++++++++++++++ 2 files changed, 165 insertions(+) create mode 100644 tests/unit/test_booking_events.py diff --git a/src/ad_seller/services/deal_service.py b/src/ad_seller/services/deal_service.py index d099d226..148c60f7 100644 --- a/src/ad_seller/services/deal_service.py +++ b/src/ad_seller/services/deal_service.py @@ -189,6 +189,31 @@ 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. + + ``source`` names the path (``quote``, ``template``, ``curated``) so + consumers of the event feed can tell how the deal came to exist. + Fail-open like every non-audit event: a bus failure never fails the + booking. + """ + 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 book_deal(request: Any) -> dict[str, Any]: """Book a deal from a previously issued quote (``DealBookingRequestModel``). @@ -341,6 +366,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 @@ -615,6 +641,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 @@ -1373,6 +1400,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, diff --git a/tests/unit/test_booking_events.py b/tests/unit/test_booking_events.py new file mode 100644 index 00000000..377829ac --- /dev/null +++ b/tests/unit/test_booking_events.py @@ -0,0 +1,137 @@ +"""Booking paths publish ``deal.created`` on the event bus. + +Only the approval flow and the negotiation service emitted events; a deal +booked over REST (quote → book, or from-template) left the event feed +empty, so an inbox polling ``GET /events`` never saw it. +""" + +import sys +from types import ModuleType, SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest + +_broken_flows = ["ad_seller.flows.execution_activation_flow"] +for _mod_name in _broken_flows: + if _mod_name not in sys.modules: + _stub = ModuleType(_mod_name) + _cls_name = _mod_name.rsplit(".", 1)[-1].replace("_", " ").title().replace(" ", "") + setattr(_stub, _cls_name, type(_cls_name, (), {})) + sys.modules[_mod_name] = _stub + +import httpx # noqa: E402 +from httpx import ASGITransport # noqa: E402 + +from ad_seller.events.bus import InMemoryEventBus # noqa: E402 +from ad_seller.events.models import EventType # noqa: E402 +from ad_seller.interfaces.api.main import _get_optional_api_key_record, app # noqa: E402 +from ad_seller.models.api_key import ApiKeyRole # noqa: E402 +from ad_seller.models.buyer_identity import BuyerIdentity # noqa: E402 +from ad_seller.services import deal_service # noqa: E402 +from tests.unit.test_curated_deal import _catalog, _curated_request, _make_product # noqa: E402 +from tests.unit.test_deal_booking_endpoints import ( # noqa: E402 + _authenticate, + _make_available_quote, +) +from tests.unit.test_trust_tier_verification import _mock_catalog # noqa: E402 + + +@pytest.fixture +def mock_storage(): + store = {} + storage = AsyncMock() + storage.get = AsyncMock(side_effect=lambda k: store.get(k)) + storage.set = AsyncMock(side_effect=lambda k, v, ttl=None: store.__setitem__(k, v)) + storage.get_quote = AsyncMock(side_effect=lambda qid: store.get(f"quote:{qid}")) + storage.set_quote = AsyncMock( + side_effect=lambda qid, data, ttl=86400: store.__setitem__(f"quote:{qid}", data) + ) + storage.get_deal = AsyncMock(side_effect=lambda did: store.get(f"deal:{did}")) + storage.set_deal = AsyncMock( + side_effect=lambda did, data: store.__setitem__(f"deal:{did}", data) + ) + storage._store = store + return storage + + +@pytest.fixture +def event_bus(): + return InMemoryEventBus() + + +@pytest.fixture +def client(): + app.dependency_overrides[_get_optional_api_key_record] = lambda: None + transport = ASGITransport(app=app) + c = httpx.AsyncClient(transport=transport, base_url="http://test") + yield c + app.dependency_overrides.clear() + + +class TestBookingEmitsDealCreated: + async def test_quote_booking_emits_deal_created(self, client, mock_storage, event_bus): + quote = _make_available_quote() + mock_storage._store[f"quote:{quote['quote_id']}"] = quote + _authenticate() # booking requires a verified buyer since #77 + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.bus.get_event_bus", AsyncMock(return_value=event_bus)), + ): + resp = await client.post( + "/api/v1/deals", + json={"idempotency_key": "idem-evt", "quote_id": quote["quote_id"]}, + ) + + assert resp.status_code == 200 + deal_id = resp.json()["deal"]["deal_id"] + events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) + assert [e.deal_id for e in events] == [deal_id] + assert events[0].payload["deal_type"] == quote["deal_type"] + assert events[0].payload["source"] == "quote" + + async def test_template_booking_emits_deal_created(self, client, mock_storage, event_bus): + # from-template requires an authenticated buyer; supply a minimal key record. + app.dependency_overrides[_get_optional_api_key_record] = lambda: SimpleNamespace( + key_id="k-test", + role=ApiKeyRole.BUYER, + identity=BuyerIdentity(agency_id="agency-001", agency_name="Agency One"), + ) + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.bus.get_event_bus", AsyncMock(return_value=event_bus)), + patch( + "ad_seller.interfaces.api.main._get_static_product_catalog", + return_value=_mock_catalog(), + ), + ): + resp = await client.post( + "/api/v1/deals/from-template", + json={ + "idempotency_key": "idem-tpl-evt", + "deal_type": "PD", + "product_id": "ctv-premium-sports", + "impressions": 1000000, + }, + ) + + assert resp.status_code == 201, resp.text + deal_id = resp.json()["deal_id"] + events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) + assert [e.deal_id for e in events] == [deal_id] + assert events[0].payload["deal_type"] == "PD" + assert events[0].payload["source"] == "template" + + async def test_curated_booking_emits_deal_created(self, mock_storage, event_bus): + catalog = _catalog(_make_product(base_cpm=45.0, floor_cpm=35.0)) + with ( + patch("ad_seller.storage.factory.get_storage", AsyncMock(return_value=mock_storage)), + patch("ad_seller.events.bus.get_event_bus", AsyncMock(return_value=event_bus)), + ): + result = await deal_service.create_curated_deal( + _curated_request(product_id="ctv-premium-sports"), catalog + ) + + events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) + assert [e.deal_id for e in events] == [result["deal_id"]] + assert events[0].payload["source"] == "curated" From 53917d20a7c4153e6f13122dd72fcb96d8fb90cf Mon Sep 17 00:00:00 2001 From: numaras Date: Tue, 15 Sep 2026 16:28:17 -0300 Subject: [PATCH 2/9] docs: state that deal.created is audit-class and pin both failure branches _emit_deal_created is fail-closed, not fail-open: deal.created is in AUDIT_EVENT_TYPES, so a bus failure writes the event to the audit fallback file and the booking still succeeds; if that write also fails, the error propagates after the deal is already persisted. No code-path change. Two tests pin the branches over HTTP on the quote path. --- src/ad_seller/services/deal_service.py | 6 ++- tests/unit/test_booking_events.py | 62 +++++++++++++++++++++++++- 2 files changed, 65 insertions(+), 3 deletions(-) diff --git a/src/ad_seller/services/deal_service.py b/src/ad_seller/services/deal_service.py index 148c60f7..ddf7ee31 100644 --- a/src/ad_seller/services/deal_service.py +++ b/src/ad_seller/services/deal_service.py @@ -194,8 +194,10 @@ async def _emit_deal_created(deal_data: dict[str, Any], source: str) -> None: ``source`` names the path (``quote``, ``template``, ``curated``) so consumers of the event feed can tell how the deal came to exist. - Fail-open like every non-audit event: a bus failure never fails the - booking. + ``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 diff --git a/tests/unit/test_booking_events.py b/tests/unit/test_booking_events.py index 377829ac..1bd45d40 100644 --- a/tests/unit/test_booking_events.py +++ b/tests/unit/test_booking_events.py @@ -7,7 +7,7 @@ import sys from types import ModuleType, SimpleNamespace -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -135,3 +135,63 @@ async def test_curated_booking_emits_deal_created(self, mock_storage, event_bus) events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) assert [e.deal_id for e in events] == [result["deal_id"]] assert events[0].payload["source"] == "curated" + + +class TestDealCreatedIsAuditClass: + """``deal.created`` is in ``AUDIT_EVENT_TYPES``: a bus failure falls back + to the audit log and the booking still succeeds; a fallback failure + propagates, but only after the deal is persisted.""" + + async def test_bus_failure_writes_fallback_and_booking_succeeds(self, client, mock_storage): + quote = _make_available_quote() + mock_storage._store[f"quote:{quote['quote_id']}"] = quote + _authenticate() + broken_bus = SimpleNamespace(publish=AsyncMock(side_effect=RuntimeError("bus down"))) + fallback = MagicMock() + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.bus.get_event_bus", AsyncMock(return_value=broken_bus)), + patch("ad_seller.events.helpers.write_audit_fallback", fallback), + ): + resp = await client.post( + "/api/v1/deals", + json={"idempotency_key": "idem-bus-down", "quote_id": quote["quote_id"]}, + ) + + assert resp.status_code == 200, resp.text + fallback.assert_called_once() + record = fallback.call_args.args[0] + assert record["event_type"] == "deal.created" + + async def test_fallback_failure_propagates_after_persist(self, mock_storage): + quote = _make_available_quote() + mock_storage._store[f"quote:{quote['quote_id']}"] = quote + _authenticate() + broken_bus = SimpleNamespace(publish=AsyncMock(side_effect=RuntimeError("bus down"))) + fallback = MagicMock(side_effect=OSError("disk full")) + + transport = ASGITransport(app=app, raise_app_exceptions=False) + try: + async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw: + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.bus.get_event_bus", AsyncMock(return_value=broken_bus)), + patch("ad_seller.events.helpers.write_audit_fallback", fallback), + ): + resp = await raw.post( + "/api/v1/deals", + json={ + "idempotency_key": "idem-fallback-down", + "quote_id": quote["quote_id"], + }, + ) + finally: + app.dependency_overrides.clear() + + assert resp.status_code == 500, resp.text + deal_ids = [k.removeprefix("deal:") for k in mock_storage._store if k.startswith("deal:")] + assert len(deal_ids) == 1 + persisted = await mock_storage.get_deal(deal_ids[0]) + assert persisted["deal_id"] == deal_ids[0] + assert persisted["quote_id"] == quote["quote_id"] From a73033cec09cd894e54cb73e00486fbfe530ef4c Mon Sep 17 00:00:00 2001 From: numaras Date: Tue, 15 Sep 2026 16:29:25 -0300 Subject: [PATCH 3/9] feat: emit deal.created from the bulk-create and migration booking paths bulk_deal_operations (create branch) and migrate_deal persisted a new deal without publishing deal.created, so the event feed missed those bookings. Both now call _emit_deal_created right after storage.set_deal, with source "bulk" and "migration"; the helper docstring lists all five sources. No emission for deprecate or for the old deal in a migration. --- src/ad_seller/services/deal_service.py | 7 ++-- tests/unit/test_booking_events.py | 47 ++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/src/ad_seller/services/deal_service.py b/src/ad_seller/services/deal_service.py index ddf7ee31..4ed14cb0 100644 --- a/src/ad_seller/services/deal_service.py +++ b/src/ad_seller/services/deal_service.py @@ -192,8 +192,9 @@ 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. - ``source`` names the path (``quote``, ``template``, ``curated``) so - consumers of the event feed can tell how the deal came to exist. + ``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 @@ -782,6 +783,7 @@ def _result(index: int, action: str, success: bool, deal_id=None, error=None): "notes": op.notes, } await storage.set_deal(deal_id, deal_data) + await _emit_deal_created(deal_data, source="bulk") # Mark quote as booked quote["status"] = QuoteStatus.BOOKED.value @@ -1493,6 +1495,7 @@ async def migrate_deal(deal_id: str, request: Any) -> dict[str, Any]: } await storage.set_deal(new_deal_id, new_deal) + await _emit_deal_created(new_deal, source="migration") # Deprecate old deal old_deal["status"] = "deprecated" diff --git a/tests/unit/test_booking_events.py b/tests/unit/test_booking_events.py index 1bd45d40..dfd20be9 100644 --- a/tests/unit/test_booking_events.py +++ b/tests/unit/test_booking_events.py @@ -136,6 +136,53 @@ async def test_curated_booking_emits_deal_created(self, mock_storage, event_bus) assert [e.deal_id for e in events] == [result["deal_id"]] assert events[0].payload["source"] == "curated" + async def test_bulk_create_emits_deal_created(self, mock_storage, event_bus): + quote = _make_available_quote() + mock_storage._store[f"quote:{quote['quote_id']}"] = quote + op = SimpleNamespace(action="create", quote_id=quote["quote_id"], deal_id=None, notes=None) + + with ( + patch("ad_seller.storage.factory.get_storage", AsyncMock(return_value=mock_storage)), + patch("ad_seller.events.bus.get_event_bus", AsyncMock(return_value=event_bus)), + ): + results = await deal_service.bulk_deal_operations([op]) + + assert results[0]["success"] is True, results + events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) + assert [e.deal_id for e in events] == [results[0]["deal_id"]] + assert events[0].payload["source"] == "bulk" + + async def test_migration_emits_deal_created(self, mock_storage, event_bus): + old_id = "DEAL-ORIG" + mock_storage._store[f"deal:{old_id}"] = { + "deal_id": old_id, + "deal_type": "PD", + "status": "confirmed", + "product_id": "ctv-premium-sports", + "actual_price_cpm": 30.0, + "impressions": 1_000_000, + } + request = SimpleNamespace( + deal_type=None, + product_id=None, + max_cpm=None, + impressions=None, + flight_start=None, + flight_end=None, + buyer_seat_ids=None, + reason="better supply path", + ) + + with ( + patch("ad_seller.storage.factory.get_storage", AsyncMock(return_value=mock_storage)), + patch("ad_seller.events.bus.get_event_bus", AsyncMock(return_value=event_bus)), + ): + result = await deal_service.migrate_deal(old_id, request) + + events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) + assert [e.deal_id for e in events] == [result["new_deal_id"]] + assert events[0].payload["source"] == "migration" + class TestDealCreatedIsAuditClass: """``deal.created`` is in ``AUDIT_EVENT_TYPES``: a bus failure falls back From 66c7c07fd7173b9a496e1d28e7ae711b390860f2 Mon Sep 17 00:00:00 2001 From: numaras Date: Tue, 15 Sep 2026 16:29:57 -0300 Subject: [PATCH 4/9] test: pin that an idempotent replay of a booking emits deal.created once Same quote booked twice over HTTP with the same idempotency_key: both responses are 200 with the same deal_id and the bus holds exactly one deal.created. Test-only; the emission already sits after the replay short-circuit. --- tests/unit/test_booking_events.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/unit/test_booking_events.py b/tests/unit/test_booking_events.py index dfd20be9..539a2c82 100644 --- a/tests/unit/test_booking_events.py +++ b/tests/unit/test_booking_events.py @@ -183,6 +183,26 @@ async def test_migration_emits_deal_created(self, mock_storage, event_bus): assert [e.deal_id for e in events] == [result["new_deal_id"]] assert events[0].payload["source"] == "migration" + async def test_idempotent_replay_emits_once(self, client, mock_storage, event_bus): + quote = _make_available_quote() + mock_storage._store[f"quote:{quote['quote_id']}"] = quote + _authenticate() + body = {"idempotency_key": "idem-replay", "quote_id": quote["quote_id"]} + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.bus.get_event_bus", AsyncMock(return_value=event_bus)), + ): + first = await client.post("/api/v1/deals", json=body) + second = await client.post("/api/v1/deals", json=body) + + assert first.status_code == 200, first.text + assert second.status_code == 200, second.text + deal_id = first.json()["deal"]["deal_id"] + assert second.json()["deal"]["deal_id"] == deal_id + events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) + assert [e.deal_id for e in events] == [deal_id] + class TestDealCreatedIsAuditClass: """``deal.created`` is in ``AUDIT_EVENT_TYPES``: a bus failure falls back From 91f0bac359ca8ac3615d6fef23963a6d923f4c7e Mon Sep 17 00:00:00 2001 From: numaras Date: Tue, 15 Sep 2026 16:31:01 -0300 Subject: [PATCH 5/9] test: pin every deal.created payload key on all five booking paths Each payload-inspecting test now asserts source, deal_type, status, product_id and quote_id (None where the path books without a quote) through one module-level _assert_payload helper shared by the five tests. --- tests/unit/test_booking_events.py | 57 +++++++++++++++++++++++++++---- 1 file changed, 50 insertions(+), 7 deletions(-) diff --git a/tests/unit/test_booking_events.py b/tests/unit/test_booking_events.py index 539a2c82..a3023dfd 100644 --- a/tests/unit/test_booking_events.py +++ b/tests/unit/test_booking_events.py @@ -68,6 +68,16 @@ def client(): app.dependency_overrides.clear() +_PAYLOAD_KEYS = ("source", "deal_type", "status", "product_id", "quote_id") + + +def _assert_payload(event, **expected): + """Pin every ``deal.created`` payload key (``quote_id`` is ``None`` on + paths that book without a quote).""" + assert set(expected) == set(_PAYLOAD_KEYS), "pin all five payload keys" + assert event.payload == expected + + class TestBookingEmitsDealCreated: async def test_quote_booking_emits_deal_created(self, client, mock_storage, event_bus): quote = _make_available_quote() @@ -87,8 +97,14 @@ async def test_quote_booking_emits_deal_created(self, client, mock_storage, even deal_id = resp.json()["deal"]["deal_id"] events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) assert [e.deal_id for e in events] == [deal_id] - assert events[0].payload["deal_type"] == quote["deal_type"] - assert events[0].payload["source"] == "quote" + _assert_payload( + events[0], + source="quote", + deal_type=quote["deal_type"], + status="proposed", + product_id="ctv-premium-sports", + quote_id=quote["quote_id"], + ) async def test_template_booking_emits_deal_created(self, client, mock_storage, event_bus): # from-template requires an authenticated buyer; supply a minimal key record. @@ -119,8 +135,14 @@ async def test_template_booking_emits_deal_created(self, client, mock_storage, e deal_id = resp.json()["deal_id"] events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) assert [e.deal_id for e in events] == [deal_id] - assert events[0].payload["deal_type"] == "PD" - assert events[0].payload["source"] == "template" + _assert_payload( + events[0], + source="template", + deal_type="PD", + status="confirmed", + product_id="ctv-premium-sports", + quote_id=None, + ) async def test_curated_booking_emits_deal_created(self, mock_storage, event_bus): catalog = _catalog(_make_product(base_cpm=45.0, floor_cpm=35.0)) @@ -134,7 +156,14 @@ async def test_curated_booking_emits_deal_created(self, mock_storage, event_bus) events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) assert [e.deal_id for e in events] == [result["deal_id"]] - assert events[0].payload["source"] == "curated" + _assert_payload( + events[0], + source="curated", + deal_type="PMP", + status="confirmed", + product_id="ctv-premium-sports", + quote_id=None, + ) async def test_bulk_create_emits_deal_created(self, mock_storage, event_bus): quote = _make_available_quote() @@ -150,7 +179,14 @@ async def test_bulk_create_emits_deal_created(self, mock_storage, event_bus): assert results[0]["success"] is True, results events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) assert [e.deal_id for e in events] == [results[0]["deal_id"]] - assert events[0].payload["source"] == "bulk" + _assert_payload( + events[0], + source="bulk", + deal_type=quote["deal_type"], + status="confirmed", + product_id="ctv-premium-sports", + quote_id=quote["quote_id"], + ) async def test_migration_emits_deal_created(self, mock_storage, event_bus): old_id = "DEAL-ORIG" @@ -181,7 +217,14 @@ async def test_migration_emits_deal_created(self, mock_storage, event_bus): events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) assert [e.deal_id for e in events] == [result["new_deal_id"]] - assert events[0].payload["source"] == "migration" + _assert_payload( + events[0], + source="migration", + deal_type="PD", + status="confirmed", + product_id="ctv-premium-sports", + quote_id=None, + ) async def test_idempotent_replay_emits_once(self, client, mock_storage, event_bus): quote = _make_available_quote() From b9a4822aab9f3a5e09597ac075d961113a5423ba Mon Sep 17 00:00:00 2001 From: numaras Date: Tue, 15 Sep 2026 16:32:36 -0300 Subject: [PATCH 6/9] test: stop the audience-plan suite from writing the audit fallback file The http_client fixture's _FakeStorage has no ``set``, so the booking's audit-class deal.created fell back to data/audit_fallback.jsonl on every run. The fixture now patches get_event_bus to an InMemoryEventBus for every test that books through it, and the fallback path is gitignored next to *.db. --- .gitignore | 1 + tests/unit/test_audience_plan_validation.py | 9 +++++++-- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/.gitignore b/.gitignore index cd538a43..943b57ba 100644 --- a/.gitignore +++ b/.gitignore @@ -77,6 +77,7 @@ cython_debug/ # Database files *.db +data/audit_fallback.jsonl *.sqlite *.sqlite3 diff --git a/tests/unit/test_audience_plan_validation.py b/tests/unit/test_audience_plan_validation.py index 3e40d1b6..81c75649 100644 --- a/tests/unit/test_audience_plan_validation.py +++ b/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, @@ -491,7 +492,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() From 6a1c9e70588e8830760c3cdcf795841fd33fa077 Mon Sep 17 00:00:00 2001 From: numaras Date: Tue, 15 Sep 2026 16:33:07 -0300 Subject: [PATCH 7/9] docs: describe the real deal.created emitters in the event-bus overview The DEAL_CREATED row said the event came from an accepted proposal; it is published by every booking path, with payload.source naming which one. --- docs/event-bus/overview.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/event-bus/overview.md b/docs/event-bus/overview.md index d21a6cd6..1be12c8f 100644 --- a/docs/event-bus/overview.md +++ b/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 | From b0e20f31bc3c2cd3205a76332d423a5e6e2725ed Mon Sep 17 00:00:00 2001 From: numaras Date: Tue, 15 Sep 2026 16:33:07 -0300 Subject: [PATCH 8/9] docs: changelog entry for deal.created on every booking path --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8dd74345..5aa9aff2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ All notable changes to the IAB Tech Lab Seller Agent are documented here. surfacing later as an uncaught `AttributeError` in `apply_change_request`'s `order_meta.update(proposed)`, a 500 with no way to repair the order. +- `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 From 4577058eb5701ab2f57b867c0200476f6e4e3d81 Mon Sep 17 00:00:00 2001 From: numaras Date: Mon, 5 Oct 2026 13:44:03 -0400 Subject: [PATCH 9/9] fix: publish deal.created after each booking path's last write Bulk create published before marking the quote booked, so a listener could read an available quote for a deal that already existed. Migration had the same shape: it published before deprecating the replaced deal. Both now publish last; the tests record what a listener reads from storage at publish time. Co-Authored-By: Claude Opus 5.5 --- src/ad_seller/services/deal_service.py | 7 +++++-- tests/unit/test_booking_events.py | 24 ++++++++++++++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/src/ad_seller/services/deal_service.py b/src/ad_seller/services/deal_service.py index 541b060a..02617d4f 100644 --- a/src/ad_seller/services/deal_service.py +++ b/src/ad_seller/services/deal_service.py @@ -192,6 +192,9 @@ 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. @@ -783,11 +786,11 @@ def _result(index: int, action: str, success: bool, deal_id=None, error=None): "notes": op.notes, } await storage.set_deal(deal_id, deal_data) - await _emit_deal_created(deal_data, source="bulk") # 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)) @@ -1496,7 +1499,6 @@ async def migrate_deal(deal_id: str, request: Any) -> dict[str, Any]: } await storage.set_deal(new_deal_id, new_deal) - await _emit_deal_created(new_deal, source="migration") # Deprecate old deal old_deal["status"] = "deprecated" @@ -1504,6 +1506,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, diff --git a/tests/unit/test_booking_events.py b/tests/unit/test_booking_events.py index a3023dfd..f924b7ef 100644 --- a/tests/unit/test_booking_events.py +++ b/tests/unit/test_booking_events.py @@ -78,6 +78,20 @@ def _assert_payload(event, **expected): assert event.payload == expected +def _observe_at_publish(bus, observe): + """Record ``observe()`` at the moment the bus publishes, i.e. the state a + listener reacting to the event would read back from storage.""" + seen = [] + publish = bus.publish + + async def _publish(event): + seen.append(observe()) + return await publish(event) + + bus.publish = _publish + return seen + + class TestBookingEmitsDealCreated: async def test_quote_booking_emits_deal_created(self, client, mock_storage, event_bus): quote = _make_available_quote() @@ -169,6 +183,9 @@ async def test_bulk_create_emits_deal_created(self, mock_storage, event_bus): quote = _make_available_quote() mock_storage._store[f"quote:{quote['quote_id']}"] = quote op = SimpleNamespace(action="create", quote_id=quote["quote_id"], deal_id=None, notes=None) + quote_status = _observe_at_publish( + event_bus, lambda: mock_storage._store[f"quote:{quote['quote_id']}"]["status"] + ) with ( patch("ad_seller.storage.factory.get_storage", AsyncMock(return_value=mock_storage)), @@ -179,6 +196,8 @@ async def test_bulk_create_emits_deal_created(self, mock_storage, event_bus): assert results[0]["success"] is True, results events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) assert [e.deal_id for e in events] == [results[0]["deal_id"]] + # The quote is already booked when a listener sees the deal. + assert quote_status == ["booked"] _assert_payload( events[0], source="bulk", @@ -208,6 +227,9 @@ async def test_migration_emits_deal_created(self, mock_storage, event_bus): buyer_seat_ids=None, reason="better supply path", ) + old_status = _observe_at_publish( + event_bus, lambda: mock_storage._store[f"deal:{old_id}"]["status"] + ) with ( patch("ad_seller.storage.factory.get_storage", AsyncMock(return_value=mock_storage)), @@ -217,6 +239,8 @@ async def test_migration_emits_deal_created(self, mock_storage, event_bus): events = await event_bus.list_events(event_type=EventType.DEAL_CREATED.value) assert [e.deal_id for e in events] == [result["new_deal_id"]] + # The replaced deal is already deprecated when a listener sees the new one. + assert old_status == ["deprecated"] _assert_payload( events[0], source="migration",