feat(media-buy): add action rights assessment

README.md

@@ -358,6 +358,7 @@ forward traffic degrades gracefully rather than failing.
 - **[Migrating from SDK 7 to 8](https://github.com/adcontextprotocol/adcp-client-python/blob/main/MIGRATION_v7_to_v8.md)** - Secure webhook defaults and telemetry changes
 - **[Migrating from AdCP 3.1 to 3.2 beta](MIGRATION_ADCP_3.1_TO_3.2.md)** - Compact lifecycle adoption and old/new compatibility matrix
 - **[Durable legacy purchase continuations](docs/legacy-purchase-continuations.md)** - Safe products-only compatibility redemption and crash recovery
+- **[Media-buy action rights](docs/media-buy-action-rights.md)** - Assess product possibilities, accepted change rights, and currently available actions
 - **[Testing your AdCP server](docs/testing-your-adcp-server.md)** - In-process harness for unit tests plus storyboard-runner compliance grading
 - **[Universal macro translation](docs/universal-macro-translation.md)** - Producer-side pixel URL translation, trust boundary, and diagnostics
 - **[Multi-tenant contract](docs/multi-tenant-contract.md)** - Scope invariants every multi-tenant agent must satisfy

docs/media-buy-action-rights.md

@@ -0,0 +1,106 @@
+# Media-buy action rights
+
+AdCP exposes three related action surfaces with different authority:
+
+1. `Product.allowed_actions` says what a product may support. It is advisory.
+2. `Proposal.commercial_terms.change_terms` records the rights accepted in a deal.
+3. `MediaBuy.available_actions` says which accepted rights are executable now.
+
+The SDK joins those surfaces without promoting product templates or legacy
+compatibility fields into authority.
+
+## Buyer assessment
+
+```python
+from adcp import assess_media_buy_action
+
+assessment = assess_media_buy_action(
+    "increase_budget",
+    product=product,
+    proposal=accepted_proposal,
+    media_buy=current_media_buy,
+    intent={
+        "current_amount": "1000",
+        "result_amount": "1100",
+        "currency": "USD",
+    },
+)
+
+if assessment.status == "available_now":
+    print(assessment.task, assessment.mode)
+else:
+    print(assessment.possible, assessment.promised, assessment.available)
+```
+
+The status is one of `available_now`, `wrong_status`, `not_negotiated`,
+`unsupported_by_product`, `currently_unavailable`, or `legacy_unknown`.
+Portable budget, flight, package-count, and effective-time bounds are checked
+when the caller supplies enough current/result state. Opaque condition IDs and
+contract references are never executed or interpreted by the SDK.
+
+For deprecated `update_media_buy` patches,
+`assess_update_media_buy_actions()` first decomposes the patch into canonical
+actions and then applies the same checks. Fine-grained beta.9 actions retain
+coarse 3.x candidates so older seller declarations remain readable without
+expanding authority.
+
+## Routing and races
+
+`route_media_buy_action()` selects the normal compact task. Operational
+controls use `control_media_buy`, commercial amendments use
+`refine_proposals`, and creative mutations use `sync_creatives`. Some actions
+are valid through either control or refinement; an authoritative live action's
+explicit `task` wins when the protocol permits it.
+
+`dispatch_media_buy_action()` is asynchronous and accepts an already validated
+assessment plus the generated request model for that task. A
+`seller_managed` action uses the ordinary asynchronous task lifecycle; it does
+not introduce a seller-review MediaBuy status.
+
+Always send the latest MediaBuy `revision` and an idempotency key. If an
+`ACTION_NOT_ALLOWED` race returns `currently_available_actions`, pass that echo
+to `reassess_media_buy_action()` for an immediate explanation, then refresh the
+full MediaBuy before retrying with its new revision.
+
+## Seller materialization and projection
+
+```python
+from adcp import ChangeTermSelection, materialize_change_terms
+
+change_terms = materialize_change_terms(
+    product.allowed_actions,
+    [
+        ChangeTermSelection(
+            action="increase_budget",
+            term_id="right_budget_1",
+            service_mode="seller_managed",
+            allowed_statuses=("active",),
+        )
+    ],
+)
+```
+
+Only explicit selections become binding terms. The builder rejects duplicate
+actions and term IDs, expanded status scopes, unadvertised modes, and
+action/constraint mismatches.
+
+Use `project_available_actions()` to derive the current surface from accepted
+terms. Optional authorization, delegation, seller-policy, product, and
+resolved-condition gates only narrow the result. Conditions must be explicitly
+resolved to `True`; missing or indeterminate condition state fails closed.
+
+## Version behavior
+
+| Wire version | Projection |
+|---|---|
+| AdCP 3.1.19 | `terms_ref` compatibility alias; no inferred proposal identity |
+| Early AdCP 3.2 beta | explicit `task`, legacy `terms_ref`, `requires_approval` compatibility mode |
+| AdCP 3.2 beta.9+ | explicit `task`, `seller_managed`, and `change_term_id` |
+
+An arbitrary inbound 3.1 `terms_ref` remains opaque even when its text matches
+a proposal term ID. When a 3.2 payload contains both aliases, unequal values
+fail closed.
+
+The language-neutral fixture at
+`tests/fixtures/media_buy_action_assessment.json` defines normalized buyer
+results for cross-SDK parity.

examples/seller_agent.py

@@ -21,12 +21,20 @@
 from datetime import datetime, timezone
 from typing import Any
 
-from adcp import Creative, Format, Product
+from adcp import (
+    ActionAvailabilityStatus,
+    Creative,
+    Format,
+    Product,
+    assess_update_media_buy_actions,
+    project_available_actions,
+)
 from adcp.canonical_formats import (
     CanonicalFormatLegacyResolutionContext,
     LegacyFormatConversionContext,
     migrated_format_option_id,
 )
+from adcp.decisioning import assert_media_buy_transition
 from adcp.server import (
     INSECURE_ALLOW_ALL,
     ADCPHandler,
@@ -452,6 +460,24 @@ def _resolve_available_actions(
     return available
 
 
+def _change_terms_for_buy(media_buy: dict[str, Any]) -> list[dict[str, Any]] | None:
+    proposal = media_buy.get("accepted_proposal")
+    if not isinstance(proposal, dict):
+        return None
+    commercial_terms = proposal.get("commercial_terms")
+    if not isinstance(commercial_terms, dict) or "change_terms" not in commercial_terms:
+        return None
+    change_terms = commercial_terms.get("change_terms")
+    return change_terms if isinstance(change_terms, list) else []
+
+
+def _available_actions_for_buy(media_buy: dict[str, Any]) -> list[dict[str, Any]]:
+    change_terms = _change_terms_for_buy(media_buy)
+    if change_terms is not None:
+        return project_available_actions(change_terms, media_buy["status"]).to_wire()
+    return _resolve_available_actions(media_buy.get("packages", []), media_buy["status"])
+
+
 def _attempted_action_for_update(
     params: dict[str, Any],
     mb: dict[str, Any],
@@ -485,13 +511,14 @@ def _action_not_allowed_response(
     attempted_action: str,
     reason: str,
     currently_available_actions: list[dict[str, Any]],
+    compact: bool = False,
 ) -> dict[str, Any]:
     recovery = (
         "terminal"
         if reason in {"not_supported_on_product", "not_supported_on_buy"}
         else "correctable"
     )
-    return {
+    response: dict[str, Any] = {
         "errors": [
             {
                 "code": "ACTION_NOT_ALLOWED",
@@ -505,6 +532,32 @@ def _action_not_allowed_response(
             }
         ]
     }
+    if compact:
+        response["status"] = "failed"
+    return response
+
+
+def _requote_required_response(
+    *,
+    field: str,
+    change_term_id: str,
+    constraint: str,
+) -> dict[str, Any]:
+    return {
+        "status": "failed",
+        "errors": [
+            {
+                "code": "REQUOTE_REQUIRED",
+                "message": "Requested change exceeds the accepted commercial envelope",
+                "recovery": "correctable",
+                "details": {
+                    "envelope_field": field,
+                    "change_term_id": change_term_id,
+                    "constraint": constraint,
+                },
+            }
+        ],
+    }
 
 
 def _products_for_request(params: dict[str, Any]) -> list[dict[str, Any]]:
@@ -993,11 +1046,107 @@ async def get_media_buys(self, params: dict[str, Any], context: Any = None) -> d
             }
             if mb.get("context") is not None:
                 result["context"] = mb["context"]
-            if mb.get("available_actions"):
-                result["available_actions"] = mb["available_actions"]
+            available_actions = _available_actions_for_buy(mb)
+            if available_actions:
+                result["available_actions"] = available_actions
+            if mb.get("accepted_proposal") is not None:
+                accepted_proposal = deepcopy(mb["accepted_proposal"])
+                result["accepted_proposal"] = accepted_proposal
+                result["accepted_proposal_id"] = accepted_proposal["proposal_id"]
+                result["accepted_proposal_terms_digest"] = accepted_proposal["terms_digest"]
             results.append(result)
         return media_buys_response(results)
 
+    async def control_media_buy(
+        self, params: dict[str, Any], context: Any = None
+    ) -> dict[str, Any]:
+        mb_id = params.get("media_buy_id")
+        mb = media_buys.get(mb_id) if isinstance(mb_id, str) else None
+        if mb is None or not isinstance(mb_id, str):
+            error = adcp_error("MEDIA_BUY_NOT_FOUND", "Media buy not found")
+            return {"status": "failed", **error}
+
+        revision = mb.get("revision", 1)
+        if params.get("revision") != revision:
+            error = adcp_error("CONFLICT", "Revision mismatch - refetch and retry")
+            return {"status": "failed", **error}
+
+        current = deepcopy(mb)
+        current["available_actions"] = _available_actions_for_buy(mb)
+        proposal = mb.get("accepted_proposal")
+        assessments = assess_update_media_buy_actions(
+            params,
+            current,
+            proposal=proposal,
+        )
+        attempted = assessments[0] if assessments else None
+        if attempted is None:
+            error = adcp_error("INVALID_REQUEST", "No supported control field supplied")
+            return {"status": "failed", **error}
+
+        violated = next(
+            (check for check in attempted.constraints if check.outcome.value == "violated"),
+            None,
+        )
+        if violated is not None:
+            return _requote_required_response(
+                field=(
+                    "total_budget.amount"
+                    if attempted.action in {"increase_budget", "decrease_budget"}
+                    else violated.field or "control"
+                ),
+                change_term_id=attempted.change_term_id or "unknown",
+                constraint=violated.constraint,
+            )
+
+        if attempted.status is not ActionAvailabilityStatus.available_now:
+            term = next(
+                (
+                    value
+                    for value in (_change_terms_for_buy(mb) or [])
+                    if value.get("action") == attempted.action
+                ),
+                None,
+            )
+            if (
+                term is not None
+                and term.get("allowed_statuses")
+                and mb["status"] not in term["allowed_statuses"]
+            ):
+                reason = "wrong_status"
+            elif term is not None and term.get("conditions"):
+                reason = "condition_unresolved"
+            else:
+                reason = "not_supported_on_buy"
+            return _action_not_allowed_response(
+                attempted_action=attempted.action,
+                reason=reason,
+                currently_available_actions=current["available_actions"],
+                compact=True,
+            )
+
+        if params.get("paused") is True:
+            assert_media_buy_transition(mb["status"], "paused", media_buy_id=mb_id)
+            mb["status"] = "paused"
+        elif params.get("paused") is False:
+            assert_media_buy_transition(mb["status"], "active", media_buy_id=mb_id)
+            mb["status"] = "active"
+        elif params.get("canceled") is True:
+            assert_media_buy_transition(mb["status"], "canceled", media_buy_id=mb_id)
+            mb["status"] = "canceled"
+        if "total_budget" in params:
+            mb["total_budget"] = deepcopy(params["total_budget"])
+
+        mb["revision"] = revision + 1
+        mb["available_actions"] = _available_actions_for_buy(mb)
+        return {
+            "status": "completed",
+            "media_buy_id": mb_id,
+            "revision": mb["revision"],
+            "media_buy_status": mb["status"],
+            "available_actions": mb["available_actions"],
+        }
+
     async def update_media_buy(self, params: dict[str, Any], context: Any = None) -> dict[str, Any]:
         mb_id = params.get("media_buy_id")
         mb = media_buys.get(mb_id) if mb_id else None
@@ -1550,6 +1699,7 @@ async def seed_media_buy(
         data.setdefault("packages", [])
         data.setdefault("confirmed_at", _now_z())
         data.setdefault("revision", 1)
+        data["available_actions"] = _available_actions_for_buy(data)
         media_buys[mb_id] = data
         return {"media_buy_id": mb_id}