From 212364a3babc3342427547d5cb26f985669ba81e Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Tue, 3 Nov 2026 07:42:45 -0500 Subject: [PATCH 1/2] fix(server): preserve valid creatives in lenient legacy batches 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. (cherry picked from commit 2bd7751cb1312efda5aefc4a11fecdfab0933c9e) --- docs/canonical-format-kinds-migration.md | 37 ++ src/adcp/canonical_formats/projection.py | 3 +- src/adcp/decisioning/handler.py | 14 + src/adcp/server/_creative_batch.py | 202 +++++++ src/adcp/server/idempotency/store.py | 16 + src/adcp/server/mcp_tools.py | 88 ++- ...test_mixed_legacy_projection_regression.py | 568 ++++++++++++++++++ 7 files changed, 917 insertions(+), 11 deletions(-) create mode 100644 src/adcp/server/_creative_batch.py create mode 100644 tests/test_mixed_legacy_projection_regression.py diff --git a/docs/canonical-format-kinds-migration.md b/docs/canonical-format-kinds-migration.md index 2eef9268af..2608ea96c9 100644 --- a/docs/canonical-format-kinds-migration.md +++ b/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. diff --git a/src/adcp/canonical_formats/projection.py b/src/adcp/canonical_formats/projection.py index 3bf9d382eb..91c6790231 100644 --- a/src/adcp/canonical_formats/projection.py +++ b/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( diff --git a/src/adcp/decisioning/handler.py b/src/adcp/decisioning/handler.py index 707b687043..d72ddfcff8 100644 --- a/src/adcp/decisioning/handler.py +++ b/src/adcp/decisioning/handler.py @@ -3202,6 +3202,10 @@ async def _merge_overlay_hook(_update_result: Any) -> None: await self._reporting_subscription_write(account.id, tool_ctx) 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, @@ -3210,6 +3214,14 @@ 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", @@ -3217,6 +3229,8 @@ async def sync_creatives( # type: ignore[override] 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) diff --git a/src/adcp/server/_creative_batch.py b/src/adcp/server/_creative_batch.py new file mode 100644 index 0000000000..c30b19f349 --- /dev/null +++ b/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)) diff --git a/src/adcp/server/idempotency/store.py b/src/adcp/server/idempotency/store.py index fb0185eb5b..0a23eae29c 100644 --- a/src/adcp/server/idempotency/store.py +++ b/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: diff --git a/src/adcp/server/mcp_tools.py b/src/adcp/server/mcp_tools.py index a68ddc4959..c1a25b77d6 100644 --- a/src/adcp/server/mcp_tools.py +++ b/src/adcp/server/mcp_tools.py @@ -3584,7 +3584,12 @@ def create_tool_caller( log_conversion_summary, ) from adcp.exceptions import ADCPFeatureUnsupportedError, ADCPTaskError + from adcp.server._creative_batch import ( + normalize_lenient_creative_batch, + partial_creative_batch, + ) from adcp.server.helpers import inject_context + from adcp.server.idempotency.store import _CREATIVE_BATCH_HASH_SOURCE from adcp.server.version_policy import project_supported_versions, resolve_supported_versions from adcp.types import Error from adcp.validation.envelope import ( @@ -4057,6 +4062,22 @@ async def _call_tool( post_adapter_validator_version = application_version ctx.metadata[CONVERSION_METADATA_KEY] = {"wire_version": wire_version, "losses": []} + if ( + method_name == "sync_creatives" + and unwrap_enum_value(params.get("validation_mode")) == "lenient" + and params.get("assignment_operations") + ): + raise ADCPTaskError( + operation=method_name, + errors=[ + Error( + code="INVALID_REQUEST", + message="assignment_operations requires strict validation", + field="validation_mode", + ) + ], + ) + creative_boundary_tools = { "create_media_buy", "get_creative_delivery", @@ -4068,6 +4089,7 @@ async def _call_tool( "update_media_buy", } legacy_projection_sources: list[Any] = [] + creative_batch = None if method_name in creative_boundary_tools and isinstance(params, dict) and wire_version: try: dialect = ( @@ -4080,11 +4102,40 @@ async def _call_tool( ) ) if dialect is CreativeDialect.LEGACY: - params = normalize_legacy_creative_request( - params, - legacy_format_converter=getattr(handler, "legacy_format_converter", None), - projection_sources=legacy_projection_sources, - ) + if ( + method_name == "sync_creatives" + and unwrap_enum_value(params.get("validation_mode")) == "lenient" + ): + params, creative_batch = normalize_lenient_creative_batch( + params, + legacy_format_converter=getattr( + handler, "legacy_format_converter", None + ), + projection_sources=legacy_projection_sources, + ) + if ( + creative_batch is not None + and params.get("creatives") + and not getattr(handler, "_supports_partial_creative_completion", False) + and ( + params.get("push_notification_config") is not None + or type(handler).get_task_status is not ADCPHandler.get_task_status + or type(handler).list_tasks is not ADCPHandler.list_tasks + ) + ): + raise LegacyCreativeProjectionError( + "Partially projected batches require inline sync_creatives " + "completion or SDK-managed PlatformHandler tasks; custom " + "asynchronous completion cannot retain projection failures" + ) + else: + params = normalize_legacy_creative_request( + params, + legacy_format_converter=getattr( + handler, "legacy_format_converter", None + ), + projection_sources=legacy_projection_sources, + ) # The normalized object is canonical handler input, not a # valid instance of the caller's legacy wire schema. post_adapter_validator_version = None @@ -4116,7 +4167,8 @@ async def _call_tool( params_model=params_model, ) - if request_mode is not None and request_mode != "off": + all_creatives_rejected = creative_batch is not None and not params.get("creatives") + if request_mode is not None and request_mode != "off" and not all_creatives_rejected: outcome = validate_request(method_name, params, version=post_adapter_validator_version) if not outcome.valid: summary = format_issues(outcome.issues) @@ -4139,7 +4191,7 @@ async def _call_tool( call_params = raw_params elif frozen_reporting_feed: call_params = params - elif params_model is not None and isinstance(params, dict): + elif params_model is not None and isinstance(params, dict) and not all_creatives_rejected: try: call_params = params_model.model_validate(params) except ValidationError as exc: @@ -4199,9 +4251,25 @@ async def _call_tool( ) ], ) from exc - if dispatched is not None: - dispatched[0] = True - result = await method(call_params, ctx) + result: Any + token = partial_creative_batch.set(creative_batch) + hash_token = _CREATIVE_BATCH_HASH_SOURCE.set( + {**raw_params, "idempotency_key": params.get("idempotency_key")} + if creative_batch is not None + else None + ) + try: + if all_creatives_rejected: + result = {"creatives": [], "dry_run": bool(params.get("dry_run"))} + else: + if dispatched is not None: + dispatched[0] = True + result = await method(call_params, ctx) + finally: + _CREATIVE_BATCH_HASH_SOURCE.reset(hash_token) + partial_creative_batch.reset(token) + if creative_batch is not None: + creative_batch.merge(result) if method_name == "list_account_changes": result = _project_account_change_feed(result, ctx, application_version) if method_name == "get_adcp_capabilities": diff --git a/tests/test_mixed_legacy_projection_regression.py b/tests/test_mixed_legacy_projection_regression.py new file mode 100644 index 0000000000..c7015e1748 --- /dev/null +++ b/tests/test_mixed_legacy_projection_regression.py @@ -0,0 +1,568 @@ +"""Integrator regression: one projection error must not reject valid siblings.""" + +from __future__ import annotations + +import asyncio +from concurrent.futures import ThreadPoolExecutor +from copy import deepcopy +from typing import Any + +import pytest + +from adcp.exceptions import ADCPTaskError +from adcp.server.base import ADCPHandler, ToolContext +from adcp.server.mcp_tools import create_tool_caller +from adcp.validation import ValidationHookConfig + + +class Seller(ADCPHandler[Any]): + adcp_capabilities = {"media_buy": {"features": {"canonical_creatives": False}}} + + def __init__(self) -> None: + self.received: list[dict[str, Any]] = [] + + async def sync_creatives(self, params: dict[str, Any], context: ToolContext) -> dict[str, Any]: + self.received.append(params) + return { + "creatives": [ + {"creative_id": creative["creative_id"], "action": "created", "status": "approved"} + for creative in params["creatives"] + ] + } + + +def batch() -> dict[str, Any]: + return { + "adcp_version": "3.1", + "idempotency_key": "mixed-creative-projection-0001", + "account": { + "brand": {"domain": "acmeoutdoor.example"}, + "operator": "pinnacle-agency.example", + "sandbox": True, + }, + "creatives": [ + { + "creative_id": "unsupported-native", + "name": "Native", + "format_id": { + "agent_url": "https://seller.example/mcp", + "id": "unsupported-native", + }, + "assets": {}, + }, + { + "creative_id": "display-one", + "name": "Display one", + "format_id": { + "agent_url": "https://seller.example/mcp", + "id": "display_300x250_image", + "width": 300, + "height": 250, + }, + "assets": {}, + }, + { + "creative_id": "display-two", + "name": "Display two", + "format_id": { + "agent_url": "https://seller.example/mcp", + "id": "display_300x250_image", + "width": 300, + "height": 250, + }, + "assets": {}, + }, + ], + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", [None, "strict"]) +async def test_strict_projection_rejects_the_batch_before_the_seller(mode: str | None) -> None: + seller = Seller() + request = batch() + if mode is not None: + request["validation_mode"] = mode + with pytest.raises(ADCPTaskError) as raised: + await create_tool_caller(seller, "sync_creatives")(request) + assert seller.received == [] + assert raised.value.errors[0].code == "INVALID_REQUEST" + assert ( + raised.value.errors[0].message + == "request.creatives[0].format_id cannot be projected (no_match)" + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("strict_validation", [False, True]) +@pytest.mark.parametrize("version", ["3.0", "3.1"]) +async def test_lenient_mixed_batch_accepts_two_creatives_and_fails_one( + strict_validation: bool, + version: str, +) -> None: + seller = Seller() + response = await create_tool_caller( + seller, + "sync_creatives", + validation=( + ValidationHookConfig(requests="strict", responses="strict") + if strict_validation + else None + ), + )({**batch(), "adcp_version": version, "validation_mode": "lenient"}) + rows = {creative["creative_id"]: creative for creative in response["creatives"]} + assert rows["unsupported-native"]["action"] == "failed" + assert rows["unsupported-native"]["errors"] + assert rows["display-one"]["action"] == "created" + assert rows["display-two"]["action"] == "created" + assert [row["creative_id"] for row in response["creatives"]] == [ + "unsupported-native", + "display-one", + "display-two", + ] + assert [row["creative_id"] for row in seller.received[0]["creatives"]] == [ + "display-one", + "display-two", + ] + + +@pytest.mark.asyncio +async def test_configured_converter_receives_original_batch_positions() -> None: + seller = Seller() + paths: list[str] = [] + + def converter(context: Any) -> dict[str, Any] | None: + paths.append(context.field) + if context.field == "request.creatives[1].format_id": + return None + return {"format_kind": "image", "params": {"width": 300, "height": 250}} + + seller.legacy_format_converter = converter + request = {**batch(), "validation_mode": "lenient"} + for index, creative in enumerate(request["creatives"]): + creative["format_id"]["id"] = f"publisher-format-{index}" + response = await create_tool_caller(seller, "sync_creatives")(request) + assert paths == [f"request.creatives[{index}].format_id" for index in range(3)] + assert [row["action"] for row in response["creatives"]] == ["created", "failed", "created"] + assert response["creatives"][1]["errors"][0]["field"] == "creatives[1].format_id" + assert response["creatives"][1]["errors"][0]["code"] == "UNSUPPORTED_FEATURE" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mutation", ["account", "creative_id", "duplicate", "format_id"]) +async def test_malformed_lenient_request_never_reaches_seller(mutation: str) -> None: + seller = Seller() + request = {**batch(), "validation_mode": "lenient"} + if mutation == "account": + request["account"] = {"sandbox": True} + elif mutation == "creative_id": + request["creatives"][0].pop("creative_id") + elif mutation == "duplicate": + request["creatives"][0]["creative_id"] = "display-one" + else: + request["creatives"][0]["format_id"] = {"id": "missing-owner"} + with pytest.raises(ADCPTaskError) as caught: + await create_tool_caller(seller, "sync_creatives")(request) + assert caught.value.errors[0].code == "INVALID_REQUEST" + assert seller.received == [] + + +@pytest.mark.asyncio +async def test_all_unprojectable_rows_fail_without_empty_handler_dispatch() -> None: + seller = Seller() + request = { + **batch(), + "validation_mode": "lenient", + "dry_run": True, + "context": {"trace": "batch"}, + } + request["creatives"] = request["creatives"][:1] + response = await create_tool_caller( + seller, + "sync_creatives", + validation=ValidationHookConfig(requests="strict", responses="strict"), + )(request) + assert response["creatives"][0]["action"] == "failed" + assert response["dry_run"] is True + assert response["context"] == {"trace": "batch"} + assert seller.received == [] + + +@pytest.mark.asyncio +async def test_delete_missing_fails_before_any_library_reconciliation() -> None: + seller = Seller() + with pytest.raises(ADCPTaskError, match="delete_missing"): + await create_tool_caller(seller, "sync_creatives")( + {**batch(), "validation_mode": "lenient", "delete_missing": True} + ) + assert seller.received == [] + # A fully projectable replacement still receives every submitted ID and + # the original reconciliation flag. No failure silently disables a sweep. + request = {**batch(), "validation_mode": "lenient", "delete_missing": True} + request["creatives"][0]["format_id"] = deepcopy(request["creatives"][1]["format_id"]) + await create_tool_caller(seller, "sync_creatives")(request) + assert seller.received[0]["delete_missing"] is True + assert len(seller.received[0]["creatives"]) == 3 + + +@pytest.mark.asyncio +async def test_assignments_cannot_use_a_rejected_existing_creative() -> None: + seller = Seller() + request = { + **batch(), + "validation_mode": "lenient", + "assignments": [ + {"creative_id": "unsupported-native", "package_id": "package_bad"}, + {"creative_id": "display-one", "package_id": "package_good"}, + ], + } + response = await create_tool_caller(seller, "sync_creatives")(request) + assert seller.received[0]["assignments"] == [ + {"creative_id": "display-one", "package_id": "package_good"} + ] + assert "package_bad" in response["creatives"][0]["assignment_errors"] + + +@pytest.mark.asyncio +async def test_partial_creative_ids_filter_fails_before_mutation() -> None: + seller = Seller() + with pytest.raises(ADCPTaskError, match="creative_ids filtering"): + await create_tool_caller(seller, "sync_creatives")( + {**batch(), "validation_mode": "lenient", "creative_ids": ["display-one"]} + ) + assert seller.received == [] + + +@pytest.mark.asyncio +async def test_all_failed_uploads_do_not_silently_drop_valid_assignment_work() -> None: + seller = Seller() + request = { + **batch(), + "validation_mode": "lenient", + "assignments": [ + {"creative_id": "already-stored", "package_id": "package_existing"}, + ], + } + request["creatives"] = request["creatives"][:1] + with pytest.raises(ADCPTaskError, match="unrelated assignments"): + await create_tool_caller(seller, "sync_creatives")(request) + assert seller.received == [] + + +@pytest.mark.asyncio +async def test_explicit_assignment_operations_remain_strict_and_atomic() -> None: + seller = Seller() + with pytest.raises(ADCPTaskError, match="requires strict"): + await create_tool_caller(seller, "sync_creatives")( + { + **batch(), + "adcp_version": "3.2", + "validation_mode": "lenient", + "assignment_operations": [ + { + "operation": "assign", + "creative_id": "display-one", + "package_id": "package_one", + } + ], + } + ) + assert seller.received == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("change", ["creative_id", "name"]) +async def test_rejected_inputs_remain_in_original_idempotency_identity(change: str) -> None: + from adcp.exceptions import IdempotencyConflictError + from adcp.server import IdempotencyStore, MemoryBackend + + store = IdempotencyStore(MemoryBackend()) + + class IdempotentSeller(Seller): + @store.wrap + async def sync_creatives( + self, params: dict[str, Any], context: ToolContext + ) -> dict[str, Any]: + return await super().sync_creatives(params, context) + + seller = IdempotentSeller() + caller = create_tool_caller(seller, "sync_creatives") + request = {**batch(), "validation_mode": "lenient", "dry_run": True} + first = await caller(request, ToolContext(caller_identity="buyer_one")) + replay = await caller(deepcopy(request), ToolContext(caller_identity="buyer_one")) + assert replay["replayed"] is True + assert replay["creatives"] == first["creatives"] + assert len(seller.received) == 1 + assert seller.received[0]["dry_run"] is True + assert seller.received[0]["idempotency_key"] == request["idempotency_key"] + changed = deepcopy(request) + changed["creatives"][0][change] = "different-rejected-input" + with pytest.raises(IdempotencyConflictError): + await caller(changed, ToolContext(caller_identity="buyer_one")) + assert len(seller.received) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("surface", ["poll", "push"]) +async def test_custom_async_surfaces_are_refused_before_dispatch(surface: str) -> None: + class PollingSeller(Seller): + async def get_task_status(self, params: Any, context: Any = None) -> Any: + return {"task_id": "custom-task"} + + seller = PollingSeller() if surface == "poll" else Seller() + request = {**batch(), "validation_mode": "lenient"} + if surface == "push": + request["push_notification_config"] = { + "url": "https://buyer.example/webhook", + "operation_id": "mixed-batch-op", + } + with pytest.raises(ADCPTaskError, match="custom asynchronous"): + await create_tool_caller(seller, "sync_creatives")(request) + assert seller.received == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("same_key,wrap_outer", [(False, True), (True, True), (False, False)]) +async def test_nested_wrapped_sync_keeps_its_own_request_identity( + same_key: bool, wrap_outer: bool +) -> None: + from adcp.server import IdempotencyStore, MemoryBackend + from adcp.server.idempotency.canonicalize import canonical_json_sha256 + + inner_backend = MemoryBackend() + inner_store = IdempotencyStore(inner_backend) + outer_store = IdempotencyStore(MemoryBackend()) + request = {**batch(), "validation_mode": "lenient"} + child_key = request["idempotency_key"] if same_key else "child-operation-0001" + child_request = {"idempotency_key": child_key, "creatives": [], "child_operation": True} + + class Inner: + @inner_store.wrap + async def sync_creatives(self, params: Any, context: Any) -> Any: + return {"creatives": []} + + class Outer(Seller): + async def sync_creatives( + self, params: dict[str, Any], context: ToolContext + ) -> dict[str, Any]: + await Inner().sync_creatives(child_request, context) + return await super().sync_creatives(params, context) + + class WrappedOuter(Outer): + @outer_store.wrap + async def sync_creatives( + self, params: dict[str, Any], context: ToolContext + ) -> dict[str, Any]: + return await super().sync_creatives(params, context) + + seller = WrappedOuter() if wrap_outer else Outer() + await create_tool_caller(seller, "sync_creatives")( + request, ToolContext(caller_identity="buyer_one") + ) + cached = await inner_backend.get("buyer_one", child_key) + assert cached is not None + assert cached.payload_hash == canonical_json_sha256(child_request) + if not same_key: + assert await inner_backend.get("buyer_one", request["idempotency_key"]) is None + + +@pytest.mark.asyncio +async def test_pre_validation_hook_key_still_deduplicates_partial_batches() -> None: + from adcp.exceptions import IdempotencyConflictError + from adcp.server import IdempotencyStore, MemoryBackend + + store = IdempotencyStore(MemoryBackend()) + + class IdempotentSeller(Seller): + @store.wrap + async def sync_creatives( + self, params: dict[str, Any], context: ToolContext + ) -> dict[str, Any]: + return await super().sync_creatives(params, context) + + seller = IdempotentSeller() + request = {**batch(), "validation_mode": "lenient"} + request.pop("idempotency_key") + caller = create_tool_caller( + seller, + "sync_creatives", + pre_validation_hook=lambda tool, params: { + **params, + "idempotency_key": "hook-provided-key-0001", + }, + ) + await caller(request, ToolContext(caller_identity="buyer_one")) + replay = await caller(deepcopy(request), ToolContext(caller_identity="buyer_one")) + assert replay["replayed"] is True + assert len(seller.received) == 1 + changed = deepcopy(request) + changed["creatives"][0]["name"] = "different rejected content" + with pytest.raises(IdempotencyConflictError): + await caller(changed, ToolContext(caller_identity="buyer_one")) + assert len(seller.received) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("completion", ["inline", "task", "workflow"]) +@pytest.mark.parametrize("typed", [False, True]) +async def test_managed_completion_preserves_rejection_rows(completion: str, typed: bool) -> None: + from adcp.decisioning import ( + AdcpError, + DecisioningCapabilities, + DecisioningPlatform, + InMemoryTaskRegistry, + SingletonAccounts, + ) + from adcp.decisioning.handler import PlatformHandler + from adcp.types import SyncCreativesSuccessResponse + + class Platform(DecisioningPlatform): + capabilities = DecisioningCapabilities(specialisms=["sales-non-guaranteed"]) + accounts = SingletonAccounts(account_id="acct_one") + + def __init__(self) -> None: + self.uploaded: list[str] = [] + self.workflows: list[str] = [] + + def sync_creatives(self, req: Any, ctx: Any) -> Any: + def finish(task_ctx: Any = None) -> Any: + self.uploaded.extend(creative.creative_id for creative in req.creatives) + result = { + "creatives": [ + { + "creative_id": creative.creative_id, + "action": "created", + "status": "approved", + } + for creative in reversed(req.creatives) + ] + } + return SyncCreativesSuccessResponse(**result) if typed else result + + if completion == "task": + return ctx.handoff_to_task(finish) + if completion == "workflow": + return ctx.handoff_to_workflow(lambda task_ctx: self.workflows.append(task_ctx.id)) + return finish() + + platform = Platform() + registry = InMemoryTaskRegistry() + with ThreadPoolExecutor(max_workers=2) as executor: + handler = PlatformHandler(platform, executor=executor, registry=registry) + handler.adcp_capabilities = Seller.adcp_capabilities + caller = create_tool_caller(handler, "sync_creatives") + request = {**batch(), "account": {"account_id": "acct_one"}, "validation_mode": "lenient"} + if completion == "workflow": + with pytest.raises(AdcpError, match="cannot use WorkflowHandoff"): + await caller(request) + assert platform.workflows == [] + assert platform.uploaded == [] + assert registry._records == {} + return + response = await caller(request) + if completion == "task": + assert response["status"] == "submitted" + for _ in range(100): + record = await registry.get( + response["task_id"], expected_account_id="acct_one:anonymous" + ) + if record is not None and record["state"] in {"completed", "failed"}: + break + await asyncio.sleep(0.01) + assert record is not None + assert record["state"] == "completed", record + response = record["result"] + assert platform.uploaded == ["display-one", "display-two"] + assert [row["creative_id"] for row in response["creatives"]] == [ + "unsupported-native", + "display-one", + "display-two", + ] + assert response["creatives"][0]["action"] == "failed" + assert [row["action"] for row in response["creatives"][1:]] == ["created", "created"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("package_id", ["package.with.dot", ""]) +@pytest.mark.parametrize("handoff", [False, True], ids=["inline", "task"]) +async def test_typed_assignment_failures_keep_all_wire_valid_package_ids( + package_id: str, handoff: bool +) -> None: + from adcp.decisioning import ( + DecisioningCapabilities, + DecisioningPlatform, + InMemoryTaskRegistry, + SingletonAccounts, + ) + from adcp.decisioning.handler import PlatformHandler + from adcp.server import IdempotencyStore, MemoryBackend + from adcp.types import SyncCreativesSuccessResponse + from adcp.validation import validate_response + + store = IdempotencyStore(MemoryBackend()) + + class Platform(DecisioningPlatform): + capabilities = DecisioningCapabilities( + specialisms=["sales-non-guaranteed"], supported_protocols=["media_buy"] + ) + accounts = SingletonAccounts(account_id="assignment_test") + + def __init__(self) -> None: + self.uploaded: list[str] = [] + self.response: Any = None + self.original_rows: list[Any] = [] + + @store.wrap + async def sync_creatives(self, req: Any, ctx: Any) -> Any: + def finish(task_ctx: Any = None) -> Any: + self.uploaded.extend(row.creative_id for row in req.creatives) + assert req.assignments is None + self.response = SyncCreativesSuccessResponse( + creatives=[ + {"creative_id": row.creative_id, "action": "created", "status": "approved"} + for row in req.creatives + ] + ) + self.original_rows = list(self.response.creatives) + return self.response + + return ctx.handoff_to_task(finish) if handoff else finish() + + platform = Platform() + registry = InMemoryTaskRegistry() + with ThreadPoolExecutor(max_workers=2) as executor: + handler = PlatformHandler(platform, executor=executor, registry=registry) + handler.adcp_capabilities = Seller.adcp_capabilities + request = { + **batch(), + "validation_mode": "lenient", + "account": {"account_id": "assignment_test"}, + "assignments": [{"creative_id": "unsupported-native", "package_id": package_id}], + } + caller = create_tool_caller( + handler, + "sync_creatives", + validation=ValidationHookConfig(requests="strict", responses="strict"), + ) + response = await caller(request) + if handoff: + for _ in range(100): + record = await registry.get( + response["task_id"], expected_account_id="assignment_test:anonymous" + ) + if record is not None and record["state"] in {"completed", "failed"}: + break + await asyncio.sleep(0.01) + assert record is not None and record["state"] == "completed", record + response = record["result"] + assert validate_response("sync_creatives", response, version="3.1").valid + assert platform.uploaded == ["display-one", "display-two"] + assert response["creatives"][0]["action"] == "failed" + assert package_id in response["creatives"][0]["assignment_errors"] + assert platform.response.creatives[1] is platform.original_rows[0] + assert platform.response.creatives[2] is platform.original_rows[1] + if not handoff: + replay = await caller(deepcopy(request)) + assert replay["replayed"] is True + assert replay["creatives"] == response["creatives"] + assert platform.uploaded == ["display-one", "display-two"] From 8f53b65ac10cd67d35d85f420f4c5f70f0f4d430 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Tue, 3 Nov 2026 12:52:04 +0000 Subject: [PATCH 2/2] fix(types): preserve JOSE boundary types across Python targets --- src/adcp/governance.py | 13 +++++++++---- src/adcp/signing/crypto.py | 5 ++++- src/adcp/signing/publisher_pins.py | 5 ++++- 3 files changed, 17 insertions(+), 6 deletions(-) diff --git a/src/adcp/governance.py b/src/adcp/governance.py index aae5bad3f7..d259f22782 100644 --- a/src/adcp/governance.py +++ b/src/adcp/governance.py @@ -1683,11 +1683,16 @@ def _jose_key(alg: str, key: Any) -> OKPKey | ECKey: Public keys go through ``crypto.import_jwk``, which also enforces the RFC 7518 P-256 coordinate length joserfc leaves open. """ + # joserfc's Python 3.11 Self return type must also support our 3.10 + # static contract. Both factories preserve their declared key class. + imported: OKPKey | ECKey if alg == "EdDSA": - return OKPKey.import_key(key) - if alg == "ES256": - return ECKey.import_key(key) - raise ValueError(f"unsupported governance alg {alg!r}") + imported = OKPKey.import_key(key) + elif alg == "ES256": + imported = ECKey.import_key(key) + else: + raise ValueError(f"unsupported governance alg {alg!r}") + return imported def _public_jwk(jwk: Mapping[str, Any]) -> dict[str, Any]: diff --git a/src/adcp/signing/crypto.py b/src/adcp/signing/crypto.py index cfa3186f9f..e0650b96e2 100644 --- a/src/adcp/signing/crypto.py +++ b/src/adcp/signing/crypto.py @@ -126,7 +126,10 @@ def import_jwk(jwk: dict[str, Any], *, private: bool = False) -> OKPKey | ECKey: member = jwk.get(name) if not isinstance(member, str) or len(b64url_decode(member)) != _P256_MEMBER_BYTES: raise ValueError(f"P-256 JWK {name} must be {_P256_MEMBER_BYTES} bytes") - return key_type.import_key(jwk) + # joserfc 1.8 uses Python 3.11 Self; retain the SDK's Python 3.10 + # static contract at this library boundary. + imported: OKPKey | ECKey = key_type.import_key(jwk) + return imported except (JoseError, ValueError, TypeError, KeyError) as exc: raise ValueError(f"unusable {kty}/{crv} JWK: {exc}") from exc diff --git a/src/adcp/signing/publisher_pins.py b/src/adcp/signing/publisher_pins.py index e8d0145488..f92a2a3eb3 100644 --- a/src/adcp/signing/publisher_pins.py +++ b/src/adcp/signing/publisher_pins.py @@ -35,7 +35,10 @@ def jwk_thumbprint(key: Mapping[str, Any]) -> str | None: if member.required and name in key } try: - return key_type.import_key({"kty": key["kty"], **public}).thumbprint() + # Keep the boundary explicit when joserfc's Self annotation is + # unavailable to the SDK's Python 3.10 type-check target. + thumbprint: str = key_type.import_key({"kty": key["kty"], **public}).thumbprint() + return thumbprint except (JoseError, ValueError, TypeError): return None