Skip to content

Commit 2bd7751

Browse files
authored
fix(server): preserve valid creatives in lenient legacy batches (#1841)
Honor explicit lenient processing for legacy creative format projection: dispatch supported siblings and return ordered per-creative failures for unprojectable formats. Preserve strict atomic rejection, original idempotency identity, rejected-assignment safety, and SDK-managed task completion results. Materialize SDK-owned failures before mutation and preserve accepted typed rows, including schema-valid assignment error keys. Unsafe destructive/filter/external-async combinations fail before dispatch and are documented. Validated with 18,863 passing repository tests and 92.16% coverage, 110 focused tests, all GitHub checks, installed-wheel probes, and independent protocol and Ladon approvals. Fixes #1839.
1 parent 536b8f8 commit 2bd7751

7 files changed

Lines changed: 915 additions & 9 deletions

File tree

‎docs/canonical-format-kinds-migration.md‎

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

‎src/adcp/canonical_formats/projection.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -662,6 +662,7 @@ def normalize_legacy_creative_request(
662662
*,
663663
legacy_format_converter: LegacyFormatConverter | None = None,
664664
projection_sources: list[Any] | None = None,
665+
field_path: str = "request",
665666
) -> dict[str, Any]:
666667
"""Upgrade legacy selectors before a primary server handler runs.
667668
@@ -766,7 +767,7 @@ def visit(item: Any, field_path: str) -> Any:
766767
}
767768
return result
768769

769-
return cast(dict[str, Any], visit(value, "request"))
770+
return cast(dict[str, Any], visit(value, field_path))
770771

771772

772773
def resolve_legacy_format_refs(

‎src/adcp/decisioning/handler.py‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2879,6 +2879,10 @@ async def _merge_overlay_hook(_update_result: Any) -> None:
28792879
self._maybe_auto_emit_sync_completion("update_media_buy", params, result)
28802880
return cast("UpdateMediaBuySuccessResponse", result)
28812881

2882+
# Partial creative admission is finalized at the managed completion seam,
2883+
# before task persistence, completion webhooks, or the inline response.
2884+
_supports_partial_creative_completion = True
2885+
28822886
async def sync_creatives( # type: ignore[override]
28832887
self,
28842888
params: SyncCreativesRequest,
@@ -2887,13 +2891,23 @@ async def sync_creatives( # type: ignore[override]
28872891
tool_ctx = context or ToolContext()
28882892
account = await self._resolve_account(params.account, tool_ctx, tool_name="sync_creatives")
28892893
ctx = self._build_ctx(tool_ctx, account)
2894+
from adcp.server._creative_batch import partial_creative_batch
2895+
2896+
batch = partial_creative_batch.get()
2897+
2898+
async def finalize_batch(result: Any) -> None:
2899+
if batch is not None:
2900+
batch.merge(result)
2901+
28902902
result = await _invoke_platform_method(
28912903
self._platform,
28922904
"sync_creatives",
28932905
params,
28942906
ctx,
28952907
executor=self._executor,
28962908
registry=self._registry,
2909+
on_complete=finalize_batch if batch is not None else None,
2910+
allow_workflow_handoff=batch is None,
28972911
**await self._handoff_webhook_kwargs(params, tool_ctx),
28982912
)
28992913
self._maybe_auto_emit_sync_completion("sync_creatives", params, result)

‎src/adcp/server/_creative_batch.py‎

Lines changed: 202 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,202 @@
1+
"""Partial admission for explicitly lenient legacy creative uploads.
2+
3+
The dispatcher owns format projection, not the seller's creative library.
4+
Never reconcile a destructive library replacement from a filtered request.
5+
"""
6+
7+
from __future__ import annotations
8+
9+
from collections.abc import Mapping
10+
from contextvars import ContextVar
11+
from copy import deepcopy
12+
from dataclasses import dataclass, field
13+
from typing import Any
14+
15+
from pydantic import ValidationError
16+
17+
from adcp.canonical_formats.projection import (
18+
LegacyCreativeProjectionError,
19+
LegacyFormatConverter,
20+
normalize_legacy_creative_request,
21+
)
22+
23+
24+
@dataclass(frozen=True)
25+
class PartialCreativeBatch:
26+
submitted_ids: tuple[str, ...]
27+
failures: tuple[dict[str, Any], ...]
28+
_typed_failures: dict[str, Any] = field(init=False, repr=False, compare=False)
29+
30+
def __post_init__(self) -> None:
31+
from adcp.types import SyncCreativeResult
32+
33+
# Materialize SDK-owned failures BEFORE dispatch can mutate anything.
34+
# The wire assignment_errors map permits arbitrary package IDs through
35+
# additionalProperties; its generated model over-restricts every key.
36+
# Validate our remaining fields, then retain our known string-valued
37+
# map without applying that extra generated key restriction.
38+
prepared = {}
39+
for failure in self.failures:
40+
row = SyncCreativeResult.model_validate(
41+
{key: value for key, value in failure.items() if key != "assignment_errors"}
42+
)
43+
if "assignment_errors" in failure:
44+
row = row.model_copy(
45+
update={"assignment_errors": deepcopy(failure["assignment_errors"])}
46+
)
47+
prepared[failure["creative_id"]] = row
48+
object.__setattr__(self, "_typed_failures", prepared)
49+
50+
def merge(self, result: Any) -> None:
51+
"""Overlay once per terminal result, before persistence or delivery.
52+
53+
Idempotency replays may already contain our rejection rows. Replacing
54+
them rather than appending makes repeated overlays idempotent too.
55+
Handler request-level errors retain their own terminal error branch.
56+
"""
57+
payload = (
58+
result.model_dump(mode="json", exclude_none=True)
59+
if hasattr(result, "model_dump")
60+
else result
61+
)
62+
if not isinstance(payload, dict) or "creatives" not in payload:
63+
return
64+
rows = payload["creatives"]
65+
if not isinstance(rows, list):
66+
raise ValueError("sync_creatives completion must contain a creatives array")
67+
by_id = {row["creative_id"]: row for row in rows}
68+
if len(by_id) != len(rows):
69+
raise ValueError("sync_creatives completion contains duplicate creative IDs")
70+
by_id.update((row["creative_id"], deepcopy(row)) for row in self.failures)
71+
if any(creative_id not in by_id for creative_id in self.submitted_ids):
72+
raise ValueError("sync_creatives completion omits an admitted creative")
73+
ordered = [by_id[creative_id] for creative_id in self.submitted_ids]
74+
submitted = set(self.submitted_ids)
75+
ordered.extend(row for row in rows if row["creative_id"] not in submitted)
76+
if isinstance(result, dict):
77+
result["creatives"] = ordered
78+
else:
79+
# Preserve the adopter's already-typed rows, including private
80+
# metadata. Rebuilding them from serialized data can reject a
81+
# conforming response or discard same-process projection state.
82+
original = {
83+
row["creative_id"] if isinstance(row, dict) else row.creative_id: row
84+
for row in result.creatives
85+
}
86+
result.creatives = [
87+
(
88+
self._typed_failures[row["creative_id"]].model_copy(deep=True)
89+
if row["creative_id"] in self._typed_failures
90+
else original[row["creative_id"]]
91+
)
92+
for row in ordered
93+
]
94+
95+
96+
partial_creative_batch: ContextVar[PartialCreativeBatch | None] = ContextVar(
97+
"adcp_partial_creative_batch", default=None
98+
)
99+
100+
101+
def normalize_lenient_creative_batch(
102+
value: Mapping[str, Any],
103+
*,
104+
legacy_format_converter: LegacyFormatConverter | None,
105+
projection_sources: list[Any],
106+
) -> tuple[dict[str, Any], PartialCreativeBatch | None]:
107+
"""Validate the whole envelope before isolating unprojectable formats."""
108+
from adcp.types.legacy import LegacySyncCreativesRequest
109+
110+
try:
111+
LegacySyncCreativesRequest.model_validate(value)
112+
except ValidationError as exc:
113+
first = exc.errors(include_input=False, include_context=False, include_url=False)[0]
114+
path = ".".join(str(part) for part in first["loc"])
115+
raise LegacyCreativeProjectionError(
116+
f"Invalid sync_creatives request at {path or 'request'}: {first['msg']}"
117+
) from exc
118+
119+
creatives = value.get("creatives")
120+
if not creatives:
121+
return (
122+
normalize_legacy_creative_request(
123+
value,
124+
legacy_format_converter=legacy_format_converter,
125+
projection_sources=projection_sources,
126+
),
127+
None,
128+
)
129+
submitted_ids = tuple(row["creative_id"] for row in creatives)
130+
if len(set(submitted_ids)) != len(submitted_ids):
131+
raise LegacyCreativeProjectionError("sync_creatives requires unique creative IDs")
132+
133+
normalized = normalize_legacy_creative_request(
134+
{key: child for key, child in value.items() if key != "creatives"},
135+
legacy_format_converter=legacy_format_converter,
136+
projection_sources=projection_sources,
137+
)
138+
admitted: list[dict[str, Any]] = []
139+
failures: list[dict[str, Any]] = []
140+
for index, creative in enumerate(creatives):
141+
creative_sources: list[Any] = []
142+
try:
143+
projected = normalize_legacy_creative_request(
144+
creative,
145+
legacy_format_converter=legacy_format_converter,
146+
projection_sources=creative_sources,
147+
field_path=f"request.creatives[{index}]",
148+
)
149+
except LegacyCreativeProjectionError as exc:
150+
message = str(exc)
151+
failures.append(
152+
{
153+
"creative_id": creative["creative_id"],
154+
"action": "failed",
155+
"errors": [
156+
{
157+
"code": "UNSUPPORTED_FEATURE",
158+
"message": message,
159+
"field": message.split(" cannot be projected", 1)[0].removeprefix(
160+
"request."
161+
),
162+
}
163+
],
164+
}
165+
)
166+
else:
167+
admitted.append(projected)
168+
projection_sources.extend(creative_sources)
169+
normalized["creatives"] = admitted
170+
if not failures:
171+
return normalized, None
172+
if value.get("delete_missing"):
173+
raise LegacyCreativeProjectionError(
174+
"delete_missing cannot reconcile a partially projected creative batch; "
175+
"retry the uploads without delete_missing, then reconcile the complete library"
176+
)
177+
if value.get("creative_ids"):
178+
raise LegacyCreativeProjectionError(
179+
"creative_ids filtering cannot be combined with a partially projected batch; "
180+
"submit only the selected creatives instead"
181+
)
182+
183+
rejected = {row["creative_id"]: row for row in failures}
184+
assignments = []
185+
for assignment in normalized.get("assignments", []):
186+
failure = rejected.get(assignment["creative_id"])
187+
if failure is None:
188+
assignments.append(assignment)
189+
else:
190+
failure.setdefault("assignment_errors", {})[
191+
assignment["package_id"]
192+
] = "Assignment refused because this submitted creative's format cannot be projected"
193+
if assignments:
194+
if not admitted:
195+
raise LegacyCreativeProjectionError(
196+
"A batch with no projectable creatives cannot also apply unrelated assignments; "
197+
"retry the assignments separately"
198+
)
199+
normalized["assignments"] = assignments
200+
else:
201+
normalized.pop("assignments", None)
202+
return normalized, PartialCreativeBatch(submitted_ids, tuple(failures))

‎src/adcp/server/idempotency/store.py‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,9 @@
6363
# also covers middleware whose wrapped callable is named call_next/execute.
6464
_RECEIPT_BATCH_DISPATCH: ContextVar[bool] = ContextVar("receipt_batch_dispatch", default=False)
6565
_FROZEN_FEED_DISPATCH: ContextVar[bool] = ContextVar("frozen_feed_dispatch", default=False)
66+
_CREATIVE_BATCH_HASH_SOURCE: ContextVar[dict[str, Any] | None] = ContextVar(
67+
"creative_batch_hash_source", default=None
68+
)
6669

6770
# Registry of functions returned by IdempotencyStore.wrap. Read by
6871
# adcp.decisioning.validate_idempotency.is_wrapped() to reconcile the
@@ -344,6 +347,19 @@ async def _wrapped(*args: Any, **kwargs: Any) -> Any:
344347
# transparent on the no-key path.
345348
return await handler(*args, **kwargs)
346349

350+
# Only the first matching boundary owns the original ingress
351+
# identity. Consuming this hint prevents nested business calls
352+
# (including same-key calls to another backend) inheriting it.
353+
# Scope/key extraction always uses the actual incoming request,
354+
# including an idempotency key supplied by a pre-validation hook.
355+
creative_source = _CREATIVE_BATCH_HASH_SOURCE.get()
356+
if (
357+
operation == "sync_creatives"
358+
and creative_source is not None
359+
and creative_source.get("idempotency_key") == idempotency_key
360+
):
361+
params_dict = creative_source
362+
_CREATIVE_BATCH_HASH_SOURCE.set(None)
347363
payload_hash = self._hash_fn(params_dict)
348364

349365
def _replay_cached(cached: CachedResponse) -> Any:

0 commit comments

Comments
 (0)