fix(server): preserve valid creatives in lenient legacy batches

docs/canonical-format-kinds-migration.md

@@ -52,3 +52,40 @@ against the selected product's declared `format_options`; use
 Adopter-defined formats can use `custom` with a corresponding declaration's
 `format_shape` and `format_schema`. Choose it when the creative satisfies that
 contract, rather than changing an unknown value to `custom` without validation.
+
+## Lenient legacy creative uploads
+
+For AdCP 3.0/3.1 legacy creative uploads, set `validation_mode: "lenient"`
+on `sync_creatives` when supported creatives should proceed even if another
+creative's `format_id` cannot be projected. The dispatcher passes the supported
+creatives to the seller and returns the unprojectable creatives as
+`action: "failed"` with `UNSUPPORTED_FEATURE` errors naming their original
+input fields. Results retain the submitted creative order. The default mode
+is `strict`; strict and omitted modes continue to reject the entire request
+before the handler runs.
+
+Malformed envelopes and duplicate creative IDs remain request errors. The
+dispatcher preserves `dry_run` and the original idempotency identity, including
+rejected inputs. Deprecated `assignments` targeting rejected submissions are
+refused and reported in those creatives' `assignment_errors`, so a stale stored
+version cannot be assigned instead. `assignment_operations` requires strict
+mode and cannot be partially processed.
+
+A partially projected batch with `delete_missing: true` is refused before the
+seller runs. The dispatcher cannot safely perform library reconciliation on
+the seller's behalf, and removing failed rows would otherwise archive existing
+creatives whose IDs the buyer did submit. Upload without `delete_missing`,
+resolve the rejected formats, then reconcile with the complete submitted ID set.
+Partially projected batches also refuse `creative_ids` filtering; submit the
+selected creatives directly instead. If every upload is unprojectable, unrelated
+valid assignments are refused before dispatch and must be retried separately.
+
+`PlatformHandler` retains projection failures in inline responses and managed
+`TaskHandoff` completion results before task persistence and webhook publication.
+`WorkflowHandoff` is refused before workflow enqueue for these partial batches.
+Custom `ADCPHandler` implementations with push configuration or overridden task
+polling tools are refused before dispatch because the SDK cannot safely overlay
+their external completion results. These checks conservatively identify custom
+async surfaces; they do not make hand-rolled `submitted` responses supported.
+Custom handlers using partial admission must finish inline, or use the managed
+`PlatformHandler` task path.

src/adcp/canonical_formats/projection.py

@@ -662,6 +662,7 @@ def normalize_legacy_creative_request(
     *,
     legacy_format_converter: LegacyFormatConverter | None = None,
     projection_sources: list[Any] | None = None,
+    field_path: str = "request",
 ) -> dict[str, Any]:
     """Upgrade legacy selectors before a primary server handler runs.
 
@@ -766,7 +767,7 @@ def visit(item: Any, field_path: str) -> Any:
                 }
         return result
 
-    return cast(dict[str, Any], visit(value, "request"))
+    return cast(dict[str, Any], visit(value, field_path))
 
 
 def resolve_legacy_format_refs(

src/adcp/decisioning/handler.py

@@ -2879,6 +2879,10 @@ async def _merge_overlay_hook(_update_result: Any) -> None:
         self._maybe_auto_emit_sync_completion("update_media_buy", params, result)
         return cast("UpdateMediaBuySuccessResponse", result)
 
+    # Partial creative admission is finalized at the managed completion seam,
+    # before task persistence, completion webhooks, or the inline response.
+    _supports_partial_creative_completion = True
+
     async def sync_creatives(  # type: ignore[override]
         self,
         params: SyncCreativesRequest,
@@ -2887,13 +2891,23 @@ async def sync_creatives(  # type: ignore[override]
         tool_ctx = context or ToolContext()
         account = await self._resolve_account(params.account, tool_ctx, tool_name="sync_creatives")
         ctx = self._build_ctx(tool_ctx, account)
+        from adcp.server._creative_batch import partial_creative_batch
+
+        batch = partial_creative_batch.get()
+
+        async def finalize_batch(result: Any) -> None:
+            if batch is not None:
+                batch.merge(result)
+
         result = await _invoke_platform_method(
             self._platform,
             "sync_creatives",
             params,
             ctx,
             executor=self._executor,
             registry=self._registry,
+            on_complete=finalize_batch if batch is not None else None,
+            allow_workflow_handoff=batch is None,
             **await self._handoff_webhook_kwargs(params, tool_ctx),
         )
         self._maybe_auto_emit_sync_completion("sync_creatives", params, result)

src/adcp/server/_creative_batch.py

@@ -0,0 +1,202 @@
+"""Partial admission for explicitly lenient legacy creative uploads.
+
+The dispatcher owns format projection, not the seller's creative library.
+Never reconcile a destructive library replacement from a filtered request.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from contextvars import ContextVar
+from copy import deepcopy
+from dataclasses import dataclass, field
+from typing import Any
+
+from pydantic import ValidationError
+
+from adcp.canonical_formats.projection import (
+    LegacyCreativeProjectionError,
+    LegacyFormatConverter,
+    normalize_legacy_creative_request,
+)
+
+
+@dataclass(frozen=True)
+class PartialCreativeBatch:
+    submitted_ids: tuple[str, ...]
+    failures: tuple[dict[str, Any], ...]
+    _typed_failures: dict[str, Any] = field(init=False, repr=False, compare=False)
+
+    def __post_init__(self) -> None:
+        from adcp.types import SyncCreativeResult
+
+        # Materialize SDK-owned failures BEFORE dispatch can mutate anything.
+        # The wire assignment_errors map permits arbitrary package IDs through
+        # additionalProperties; its generated model over-restricts every key.
+        # Validate our remaining fields, then retain our known string-valued
+        # map without applying that extra generated key restriction.
+        prepared = {}
+        for failure in self.failures:
+            row = SyncCreativeResult.model_validate(
+                {key: value for key, value in failure.items() if key != "assignment_errors"}
+            )
+            if "assignment_errors" in failure:
+                row = row.model_copy(
+                    update={"assignment_errors": deepcopy(failure["assignment_errors"])}
+                )
+            prepared[failure["creative_id"]] = row
+        object.__setattr__(self, "_typed_failures", prepared)
+
+    def merge(self, result: Any) -> None:
+        """Overlay once per terminal result, before persistence or delivery.
+
+        Idempotency replays may already contain our rejection rows. Replacing
+        them rather than appending makes repeated overlays idempotent too.
+        Handler request-level errors retain their own terminal error branch.
+        """
+        payload = (
+            result.model_dump(mode="json", exclude_none=True)
+            if hasattr(result, "model_dump")
+            else result
+        )
+        if not isinstance(payload, dict) or "creatives" not in payload:
+            return
+        rows = payload["creatives"]
+        if not isinstance(rows, list):
+            raise ValueError("sync_creatives completion must contain a creatives array")
+        by_id = {row["creative_id"]: row for row in rows}
+        if len(by_id) != len(rows):
+            raise ValueError("sync_creatives completion contains duplicate creative IDs")
+        by_id.update((row["creative_id"], deepcopy(row)) for row in self.failures)
+        if any(creative_id not in by_id for creative_id in self.submitted_ids):
+            raise ValueError("sync_creatives completion omits an admitted creative")
+        ordered = [by_id[creative_id] for creative_id in self.submitted_ids]
+        submitted = set(self.submitted_ids)
+        ordered.extend(row for row in rows if row["creative_id"] not in submitted)
+        if isinstance(result, dict):
+            result["creatives"] = ordered
+        else:
+            # Preserve the adopter's already-typed rows, including private
+            # metadata. Rebuilding them from serialized data can reject a
+            # conforming response or discard same-process projection state.
+            original = {
+                row["creative_id"] if isinstance(row, dict) else row.creative_id: row
+                for row in result.creatives
+            }
+            result.creatives = [
+                (
+                    self._typed_failures[row["creative_id"]].model_copy(deep=True)
+                    if row["creative_id"] in self._typed_failures
+                    else original[row["creative_id"]]
+                )
+                for row in ordered
+            ]
+
+
+partial_creative_batch: ContextVar[PartialCreativeBatch | None] = ContextVar(
+    "adcp_partial_creative_batch", default=None
+)
+
+
+def normalize_lenient_creative_batch(
+    value: Mapping[str, Any],
+    *,
+    legacy_format_converter: LegacyFormatConverter | None,
+    projection_sources: list[Any],
+) -> tuple[dict[str, Any], PartialCreativeBatch | None]:
+    """Validate the whole envelope before isolating unprojectable formats."""
+    from adcp.types.legacy import LegacySyncCreativesRequest
+
+    try:
+        LegacySyncCreativesRequest.model_validate(value)
+    except ValidationError as exc:
+        first = exc.errors(include_input=False, include_context=False, include_url=False)[0]
+        path = ".".join(str(part) for part in first["loc"])
+        raise LegacyCreativeProjectionError(
+            f"Invalid sync_creatives request at {path or 'request'}: {first['msg']}"
+        ) from exc
+
+    creatives = value.get("creatives")
+    if not creatives:
+        return (
+            normalize_legacy_creative_request(
+                value,
+                legacy_format_converter=legacy_format_converter,
+                projection_sources=projection_sources,
+            ),
+            None,
+        )
+    submitted_ids = tuple(row["creative_id"] for row in creatives)
+    if len(set(submitted_ids)) != len(submitted_ids):
+        raise LegacyCreativeProjectionError("sync_creatives requires unique creative IDs")
+
+    normalized = normalize_legacy_creative_request(
+        {key: child for key, child in value.items() if key != "creatives"},
+        legacy_format_converter=legacy_format_converter,
+        projection_sources=projection_sources,
+    )
+    admitted: list[dict[str, Any]] = []
+    failures: list[dict[str, Any]] = []
+    for index, creative in enumerate(creatives):
+        creative_sources: list[Any] = []
+        try:
+            projected = normalize_legacy_creative_request(
+                creative,
+                legacy_format_converter=legacy_format_converter,
+                projection_sources=creative_sources,
+                field_path=f"request.creatives[{index}]",
+            )
+        except LegacyCreativeProjectionError as exc:
+            message = str(exc)
+            failures.append(
+                {
+                    "creative_id": creative["creative_id"],
+                    "action": "failed",
+                    "errors": [
+                        {
+                            "code": "UNSUPPORTED_FEATURE",
+                            "message": message,
+                            "field": message.split(" cannot be projected", 1)[0].removeprefix(
+                                "request."
+                            ),
+                        }
+                    ],
+                }
+            )
+        else:
+            admitted.append(projected)
+            projection_sources.extend(creative_sources)
+    normalized["creatives"] = admitted
+    if not failures:
+        return normalized, None
+    if value.get("delete_missing"):
+        raise LegacyCreativeProjectionError(
+            "delete_missing cannot reconcile a partially projected creative batch; "
+            "retry the uploads without delete_missing, then reconcile the complete library"
+        )
+    if value.get("creative_ids"):
+        raise LegacyCreativeProjectionError(
+            "creative_ids filtering cannot be combined with a partially projected batch; "
+            "submit only the selected creatives instead"
+        )
+
+    rejected = {row["creative_id"]: row for row in failures}
+    assignments = []
+    for assignment in normalized.get("assignments", []):
+        failure = rejected.get(assignment["creative_id"])
+        if failure is None:
+            assignments.append(assignment)
+        else:
+            failure.setdefault("assignment_errors", {})[
+                assignment["package_id"]
+            ] = "Assignment refused because this submitted creative's format cannot be projected"
+    if assignments:
+        if not admitted:
+            raise LegacyCreativeProjectionError(
+                "A batch with no projectable creatives cannot also apply unrelated assignments; "
+                "retry the assignments separately"
+            )
+        normalized["assignments"] = assignments
+    else:
+        normalized.pop("assignments", None)
+    return normalized, PartialCreativeBatch(submitted_ids, tuple(failures))

src/adcp/server/idempotency/store.py

@@ -63,6 +63,9 @@
 # also covers middleware whose wrapped callable is named call_next/execute.
 _RECEIPT_BATCH_DISPATCH: ContextVar[bool] = ContextVar("receipt_batch_dispatch", default=False)
 _FROZEN_FEED_DISPATCH: ContextVar[bool] = ContextVar("frozen_feed_dispatch", default=False)
+_CREATIVE_BATCH_HASH_SOURCE: ContextVar[dict[str, Any] | None] = ContextVar(
+    "creative_batch_hash_source", default=None
+)
 
 # Registry of functions returned by IdempotencyStore.wrap. Read by
 # adcp.decisioning.validate_idempotency.is_wrapped() to reconcile the
@@ -344,6 +347,19 @@ async def _wrapped(*args: Any, **kwargs: Any) -> Any:
                 # transparent on the no-key path.
                 return await handler(*args, **kwargs)
 
+            # Only the first matching boundary owns the original ingress
+            # identity. Consuming this hint prevents nested business calls
+            # (including same-key calls to another backend) inheriting it.
+            # Scope/key extraction always uses the actual incoming request,
+            # including an idempotency key supplied by a pre-validation hook.
+            creative_source = _CREATIVE_BATCH_HASH_SOURCE.get()
+            if (
+                operation == "sync_creatives"
+                and creative_source is not None
+                and creative_source.get("idempotency_key") == idempotency_key
+            ):
+                params_dict = creative_source
+                _CREATIVE_BATCH_HASH_SOURCE.set(None)
             payload_hash = self._hash_fn(params_dict)
 
             def _replay_cached(cached: CachedResponse) -> Any: