Stage 1 — Security & Guardrail Hardening (#110)

* Add deterministic budget/CPM spend ceiling to the booking path

Nothing on the buyer's money-committing path previously compared the
final price to the campaign's limits: max_cpm reached the selection LLM
only as prose and the seller only as an advisory search filter, so a
seller quoting basePrice 200.0 against a target CPM of 20 was still
issued a Deal ID at ~8.5x the ceiling.

Add a pure, deterministic guard (enforce_spend_ceiling /
SpendCeilingExceeded in ad_buyer.booking.spend_ceiling) and call it at
both money-commit points:

- RequestDealTool._create_deal_response: the computed final CPM is
  checked against the buyer's max_cpm BEFORE a Deal ID is minted; a
  breach returns a structured DEAL REJECTED error. BuyerDealFlow
  threads state.max_cpm onto the tool deterministically (not via an
  LLM-visible argument).
- DealBookingFlow._execute_bookings: the total cost of LLM-parsed
  approved recommendations is checked against the campaign budget
  before any line is booked; a breach rejects the booking and marks
  the flow FAILED.

Missing limits (max_cpm/budget is None) fail open with a warning log —
an explicit choice to preserve current demo behavior; a supplied limit
is always enforced and there is no flag to disable the guard.

MultiSellerOrchestrator is intentionally untouched: evaluate_and_rank
already filters by max_cpm and select_and_book already skips quotes
whose minimum spend exceeds the remaining budget.

bead: ar-70eh

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Fail-closed delivery for audit-class events

Event emission was fail-open across the board: emit_event and
emit_event_sync swallowed every exception, so audit-trail events for
money decisions (deal booked/cancelled, booking submitted, budget
allocation, negotiation rounds, approval decisions) could silently
drop when the event bus failed.

Surgical hotfix, not a bus redesign:

- AUDIT_EVENT_TYPES (events/models.py): the money-relevant event types
  whose loss would break the audit trail.
- events/audit_fallback.py: append-only JSONL fallback writer with
  fsync-on-write; path from settings (audit_fallback_path, default
  data/audit_fallback.jsonl).
- emit_event / emit_event_sync (events/helpers.py): on bus failure for
  an audit-class event, write the event to the fallback JSONL and
  continue; if the fallback write also fails, re-raise so the
  transaction surfaces the error (fail-closed). The fire-and-forget
  running-loop path in emit_event_sync gets a done-callback that
  writes the fallback on publish failure (best effort; cannot raise
  into a caller that already moved on). Non-audit events keep the
  existing fail-open behavior unchanged.

bead: ar-z64h

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* test: restore default event loop after audit fail-closed tests

emit_event_sync consumes the thread default loop; test_event_bus.py's
deprecated asyncio.get_event_loop() then finds none when the files run
adjacently. Fresh loop on fixture teardown removes the order dependency.

bead: ar-z64h

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SPyrrnxpTatTciiNo7eK2y

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: atc964

src/ad_buyer/booking/__init__.py

@@ -13,12 +13,15 @@
     QuoteFlowClient - Quote-then-book flow for deal creation
     QuoteNormalizer - Normalizes quotes from different sellers for comparison
     TemplateFlowClient - Template-based booking (stub)
+    enforce_spend_ceiling - Deterministic budget/CPM ceiling guard
+    SpendCeilingExceeded - Raised when a spend limit would be breached
 """
 
 from .deal_id import generate_deal_id
 from .pricing import PricingCalculator, PricingResult
 from .quote_flow import QuoteFlowClient
 from .quote_normalizer import NormalizedQuote, QuoteNormalizer, SupplyPathInfo
+from .spend_ceiling import SpendCeilingExceeded, enforce_spend_ceiling
 from .template_flow import TemplateFlowClient
 
 __all__ = [
@@ -27,7 +30,9 @@
     "PricingResult",
     "QuoteFlowClient",
     "QuoteNormalizer",
+    "SpendCeilingExceeded",
     "SupplyPathInfo",
     "TemplateFlowClient",
+    "enforce_spend_ceiling",
     "generate_deal_id",
 ]

src/ad_buyer/booking/spend_ceiling.py

@@ -0,0 +1,119 @@
+# Author: Green Mountain Systems AI Inc.
+# Donated to IAB Tech Lab
+
+"""Deterministic spend-ceiling guard for money-committing paths.
+
+Nothing on the buyer's booking path previously compared the final price
+to the campaign's limits: ``max_cpm`` reached the selection LLM only as
+prose and the seller only as an advisory search filter, so a seller
+quoting far above the buyer's ceiling would still be issued a Deal ID.
+
+``enforce_spend_ceiling`` closes that hole. It is pure, deterministic
+code — no LLM involvement, no configuration flag to disable it — and is
+called at every point where the buyer commits money:
+
+* ``RequestDealTool._create_deal_response`` (before a Deal ID is minted)
+* ``DealBookingFlow._execute_bookings`` (before booked lines are created)
+
+Note: ``MultiSellerOrchestrator`` already enforces its own bounds
+(``evaluate_and_rank`` filters quotes by ``max_cpm``; ``select_and_book``
+skips quotes whose minimum spend exceeds the remaining budget), so it is
+intentionally not routed through this guard.
+
+Fail-open policy: when a limit is None (the caller never supplied a
+max_cpm or budget), the corresponding check is skipped and a warning is
+logged. This is an explicit choice to preserve current demo behavior for
+callers that never configure limits; it is NOT a bypass for configured
+limits — a supplied limit is always enforced.
+"""
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+__all__ = ["SpendCeilingExceeded", "enforce_spend_ceiling"]
+
+
+class SpendCeilingExceeded(Exception):
+    """Raised when a deal or booking would breach a spend limit.
+
+    Deliberately does NOT subclass ValueError/RuntimeError so that broad
+    ``except (OSError, ValueError, RuntimeError)`` handlers on the tool
+    paths cannot swallow it by accident — call sites must handle the
+    rejection explicitly.
+
+    Attributes:
+        final_cpm: Computed final CPM that was checked (or None).
+        max_cpm: CPM ceiling that was breached (or None).
+        total_cost: Computed total cost that was checked (or None).
+        budget: Budget ceiling that was breached (or None).
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        final_cpm: float | None = None,
+        max_cpm: float | None = None,
+        total_cost: float | None = None,
+        budget: float | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.final_cpm = final_cpm
+        self.max_cpm = max_cpm
+        self.total_cost = total_cost
+        self.budget = budget
+
+
+def enforce_spend_ceiling(
+    final_cpm: float | None = None,
+    total_cost: float | None = None,
+    max_cpm: float | None = None,
+    budget: float | None = None,
+) -> None:
+    """Enforce CPM and budget ceilings before money is committed.
+
+    Pure deterministic guard: compares actuals against limits and raises
+    when a limit would be breached. At-or-under a limit is allowed.
+
+    Args:
+        final_cpm: The computed final CPM about to be committed.
+        total_cost: The total cost about to be committed.
+        max_cpm: The campaign's maximum acceptable CPM (limit).
+        budget: The campaign's budget (limit).
+
+    Raises:
+        SpendCeilingExceeded: When final_cpm > max_cpm or
+            total_cost > budget (both comparisons only run when both
+            sides are present).
+    """
+    if max_cpm is None and budget is None:
+        # Fail-open (explicit choice): no limits were supplied, so there
+        # is nothing to enforce. Allow the spend but log a warning so
+        # unbounded money paths are visible in logs. This preserves
+        # current demo behavior for callers without configured limits.
+        logger.warning(
+            "Spend ceiling check skipped: no max_cpm or budget limit supplied "
+            "(final_cpm=%s, total_cost=%s). Spend is unbounded.",
+            final_cpm,
+            total_cost,
+        )
+        return
+
+    if max_cpm is not None and final_cpm is not None and final_cpm > max_cpm:
+        raise SpendCeilingExceeded(
+            f"Final CPM ${final_cpm:.2f} exceeds max CPM ceiling ${max_cpm:.2f}",
+            final_cpm=final_cpm,
+            max_cpm=max_cpm,
+            total_cost=total_cost,
+            budget=budget,
+        )
+
+    if budget is not None and total_cost is not None and total_cost > budget:
+        raise SpendCeilingExceeded(
+            f"Total cost ${total_cost:,.2f} exceeds budget ${budget:,.2f}",
+            final_cpm=final_cpm,
+            max_cpm=max_cpm,
+            total_cost=total_cost,
+            budget=budget,
+        )

src/ad_buyer/config/settings.py

@@ -74,6 +74,11 @@ def get_seller_endpoints(self) -> list[str]:
     postgres_pool_min: int = 2
     postgres_pool_max: int = 10
 
+    # Durable fallback for audit-class events (see events/audit_fallback.py).
+    # When the event bus fails for an audit-class event, the event is appended
+    # to this JSONL file (fsynced per write) instead of being dropped.
+    audit_fallback_path: str = "data/audit_fallback.jsonl"
+
     # CrewAI Settings
     crew_memory_enabled: bool = True
     crew_verbose: bool = True

src/ad_buyer/events/__init__.py

@@ -3,11 +3,13 @@
 
 """Event bus for buyer workflow observability and control."""
 
+from .audit_fallback import get_audit_fallback_path, write_audit_fallback
 from .bus import EventBus, InMemoryEventBus, close_event_bus, get_event_bus
 from .helpers import emit_event, emit_event_sync
-from .models import Event, EventType
+from .models import AUDIT_EVENT_TYPES, Event, EventType
 
 __all__ = [
+    "AUDIT_EVENT_TYPES",
     "Event",
     "EventType",
     "EventBus",
@@ -16,4 +18,6 @@
     "close_event_bus",
     "emit_event",
     "emit_event_sync",
+    "get_audit_fallback_path",
+    "write_audit_fallback",
 ]

src/ad_buyer/events/audit_fallback.py

@@ -0,0 +1,46 @@
+# Author: Green Mountain Systems AI Inc.
+# Donated to IAB Tech Lab
+
+"""Durable fallback writer for audit-class events.
+
+When the event bus fails while emitting an audit-class event (see
+AUDIT_EVENT_TYPES in models.py), the event is appended to a local JSONL
+file so the audit trail survives bus outages. Each write is flushed and
+fsynced so the record is on disk before the calling transaction proceeds.
+
+This is an interim safety net, not an event store: a later refactor of
+the event architecture replaces it with durable bus semantics.
+"""
+
+import json
+import os
+from pathlib import Path
+from typing import Any
+
+DEFAULT_AUDIT_FALLBACK_PATH = "data/audit_fallback.jsonl"
+
+
+def get_audit_fallback_path() -> Path:
+    """Resolve the fallback JSONL path from settings, with a sane default."""
+    try:
+        from ..config import get_settings
+
+        path = getattr(get_settings(), "audit_fallback_path", DEFAULT_AUDIT_FALLBACK_PATH)
+    except Exception:  # noqa: BLE001 - settings failure must not block the fallback write
+        path = DEFAULT_AUDIT_FALLBACK_PATH
+    return Path(path)
+
+
+def write_audit_fallback(record: dict[str, Any]) -> None:
+    """Append one event record to the fallback JSONL file, fsync-on-write.
+
+    Raises on failure -- the caller (emit_event) treats a fallback-write
+    failure as fatal for audit-class events (fail-closed).
+    """
+    path = get_audit_fallback_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    line = json.dumps(record, default=str)
+    with open(path, "a", encoding="utf-8") as f:
+        f.write(line + "\n")
+        f.flush()
+        os.fsync(f.fileno())