Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions src/ad_buyer/booking/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__ = [
Expand All @@ -27,7 +30,9 @@
"PricingResult",
"QuoteFlowClient",
"QuoteNormalizer",
"SpendCeilingExceeded",
"SupplyPathInfo",
"TemplateFlowClient",
"enforce_spend_ceiling",
"generate_deal_id",
]
119 changes: 119 additions & 0 deletions src/ad_buyer/booking/spend_ceiling.py
Original file line number Diff line number Diff line change
@@ -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,
)
5 changes: 5 additions & 0 deletions src/ad_buyer/config/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
6 changes: 5 additions & 1 deletion src/ad_buyer/events/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -16,4 +18,6 @@
"close_event_bus",
"emit_event",
"emit_event_sync",
"get_audit_fallback_path",
"write_audit_fallback",
]
46 changes: 46 additions & 0 deletions src/ad_buyer/events/audit_fallback.py
Original file line number Diff line number Diff line change
@@ -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())
Loading
Loading