diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a408c9a..e43397e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,26 @@ All notable changes to the IAB Tech Lab Seller Agent are documented here. orphaned worker thread (CrewAI has no cancellation API). Budget <= 0 still means "no bound" and always runs the crew; wire answers on the timeout path are unchanged. +- **A negotiated price now actually books.** Booking looked for an accepted + negotiation with `get_negotiation(quote_id)`, which only ever hit for a + quote-led negotiation, since that one happens to be stored under the quote + id. A proposal-led negotiation is stored under its `prop-` id, so the + lookup missed and the deal was booked at the seller's standard price with + the negotiation reported as successful. `NegotiationHistory` now retains + the `quote_id` that already arrived on `NegotiationMessage` and was being + dropped, an accepted negotiation indexes itself as + `negotiation_by_quote:{quote_id}`, and booking resolves through that index + (falling back to the direct read, so quote-led negotiations recorded before + the index existed are still honored). **Behavior change:** bookings that + silently ignored an accepted negotiation now honor it, so a booked + `final_cpm` can differ from the quoted one — it carries the agreed price + and says so in its rationale. An accepted negotiation carrying no agreed + price is now refused (`negotiation_price_unresolved`) rather than quietly + booked at the un-negotiated quoted price. Relatedly, the negotiation + storage key is now resolved against the store instead of being read off + whichever ids a given message happened to carry, so a continuation leading + with a different id continues the same negotiation rather than silently + restarting it at round one. ### Fixed diff --git a/src/ad_seller/interfaces/api/routers/negotiation.py b/src/ad_seller/interfaces/api/routers/negotiation.py index a6ecceb8..971f5020 100644 --- a/src/ad_seller/interfaces/api/routers/negotiation.py +++ b/src/ad_seller/interfaces/api/routers/negotiation.py @@ -176,8 +176,17 @@ async def post_negotiation_message( # The seller keys negotiations by proposal_id; negotiation_id doubles as # that key for continuation, and quote-led opens key off the quote_id - # (the service resolves the stored quote to its product). - proposal_id = message.proposal_id or message.negotiation_id or message.quote_id + # (the service resolves the stored quote to its product). Which of those + # is the key is resolved against the STORE rather than read off this + # message: taking `proposal_id or negotiation_id or quote_id` made the key + # depend on the ids a given message happened to carry, so a continuation + # leading with a different id resolved to a different record and silently + # restarted the negotiation. + proposal_id = await negotiation_service.resolve_negotiation_key( + proposal_id=message.proposal_id, + negotiation_id=message.negotiation_id, + quote_id=message.quote_id, + ) if proposal_id is None: # unreachable: the shared model requires one key raise HTTPException( status_code=400, @@ -226,15 +235,21 @@ async def post_negotiation_message( proposal_id=proposal_id, buyer_price=buyer_price, buyer_context=buyer_context, + quote_id=message.quote_id, ) response = cm.negotiation_round_to_response(result) else: # accept / reject — terminal moves off the recorded history; the # price engine is not run (it stays untouched). The move is # PERSISTED onto the stored negotiation so downstream booking sees - # the agreed state. + # the agreed state, and an accept indexes the negotiation by its + # quote so booking that quote can find the agreed price. status_data = await negotiation_service.apply_terminal_action( - proposal_id, message.action.value, buyer_price + proposal_id, + message.action.value, + buyer_price, + quote_id=message.quote_id, + buyer_context=buyer_context, ) response = cm.terminal_round_response(status_data, message.action, buyer_price) diff --git a/src/ad_seller/models/negotiation.py b/src/ad_seller/models/negotiation.py index d0579176..53ab2162 100644 --- a/src/ad_seller/models/negotiation.py +++ b/src/ad_seller/models/negotiation.py @@ -101,6 +101,11 @@ class NegotiationHistory(BaseModel): negotiation_id: str = Field(default_factory=lambda: f"neg-{uuid.uuid4().hex[:8]}") proposal_id: str + # The quote this negotiation concerns, when one was named. It arrives on + # the wire (``NegotiationMessage.quote_id``) and used to be dropped here, + # so nothing could correlate a negotiation to the quote a buyer books — + # booking then struck the stale quoted price and logged success. + quote_id: Optional[str] = None product_id: str buyer_tier: AccessTier strategy: NegotiationStrategy diff --git a/src/ad_seller/services/deal_service.py b/src/ad_seller/services/deal_service.py index 5609b066..4c234730 100644 --- a/src/ad_seller/services/deal_service.py +++ b/src/ad_seller/services/deal_service.py @@ -189,6 +189,34 @@ def match_agentic_audience(ref: dict[str, Any]) -> dict[str, Any]: # ============================================================================= +async def _resolve_accepted_negotiation(storage: Any, quote_id: str) -> Optional[dict[str, Any]]: + """Return the ACCEPTED negotiation about ``quote_id``, or None. + + ``DealBookingRequest`` carries only the quote id, so the quote is the only + handle the protocol offers for correlating a booking to a negotiation. + An accepted negotiation indexes itself under + ``negotiation_by_quote:{quote_id}`` (see + ``negotiation_service._persist_negotiation``); that pointer names the key + its record is stored under, whichever id the buyer led with. + + The direct read by quote id is retained as a fallback: a quote-led + negotiation is itself stored under the quote id, and records written + before the index existed have no pointer. + """ + from ..storage.base import negotiation_by_quote_key + + negotiation = None + pointer = await storage.get(negotiation_by_quote_key(quote_id)) + if isinstance(pointer, str) and pointer: + negotiation = await storage.get_negotiation(pointer) + if not negotiation: + negotiation = await storage.get_negotiation(quote_id) + + if isinstance(negotiation, dict) and negotiation.get("status") == "accepted": + return negotiation + return None + + async def book_deal(request: Any) -> dict[str, Any]: """Book a deal from a previously issued quote (``DealBookingRequestModel``). @@ -238,27 +266,51 @@ async def book_deal(request: Any) -> dict[str, Any]: }, ) - # Honor ACCEPTED negotiation state on this quote: a - # quote-led negotiation is keyed by the quote_id; when it concluded - # accepted, the booking strikes the AGREED price, not the stale quoted - # price. A re-quote at the agreed price (the buyer's historical - # workaround) carries no negotiation and books unchanged. - negotiation = await storage.get_negotiation(request.quote_id) - if negotiation and negotiation.get("status") == "accepted": + # Honor ACCEPTED negotiation state on this quote: when a negotiation about + # this quote concluded accepted, the booking strikes the AGREED price, not + # the stale quoted price. A quote with no accepted negotiation (including a + # re-quote at the agreed price, the buyer's historical workaround) books + # unchanged. + # + # Resolution goes through the quote index an accepted negotiation writes + # (`negotiation_by_quote:{quote_id}` -> the key its record is stored + # under). Passing the quote id straight to get_negotiation only ever + # worked for a QUOTE-LED negotiation, which happens to be stored under the + # quote id; a proposal-led one is stored under its `prop-` id, so the + # lookup missed every time and the agreed price was silently dropped. The + # direct read is kept as a fallback so quote-led negotiations recorded + # before this index existed are still honored. + negotiation = await _resolve_accepted_negotiation(storage, request.quote_id) + if negotiation: rounds = negotiation.get("rounds") or [] agreed_cpm = rounds[-1].get("seller_price") if rounds else None - if agreed_cpm is not None: - quote = { - **quote, - "pricing": { - **quote["pricing"], - "final_cpm": agreed_cpm, - "rationale": ( - f"{quote['pricing'].get('rationale', '')} | Negotiated to " - f"${agreed_cpm:.2f} CPM ({negotiation.get('negotiation_id')})" - ).strip(" |"), + if agreed_cpm is None: + # A negotiation that concluded accepted with no price is a corrupt + # record. Booking it at list price is the worst available outcome — + # it is exactly the silent overpayment this whole path exists to + # prevent — so refuse instead. + raise HTTPException( + status_code=500, + detail={ + "error": "negotiation_price_unresolved", + "message": ( + f"Negotiation '{negotiation.get('negotiation_id')}' for quote " + f"'{request.quote_id}' is accepted but carries no agreed price. " + "Refusing to book at the un-negotiated quoted price." + ), }, - } + ) + quote = { + **quote, + "pricing": { + **quote["pricing"], + "final_cpm": agreed_cpm, + "rationale": ( + f"{quote['pricing'].get('rationale', '')} | Negotiated to " + f"${agreed_cpm:.2f} CPM ({negotiation.get('negotiation_id')})" + ).strip(" |"), + }, + } # Pre-flight: if the buyer sent an audience_plan with this booking, validate # it against the seller's capability block. Per proposal §5.7 layer 3, any diff --git a/src/ad_seller/services/negotiation_service.py b/src/ad_seller/services/negotiation_service.py index 32f27b4f..839a9980 100644 --- a/src/ad_seller/services/negotiation_service.py +++ b/src/ad_seller/services/negotiation_service.py @@ -38,6 +38,133 @@ def build_negotiation_engine(): return NegotiationEngine(pricing_engine, yield_opt) +async def _persist_negotiation(storage: Any, proposal_id: str, data: dict[str, Any]) -> None: + """Store a negotiation, and index it by the quote it concerns. + + Single write path for every stored negotiation so the quote index cannot + be forgotten at one of the three call sites. The index entry is written + only once the negotiation has concluded ``accepted`` and carries a + ``quote_id``: only an accepted negotiation changes a booking price, so + presence in the index means exactly "this quote has an agreed price" + (resolved at booking by ``deal_service.book_deal``). + + A pointer, not a duplicated record — the record is mutable because rounds + are appended, so two copies of it would drift on the first append. + """ + from ..storage.base import negotiation_by_quote_key + + await storage.set_negotiation(proposal_id, data) + + quote_id = data.get("quote_id") + if data.get("status") == "accepted" and quote_id: + await storage.set(negotiation_by_quote_key(quote_id), proposal_id) + + +async def resolve_negotiation_key( + proposal_id: Optional[str] = None, + negotiation_id: Optional[str] = None, + quote_id: Optional[str] = None, +) -> Optional[str]: + """Resolve a negotiation message's ids to the key its history is stored under. + + The seller keys negotiations by ``proposal_id``, and a quote-led open keys + off the ``quote_id``. Deriving that key as ``proposal_id or negotiation_id + or quote_id`` made it depend on which ids a given MESSAGE happened to + carry rather than on the negotiation: a round-one open keyed on one id + followed by a continuation carrying a different id resolved to two + different records, and the continuation silently restarted the + negotiation from round one. + + So every supplied id is tried against the store first and an existing + negotiation wins, whichever id the continuation leads with. A new + negotiation is minted under the first supplied id — the previous + or-chain order, unchanged — only when none of them resolves. Which id a + negotiation is stored under is deliberately NOT changed here; re-keying + storage on ``negotiation_id`` is the structural fix. + + Returns ``None`` only when no id was supplied at all. + """ + from ..storage.factory import get_storage + + candidates = [candidate for candidate in (proposal_id, negotiation_id, quote_id) if candidate] + if not candidates: + return None + + storage = await get_storage() + for candidate in candidates: + if await storage.get_negotiation(candidate): + return candidate + + return candidates[0] + + +def _quote_not_found_detail(quote_id: str) -> dict[str, str]: + """The exact detail shape ``quote_service.get_quote`` emits for a missing + quote — reused verbatim so an unowned quote is indistinguishable from a + nonexistent one (no existence oracle; house pattern, PR #98).""" + return {"error": "quote_not_found", "message": f"Quote '{quote_id}' not found."} + + +async def _verify_quote_binding(storage: Any, quote_id: str, buyer_context: Any) -> None: + """Gate binding a ``quote_id`` onto a negotiation on quote ownership. + + A negotiation message may name the quote its agreed price will apply to; + an accepted negotiation is then indexed by that quote and REWRITES the + booked ``final_cpm`` when the quote is booked. Without this check, buyer A + could accept its own negotiation while naming buyer B's quote id, and B's + booking would silently carry A's price. The caller must own the quote it + binds — the same ownership the booking route enforces before honouring a + quote (``routers/deals.py``). + + Rules, mirroring the booking route's, fail-closed on identity: + + - The quote must resolve at all (the live quote record or its history + entry). Binding an unresolvable id is refused loudly — it could never + be booked, so the binding is either a buyer bug or an attack, and + silently dropping it would hide both. + - Ownership comes from the quote-history record's ``buyer_id`` (written + at quote issue time from the verified pricing key). No record, or no + recorded buyer: legacy data, nothing to enforce — allowed, as at + booking. A ``"public"`` buyer asserted no identity at quote time, so + there is no ownership to enforce. + - The caller's pricing key counts only when it is key-derived + (``authentication_method == "api_key"``). This route's auth is + OPTIONAL, and an anonymous caller's identity fields are self-asserted + wire data — EP-5.2 floors the TIER but the pricing key would still + echo whatever ids the body claimed, so trusting it here would let an + anonymous caller impersonate the victim's identity and pass the very + check aimed at it. Every other caller is treated as public. + - A mismatch reads as 404 ``quote_not_found``, byte-identical to the + nonexistent case: a distinguishable "exists but not yours" would turn + this route into an existence oracle over every tenant's quote ids. + """ + from ..storage.quote_history import QuoteHistoryStore + + quote = await storage.get_quote(quote_id) + try: + history_record = await QuoteHistoryStore(storage).get_quote(quote_id) + except Exception: + history_record = None + if not isinstance(history_record, dict): + history_record = None + + if quote is None and history_record is None: + raise HTTPException(status_code=404, detail=_quote_not_found_detail(quote_id)) + + quoted_buyer_id = history_record.get("buyer_id") if history_record else None + if not quoted_buyer_id or quoted_buyer_id == "public": + return + + caller_key = ( + buyer_context.get_pricing_key() + if buyer_context is not None + and getattr(buyer_context, "authentication_method", None) == "api_key" + else "public" + ) + if quoted_buyer_id != caller_key: + raise HTTPException(status_code=404, detail=_quote_not_found_detail(quote_id)) + + async def submit_proposal( request: Any, buyer_context: Any, catalog: dict[str, Any] ) -> dict[str, Any]: @@ -119,7 +246,7 @@ async def submit_proposal( # round numbering) instead of silently starting over. flow_history = result.pop("_negotiation_history", None) if flow_history: - await storage.set_negotiation(proposal_id, flow_history) + await _persist_negotiation(storage, proposal_id, flow_history) quote_history = QuoteHistoryStore(storage) verification = await quote_history.verify_pricing( buyer_id=buyer_context.get_pricing_key(), @@ -190,11 +317,18 @@ async def counter_proposal( proposal_id: str, buyer_price: float, buyer_context: Any, + quote_id: Optional[str] = None, ) -> dict[str, Any]: """Submit a counter-offer in an ongoing negotiation. Loads or creates a NegotiationHistory, evaluates the buyer's offer, persists the updated history, and emits a NEGOTIATION_ROUND event. + + ``quote_id`` is the quote the buyer is negotiating, as supplied on the + inbound message. It is recorded on the history so an accepted negotiation + can be found again from the quote that gets booked. The first one seen + wins: a later round naming a different quote must not silently move which + quote the agreed price applies to. """ from ..events.helpers import emit_event from ..events.models import EventType @@ -224,6 +358,9 @@ async def counter_proposal( if not quote_data: raise HTTPException(status_code=404, detail="Proposal not found") product_id = (quote_data.get("product") or {}).get("product_id", "") + # Quote-led: the key IS the quote, so the quote id has arrived + # even when the message did not spell it out in its own field. + quote_id = quote_id or proposal_id product_data = await storage.get_product(product_id) if not product_data: @@ -262,12 +399,21 @@ async def counter_proposal( }, ) + # Retain the quote this negotiation concerns. Backfills a negotiation + # opened before the quote was named; first one seen wins. Binding is + # gated on ownership: this id is what routes an agreed price onto a + # booked deal, so a caller may only bind a quote issued to it — whether + # the id arrived in the message's quote_id field or as a quote-led key. + if quote_id and not history.quote_id: + await _verify_quote_binding(storage, quote_id, buyer_context) + history.quote_id = quote_id + # Evaluate buyer's offer round_result = neg_engine.evaluate_buyer_offer(history, buyer_price, buyer_context) history = neg_engine.record_round(history, round_result) - # Persist - await storage.set_negotiation(proposal_id, history.model_dump(mode="json")) + # Persist (and index by quote, if this round concluded accepted) + await _persist_negotiation(storage, proposal_id, history.model_dump(mode="json")) # Emit round event await emit_event( @@ -313,6 +459,8 @@ async def apply_terminal_action( proposal_id: str, action: str, buyer_price: Optional[float] = None, + quote_id: Optional[str] = None, + buyer_context: Optional[Any] = None, ) -> dict[str, Any]: """Record a buyer's terminal accept/reject on the stored negotiation. @@ -321,6 +469,14 @@ async def apply_terminal_action( the agreed state. The price engine is not run: accept strikes at the seller's last stated price; reject is a walk-away. + ``quote_id`` backfills the quote this negotiation concerns when the + accept is the first message to name it. On accept, this is the point the + quote index is written, which is what lets booking find the agreed price. + Binding a NEW quote id is gated on quote ownership against + ``buyer_context`` (see ``_verify_quote_binding``); a ``None`` context is + treated as an anonymous (public) caller, so it fails closed against + buyer-specific quotes rather than skipping the check. + No-ops (returning current status) when no stored negotiation exists or it is already terminal, preserving the read-only legacy behavior. """ @@ -332,6 +488,9 @@ async def apply_terminal_action( if data: history = NegotiationHistory(**data) if history.status == "active" and action in ("accept", "reject"): + if quote_id and not history.quote_id: + await _verify_quote_binding(storage, quote_id, buyer_context) + history.quote_id = quote_id last = history.rounds[-1] if history.rounds else None seller_price = last.seller_price if last else history.base_price history.rounds.append( @@ -349,7 +508,7 @@ async def apply_terminal_action( ) history.status = "accepted" if action == "accept" else "rejected" history.completed_at = datetime.utcnow() - await storage.set_negotiation(proposal_id, history.model_dump(mode="json")) + await _persist_negotiation(storage, proposal_id, history.model_dump(mode="json")) return await get_negotiation_status(proposal_id) @@ -379,6 +538,7 @@ async def get_negotiation_status(proposal_id: str) -> dict[str, Any]: return { "negotiation_id": history.negotiation_id, "proposal_id": history.proposal_id, + "quote_id": history.quote_id, "product_id": history.product_id, "buyer_tier": history.buyer_tier.value, "status": history.status, diff --git a/src/ad_seller/storage/base.py b/src/ad_seller/storage/base.py index c09167e3..cb5d5c50 100644 --- a/src/ad_seller/storage/base.py +++ b/src/ad_seller/storage/base.py @@ -7,6 +7,21 @@ from typing import Any, Optional +def negotiation_by_quote_key(quote_id: str) -> str: + """Key of the pointer from a quote to the negotiation about it. + + The value stored under this key is the id the negotiation record is keyed + by (see :meth:`StorageBackend.get_negotiation`), so booking can resolve a + quote to the price agreed for it. A pointer rather than a second copy of + the record: the record is mutable because rounds are appended to it, so + two copies would drift on the first append. + + Written only when a negotiation concludes ``accepted``, so presence in + this index means exactly "this quote has an agreed price". + """ + return f"negotiation_by_quote:{quote_id}" + + class StorageBackend(ABC): """Abstract base class for storage backends.""" diff --git a/tests/unit/test_audience_plan_validation.py b/tests/unit/test_audience_plan_validation.py index 3e40d1b6..1c94cd28 100644 --- a/tests/unit/test_audience_plan_validation.py +++ b/tests/unit/test_audience_plan_validation.py @@ -455,6 +455,10 @@ def http_client(): ), "get_deal": staticmethod(lambda did: _async_return(store.get(f"deal:{did}"))), "set_deal": staticmethod(lambda did, data: _async_set(store, f"deal:{did}", data)), + # Generic KV path: booking resolves a quote's negotiation through + # the negotiation_by_quote pointer before reading the record. + "get": staticmethod(lambda key: _async_return(store.get(key))), + "set": staticmethod(lambda key, value, ttl=None: _async_set(store, key, value)), # Booking consults negotiation state; none stored here. "get_negotiation": staticmethod( lambda pid: _async_return(store.get(f"negotiation:{pid}")) diff --git a/tests/unit/test_negotiation_quote_correlation.py b/tests/unit/test_negotiation_quote_correlation.py new file mode 100644 index 00000000..41e7dd88 --- /dev/null +++ b/tests/unit/test_negotiation_quote_correlation.py @@ -0,0 +1,959 @@ +# Author: Green Mountain Systems AI Inc. +# Donated to IAB Tech Lab + +"""A negotiated price must actually book. + +The defect: ``deal_service.book_deal`` honored an accepted negotiation by +calling ``storage.get_negotiation(request.quote_id)``. Negotiations are keyed +by the id the opening message led with, so that lookup only ever hit for a +QUOTE-LED negotiation. A proposal-led buyer -- which is what ours is -- has +its negotiation stored under a ``prop-`` id, so the lookup missed every time: +the booked deal carried the seller's standard price while the log said +"Negotiation succeeded". + +Three things are pinned here: + +1. ``NegotiationHistory.quote_id`` -- the quote id arrives on + ``NegotiationMessage`` and used to be dropped, so nothing could correlate + a negotiation to the quote a buyer books. +2. The quote index -- an accepted negotiation writes + ``negotiation_by_quote:{quote_id}`` naming the key its record lives under, + and booking resolves through it whichever id the buyer led with. A + negotiation that concluded accepted but carries no price refuses to book + rather than silently falling back to the quoted price. +3. Key resolution (fault A2) -- the storage key is resolved against the store + rather than read off whichever ids a given message happened to carry, so a + continuation leading with a different id continues the SAME negotiation + instead of silently restarting it. +""" + +import os +import sys +from datetime import datetime, timedelta +from types import ModuleType +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +# Settings require an Anthropic key even for key-less unit runs; same idiom +# as test_negotiation_cold_start.py (no LLM call is ever made here). +os.environ.setdefault("ANTHROPIC_API_KEY", "test-key-for-unit-tests") + +# Stub execution_activation_flow (cancel-scope leak on ad-server +# connection failure, unresolved -- issue #60 part 2). +_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 fastapi import HTTPException # noqa: E402 +from httpx import ASGITransport # noqa: E402 + +from ad_seller.interfaces.api.main import _get_optional_api_key_record, app # noqa: E402 +from ad_seller.models.buyer_identity import BuyerIdentity # noqa: E402 +from ad_seller.services import deal_service, negotiation_service # noqa: E402 + +pytestmark = pytest.mark.asyncio + +QUOTE_ID = "qt-corr123456" +PROPOSAL_ID = "prop-corr01" +PRODUCT_ID = "ctv-premium-sports" +QUOTED_CPM = 28.26 + + +# ============================================================================= +# Helpers +# ============================================================================= + + +def _available_quote(quote_id=QUOTE_ID, product_id=PRODUCT_ID, final_cpm=QUOTED_CPM): + return { + "quote_id": quote_id, + "status": "available", + "deal_type": "PD", + "product": { + "product_id": product_id, + "name": "Premium CTV - Sports", + "inventory_type": "ctv", + }, + "pricing": { + "base_cpm": 35.0, + "tier_discount_pct": 15.0, + "volume_discount_pct": 5.0, + "final_cpm": final_cpm, + "currency": "USD", + "pricing_model": "cpm", + "rationale": "Base price: $35.00 CPM | Advertiser tier: -15%", + }, + "terms": { + "impressions": 5000000, + "flight_start": "2026-04-01", + "flight_end": "2026-04-30", + "guaranteed": False, + }, + "buyer_tier": "advertiser", + "expires_at": (datetime.utcnow() + timedelta(hours=23)).isoformat() + "Z", + "created_at": datetime.utcnow().isoformat() + "Z", + } + + +def _stored_product(product_id=PRODUCT_ID): + return {"product_id": product_id, "base_cpm": 35.0, "floor_cpm": 20.0} + + +def _stored_proposal(proposal_id=PROPOSAL_ID, product_id=PRODUCT_ID): + return { + "proposal_id": proposal_id, + "product_id": product_id, + "deal_type": "preferred_deal", + "price": 24.0, + "impressions": 5_000_000, + "start_date": "2026-04-01", + "end_date": "2026-04-30", + "buyer_id": "buyer-1", + } + + +def _accepted_negotiation( + proposal_id=PROPOSAL_ID, + quote_id=QUOTE_ID, + product_id=PRODUCT_ID, + agreed=24.0, + rounds=None, +): + """A negotiation record that concluded ``accepted``. + + ``rounds=[]`` models the corrupt state step 3 guards: accepted with no + agreed price anywhere on the record. + """ + if rounds is None: + rounds = [ + { + "round_number": 1, + "buyer_price": 22.0, + "seller_price": 26.0, + "action": "counter", + "rationale": "Counter at $26.00", + "timestamp": datetime.utcnow().isoformat(), + }, + { + "round_number": 2, + "buyer_price": agreed, + "seller_price": agreed, + "action": "accept", + "rationale": f"Buyer accepted at ${agreed:.2f} CPM.", + "timestamp": datetime.utcnow().isoformat(), + }, + ] + return { + "negotiation_id": "neg-agreed01", + "proposal_id": proposal_id, + "quote_id": quote_id, + "product_id": product_id, + "buyer_tier": "advertiser", + "strategy": "premium", + "limits": { + "max_rounds": 6, + "per_round_concession_cap": 0.06, + "total_concession_cap": 0.20, + "gap_split_buyer_share": 0.65, + }, + "base_price": 29.75, + "floor_price": 20.0, + "rounds": rounds, + "status": "accepted", + "started_at": datetime.utcnow().isoformat(), + "completed_at": datetime.utcnow().isoformat(), + "package_id": None, + } + + +def _buyer_context(buyer_tier="advertiser"): + from ad_seller.interfaces.api.deps import _build_buyer_context + + return _build_buyer_context( + buyer_tier=buyer_tier, agency_id="agency-test-1", advertiser_id="adv-test-1" + ) + + +def _booking_request(quote_id=QUOTE_ID): + request = MagicMock() + request.quote_id = quote_id + request.audience_plan = None + return request + + +@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.delete = AsyncMock(side_effect=lambda k: store.pop(k, None)) + 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.get_proposal = AsyncMock(side_effect=lambda pid: store.get(f"proposal:{pid}")) + storage.set_proposal = AsyncMock( + side_effect=lambda pid, data: store.__setitem__(f"proposal:{pid}", data) + ) + storage.get_product = AsyncMock(side_effect=lambda pid: store.get(f"product:{pid}")) + storage.set_product = AsyncMock( + side_effect=lambda pid, data: store.__setitem__(f"product:{pid}", data) + ) + storage.get_negotiation = AsyncMock(side_effect=lambda pid: store.get(f"negotiation:{pid}")) + storage.set_negotiation = AsyncMock( + side_effect=lambda pid, data: store.__setitem__(f"negotiation:{pid}", data) + ) + storage._store = store + return storage + + +@pytest.fixture +def client(): + """An authenticated ADVERTISER-tier buyer. + + ``BuyerContext.effective_tier`` needs agency_id AND advertiser_id to + reach ADVERTISER (progressive revelation), which is what the fixture + quote's ``buyer_tier`` requires of a caller booking it. + """ + app.dependency_overrides[_get_optional_api_key_record] = lambda: MagicMock( + identity=BuyerIdentity( + seat_id="seat-test-1", agency_id="agency-test-1", advertiser_id="adv-test-1" + ) + ) + transport = ASGITransport(app=app) + c = httpx.AsyncClient(transport=transport, base_url="http://test") + yield c + app.dependency_overrides.clear() + + +def _seed_negotiable(mock_storage): + """Quote + product + proposal, the state a proposal-led buyer negotiates from.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"product:{PRODUCT_ID}"] = _stored_product() + mock_storage._store[f"proposal:{PROPOSAL_ID}"] = _stored_proposal() + + +# ============================================================================= +# Step 1: the quote id arrives on the wire and must be retained +# ============================================================================= + + +class TestQuoteIdRetained: + async def test_open_records_the_quote_id_from_the_message(self, mock_storage): + """A negotiation opened for a quote persists that quote id.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, + buyer_price=24.0, + buyer_context=_buyer_context(), + quote_id=QUOTE_ID, + ) + status = await negotiation_service.get_negotiation_status(PROPOSAL_ID) + + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["quote_id"] == QUOTE_ID + # Readable back out, not merely written. + assert status["quote_id"] == QUOTE_ID + + async def test_quote_id_survives_open_counter_accept(self, mock_storage): + """The id survives a full open -> counter -> accept cycle.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=_buyer_context(), quote_id=QUOTE_ID + ) + # A later round need not repeat the quote id. + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=23.5, buyer_context=_buyer_context() + ) + status = await negotiation_service.apply_terminal_action(PROPOSAL_ID, "accept") + + assert status["quote_id"] == QUOTE_ID + assert status["status"] == "accepted" + + async def test_proposal_only_negotiation_stores_none(self, mock_storage): + """A pure proposal-led negotiation still works, with quote_id None.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + result = await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=24.0, buyer_context=_buyer_context() + ) + + assert result["round_number"] == 1 + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["quote_id"] is None + + async def test_quote_id_is_not_overwritten_by_a_later_round(self, mock_storage): + """A later round naming a DIFFERENT quote must not silently move which + quote the agreed price applies to.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=_buyer_context(), quote_id=QUOTE_ID + ) + await negotiation_service.counter_proposal( + PROPOSAL_ID, + buyer_price=23.0, + buyer_context=_buyer_context(), + quote_id="qt-someotherquote", + ) + + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["quote_id"] == QUOTE_ID + + +# ============================================================================= +# Step 2: an accepted negotiation is findable from the quote it concerns +# ============================================================================= + + +class TestQuoteIndex: + async def test_accept_writes_the_quote_pointer(self, mock_storage): + """Accepting indexes the negotiation by its quote, pointing at the key + the record is actually stored under.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=_buyer_context(), quote_id=QUOTE_ID + ) + await negotiation_service.apply_terminal_action(PROPOSAL_ID, "accept") + + assert mock_storage._store[f"negotiation_by_quote:{QUOTE_ID}"] == PROPOSAL_ID + + async def test_active_negotiation_writes_no_pointer(self, mock_storage): + """Only an ACCEPTED negotiation changes a booking price, so only an + accepted one appears in the index.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=_buyer_context(), quote_id=QUOTE_ID + ) + + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["status"] == "active" + assert f"negotiation_by_quote:{QUOTE_ID}" not in mock_storage._store + + async def test_booking_resolves_a_proposal_led_negotiation(self, mock_storage): + """THE bug, at the service boundary: the negotiation is stored under a + ``prop-`` id, so booking can only find it through the index.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{PROPOSAL_ID}"] = _accepted_negotiation(agreed=24.0) + mock_storage._store[f"negotiation_by_quote:{QUOTE_ID}"] = PROPOSAL_ID + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + deal = await deal_service.book_deal(_booking_request()) + + assert deal["pricing"]["final_cpm"] == 24.0 + assert deal["openrtb_params"]["bidfloor"] == 24.0 + assert "Negotiated to $24.00 CPM" in deal["pricing"]["rationale"] + + async def test_booking_ignores_a_pointer_to_an_active_negotiation(self, mock_storage): + """A resolvable but still-active negotiation must not change the price.""" + neg = _accepted_negotiation() + neg["status"] = "active" + neg["rounds"] = neg["rounds"][:1] + neg["completed_at"] = None + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{PROPOSAL_ID}"] = neg + mock_storage._store[f"negotiation_by_quote:{QUOTE_ID}"] = PROPOSAL_ID + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + deal = await deal_service.book_deal(_booking_request()) + + assert deal["pricing"]["final_cpm"] == QUOTED_CPM + assert "Negotiated to" not in deal["pricing"]["rationale"] + + async def test_quote_led_negotiation_still_honored_without_a_pointer(self, mock_storage): + """A quote-led negotiation is stored under the quote id itself. That + path worked before the index existed and must keep working, including + for records written before it.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{QUOTE_ID}"] = _accepted_negotiation( + proposal_id=QUOTE_ID, agreed=25.5 + ) + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + deal = await deal_service.book_deal(_booking_request()) + + assert deal["pricing"]["final_cpm"] == 25.5 + + async def test_negotiated_price_books_end_to_end(self, client, mock_storage): + """The end-to-end case the defect broke: negotiate a price down over + the wire, accept, then book that quote -- the deal must carry the + agreed price and say so in its rationale.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + async with client as c: + # Proposal-led, exactly what our buyer sends: proposal_id as + # the negotiation key, quote_id naming the quote at stake. + counter = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-corr-counter", + "action": "counter", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + "buyer_price": {"amount_micros": 22_000_000, "currency": "USD"}, + }, + ) + assert counter.status_code == 200, counter.text + + accept = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-corr-accept", + "action": "accept", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + }, + ) + assert accept.status_code == 200, accept.text + assert accept.json()["status"] == "accepted" + + booked = await c.post( + "/api/v1/deals", + json={ + "idempotency_key": "idem-corr-book", + "quote_id": QUOTE_ID, + "buyer_identity": { + "seat_id": "seat-test-1", + "agency_id": "agency-test-1", + "advertiser_id": "adv-test-1", + }, + }, + ) + + assert booked.status_code == 200, booked.text + stored_negotiation = mock_storage._store[f"negotiation:{PROPOSAL_ID}"] + agreed = stored_negotiation["rounds"][-1]["seller_price"] + # The agreed price must actually differ from the quoted one, or this + # test could pass without the fix doing anything. + assert agreed != QUOTED_CPM + + pricing = booked.json()["deal"]["pricing"] + assert pricing["final_cpm"]["amount_micros"] == round(agreed * 1_000_000) + assert f"Negotiated to ${agreed:.2f} CPM" in pricing["rationale"] + + async def test_quote_with_no_negotiation_books_at_the_quoted_price(self, client, mock_storage): + """The negative case: no negotiation, no change. A quote books at the + price it was quoted at.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + async with client as c: + booked = await c.post( + "/api/v1/deals", + json={ + "idempotency_key": "idem-corr-plain", + "quote_id": QUOTE_ID, + "buyer_identity": { + "seat_id": "seat-test-1", + "agency_id": "agency-test-1", + "advertiser_id": "adv-test-1", + }, + }, + ) + + assert booked.status_code == 200, booked.text + pricing = booked.json()["deal"]["pricing"] + assert pricing["final_cpm"]["amount_micros"] == round(QUOTED_CPM * 1_000_000) + assert "Negotiated to" not in pricing["rationale"] + + +# ============================================================================= +# Step 2b (fault A2): the storage key is resolved, not read off the message +# ============================================================================= + + +class TestNegotiationKeyResolution: + async def test_continuation_with_a_different_id_continues_the_same_negotiation( + self, client, mock_storage + ): + """Round one opened quote-led, so the record is keyed by the quote. A + continuation leading with the proposal_id must resolve to the SAME + negotiation. + + Reading the key off the message (``proposal_id or negotiation_id or + quote_id``) picked the proposal id here, found no record under it, + found the stored PROPOSAL under it, and opened a second negotiation at + round one -- a silent restart that discards every concession already + made. + """ + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + async with client as c: + first = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-cont-1", + "action": "counter", + "quote_id": QUOTE_ID, + "buyer_price": {"amount_micros": 22_000_000, "currency": "USD"}, + }, + ) + assert first.status_code == 200, first.text + negotiation_id = first.json()["negotiation_id"] + + second = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-cont-2", + "action": "counter", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + "buyer_price": {"amount_micros": 23_000_000, "currency": "USD"}, + }, + ) + + assert second.status_code == 200, second.text + body = second.json() + assert body["negotiation_id"] == negotiation_id, "continuation started a new negotiation" + assert body["round"]["round_number"] == 2, "continuation restarted the round numbering" + # One record, under the id round one opened it with. No second record + # minted under the proposal id the continuation led with. + assert len(mock_storage._store[f"negotiation:{QUOTE_ID}"]["rounds"]) == 2 + assert f"negotiation:{PROPOSAL_ID}" not in mock_storage._store + + async def test_resolution_prefers_an_existing_record_over_message_order(self, mock_storage): + """Unit view of the same thing: an id that resolves to a stored + negotiation wins over an id that merely came first on the message.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{QUOTE_ID}"] = _accepted_negotiation( + proposal_id=QUOTE_ID, rounds=[] + ) + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + resolved = await negotiation_service.resolve_negotiation_key( + proposal_id=PROPOSAL_ID, negotiation_id="neg-agreed01", quote_id=QUOTE_ID + ) + + assert resolved == QUOTE_ID + + async def test_a_negotiation_id_alone_is_not_a_storage_key(self, mock_storage): + """Known limitation, deliberately left to the structural fix: no + negotiation is stored under its own ``neg-`` id, so a message carrying + ONLY negotiation_id has nothing to resolve against. + + This is not a silent restart -- there is no proposal or quote under a + ``neg-`` id either, so the round is refused with a 404 rather than + quietly beginning again. Re-keying storage on negotiation_id is what + makes this case resolvable. + """ + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{QUOTE_ID}"] = _accepted_negotiation( + proposal_id=QUOTE_ID, rounds=[] + ) + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + resolved = await negotiation_service.resolve_negotiation_key( + negotiation_id="neg-agreed01" + ) + assert resolved == "neg-agreed01" + + with pytest.raises(HTTPException) as exc: + await negotiation_service.counter_proposal( + resolved, buyer_price=23.0, buyer_context=_buyer_context() + ) + + assert exc.value.status_code == 404 + + async def test_unresolvable_ids_mint_under_the_first_supplied_id(self, mock_storage): + """Nothing stored under any supplied id: the key is the first one + supplied, the previous or-chain order, so opens are unchanged.""" + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + resolved = await negotiation_service.resolve_negotiation_key( + proposal_id=PROPOSAL_ID, negotiation_id="neg-fresh01", quote_id=QUOTE_ID + ) + + assert resolved == PROPOSAL_ID + + async def test_no_ids_at_all_resolves_to_none(self, mock_storage): + """The router's 400 path: no negotiation context supplied.""" + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + assert await negotiation_service.resolve_negotiation_key() is None + + +# ============================================================================= +# Step 2c: only the quote's own buyer may bind it to a negotiation +# ============================================================================= + + +VICTIM_PRICING_KEY = "advertiser:adv-victim-9" + + +def _quote_history_record(quote_id=QUOTE_ID, buyer_id=VICTIM_PRICING_KEY): + """The record the quotes route writes at issue time (QuoteHistoryStore).""" + return { + "quote_id": quote_id, + "buyer_id": buyer_id, + "product_id": PRODUCT_ID, + "quoted_cpm": QUOTED_CPM, + "quoted_at": datetime.utcnow().isoformat(), + "expires_at": (datetime.utcnow() + timedelta(hours=23)).isoformat(), + } + + +def _api_key_buyer_context(seat="seat-test-1", agency="agency-test-1", adv="adv-test-1"): + """A KEY-derived context — the only identity quote ownership trusts.""" + from ad_seller.interfaces.api.deps import _build_buyer_context + + record = MagicMock() + record.identity = BuyerIdentity(seat_id=seat, agency_id=agency, advertiser_id=adv) + return _build_buyer_context(api_key_record=record) + + +class TestQuoteBindingOwnership: + """The quote a negotiation binds is the quote whose booked price it will + REWRITE, so binding is gated on ownership. An unowned quote reads as 404 + ``quote_not_found``, byte-identical to a nonexistent one (no existence + oracle — house pattern, PR #98).""" + + async def test_cross_tenant_accept_cannot_rewrite_the_victims_booked_price( + self, client, mock_storage + ): + """THE attack, end to end: buyer A negotiates its own proposal down, + accepts while naming buyer B's quote id, then B books its quote. The + accept must be refused, no quote pointer written, and B's deal must + book at B's quoted price -- not A's negotiated one.""" + _seed_negotiable(mock_storage) + # The quote belongs to the VICTIM, not to the caller the client + # fixture authenticates (advertiser:adv-test-1). + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record() + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + async with client as c: + # A negotiates its OWN proposal -- no quote named, this is fine. + counter = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-xt-counter", + "action": "counter", + "proposal_id": PROPOSAL_ID, + "buyer_price": {"amount_micros": 22_000_000, "currency": "USD"}, + }, + ) + assert counter.status_code == 200, counter.text + + # The accept names B's quote: refused, and as NOT-FOUND, so a + # probe cannot distinguish "exists but not yours" from + # "does not exist" (no existence oracle). + accept = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-xt-accept", + "action": "accept", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + }, + ) + assert accept.status_code == 404, accept.text + assert accept.json()["detail"]["error"] == "quote_not_found" + + # Nothing bound, nothing indexed, negotiation NOT concluded. + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["quote_id"] is None + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["status"] == "active" + assert f"negotiation_by_quote:{QUOTE_ID}" not in mock_storage._store + + # B books its quote: full price, no negotiation suffix. + app.dependency_overrides[_get_optional_api_key_record] = lambda: MagicMock( + identity=BuyerIdentity( + seat_id="seat-victim-9", + agency_id="agency-victim-9", + advertiser_id="adv-victim-9", + ) + ) + victim = httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test") + async with victim as v: + booked = await v.post( + "/api/v1/deals", + json={ + "idempotency_key": "idem-xt-book", + "quote_id": QUOTE_ID, + "buyer_identity": { + "seat_id": "seat-victim-9", + "agency_id": "agency-victim-9", + "advertiser_id": "adv-victim-9", + }, + }, + ) + + assert booked.status_code == 200, booked.text + pricing = booked.json()["deal"]["pricing"] + assert pricing["final_cpm"]["amount_micros"] == round(QUOTED_CPM * 1_000_000) + assert "Negotiated to" not in pricing["rationale"] + + async def test_own_quote_still_binds_and_books_the_negotiated_price(self, client, mock_storage): + """The legitimate flow with an ownership record present: the caller + negotiates naming ITS OWN quote, accepts, books -- the deal carries + the agreed price.""" + _seed_negotiable(mock_storage) + # The quote was issued to the very buyer the client authenticates. + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record( + buyer_id="advertiser:adv-test-1" + ) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + async with client as c: + counter = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-own-counter", + "action": "counter", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + "buyer_price": {"amount_micros": 22_000_000, "currency": "USD"}, + }, + ) + assert counter.status_code == 200, counter.text + + accept = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-own-accept", + "action": "accept", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + }, + ) + assert accept.status_code == 200, accept.text + assert accept.json()["status"] == "accepted" + + booked = await c.post( + "/api/v1/deals", + json={ + "idempotency_key": "idem-own-book", + "quote_id": QUOTE_ID, + "buyer_identity": { + "seat_id": "seat-test-1", + "agency_id": "agency-test-1", + "advertiser_id": "adv-test-1", + }, + }, + ) + + assert booked.status_code == 200, booked.text + agreed = mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["rounds"][-1]["seller_price"] + assert agreed != QUOTED_CPM + pricing = booked.json()["deal"]["pricing"] + assert pricing["final_cpm"]["amount_micros"] == round(agreed * 1_000_000) + assert f"Negotiated to ${agreed:.2f} CPM" in pricing["rationale"] + + async def test_binding_an_unresolvable_quote_id_is_refused(self, client, mock_storage): + """A quote_id resolving to neither a quote nor a history record could + never be booked, so binding it is a buyer bug or an attack -- refused + loudly rather than silently dropped.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + async with client as c: + counter = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-ghost-counter", + "action": "counter", + "proposal_id": PROPOSAL_ID, + "quote_id": "qt-nosuchquote", + "buyer_price": {"amount_micros": 22_000_000, "currency": "USD"}, + }, + ) + + assert counter.status_code == 404, counter.text + assert counter.json()["detail"]["error"] == "quote_not_found" + # The refused round left nothing behind. + assert f"negotiation:{PROPOSAL_ID}" not in mock_storage._store + + async def test_quote_led_open_of_a_foreign_quote_is_refused(self, mock_storage): + """The sibling vector: the same foreign quote id sent as the + negotiation KEY (quote-led open) rather than in the quote_id field + binds through the same site and is refused the same way.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"product:{PRODUCT_ID}"] = _stored_product() + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record() + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + with pytest.raises(HTTPException) as exc: + await negotiation_service.counter_proposal( + QUOTE_ID, buyer_price=22.0, buyer_context=_api_key_buyer_context() + ) + + assert exc.value.status_code == 404 + assert exc.value.detail["error"] == "quote_not_found" + assert f"negotiation:{QUOTE_ID}" not in mock_storage._store + + async def test_anonymous_self_asserted_identity_cannot_bind_a_buyer_quote(self, mock_storage): + """This route's auth is OPTIONAL and an anonymous caller's identity + fields are self-asserted wire data. Asserting the victim's ids must + not pass the ownership check aimed at exactly that caller: only a + KEY-derived identity counts, everyone else is public.""" + _seed_negotiable(mock_storage) + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record() + + from ad_seller.interfaces.api.deps import _build_buyer_context + + impostor = _build_buyer_context( + buyer_tier="advertiser", agency_id="agency-victim-9", advertiser_id="adv-victim-9" + ) + assert impostor.get_pricing_key() == VICTIM_PRICING_KEY # the claim... + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + with pytest.raises(HTTPException) as exc: # ...does not bind. + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=impostor, quote_id=QUOTE_ID + ) + + assert exc.value.status_code == 404 + assert exc.value.detail["error"] == "quote_not_found" + + async def test_a_public_quote_stays_negotiable_without_ownership(self, mock_storage): + """A quote issued to ``buyer_id="public"`` asserted no identity at + quote time -- nothing buyer-specific was priced, so there is no + ownership to enforce (same rule as booking).""" + _seed_negotiable(mock_storage) + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record(buyer_id="public") + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + result = await negotiation_service.counter_proposal( + PROPOSAL_ID, + buyer_price=22.0, + buyer_context=_buyer_context(), + quote_id=QUOTE_ID, + ) + + assert result["round_number"] == 1 + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["quote_id"] == QUOTE_ID + + async def test_terminal_bind_without_a_context_fails_closed(self, mock_storage): + """``apply_terminal_action`` called with a quote to bind but no buyer + context treats the caller as public: a buyer-specific quote does not + bind, rather than the check being skipped.""" + _seed_negotiable(mock_storage) + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record() + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=_buyer_context() + ) + with pytest.raises(HTTPException) as exc: + await negotiation_service.apply_terminal_action( + PROPOSAL_ID, "accept", quote_id=QUOTE_ID + ) + + assert exc.value.status_code == 404 + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["status"] == "active" + assert f"negotiation_by_quote:{QUOTE_ID}" not in mock_storage._store + + +# ============================================================================= +# Step 3: an accepted negotiation with no price must not book at list price +# ============================================================================= + + +class TestAcceptedWithoutPriceFailsLoudly: + async def test_accepted_negotiation_with_no_rounds_refuses_to_book(self, mock_storage): + """A negotiation that concluded accepted with no agreed price is a + corrupt record. Booking it at the un-negotiated quoted price is the + worst available outcome, so booking fails instead.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{PROPOSAL_ID}"] = _accepted_negotiation(rounds=[]) + mock_storage._store[f"negotiation_by_quote:{QUOTE_ID}"] = PROPOSAL_ID + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + with pytest.raises(HTTPException) as exc: + await deal_service.book_deal(_booking_request()) + + assert exc.value.status_code == 500 + assert exc.value.detail["error"] == "negotiation_price_unresolved" + # And nothing was booked at the list price behind the buyer's back. + assert not [k for k in mock_storage._store if k.startswith("deal:")] + assert mock_storage._store[f"quote:{QUOTE_ID}"]["status"] == "available" + + async def test_accepted_negotiation_with_a_null_seller_price_refuses_to_book( + self, mock_storage + ): + """Same guard, reached via a final round whose price is missing rather + than via an empty rounds list.""" + neg = _accepted_negotiation( + rounds=[ + { + "round_number": 1, + "buyer_price": 22.0, + "seller_price": None, + "action": "accept", + "rationale": "corrupt", + "timestamp": datetime.utcnow().isoformat(), + } + ] + ) + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{PROPOSAL_ID}"] = neg + mock_storage._store[f"negotiation_by_quote:{QUOTE_ID}"] = PROPOSAL_ID + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + with pytest.raises(HTTPException) as exc: + await deal_service.book_deal(_booking_request()) + + assert exc.value.status_code == 500 + assert exc.value.detail["error"] == "negotiation_price_unresolved"