fix: send quote_id on negotiation messages and assert the booked price equals the agreed price

src/ad_buyer/clients/contract_mappers.py

@@ -481,8 +481,11 @@ def to_wire_negotiation_message(
     the same model. Money-mutating (FD-12): carries a required idempotency_key.
 
     Action rules enforced by the shared model: 'counter'/'final_offer' require
-    ``buyer_price``; 'reject' must omit it; 'accept' may echo it. Exactly one of
-    negotiation_id/proposal_id/quote_id must be present.
+    ``buyer_price``; 'reject' must omit it; 'accept' may echo it. AT LEAST one
+    of negotiation_id/proposal_id/quote_id must be present -- the shared
+    validator rejects only the all-None case, so they are NOT mutually
+    exclusive and a proposal-led negotiation can carry the ``quote_id`` it
+    concerns alongside ``proposal_id``/``negotiation_id``.
     """
     return WireNegotiationMessage(
         idempotency_key=idempotency_key or uuid4().hex,

src/ad_buyer/negotiation/client.py

@@ -156,6 +156,7 @@ async def start_negotiation(
         initial_price: float,
         strategy: NegotiationStrategy,
         negotiation_enabled: bool = True,
+        quote_id: str | None = None,
     ) -> NegotiationSession:
         """Start a negotiation by sending the first counter-offer.
 
@@ -169,6 +170,11 @@ async def start_negotiation(
             strategy: The negotiation strategy (stored for reference).
             negotiation_enabled: Whether the seller's package allows
                 negotiation.  When False, raises ValueError.
+            quote_id: The quote this negotiation concerns, when the buyer
+                holds one. Sent on this and every subsequent message of the
+                negotiation so the seller can correlate the agreed price to
+                the quote that will be booked. Stored on the returned
+                session.
 
         Returns:
             A NegotiationSession tracking the negotiation state.
@@ -185,10 +191,15 @@ async def start_negotiation(
         url = f"{seller_url}{_NEGOTIATION_MESSAGES_PATH}"
         # Opening move: a shared NegotiationMessage with action="counter" and
         # negotiation_id=None (the seller mints the negotiation on open). Money
-        # crosses as the shared Money via buyer_price.
+        # crosses as the shared Money via buyer_price. quote_id rides along
+        # when known: the shared model requires AT LEAST one of
+        # negotiation_id/proposal_id/quote_id, so sending the quote id
+        # alongside proposal_id is valid and leaves the seller's
+        # proposal-first key derivation untouched.
         payload = to_wire_negotiation_message(
             action="counter",
             proposal_id=proposal_id,
+            quote_id=quote_id,
             buyer_price=initial_price,
         ).model_dump(mode="json", exclude_none=True)
 
@@ -203,6 +214,7 @@ async def start_negotiation(
             negotiation_id=data.get("negotiation_id", f"neg-{proposal_id}"),
             current_seller_price=data.get("seller_price", data.get("current_price", 0.0)),
             our_last_offer=initial_price,
+            quote_id=quote_id,
             rounds=[],
         )
 
@@ -244,6 +256,7 @@ async def counter_offer(
             action="counter",
             proposal_id=session.proposal_id,
             negotiation_id=session.negotiation_id,
+            quote_id=session.quote_id,
             buyer_price=price,
         ).model_dump(mode="json", exclude_none=True)
 
@@ -293,6 +306,7 @@ async def accept(self, session: NegotiationSession) -> dict[str, Any]:
             action="accept",
             proposal_id=session.proposal_id,
             negotiation_id=session.negotiation_id,
+            quote_id=session.quote_id,
             buyer_price=session.current_seller_price,
         ).model_dump(mode="json", exclude_none=True)
 
@@ -322,6 +336,7 @@ async def decline(self, session: NegotiationSession) -> None:
             action="reject",
             proposal_id=session.proposal_id,
             negotiation_id=session.negotiation_id,
+            quote_id=session.quote_id,
         ).model_dump(mode="json", exclude_none=True)
 
         async with httpx.AsyncClient(timeout=self._timeout) as client:
@@ -336,6 +351,7 @@ async def auto_negotiate(
         proposal_id: str,
         strategy: NegotiationStrategy,
         negotiation_enabled: bool = True,
+        quote_id: str | None = None,
     ) -> NegotiationResult:
         """Run a full negotiation loop automatically using the strategy.
 
@@ -353,6 +369,8 @@ async def auto_negotiate(
             strategy: The negotiation strategy to use.
             negotiation_enabled: Whether the seller's package allows
                 negotiation.  When False, returns DECLINED immediately.
+            quote_id: The quote this negotiation concerns, when the buyer
+                holds one; sent on every outbound message of the loop.
 
         Returns:
             NegotiationResult with the outcome and history.
@@ -385,6 +403,7 @@ async def auto_negotiate(
             proposal_id=proposal_id,
             initial_price=initial_price,
             strategy=strategy,
+            quote_id=quote_id,
         )
 
         # Check if seller already accepted or if we should accept their counter

src/ad_buyer/negotiation/models.py

@@ -47,6 +47,12 @@ class NegotiationSession(BaseModel):
     negotiation_id: str
     current_seller_price: float
     our_last_offer: float | None = None
+    #: The quote this negotiation concerns, when the buyer holds one. Sent on
+    #: every outbound NegotiationMessage as ``quote_id`` so the seller can
+    #: correlate the agreed price back to the quote that gets booked -- the
+    #: only correlation route the booking contract offers (DealBookingRequest
+    #: carries a quote_id and no negotiation_id).
+    quote_id: str | None = None
     rounds: list[NegotiationRound] = Field(default_factory=list)
     started_at: datetime = Field(default_factory=utc_now)
 

src/ad_buyer/orchestration/multi_seller.py

@@ -43,6 +43,7 @@
     CapabilityClient,
     CapabilityDiscoveryResult,
 )
+from ..clients.contract_mappers import _money_from_float
 from ..clients.deals_client import DealsClientError
 from ..clients.sgp_client import SGPClient
 from ..events.models import Event, EventType
@@ -409,8 +410,8 @@ class OrchestrationResult:
             original quote, opening price, per-round history, outcome
             (``accepted`` / ``walked_away`` / ``rejected`` /
             ``unavailable`` / ``requote_failed`` /
-            ``requote_above_ceiling``), agreed CPM, and the fresh quote
-            id when re-quoted.
+            ``requote_above_ceiling`` / ``requote_price_mismatch``),
+            agreed CPM, and the fresh quote id when re-quoted.
     """
 
     discovered_sellers: list[AgentCard]
@@ -1337,9 +1338,14 @@ async def _negotiate_quote(
            quote with ``target_cpm=P``. The seller's quote engine honors
            an acceptable target as the quote's ``final_cpm``, and booking
            strikes the deal at the fresh quote's price -- the only
-           contract-clean way to book a negotiated price today. The fresh
-           quote is verified <= ``max_cpm`` before it enters the ranked
-           list; the budget/spend-ceiling gates remain the final arbiter.
+           contract-clean way to book a negotiated price today. Before the
+           fresh quote enters the ranked list -- i.e. before anything can
+           book it -- it must clear TWO independent guards: it must be
+           <= ``max_cpm`` (never book above the ceiling), and its
+           ``final_cpm`` must EQUAL P (never book a price we did not agree
+           to), compared as integer micros -- the unit money is defined in
+           on the wire -- with no tolerance window. The budget/spend-ceiling
+           gates remain the final arbiter.
 
         Returns:
             (negotiated_result, record). ``negotiated_result`` is a
@@ -1434,6 +1440,11 @@ async def _negotiate_quote(
             negotiation_id=str(counter_terms.get("negotiation_id") or proposal_id),
             current_seller_price=seller_price if seller_price is not None else 0.0,
             our_last_offer=open_price,
+            # Correlation: every message of this negotiation carries the quote
+            # it concerns, so the seller can resolve the agreed price from the
+            # quote that gets booked. DealBookingRequest has no negotiation_id,
+            # so the quote is the only correlation key the contract offers.
+            quote_id=quote.quote_id,
         )
 
         agreed: float | None = None
@@ -1565,6 +1576,64 @@ async def _negotiate_quote(
             await self._emit_negotiation_concluded(record)
             return None, record
 
+        # Money invariant, INDEPENDENT of the ceiling check above: the bookable
+        # quote must carry the price that was actually AGREED, not merely some
+        # price under the buyer's ceiling. The ceiling test alone passes for any
+        # price below the limit, including the seller's undiscounted list price,
+        # so a dropped negotiation books silently and logs as a success. This is
+        # the assertion that makes that class of defect loud.
+        #
+        # It is checked BEFORE the quote enters the ranked list -- i.e. before
+        # the booking call -- so a mismatch refuses to book rather than
+        # surfacing once the money is committed.
+        #
+        # The two guards are deliberately separate and separately reported. The
+        # ceiling breach is checked first because it is the more severe and more
+        # specific diagnosis ("the seller reneged and went above our limit"),
+        # and because it also disposes of the unpriced (`final_cpm is None`)
+        # case. Collapsing the two would lose that distinction; reordering them
+        # would require handling the unpriced case here instead.
+        #
+        # The comparison is on integer micros, NOT on floats with a tolerance
+        # window. Micros are the unit money is actually defined in on the wire
+        # (the shared `Money`), so quantizing both sides and comparing integers
+        # asks exactly the right question: "is this the same money?" There is no
+        # chosen fudge factor to widen, and no band in which a real pricing
+        # difference stays silent. It is also the only comparison that is robust
+        # for the right reason: a bit-exact `!=` on floats happens to be correct
+        # today only because both sides use the same two operations, and would
+        # refuse a perfectly good booking if `agreed` ever carried more than six
+        # decimal places, because the micros quantization then produces a real
+        # ~1e-7 delta. Quantizing first absorbs that -- the only delta this path
+        # can actually produce -- while still rejecting a difference of one
+        # micro and upward.
+        agreed_money = _money_from_float(agreed)
+        final_money = _money_from_float(final_cpm)
+        # Both inputs are non-None floats by the guards above, so neither
+        # conversion can return None. The None arms are spelled out anyway so
+        # the guard fails CLOSED rather than booking an unverifiable price.
+        if (
+            agreed_money is None
+            or final_money is None
+            or final_money.amount_micros != agreed_money.amount_micros
+        ):
+            logger.warning(
+                "Re-quote from seller %s came back at %.2f but the negotiated "
+                "price was %.2f; refusing to book a price we did not agree to "
+                "(ceiling %.2f)",
+                seller_id,
+                final_cpm,
+                agreed,
+                max_cpm,
+            )
+            record["outcome"] = "requote_price_mismatch"
+            record["error"] = (
+                f"re-quote priced at {final_cpm:.2f} but the agreed price was "
+                f"{agreed:.2f}; the negotiated price did not reach the bookable quote"
+            )
+            await self._emit_negotiation_concluded(record)
+            return None, record
+
         record["outcome"] = "accepted"
         record["requoted_quote_id"] = requote.quote_id
         await self._emit_negotiation_concluded(record)