fix: harden media buy response normalization

scripts/post_generate_fixes.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+# ruff: noqa: E501
 """
 Post-generation fixes for generated Pydantic models.
 
@@ -1517,32 +1518,23 @@ def restore_response_variant_aliases() -> None:
 # Backward-compatible SDK response arms. Upstream beta 3 schemas collapse this
 # task response to the common protocol envelope, but the Python SDK keeps the
 # historical numbered variants as ergonomic construction/parsing aliases.
-from collections.abc import Sequence
 from typing import Any, Literal, TypeAlias
 
 from pydantic import ConfigDict, model_validator
 
+from adcp.types.media_buy_status_helpers import MEDIA_BUY_LEGACY_STATUS_VALUES, unwrap_enum_value
+
 from ..core import error as error_1
 from ..core import ext as ext_1
 from ..core import package as package_1
 from ..core.protocol_envelope import ProtocolEnvelope
 from ..enums import media_buy_status as media_buy_status_1
 from ..enums import task_status as task_status_1
-
-
-_MEDIA_BUY_STATUS_VALUES = {
-    "pending_creatives",
-    "pending_start",
-    "active",
-    "paused",
-    "rejected",
-    "canceled",
-}
-
-
-def _value(value: Any) -> Any:
-    return getattr(value, "value", value)
 """
+    update_media_header = media_header.replace(
+        "from typing import Any, Literal, TypeAlias",
+        "from collections.abc import Sequence\nfrom typing import Any, Literal, TypeAlias",
+    )
 
     simple_error_arms: dict[str, tuple[str, str, str]] = {
         "media_buy/build_creative_response.py": (
@@ -1935,12 +1927,15 @@ class CreateMediaBuyResponse1(AdcpVersionEnvelope):
     def _normalize_legacy_status(cls, data: Any) -> Any:
         if not isinstance(data, dict):
             return data
-        raw_status = _value(data.get("status"))
-        media_buy_status = _value(data.get("media_buy_status"))
+        raw_status = unwrap_enum_value(data.get("status"))
+        media_buy_status = unwrap_enum_value(data.get("media_buy_status"))
         if raw_status is None:
             data = dict(data)
             data["status"] = "completed"
-        elif media_buy_status is None and raw_status in _MEDIA_BUY_STATUS_VALUES:
+        elif raw_status == "completed":
+            data = dict(data)
+            data["status"] = "completed"
+        elif media_buy_status is None and raw_status in MEDIA_BUY_LEGACY_STATUS_VALUES:
             data = dict(data)
             data["media_buy_status"] = raw_status
             data["status"] = "completed"
@@ -1988,7 +1983,7 @@ def _normalize_submitted_status(cls, data: Any) -> Any:
         "media_buy/update_media_buy_response.py",
         "UpdateMediaBuyResponse",
         "class UpdateMediaBuyResponse1",
-        media_header
+        update_media_header
         + """
 
 class UpdateMediaBuyResponse1(AdcpVersionEnvelope):
@@ -2005,12 +2000,15 @@ class UpdateMediaBuyResponse1(AdcpVersionEnvelope):
     def _normalize_legacy_status(cls, data: Any) -> Any:
         if not isinstance(data, dict):
             return data
-        raw_status = _value(data.get("status"))
-        media_buy_status = _value(data.get("media_buy_status"))
+        raw_status = unwrap_enum_value(data.get("status"))
+        media_buy_status = unwrap_enum_value(data.get("media_buy_status"))
         if raw_status is None:
             data = dict(data)
             data["status"] = "completed"
-        elif media_buy_status is None and raw_status in _MEDIA_BUY_STATUS_VALUES:
+        elif raw_status == "completed":
+            data = dict(data)
+            data["status"] = "completed"
+        elif media_buy_status is None and raw_status in MEDIA_BUY_LEGACY_STATUS_VALUES:
             data = dict(data)
             data["media_buy_status"] = raw_status
             data["status"] = "completed"

src/adcp/__init__.py

@@ -1077,8 +1077,8 @@ def get_adcp_version() -> str:
     "UpdateContentStandardsErrorResponse",
     "UpdateMediaBuySuccessResponse",
     "UpdateMediaBuyResponse1",
-    "UpdateMediaBuyResponse3",
     "UpdateMediaBuyErrorResponse",
+    "UpdateMediaBuyResponse3",
     "UpdateMediaBuyPackagesRequest",
     "UpdateMediaBuyPropertiesRequest",
     "UpdateMediaBuySubmittedResponse",

src/adcp/server/mcp_tools.py

@@ -27,29 +27,16 @@
 
 from adcp.server.base import ADCPHandler, ToolContext
 from adcp.server.test_controller import SCENARIOS as _CONTROLLER_SCENARIOS
+from adcp.types import (
+    MEDIA_BUY_LEGACY_STATUS_VALUES,
+    unwrap_enum_value,
+)
 from adcp.types.error_narrowing import narrow_union_errors
 from adcp.validation.client_hooks import ValidationHookConfig
 
 logger = logging.getLogger(__name__)
 
 
-_MEDIA_BUY_STATUS_VALUES = {
-    "draft",
-    "pending_creatives",
-    "pending_start",
-    "active",
-    "paused",
-    "completed",
-    "canceled",
-    "cancelled",
-    "rejected",
-}
-
-
-def _enum_value(value: Any) -> Any:
-    return getattr(value, "value", value)
-
-
 def _looks_like_sync_media_buy_success(method_name: str, result: dict[str, Any]) -> bool:
     return (
         method_name in {"create_media_buy", "update_media_buy"}
@@ -71,20 +58,20 @@ def _normalize_response_envelope(
     account-specific inventory as globally cacheable.
     """
     if _looks_like_sync_media_buy_success(method_name, result):
-        raw_status = _enum_value(result.get("status"))
-        media_buy_status = _enum_value(result.get("media_buy_status"))
+        raw_status = unwrap_enum_value(result.get("status"))
+        media_buy_status = unwrap_enum_value(result.get("media_buy_status"))
         if raw_status is not None:
             result["status"] = raw_status
         if media_buy_status is not None:
             result["media_buy_status"] = media_buy_status
-        if media_buy_status is None and raw_status in _MEDIA_BUY_STATUS_VALUES:
-            if raw_status != "completed":
-                result["media_buy_status"] = raw_status
-                result["status"] = "completed"
+        if media_buy_status is None and raw_status in MEDIA_BUY_LEGACY_STATUS_VALUES:
+            result["media_buy_status"] = raw_status
+            result["status"] = "completed"
         elif media_buy_status is not None and raw_status in {None, media_buy_status}:
             result["status"] = "completed"
 
-    result.setdefault("status", "completed")
+    if "status" not in result and "task_id" not in result:
+        result["status"] = "completed"
     if (
         method_name in {"get_products", "get_signals"}
         and "cache_scope" not in result

src/adcp/types/__init__.py

@@ -784,6 +784,17 @@ def __init__(self, *args: object, **kwargs: object) -> None:
 UrlType = UrlAssetType
 AvailableReportingFrequency = ReportingFrequency
 
+# Internal normalization helpers used by generated response models and server
+# envelope handling. Keep these importable through ``adcp.types`` for the
+# repo's generated-type layering rule, but leave them out of ``__all__`` so they
+# do not become part of the documented public API.
+from adcp.types.media_buy_status_helpers import (  # noqa: E402
+    MEDIA_BUY_LEGACY_STATUS_VALUES as MEDIA_BUY_LEGACY_STATUS_VALUES,
+)
+from adcp.types.media_buy_status_helpers import (
+    unwrap_enum_value as unwrap_enum_value,
+)
+
 __all__ = [
     # A2UI types
     "A2UiComponent",

src/adcp/types/aliases.py

@@ -1877,6 +1877,7 @@ class UnknownGroupAsset(_BaseGroupAsset):
     # Update media buy responses
     "UpdateMediaBuySuccessResponse",
     "UpdateMediaBuyErrorResponse",
+    "UpdateMediaBuyResponse3",
     "UpdateMediaBuySubmittedResponse",
     # Validate content delivery responses
     "ValidateContentDeliverySuccessResponse",

src/adcp/types/generated_poc/media_buy/create_media_buy_response.py

@@ -4,39 +4,24 @@
 
 from __future__ import annotations
 
-from ..core.version_envelope import AdcpVersionEnvelope
-
-
-
 # Backward-compatible SDK response arms. Upstream beta 3 schemas collapse this
 # task response to the common protocol envelope, but the Python SDK keeps the
 # historical numbered variants as ergonomic construction/parsing aliases.
 from typing import Any, Literal, TypeAlias
 
 from pydantic import ConfigDict, model_validator
 
+from adcp.types.media_buy_status_helpers import MEDIA_BUY_LEGACY_STATUS_VALUES, unwrap_enum_value
+
 from ..core import error as error_1
 from ..core import ext as ext_1
 from ..core import package as package_1
 from ..core.protocol_envelope import ProtocolEnvelope
+from ..core.version_envelope import AdcpVersionEnvelope
 from ..enums import media_buy_status as media_buy_status_1
 from ..enums import task_status as task_status_1
 
 
-_MEDIA_BUY_STATUS_VALUES = {
-    "pending_creatives",
-    "pending_start",
-    "active",
-    "paused",
-    "rejected",
-    "canceled",
-}
-
-
-def _value(value: Any) -> Any:
-    return getattr(value, "value", value)
-
-
 class CreateMediaBuyResponse1(AdcpVersionEnvelope):
     model_config = ConfigDict(extra='allow')
     media_buy_id: str
@@ -50,12 +35,15 @@ class CreateMediaBuyResponse1(AdcpVersionEnvelope):
     def _normalize_legacy_status(cls, data: Any) -> Any:
         if not isinstance(data, dict):
             return data
-        raw_status = _value(data.get("status"))
-        media_buy_status = _value(data.get("media_buy_status"))
+        raw_status = unwrap_enum_value(data.get("status"))
+        media_buy_status = unwrap_enum_value(data.get("media_buy_status"))
         if raw_status is None:
             data = dict(data)
             data["status"] = "completed"
-        elif media_buy_status is None and raw_status in _MEDIA_BUY_STATUS_VALUES:
+        elif raw_status == "completed":
+            data = dict(data)
+            data["status"] = "completed"
+        elif media_buy_status is None and raw_status in MEDIA_BUY_LEGACY_STATUS_VALUES:
             data = dict(data)
             data["media_buy_status"] = raw_status
             data["status"] = "completed"

src/adcp/types/generated_poc/media_buy/update_media_buy_response.py

@@ -4,10 +4,6 @@
 
 from __future__ import annotations
 
-from ..core.version_envelope import AdcpVersionEnvelope
-
-
-
 # Backward-compatible SDK response arms. Upstream beta 3 schemas collapse this
 # task response to the common protocol envelope, but the Python SDK keeps the
 # historical numbered variants as ergonomic construction/parsing aliases.
@@ -16,28 +12,17 @@
 
 from pydantic import ConfigDict, model_validator
 
+from adcp.types.media_buy_status_helpers import MEDIA_BUY_LEGACY_STATUS_VALUES, unwrap_enum_value
+
 from ..core import error as error_1
 from ..core import ext as ext_1
 from ..core import package as package_1
 from ..core.protocol_envelope import ProtocolEnvelope
+from ..core.version_envelope import AdcpVersionEnvelope
 from ..enums import media_buy_status as media_buy_status_1
 from ..enums import task_status as task_status_1
 
 
-_MEDIA_BUY_STATUS_VALUES = {
-    "pending_creatives",
-    "pending_start",
-    "active",
-    "paused",
-    "rejected",
-    "canceled",
-}
-
-
-def _value(value: Any) -> Any:
-    return getattr(value, "value", value)
-
-
 class UpdateMediaBuyResponse1(AdcpVersionEnvelope):
     model_config = ConfigDict(extra='allow')
     media_buy_id: str
@@ -52,12 +37,15 @@ class UpdateMediaBuyResponse1(AdcpVersionEnvelope):
     def _normalize_legacy_status(cls, data: Any) -> Any:
         if not isinstance(data, dict):
             return data
-        raw_status = _value(data.get("status"))
-        media_buy_status = _value(data.get("media_buy_status"))
+        raw_status = unwrap_enum_value(data.get("status"))
+        media_buy_status = unwrap_enum_value(data.get("media_buy_status"))
         if raw_status is None:
             data = dict(data)
             data["status"] = "completed"
-        elif media_buy_status is None and raw_status in _MEDIA_BUY_STATUS_VALUES:
+        elif raw_status == "completed":
+            data = dict(data)
+            data["status"] = "completed"
+        elif media_buy_status is None and raw_status in MEDIA_BUY_LEGACY_STATUS_VALUES:
             data = dict(data)
             data["media_buy_status"] = raw_status
             data["status"] = "completed"

src/adcp/types/media_buy_status_helpers.py

@@ -0,0 +1,26 @@
+"""Shared helpers for media-buy lifecycle status compatibility.
+
+The sync create/update media-buy response arms accept legacy constructor and
+handler payloads where ``status`` carried lifecycle state. ``completed`` is a
+valid lifecycle enum value, but it is also the 3.1 task-envelope success status,
+so the SDK only infers lifecycle status from the unambiguous subset below.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+MEDIA_BUY_LEGACY_STATUS_VALUES = frozenset(
+    {
+        "active",
+        "canceled",
+        "paused",
+        "pending_creatives",
+        "pending_start",
+        "rejected",
+    }
+)
+
+
+def unwrap_enum_value(value: Any) -> Any:
+    return getattr(value, "value", value)