fix(types): preserve open canonical format kinds

scripts/post_generate_fixes.py

@@ -3588,6 +3588,36 @@ def _reject_seller_bound_manifest_fields(cls, data: Any) -> Any:
     print("  core/creative_representation.py: restored canonical format contract")
 
 
+def preserve_open_delivery_format_kind() -> None:
+    """Keep the public delivery-creative consumer boundary forward-compatible."""
+    target = OUTPUT_DIR / "creative" / "get_creative_delivery_response.py"
+    if not target.exists():
+        print("  creative/get_creative_delivery_response.py not found (skipping open enum)")
+        return
+
+    source = target.read_text()
+    generated = """    format_kind: Annotated[
+        canonical_format_kind.CanonicalFormatKind | None,
+        Field(description='Canonical format kind delivered for this creative.'),
+    ] = None"""
+    replacement = """    format_kind: Annotated[
+        canonical_format_kind.CanonicalFormatKind | str | None,
+        Field(
+            description='Canonical format kind delivered for this creative.',
+            union_mode='left_to_right',
+        ),
+    ] = None"""
+    if generated in source:
+        target.write_text(source.replace(generated, replacement, 1))
+        print("  creative/get_creative_delivery_response.py: opened format_kind enum")
+    elif replacement in source:
+        print("  creative/get_creative_delivery_response.py: format_kind already open")
+    else:
+        raise RuntimeError(
+            "get_creative_delivery_response.py: expected format_kind field not found"
+        )
+
+
 def enforce_transformer_output_contract() -> None:
     """Require a transformer to declare canonical or legacy output formats."""
     target = OUTPUT_DIR / "core" / "transformer.py"
@@ -5742,6 +5772,7 @@ def main(argv: list[str] | None = None):
         restore_principal_result_aliases,
         disambiguate_comply_response_arm,
         restore_flattened_contract_field_types,
+        preserve_open_delivery_format_kind,
         enforce_transformer_output_contract,
         restore_constructible_response_bases,
         restore_response_variant_aliases,

src/adcp/types/_forward_compat.py

@@ -27,9 +27,9 @@
 from __future__ import annotations
 
 from copy import copy
-from typing import Any, cast, get_args
+from typing import Annotated, Any, cast, get_args
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from pydantic.fields import FieldInfo
 
 from adcp.types.aliases import FormatAssetUnion, GroupFormatAssetUnion, RepeatableAssetGroup
@@ -48,14 +48,24 @@
 from adcp.types.generated_poc.bundled.protocol.get_adcp_capabilities_response import (
     PublisherDomain as BundledPublisherDomain,
 )
+from adcp.types.generated_poc.core.canonical_format_kind import CanonicalFormatKind
 from adcp.types.generated_poc.core.canonical_product import PublisherDomain
+from adcp.types.generated_poc.core.creative_manifest import CreativeManifest
 from adcp.types.generated_poc.core.format import Format
 from adcp.types.generated_poc.core.media_buy_features import MediaBuyFeatures
+from adcp.types.generated_poc.creative.get_creative_delivery_response import (
+    Creative as DeliveryCreative,
+)
 from adcp.types.generated_poc.protocol.get_adcp_capabilities_response import (
     AcceptancePolicyDiscovery,
     PrimaryCountry,
 )
 
+_OpenCanonicalFormatKind = Annotated[
+    CanonicalFormatKind | str,
+    Field(union_mode="left_to_right"),
+]
+
 
 def _patch_model_field(model: type[BaseModel], field_name: str, new_annotation: Any) -> None:
     """Replace a Pydantic model field's annotation in-place.
@@ -103,6 +113,23 @@ def _patch_equivalent_model_field(
 
 def _apply_forward_compat() -> None:
     """Apply open-union, capability, and public-model compatibility patches."""
+    # Canonical format kinds are an open enum on consumer boundaries. Preserve
+    # values introduced by a newer protocol revision instead of rejecting the
+    # entire creative manifest. Known values still coerce to the StrEnum arm.
+    _patch_model_field(
+        CreativeManifest,
+        "format_kind",
+        _OpenCanonicalFormatKind | None,
+    )
+    CreativeManifest.model_rebuild(force=True)
+
+    _patch_model_field(
+        DeliveryCreative,
+        "format_kind",
+        _OpenCanonicalFormatKind | None,
+    )
+    DeliveryCreative.model_rebuild(force=True)
+
     _patch_model_field(Format, "assets", list[FormatAssetUnion] | None)
     Format.model_rebuild(force=True)
 

src/adcp/types/canonical_creative.py

@@ -117,6 +117,11 @@
     unwrap_enum_value,
 )
 
+_OpenCanonicalFormatKind = Annotated[
+    CanonicalFormatKind | str,
+    Field(union_mode="left_to_right"),
+]
+
 _LEGACY_IDENTITY_KEY = re.compile(r"(^|_)(?:format_ids?|v1_format_ref)($|_)")
 _CREDENTIAL_SHAPED_KEY_SUFFIXES = (
     "credential",
@@ -518,6 +523,8 @@ def _validate_custom_shape(self) -> Format:
         if self.format_kind is CanonicalFormatKind.custom:
             if not self.format_shape:
                 raise ValueError("custom formats require format_shape")
+            if self.format_schema is None:
+                raise ValueError("custom formats require format_schema")
         elif self.format_shape is not None or self.format_schema is not None:
             raise ValueError("format_shape and format_schema are only valid for custom formats")
         return self
@@ -553,16 +560,25 @@ def _validate_custom_shape(self) -> Format:
 CreativeAsset = _canonical_clone(
     "CreativeAsset",
     _CanonicalCreativeWire,
-    overrides={"format_kind": (CanonicalFormatKind, Field())},
+    overrides={"format_kind": (_OpenCanonicalFormatKind, Field())},
 )
 
 Creative = _canonical_clone(
     "Creative",
     _CanonicalListedCreative,
-    overrides={"format_kind": (CanonicalFormatKind, Field())},
+    overrides={"format_kind": (_OpenCanonicalFormatKind, Field())},
 )
 
-_CreativeManifestBase = _canonical_clone("_CreativeManifestBase", _CanonicalCreativeManifestWire)
+_CreativeManifestBase = _canonical_clone(
+    "_CreativeManifestBase",
+    _CanonicalCreativeManifestWire,
+    overrides={
+        "format_kind": (
+            _OpenCanonicalFormatKind | None,
+            copy.deepcopy(_CanonicalCreativeManifestWire.model_fields["format_kind"]),
+        )
+    },
+)
 
 
 class CreativeManifest(_CreativeManifestBase):
@@ -604,7 +620,7 @@ def wire_value(value: Any) -> Any:
     "DeliveryCreative",
     _LegacyDeliveryCreative,
     overrides={
-        "format_kind": (CanonicalFormatKind | None, Field(default=None)),
+        "format_kind": (_OpenCanonicalFormatKind | None, Field(default=None)),
         "variants": (list[CreativeVariant], Field()),
     },
 )

src/adcp/types/canonical_creative.pyi

@@ -46,22 +46,24 @@ class Product(CanonicalBoundaryModel):
 
 class CreativeAsset(CanonicalBoundaryModel):
     creative_id: str
-    format_kind: CanonicalFormatKind
+    format_kind: CanonicalFormatKind | str
     format_option_ref: Any
 
 class Creative(CanonicalBoundaryModel):
     creative_id: str
-    format_kind: CanonicalFormatKind
+    format_kind: CanonicalFormatKind | str
     format_option_ref: Any
 
-class CreativeManifest(CanonicalBoundaryModel): ...
+class CreativeManifest(CanonicalBoundaryModel):
+    format_kind: CanonicalFormatKind | str | None = ...
+    assets: dict[str, Any]
 
 class CreativeVariant(CanonicalBoundaryModel):
     manifest: CreativeManifest | None
 
 class DeliveryCreative(CanonicalBoundaryModel):
     creative_id: str
-    format_kind: CanonicalFormatKind | None
+    format_kind: CanonicalFormatKind | str | None
     variants: list[CreativeVariant]
 
 class CreativeFilters(CanonicalBoundaryModel): ...

src/adcp/types/generated_poc/creative/get_creative_delivery_response.py

@@ -79,8 +79,11 @@ class Creative(AdCPBaseModel):
         ),
     ] = None
     format_kind: Annotated[
-        canonical_format_kind.CanonicalFormatKind | None,
-        Field(description='Canonical format kind delivered for this creative.'),
+        canonical_format_kind.CanonicalFormatKind | str | None,
+        Field(
+            description='Canonical format kind delivered for this creative.',
+            union_mode='left_to_right',
+        ),
     ] = None
     format_option_ref: Annotated[
         format_option_ref_1.FormatOptionReference | None,

tests/test_canonical_formats_v2_to_v1.py

@@ -27,6 +27,13 @@ def _ref(id_: str = "display_300x250_image") -> FormatId:
     )
 
 
+def _format_schema() -> dict[str, str]:
+    return {
+        "uri": "https://example.com/custom-format.json",
+        "digest": f"sha256:{'0' * 64}",
+    }
+
+
 # ---------------------------------------------------------------------------
 # Step 1 — explicit v1-unreachability is silent (no refs, no advisories)
 # ---------------------------------------------------------------------------
@@ -57,6 +64,7 @@ def test_custom_without_refs_is_silent() -> None:
         format_kind=CanonicalFormatKind.custom,
         params={},
         format_shape="multi_placement_takeover",
+        format_schema=_format_schema(),
     )
 
     result = project_declaration_to_v1(decl)
@@ -72,6 +80,7 @@ def test_custom_with_v1_format_ref_emits_refs() -> None:
         format_kind=CanonicalFormatKind.custom,
         params={},
         format_shape="multi_placement_takeover",
+        format_schema=_format_schema(),
         v1_format_ref=refs,
     )
 
@@ -166,6 +175,7 @@ def test_non_translatable_canonicals_are_silent_with_no_ref(kind: CanonicalForma
         format_kind=kind,
         params={},
         format_shape="test_custom" if kind is CanonicalFormatKind.custom else None,
+        format_schema=_format_schema() if kind is CanonicalFormatKind.custom else None,
     )
 
     result = project_declaration_to_v1(decl)

tests/test_code_generation.py

@@ -1195,6 +1195,35 @@ def test_schema_derived_response_arms_preserve_nested_validation():
         )
 
 
+def test_post_generate_preserves_open_delivery_format_kind(tmp_path, monkeypatch) -> None:
+    """The public delivery alias must stay open after clean code generation."""
+    from scripts import post_generate_fixes
+
+    generated_dir = tmp_path / "generated_poc"
+    target = generated_dir / "creative" / "get_creative_delivery_response.py"
+    target.parent.mkdir(parents=True)
+    target.write_text(
+        "from typing import Annotated\n"
+        "from pydantic import Field\n"
+        "from ..core import canonical_format_kind\n\n"
+        "class Creative:\n"
+        "    format_kind: Annotated[\n"
+        "        canonical_format_kind.CanonicalFormatKind | None,\n"
+        "        Field(description='Canonical format kind delivered for this creative.'),\n"
+        "    ] = None\n"
+    )
+    monkeypatch.setattr(post_generate_fixes, "OUTPUT_DIR", generated_dir)
+
+    post_generate_fixes.preserve_open_delivery_format_kind()
+    generated_source = target.read_text()
+    post_generate_fixes.preserve_open_delivery_format_kind()
+
+    assert target.read_text() == generated_source
+    assert "CanonicalFormatKind | str | None" in generated_source
+    assert "union_mode='left_to_right'" in generated_source
+    compile(generated_source, str(target), "exec")
+
+
 def test_post_generate_sync_creatives_response_arms_match_schema_creative_fields(
     tmp_path, monkeypatch
 ):

tests/test_creative_asset_regression.py

@@ -0,0 +1,21 @@
+"""Regression tests for the public CreativeAsset binding (issue #1141)."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+import adcp.types
+from adcp.types.canonical_creative import CanonicalBoundaryModel
+
+
+def test_creative_asset_is_concrete_canonical_class() -> None:
+    cls = adcp.types.CreativeAsset
+
+    assert isinstance(cls, type), "CreativeAsset must be a class, not a union"
+    assert issubclass(cls, CanonicalBoundaryModel)
+    assert issubclass(cls, BaseModel)
+    assert cls.__name__ == "CreativeAsset"
+    assert cls is not adcp.types.LegacyCreativeAsset
+    assert "format_id" not in cls.model_fields
+    assert "format_kind" in cls.model_fields
+    assert cls.model_fields["format_kind"].is_required()