Package adcp

async def challenge_webhook_destination(
    *,
    url: str | AnyUrl,
    account_id: str,
    subscriber_id: str,
    sender: WebhookSender | None = None,
    authentication: AdCPBaseModel | Mapping[str, Any] | None = None,
    challenge: str | None = None,
    timeout_seconds: float | None = None,
    policy: WebhookDestinationPolicy | None = None,
    field: str | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> WebhookChallengeResult:
    """Validate and prove control of a durable webhook destination.

    Use before activating a new or changed active
    ``sync_accounts.accounts[].notification_configs[]`` entry. Inactive
    configs can be persisted without calling this helper.

    ``authentication`` follows the durable config's legacy auth selector:
    when present, the challenge is sent with Bearer or HMAC-SHA256. When
    omitted, pass an RFC 9421 :class:`WebhookSender`; the helper uses that
    sender's webhook-signing key and the SDK-managed pinned transport.
    """

    error_url = str(url) if isinstance(url, (str, AnyUrl)) else None
    if sender is not None and authentication is not None:
        raise WebhookChallengeError(
            "pass either sender= for RFC 9421 or authentication= for legacy auth, not both",
            reason="ambiguous_auth_mode",
            field=field,
            url=error_url,
        )
    sender_owns_client = bool(getattr(cast(Any, sender), "_owns_client", False))
    sender_transport_hooks = tuple(getattr(cast(Any, sender), "_transport_hooks", ()))
    if sender is not None and not sender_owns_client:
        raise WebhookChallengeError(
            "proof-of-control requires a WebhookSender constructed without client=",
            reason="unsafe_sender_client",
            field=field,
            url=error_url,
        )
    if sender is not None and sender_transport_hooks:
        raise WebhookChallengeError(
            "proof-of-control does not support sender transport_hooks",
            reason="unsupported_sender_hooks",
            field=field,
            url=error_url,
        )
    if sender is not None and not sender.signs_with_rfc9421:
        raise WebhookChallengeError(
            "proof-of-control requires an RFC 9421 WebhookSender when authentication is omitted",
            reason="sender_auth_mode_mismatch",
            field=field,
            url=error_url,
            suggestion=(
                "Use WebhookSender.from_jwk(...) for default durable configs, "
                "or pass config.authentication for legacy Bearer/HMAC configs."
            ),
        )
    if sender is None and authentication is None:
        raise WebhookChallengeError(
            "webhook challenge requires sender= when authentication is omitted",
            reason="sender_required",
            field=field,
            url=error_url,
            suggestion=(
                "Pass the seller's WebhookSender, or pass config.authentication " "for legacy auth."
            ),
        )
    try:
        destination = validate_webhook_destination_url(url, policy=policy, field=field)
        payload = create_webhook_challenge_payload(
            account_id=account_id,
            subscriber_id=subscriber_id,
            challenge=challenge,
        )
    except WebhookDestinationValidationError as exc:
        raise WebhookChallengeError(
            str(exc),
            reason=exc.reason,
            field=exc.field,
            url=exc.url,
            suggestion=exc.suggestion,
        ) from exc
    except ValueError as exc:
        raise WebhookChallengeError(
            f"webhook challenge configuration is invalid: {exc}",
            reason="invalid_configuration",
            field=field,
            url=error_url,
        ) from exc
    challenge_value = payload["challenge"]

    try:
        if sender is not None:
            effective_timeout = (
                timeout_seconds
                if timeout_seconds is not None
                else float(getattr(cast(Any, sender), "_timeout", _DEFAULT_TIMEOUT_SECONDS))
            )
            response = await _send_sender_webhook_challenge(
                url=destination.effective_url,
                sender=sender,
                payload=payload,
                timeout_seconds=effective_timeout,
                policy=destination.policy,
                extra_headers=extra_headers,
            )
            status_code = response.status_code
            response_headers = dict(response.headers)
            response_body = response.content
        else:
            auth_config = _authentication_to_config(cast(Any, authentication))
            response = await _send_legacy_webhook_challenge(
                url=destination.effective_url,
                authentication=auth_config,
                payload=payload,
                extra_headers=extra_headers,
                timeout_seconds=timeout_seconds,
                policy=destination.policy,
            )
            status_code = response.status_code
            response_headers = dict(response.headers)
            response_body = response.content
    except httpx.TimeoutException as exc:
        raise WebhookChallengeError(
            "webhook challenge timed out",
            reason="timeout",
            field=field,
            url=destination.original_url,
        ) from exc
    except httpx.HTTPError as exc:
        raise WebhookChallengeError(
            f"webhook challenge request failed: {exc}",
            reason="request_failed",
            field=field,
            url=destination.original_url,
        ) from exc
    except ValueError as exc:
        raise WebhookChallengeError(
            f"webhook challenge configuration is invalid: {exc}",
            reason="invalid_configuration",
            field=field,
            url=destination.original_url,
        ) from exc

    if not 200 <= status_code < 300:
        raise WebhookChallengeError(
            f"webhook challenge failed with HTTP {status_code}",
            reason="http_status",
            field=field,
            url=destination.original_url,
            status_code=status_code,
        )

    echoed_field = validate_webhook_challenge_response(
        response_body,
        challenge=challenge_value,
        field=field,
        url=destination.original_url,
    )
    return WebhookChallengeResult(
        challenge=challenge_value,
        echoed_field=echoed_field,
        destination=destination,
        status_code=status_code,
        response_headers=response_headers,
        response_body=response_body,
    )

def create_a2a_webhook_payload(
    task_id: str,
    status: GeneratedTaskStatus,
    context_id: str,
    result: PydanticBaseModel | dict[str, Any],
    timestamp: datetime | None = None,
) -> Task | TaskStatusUpdateEvent:
    """
    Create A2A webhook payload (Task or TaskStatusUpdateEvent).

    Per A2A specification:
    - Terminated statuses (completed, failed, canceled, rejected): Returns Task
      with artifacts[].parts[]
    - Intermediate statuses (working, input-required, submitted, auth-required):
      Returns TaskStatusUpdateEvent with status.message.parts[]

    This function helps agent implementations construct properly formatted A2A webhook
    payloads for sending to clients.

    Args:
        task_id: Unique identifier for the task
        status: Current task status
        context_id: Session/conversation identifier (required by A2A protocol)
        timestamp: When the webhook was generated (defaults to current UTC time)
        result: Task-specific payload — any Pydantic model or plain dict

    Returns:
        Task object for terminated statuses, TaskStatusUpdateEvent for intermediate statuses

    Examples:
        Create a completed Task webhook:
        >>> from adcp.webhooks import create_a2a_webhook_payload
        >>> from adcp.types import GeneratedTaskStatus
        >>>
        >>> task = create_a2a_webhook_payload(
        ...     task_id="task_123",
        ...     context_id="ctx_123",
        ...     status=GeneratedTaskStatus.completed,
        ...     result={"products": [...]},
        ... )
        >>> # task is a Task object with artifacts containing the result

        Create a working status update:
        >>> event = create_a2a_webhook_payload(
        ...     task_id="task_456",
        ...     context_id="ctx_456",
        ...     status=GeneratedTaskStatus.working,
        ...     result={"current_step": "processing", "percentage": 30},
        ... )
        >>> # event is a TaskStatusUpdateEvent with status.message

        Send A2A webhook via HTTP POST:
        >>> import httpx
        >>> from a2a.types import Task
        >>>
        >>> payload = create_a2a_webhook_payload(...)
        >>> # Serialize to dict for JSON
        >>> if isinstance(payload, Task):
        ...     payload_dict = payload.model_dump(mode='json')
        ... else:
        ...     payload_dict = payload.model_dump(mode='json')
        >>>
        >>> response = await httpx.post(webhook_url, json=payload_dict)
    """
    if timestamp is None:
        timestamp = datetime.now(timezone.utc)

    # Convert datetime to ISO string for A2A protocol
    timestamp_str = timestamp.isoformat() if isinstance(timestamp, datetime) else timestamp
    timestamp_proto = _isoformat_to_proto_timestamp(timestamp_str) if timestamp_str else None

    # Map GeneratedTaskStatus to A2A TaskState enum value.
    # GeneratedTaskStatus is always an Enum so .value is guaranteed.
    status_value = status.value
    adcp_to_task_state: dict[str, int] = {
        "completed": pb.TaskState.TASK_STATE_COMPLETED,
        "failed": pb.TaskState.TASK_STATE_FAILED,
        "canceled": pb.TaskState.TASK_STATE_CANCELED,
        "rejected": pb.TaskState.TASK_STATE_REJECTED,
        "working": pb.TaskState.TASK_STATE_WORKING,
        "submitted": pb.TaskState.TASK_STATE_SUBMITTED,
        # GeneratedTaskStatus enum values are hyphenated ("input-required",
        # "auth-required"). The underscore forms are accepted as a convenience
        # for callers passing raw strings rather than enum members.
        "input_required": pb.TaskState.TASK_STATE_INPUT_REQUIRED,
        "input-required": pb.TaskState.TASK_STATE_INPUT_REQUIRED,
        "auth_required": pb.TaskState.TASK_STATE_AUTH_REQUIRED,
        "auth-required": pb.TaskState.TASK_STATE_AUTH_REQUIRED,
    }
    task_state_enum = adcp_to_task_state.get(status_value)
    if task_state_enum is None:
        # Falling back to TASK_STATE_UNSPECIFIED (proto3 zero) would be
        # silently omitted by MessageToDict, producing an invalid wire
        # shape ``{"status": {}}`` that A2A v0.3 receivers reject as
        # missing the required ``state`` field. Fail loud at the builder
        # boundary so callers can't ship a broken envelope.
        known = [
            "submitted",
            "working",
            "input-required",
            "completed",
            "canceled",
            "failed",
            "rejected",
            "auth-required",
        ]
        raise ValueError(
            f"create_a2a_webhook_payload: unknown status {status_value!r}. "
            f"Known AdCP→A2A states: {known}. "
            "Note: 'unknown' has no a2a-sdk 1.0 protobuf constant; build a "
            "Task manually and pass it through to_wire_dict if you need to "
            "emit that state."
        )

    # Build parts for the message/artifact.
    parts: list[pb.Part] = []

    # Convert Pydantic model to dict if needed
    if hasattr(result, "model_dump"):
        result_dict: dict[str, Any] = result.model_dump(mode="json")
    else:
        result_dict = result

    value = Value()
    ParseDict(result_dict, value)
    parts.append(pb.Part(data=value))

    # Determine if this is a terminated status (Task) or intermediate (TaskStatusUpdateEvent).
    # canceled and rejected are terminal: the task will not continue.
    is_terminated = status in (
        GeneratedTaskStatus.completed,
        GeneratedTaskStatus.failed,
        GeneratedTaskStatus.canceled,
        GeneratedTaskStatus.rejected,
    )

    if is_terminated:
        status_kwargs: dict[str, Any] = {"state": task_state_enum}
        if timestamp_proto is not None:
            status_kwargs["timestamp"] = timestamp_proto
        task_status = pb.TaskStatus(**status_kwargs)

        artifacts = (
            [
                pb.Artifact(
                    artifact_id=f"{task_id}_result",
                    parts=parts,
                )
            ]
            if parts
            else []
        )

        return pb.Task(
            id=task_id,
            status=task_status,
            artifacts=artifacts,
            context_id=context_id,
        )

    # Intermediate status: build a Message carrying the parts and nest it
    # inside TaskStatus.message so the event mirrors the spec shape.
    message_obj = None
    if parts:
        message_obj = pb.Message(
            message_id=f"{task_id}_msg",
            role=pb.Role.ROLE_AGENT,
            parts=parts,
        )

    status_kwargs = {"state": task_state_enum}
    if timestamp_proto is not None:
        status_kwargs["timestamp"] = timestamp_proto
    if message_obj is not None:
        status_kwargs["message"] = message_obj
    task_status = pb.TaskStatus(**status_kwargs)

    return pb.TaskStatusUpdateEvent(
        task_id=task_id,
        status=task_status,
        context_id=context_id,
    )

def create_mcp_webhook_payload(
    task_id: str,
    status: GeneratedTaskStatus | str,
    task_type: TaskType | str,
    *,
    result: PydanticBaseModel | dict[str, Any] | None = None,
    timestamp: datetime | None = None,
    operation_id: str,
    notification_id: str | None = None,
    message: str | None = None,
    context_id: str | None = None,
    protocol: AdcpProtocol | str | None = None,
    idempotency_key: str | None = None,
    token: str | None = None,
) -> McpWebhookPayload:
    """
    Build an :class:`McpWebhookPayload` for a tracked async task.

    Pair with :func:`to_wire_dict` for HTTP transport — Pydantic-typed at
    construction so the publisher catches schema drift before it leaves
    the process.

    ``task_type`` is restricted to the closed :class:`TaskType` enum (the
    spec's complete set of async/tracked operations). Passing a value not
    present in the enum produces a validation error before an invalid webhook
    payload can leave the process.

    Args:
        task_id: Unique identifier for the task.
        status: Current task status.
        task_type: Type of AdCP async operation (see :class:`TaskType`).
        result: Task-specific payload — any Pydantic model or plain dict.
            Plain dicts are validated against
            :class:`AdcpAsyncResponseData`'s discriminated union.
        timestamp: When the webhook was generated. Defaults to current UTC.
        operation_id: Client-generated identifier supplied through
            ``push_notification_config.operation_id``. Required for every
            task webhook; publishers echo it verbatim and MUST NOT derive it
            from the receiver URL.
        notification_id: Stable identity for one logical notification.
            Terminal task webhooks default to ``"{task_id}.terminal"`` so
            re-emissions under different delivery keys still converge.
        message: Human-readable summary of task state.
        context_id: Session/conversation identifier.
        protocol: AdCP protocol this task belongs to (see :class:`AdcpProtocol`).
            Auto-derived from ``task_type`` when omitted, matching the JS
            SDK's ``protocolForTool`` so cross-SDK bodies classify
            operations identically. Pass an explicit value to override.
        idempotency_key: Sender-generated key stable across retries of the
            same event. Defaults to a freshly-generated UUID v4 — callers
            retrying delivery of the same event MUST pass the key from
            their first attempt; passing None twice mints two keys and
            defeats dedup.
        token: Buyer-supplied token from ``push_notification_config.token``,
            echoed back per spec for authenticity validation.

    Returns:
        :class:`McpWebhookPayload` instance. Use :func:`to_wire_dict` (or
        ``payload.model_dump(mode="json", exclude_none=True)``) to get the
        JSON-ready dict for HTTP transport.

    Examples:
        Create a completed webhook with results:
        >>> from adcp.webhooks import create_mcp_webhook_payload, to_wire_dict
        >>> from adcp.types import GeneratedTaskStatus
        >>>
        >>> payload = create_mcp_webhook_payload(
        ...     task_id="task_123",
        ...     operation_id="op_123",
        ...     status=GeneratedTaskStatus.completed,
        ...     task_type="create_media_buy",
        ...     result={"media_buy_id": "mb_1", "buyer_ref": "ref_1"},
        ...     message="Created campaign"
        ... )
        >>> wire = to_wire_dict(payload)

        Create a failed webhook with error:
        >>> payload = create_mcp_webhook_payload(
        ...     task_id="task_456",
        ...     operation_id="op_456",
        ...     status=GeneratedTaskStatus.failed,
        ...     task_type="create_media_buy",
        ...     result={"errors": [{"code": "INVALID_INPUT", "message": "..."}]},
        ...     message="Validation failed"
        ... )

        Create a working status update:
        >>> payload = create_mcp_webhook_payload(
        ...     task_id="task_789",
        ...     operation_id="op_789",
        ...     status=GeneratedTaskStatus.working,
        ...     task_type="sync_creatives",
        ...     message="Processing 3 of 10 creatives"
        ... )
    """
    if timestamp is None:
        timestamp = datetime.now(timezone.utc)
    if idempotency_key is None:
        idempotency_key = generate_webhook_idempotency_key()
    if not operation_id:
        raise ValueError(
            "operation_id is required for AdCP task webhooks; copy "
            "push_notification_config.operation_id verbatim"
        )

    status_value = status.value if hasattr(status, "value") else str(status)
    if notification_id is None and status_value in {
        "completed",
        "failed",
        "canceled",
        "rejected",
    }:
        notification_id = f"{task_id}.terminal"

    # Auto-derive `protocol` from `task_type` when caller doesn't override.
    # Matches `protocolForTool` in the JS reference SDK so cross-SDK bodies
    # classify operations identically.
    if protocol is None:
        try:
            task_type_enum = task_type if isinstance(task_type, TaskType) else TaskType(task_type)
        except ValueError:
            # Unknown string — let `model_validate` raise the canonical
            # task_type error below rather than swallow it here.
            task_type_enum = None
        if task_type_enum is not None:
            protocol = _TASK_TYPE_TO_PROTOCOL.get(task_type_enum)

    # Foreign BaseModel subclasses (anything outside AdcpAsyncResponseData)
    # don't match the discriminated-union variants by identity — dump to a
    # dict so the union picks by shape, matching the dict path.
    result_value: PydanticBaseModel | dict[str, Any] | None
    if isinstance(result, PydanticBaseModel):
        result_value = result.model_dump(mode="json")
    else:
        result_value = result

    payload = McpWebhookPayload.model_validate(
        {
            "idempotency_key": idempotency_key,
            "notification_id": notification_id,
            "task_id": task_id,
            "task_type": task_type,
            "protocol": protocol,
            "status": status_value,
            "timestamp": timestamp,
            "operation_id": operation_id,
            "message": message,
            "context_id": context_id,
            "token": token,
        }
    )
    # Preserve task result payloads byte-for-byte. Validating through the
    # generated AdcpAsyncResponseData union can coerce arbitrary dicts into
    # typed response models and inject response defaults, changing webhook
    # bodies before signing.
    payload.result = result_value  # type: ignore[assignment]
    return payload

def create_test_agent(**overrides: Any) -> AgentConfig:
    """Create a custom test agent configuration.

    Useful when you need to modify the default test agent setup.

    Args:
        **overrides: Keyword arguments to override default config values

    Returns:
        Complete agent configuration

    Example:
        ```python
        from adcp.testing import create_test_agent
        from adcp.client import ADCPClient

        # Use default test agent with custom ID
        config = create_test_agent(id="my-test-agent")
        client = ADCPClient(config)
        ```

    Example:
        ```python
        # Use A2A protocol instead of MCP
        from adcp.types.core import Protocol

        config = create_test_agent(
            protocol=Protocol.A2A,
            agent_uri="https://test-agent.adcontextprotocol.org"
        )
        ```
    """
    base_config = TEST_AGENT_MCP_CONFIG.model_dump()
    base_config.update(overrides)
    return AgentConfig(**base_config)

def create_webhook_challenge_payload(
    *,
    account_id: str,
    subscriber_id: str,
    challenge: str | None = None,
) -> dict[str, str]:
    """Build the durable ``notification_configs[]`` challenge payload."""

    if not isinstance(account_id, str) or not account_id:
        raise ValueError("account_id must be a non-empty string")
    if not isinstance(subscriber_id, str) or not subscriber_id:
        raise ValueError("subscriber_id must be a non-empty string")
    challenge_value = generate_webhook_challenge_value() if challenge is None else challenge
    if not isinstance(challenge_value, str) or not challenge_value:
        raise ValueError("challenge must be a non-empty string")
    return {
        "type": "webhook.challenge",
        "challenge": challenge_value,
        "account_id": account_id,
        "subscriber_id": subscriber_id,
    }

async def detect_publisher_properties_divergence(
    agent_url: str,
    *,
    directory_url: str,
    sample_size: int | None = 200,
    max_concurrency: int = 20,
    timeout: float = 30.0,
    client: httpx.AsyncClient | None = None,
) -> DivergenceReport:
    """Compare directory's inline resolution against per-publisher federated fetches.

    For each publisher the directory lists under ``agent_url``, fetches
    that publisher's own ``adagents.json`` and compares the property set
    against the directory's claim. Returns only publishers where the two
    paths disagree (or where the child fetch failed).

    Always requests ``include=["properties"]`` from the directory so the
    full ``(publisher_domain, property_id)`` set-diff lights up on
    directories that support adcp#4894. Against older directories that
    return only ``properties_authorized`` counts, falls back to count-
    comparison; ``missing_in_inline`` / ``missing_in_federated`` are
    None in that fallback path.

    Per adcp#4827 §Resolution-paths, the federated result is
    authoritative when the two paths disagree.

    Args:
        agent_url: agent to check.
        directory_url: AAO directory base URL (HTTPS only — same SSRF
            gate as :func:`fetch_agent_authorizations_from_directory`).
        sample_size: cap the sweep at N publishers (drawn from the first
            page of directory results). None opts into a full sweep
            across all pages — only do this for small networks. Default
            200 keeps the divergence sweep bounded by default.
        max_concurrency: semaphore-capped concurrent federated fetches.
            Default 20 — caps the burst against publisher origins.
        timeout: per-request timeout (directory + child fetches).
        client: optional shared ``httpx.AsyncClient``.

    Returns:
        :data:`DivergenceReport` (``list[PublisherDivergence]``). Empty
        list = no divergence detected. Note in count-only fallback mode,
        an empty list means counts agree but set-equality is not
        guaranteed.
    """
    http = client
    collected: list[DirectoryPublisherEntry] = []
    cursor: str | None = None
    seen_cursors: set[str] = set()
    page_count = 0
    while True:
        page = await fetch_agent_authorizations_from_directory(
            agent_url,
            directory_url=directory_url,
            cursor=cursor,
            include=["properties"],
            timeout=timeout,
            client=http,
        )
        page_count += 1
        collected.extend(page.publishers)
        if sample_size is not None and len(collected) >= sample_size:
            collected = collected[:sample_size]
            break
        cursor = page.next_cursor
        if not cursor:
            break
        if cursor in seen_cursors:
            raise AdagentsValidationError(
                f"Directory page cursor {cursor!r} repeated — refusing to loop forever."
            )
        seen_cursors.add(cursor)
        if page_count >= MAX_DIRECTORY_PAGES:
            raise AdagentsValidationError(
                f"Directory pagination exceeded {MAX_DIRECTORY_PAGES} pages — aborting sweep."
            )

    # Dedupe by publisher_domain before fan-out: a hostile directory
    # returning N rows for the same publisher would otherwise amplify
    # into N concurrent fetches against a single victim host. First
    # occurrence wins (deterministic) — conflicting property_ids /
    # properties_authorized across duplicates are dropped here; the
    # directory's behavior is itself a divergence signal for ops.
    seen_domains: set[str] = set()
    deduped: list[DirectoryPublisherEntry] = []
    for entry in collected:
        if entry.publisher_domain in seen_domains:
            continue
        seen_domains.add(entry.publisher_domain)
        deduped.append(entry)
    collected = deduped

    # Emit a one-shot warning when the entire sample comes back without
    # property_ids[]. In count-only mode, same-count substitutions are
    # undetectable — adopters should pin include=["properties"] support
    # on directories that offer it.
    if collected and all(e.property_ids is None for e in collected):
        logger.warning(
            "AAO directory %s did not return property_ids[] on any publisher "
            "entry — falling back to count-only divergence detection. Same-count "
            "substitutions are undetectable in this mode. Upgrade the directory "
            "or pin include=['properties'] support.",
            directory_url,
        )

    sem = asyncio.Semaphore(max_concurrency)

    async def _probe(entry: DirectoryPublisherEntry) -> PublisherDivergence | None:
        async with sem:
            try:
                data = await fetch_adagents(entry.publisher_domain, timeout=timeout, client=http)
                federated_props = get_properties_by_agent(data, agent_url)
                # Falsy/empty property_id is silently dropped: upstream
                # schema requires a non-empty string, so an empty value
                # is a structural violation that belongs in
                # validate_adagents, not a divergence signal. Federated
                # properties with valid IDs only.
                federated_ids = {
                    str(p.get("property_id")) for p in federated_props if p.get("property_id")
                }
            except (
                AdagentsNotFoundError,
                AdagentsValidationError,
                AdagentsTimeoutError,
                httpx.HTTPError,
                OSError,
                ValueError,
            ) as exc:
                return PublisherDivergence(
                    publisher_domain=entry.publisher_domain,
                    directory_properties_authorized=entry.properties_authorized,
                    federated_properties_found=0,
                    missing_in_inline=None,
                    missing_in_federated=None,
                    child_fetch_error=str(exc),
                )

        if entry.property_ids is not None:
            # Full set-diff path (adcp#4894).
            dir_ids = set(entry.property_ids)
            missing_in_inline = sorted(federated_ids - dir_ids)
            missing_in_federated = sorted(dir_ids - federated_ids)
            if not missing_in_inline and not missing_in_federated:
                return None
            return PublisherDivergence(
                publisher_domain=entry.publisher_domain,
                directory_properties_authorized=entry.properties_authorized,
                federated_properties_found=len(federated_ids),
                missing_in_inline=missing_in_inline,
                missing_in_federated=missing_in_federated,
            )

        # Count-only fallback (older directories).
        if len(federated_ids) == entry.properties_authorized:
            return None
        return PublisherDivergence(
            publisher_domain=entry.publisher_domain,
            directory_properties_authorized=entry.properties_authorized,
            federated_properties_found=len(federated_ids),
            missing_in_inline=None,
            missing_in_federated=None,
        )

    probes = await asyncio.gather(*[_probe(e) for e in collected])

    return [p for p in probes if p is not None]

def domain_matches(property_domain: str, agent_domain_pattern: str) -> bool:
    """Check if domains match per AdCP rules.

    Rules:
    - Exact match always succeeds
    - 'example.com' matches www.example.com, m.example.com (common subdomains)
    - 'subdomain.example.com' matches that specific subdomain only
    - '*.example.com' matches all subdomains

    Args:
        property_domain: Domain from property
        agent_domain_pattern: Domain pattern from adagents.json

    Returns:
        True if domains match per AdCP rules
    """
    # Normalize both domains for comparison
    try:
        property_domain = _normalize_domain(property_domain)
        agent_domain_pattern = _normalize_domain(agent_domain_pattern)
    except AdagentsValidationError:
        # Invalid domain format - no match
        return False

    # Exact match
    if property_domain == agent_domain_pattern:
        return True

    # Wildcard pattern (*.example.com)
    if agent_domain_pattern.startswith("*."):
        base_domain = agent_domain_pattern[2:]
        return property_domain.endswith(f".{base_domain}")

    # Bare domain matches common subdomains (www, m)
    # If agent pattern is a bare domain (no subdomain), match www/m subdomains
    if "." in agent_domain_pattern and not agent_domain_pattern.startswith("www."):
        # Check if this looks like a bare domain (e.g., example.com)
        parts = agent_domain_pattern.split(".")
        if len(parts) == 2:  # Looks like bare domain
            common_subdomains = ["www", "m"]
            for subdomain in common_subdomains:
                if property_domain == f"{subdomain}.{agent_domain_pattern}":
                    return True

    return False

def encode_unreserved(raw: str) -> str:
    """UTF-8 encode ``raw``, escaping every byte outside RFC 3986 unreserved.

    Percent escapes use uppercase hexadecimal. Unlike
    :func:`urllib.parse.quote`, this helper does not preserve ``/``.
    """

    return "".join(
        chr(byte) if byte in _UNRESERVED_BYTES else f"%{byte:02X}" for byte in raw.encode("utf-8")
    )

def extract_webhook_result_data(webhook_payload: dict[str, Any]) -> dict[str, Any] | None:
    """
    Extract result data from webhook payload (MCP or A2A format).

    This utility function handles webhook payloads from both MCP and A2A protocols,
    extracting the result data regardless of the webhook format. Useful for quick
    inspection, logging, or custom webhook routing logic without requiring full
    client initialization.

    Protocol Detection:
    - A2A Task: Has "artifacts" field (terminated statuses: completed, failed, canceled, rejected)
    - A2A TaskStatusUpdateEvent: Has nested "status.message" structure (intermediate statuses)
    - MCP: Has "result" field directly

    Args:
        webhook_payload: Raw webhook dictionary from HTTP request (JSON-deserialized)

    Returns:
        dict[str, Any] containing the extracted AdCP response data, or None if no
        result is present. For A2A webhooks, unwraps data from artifacts/message parts
        structure. For MCP webhooks, returns the result field directly.

    Examples:
        Extract from MCP webhook:
        >>> mcp_payload = {
        ...     "task_id": "task_123",
        ...     "task_type": "create_media_buy",
        ...     "status": "completed",
        ...     "timestamp": "2025-01-15T10:00:00Z",
        ...     "result": {"media_buy_id": "mb_123", "buyer_ref": "ref_123", "packages": []}
        ... }
        >>> result = extract_webhook_result_data(mcp_payload)
        >>> print(result["media_buy_id"])
        mb_123

        Extract from A2A Task webhook:
        >>> a2a_task_payload = {
        ...     "id": "task_456",
        ...     "context_id": "ctx_456",
        ...     "status": {"state": "completed", "timestamp": "2025-01-15T10:00:00Z"},
        ...     "artifacts": [
        ...         {
        ...             "artifact_id": "artifact_456",
        ...             "parts": [
        ...                 {
        ...                     "data": {
        ...                         "media_buy_id": "mb_456",
        ...                         "buyer_ref": "ref_456",
        ...                         "packages": []
        ...                     }
        ...                 }
        ...             ]
        ...         }
        ...     ]
        ... }
        >>> result = extract_webhook_result_data(a2a_task_payload)
        >>> print(result["media_buy_id"])
        mb_456

        Extract from A2A TaskStatusUpdateEvent webhook:
        >>> a2a_event_payload = {
        ...     "task_id": "task_789",
        ...     "context_id": "ctx_789",
        ...     "status": {
        ...         "state": "working",
        ...         "timestamp": "2025-01-15T10:00:00Z",
        ...         "message": {
        ...             "message_id": "msg_789",
        ...             "role": "agent",
        ...             "parts": [
        ...                 {"data": {"current_step": "processing", "percentage": 50}}
        ...             ]
        ...         }
        ...     },
        ...     "final": False
        ... }
        >>> result = extract_webhook_result_data(a2a_event_payload)
        >>> print(result["percentage"])
        50

        Handle webhook with no result:
        >>> empty_payload = {"task_id": "task_000", "status": "working", "timestamp": "..."}
        >>> result = extract_webhook_result_data(empty_payload)
        >>> print(result)
        None
    """
    # Detect A2A Task format (has "artifacts" field)
    if "artifacts" in webhook_payload:
        # Extract from task.artifacts[].parts[]
        artifacts = webhook_payload.get("artifacts", [])
        if not artifacts:
            return None

        # Use last artifact (most recent)
        target_artifact = artifacts[-1]
        parts = target_artifact.get("parts", [])
        if not parts:
            return None

        # Find DataPart (skip TextPart)
        for part in parts:
            # Check if this part has "data" field (DataPart)
            if "data" in part:
                data = part["data"]
                # Unwrap {"response": {...}} wrapper if present (A2A convention)
                if isinstance(data, dict) and "response" in data and len(data) == 1:
                    return cast(dict[str, Any], data["response"])
                return cast(dict[str, Any], data)

        return None

    # Detect A2A TaskStatusUpdateEvent format (has nested "status.message")
    status = webhook_payload.get("status")
    if isinstance(status, dict):
        message = status.get("message")
        if isinstance(message, dict):
            # Extract from status.message.parts[]
            parts = message.get("parts", [])
            if not parts:
                return None

            # Find DataPart
            for part in parts:
                if "data" in part:
                    data = part["data"]
                    # Unwrap {"response": {...}} wrapper if present
                    if isinstance(data, dict) and "response" in data and len(data) == 1:
                        return cast(dict[str, Any], data["response"])
                    return cast(dict[str, Any], data)

            return None

    # MCP format: result field directly
    return cast(dict[str, Any] | None, webhook_payload.get("result"))

async def fetch_adagents(
    publisher_domain: str,
    timeout: float = 10.0,
    user_agent: str = "AdCP-Client/1.0",
    client: httpx.AsyncClient | None = None,
) -> dict[str, Any]:
    """Fetch and parse adagents.json from publisher domain.

    Discovery order:

    1. ``https://{publisher}/.well-known/adagents.json`` (direct).
    2. ``authoritative_location`` redirect, if the direct response is a
       URL reference.
    3. RFC 4175 ads.txt MANAGERDOMAIN fallback, on direct 404 only:
       fetches ``https://{publisher}/ads.txt`` for a
       ``MANAGERDOMAIN=`` directive and, if present, tries
       ``https://{manager}/.well-known/adagents.json``.

    The fallback is one-hop only. If the manager domain also 404s,
    this raises :class:`AdagentsNotFoundError` for the original
    publisher — not a silent pass.

    Args:
        publisher_domain: Domain hosting the adagents.json file.
        timeout: Request timeout in seconds.
        user_agent: User-Agent header for HTTP request.
        client: Optional httpx.AsyncClient for connection pooling.
            If provided, caller is responsible for client lifecycle.
            If None, a new client is created for this request.

    Returns:
        Parsed adagents.json data (resolved via authoritative_location
        or ads.txt MANAGERDOMAIN if applicable).

    Raises:
        AdagentsNotFoundError: If adagents.json was not found via any
            discovery path.
        AdagentsAccessBlockedError: If the publisher's CDN returns HTTP
            403 with ``cf-mitigated: challenge`` (Cloudflare bot-management
            block). Subclass of ``AdagentsValidationError``.
        AdagentsValidationError: If JSON is invalid, malformed, or
            redirects exceed maximum depth or form a loop.
        AdagentsTimeoutError: If request times out.

    Notes:
        For production use with multiple requests, pass a shared
        httpx.AsyncClient to enable connection pooling.

        Callers who need to know which discovery path produced the
        data (direct, authoritative_location, or ads_txt_managerdomain)
        should call :func:`validate_adagents_domain` instead.

        ``fetch_adagents`` performs only minimal structural checks. To
        report per-entry schema violations (e.g., bare entries missing
        ``authorization_type``) without raising, pass the returned data
        to :func:`validate_adagents_structure`.
    """
    publisher_domain = _validate_publisher_domain(publisher_domain)

    try:
        data, *_ = await _resolve_direct(publisher_domain, timeout, user_agent, client)
        return data
    except AdagentsNotFoundError:
        manager_data = await _try_managerdomain_fallback(
            publisher_domain, timeout, user_agent, client
        )
        if manager_data is not None:
            return manager_data
        raise

async def fetch_adagents_with_cache(
    publisher_domain: str,
    cache_entry: AdagentsCacheEntry | None = None,
    timeout: float = 10.0,
    user_agent: str = "AdCP-Client/1.0",
    client: httpx.AsyncClient | None = None,
) -> AdagentsFetchResult:
    """Fetch with conditional refresh — returns body plus refreshed validators.

    Pass the previous fetch's :class:`AdagentsCacheEntry` to send
    ``If-None-Match`` / ``If-Modified-Since`` on the next fetch. A 304
    from the publisher is treated as a successful refresh: the cached
    ``body`` is returned with ``not_modified=True``, satisfying the
    7-day cache window described in adcp#4504.

    The first hop (``/.well-known/adagents.json``) is capped at 5 MiB;
    a dereferenced ``authoritative_location`` file is capped at 20 MiB.
    Both caps fail closed — oversized responses raise
    :class:`AdagentsValidationError` rather than truncate.

    Does NOT perform the ads.txt ``managerdomain`` fallback; the
    fallback is best-effort discovery, not cache-aware refresh, and
    bypassing it on 304 keeps the path simple. Callers that need both
    behaviors should compose this helper with
    :func:`validate_adagents_domain`.
    """
    publisher_domain = _validate_publisher_domain(publisher_domain)
    data, discovery, etag, last_modified, not_modified = await _resolve_direct(
        publisher_domain, timeout, user_agent, client, cache_entry=cache_entry
    )
    return AdagentsFetchResult(
        data=data,
        discovery_method=discovery,
        etag=etag,
        last_modified=last_modified,
        not_modified=not_modified,
    )

async def fetch_agent_authorizations(
    agent_url: str,
    publisher_domains: list[str],
    timeout: float = 10.0,
    client: httpx.AsyncClient | None = None,
) -> dict[str, AuthorizationContext]:
    """Fetch authorization contexts by checking publisher adagents.json files.

    This function discovers what publishers have authorized your agent by fetching
    their adagents.json files from the .well-known directory and extracting the
    properties your agent can access.

    This is the "pull" approach - you query publishers to see if they've authorized you.

    Args:
        agent_url: URL of your sales agent
        publisher_domains: List of publisher domains to check (e.g., ["nytimes.com", "wsj.com"])
        timeout: Request timeout in seconds for each fetch
        client: Optional httpx.AsyncClient for connection pooling

    Returns:
        Dictionary mapping publisher domain to AuthorizationContext.
        Only includes domains where the agent is authorized.

    Example:
        >>> # "Pull" approach - check what publishers have authorized you
        >>> contexts = await fetch_agent_authorizations(
        ...     "https://our-sales-agent.com",
        ...     ["nytimes.com", "wsj.com", "cnn.com"]
        ... )
        >>> for domain, ctx in contexts.items():
        ...     print(f"{domain}:")
        ...     print(f"  Property IDs: {ctx.property_ids}")
        ...     print(f"  Tags: {ctx.property_tags}")

    Notes:
        - Silently skips domains where adagents.json is not found or invalid
        - Only returns domains where the agent is explicitly authorized
        - For production use with many domains, pass a shared httpx.AsyncClient
          to enable connection pooling
    """
    import asyncio

    # Create tasks to fetch all adagents.json files in parallel
    async def fetch_authorization_for_domain(
        domain: str,
    ) -> tuple[str, AuthorizationContext | None]:
        """Fetch authorization context for a single domain."""
        try:
            adagents_data = await fetch_adagents(domain, timeout=timeout, client=client)

            # Check if agent is authorized
            if not verify_agent_authorization(adagents_data, agent_url):
                return (domain, None)

            # Get properties for this agent
            properties = get_properties_by_agent(adagents_data, agent_url)

            # Create authorization context
            return (domain, AuthorizationContext(properties))

        except (AdagentsNotFoundError, AdagentsValidationError, AdagentsTimeoutError):
            # Silently skip domains with missing or invalid adagents.json.
            # AdagentsAccessBlockedError (AdagentsValidationError subclass) is
            # intentionally swallowed: a bot-blocked domain is treated as
            # authorization-unavailable, same as a missing file.
            return (domain, None)

    # Fetch all domains in parallel
    tasks = [fetch_authorization_for_domain(domain) for domain in publisher_domains]
    results = await asyncio.gather(*tasks)

    # Build result dictionary, filtering out None values
    return {domain: ctx for domain, ctx in results if ctx is not None}

async def fetch_agent_authorizations_from_directory(
    agent_url: str,
    *,
    directory_url: str,
    since: str | None = None,
    cursor: str | None = None,
    include: list[str] | None = None,
    timeout: float = 10.0,
    client: httpx.AsyncClient | None = None,
) -> AgentAuthorizationsDirectoryResult:
    """Query an AAO directory for publishers that authorize ``agent_url``.

    Calls ``GET {directory_url}/v1/agents/{agent_url}/publishers`` per the
    AAO inverse-lookup contract (adcp#4823 / #4828) and returns the parsed
    response. The directory's answer is *discovery*, not authorization:
    callers should still verify each returned ``publisher_domain`` via
    :func:`fetch_adagents` before treating an edge as trusted.

    Args:
        agent_url: The agent whose publisher authorizations are being
            queried. Passed verbatim in the path; the directory echoes
            back a canonicalized form on the response.
        directory_url: HTTPS base URL of the AAO directory
            (e.g. ``"https://aao.example.com"``). The ``/v1/agents/...``
            path is appended; pass the directory's root, not a
            request-specific path.
        since: Optional RFC 3339 timestamp from a prior
            ``directory_indexed_at`` — passed through as ``?since=...``
            to limit the result to edges that changed since that point.
        cursor: Optional opaque pagination cursor from a prior response's
            ``next_cursor`` — passed through as ``?cursor=...`` to fetch
            the next page.
        include: Optional list of expansion keys per the AAO directory
            API spec (adcp#4894). Each value is emitted as a separate
            ``?include=<value>`` query parameter (repeated-key form, not
            comma-joined). Pass ``["properties"]`` against directories
            that support it to receive per-publisher ``property_ids[]``
            on each row, enabling full set-diff against the publisher's
            own adagents.json. Directories that don't support a given
            expansion key simply omit the corresponding fields from the
            response; callers should treat absence as count-only mode.
        timeout: Request timeout in seconds.
        client: Optional shared ``httpx.AsyncClient`` for connection
            pooling. Caller owns the client lifecycle.

    Returns:
        :class:`AgentAuthorizationsDirectoryResult`. On 404 from the
        directory the function returns a result with ``publishers=[]``
        and ``directory_indexed_at=None`` — directories MUST be allowed
        to answer "I do not index this agent" without callers needing
        to branch on exception type.

    Raises:
        AdagentsValidationError: If ``directory_url`` is malformed, the
            response status is non-200/non-404, the body is not valid
            JSON, or the body does not match the directory result schema.
        AdagentsTimeoutError: If the request times out.

    Notes:
        - ``directory_url`` is gated through the same SSRF protection
          (HTTPS only, DNS pre-check, private/reserved address ban) as
          publisher-side fetches.
        - Response bodies are capped at 5 MiB. Bulk responses paginate
          via ``next_cursor``; pass that value as ``cursor`` on the next
          call.
    """
    if not isinstance(agent_url, str) or not agent_url:
        raise AdagentsValidationError("agent_url must be a non-empty string")
    if not isinstance(directory_url, str) or not directory_url:
        raise AdagentsValidationError("directory_url must be a non-empty string")

    base = directory_url.rstrip("/")
    if not base.startswith("https://"):
        raise AdagentsValidationError(f"directory_url must be an HTTPS URL, got: {directory_url!r}")
    _validate_redirect_url(f"{base}/v1/agents/_/publishers")

    request_url = f"{base}/v1/agents/{quote(agent_url, safe='')}/publishers"
    query_pairs: list[tuple[str, str]] = []
    if since is not None:
        query_pairs.append(("since", since))
    if cursor is not None:
        query_pairs.append(("cursor", cursor))
    if include:
        # Repeated-key form per docs/aao/directory-api.mdx (style: form,
        # explode: true). Comma-joined NOT accepted by spec-conformant
        # directories.
        for value in include:
            query_pairs.append(("include", value))
    if query_pairs:
        query_string = "&".join(f"{quote(k, safe='')}={quote(v, safe='')}" for k, v in query_pairs)
        request_url = f"{request_url}?{query_string}"

    parsed = urlparse(request_url)
    await _dns_validate_host(
        parsed.hostname or "", parsed.port or (443 if parsed.scheme == "https" else 80)
    )

    headers = {"User-Agent": "AdCP-Client/1.0", "Accept": "application/json"}

    # SDK-owned client is pinned to the validated IP (see _fetch_adagents_url).
    # A failed resolve/SSRF check raises AdagentsValidationError, which
    # propagates past the httpx handlers below — the correct fail-closed
    # outcome (we do not convert it into an empty result).
    try:
        if client is not None:
            body, status_code, _ = await _stream_capped(
                client, request_url, headers, timeout, MAX_DIRECTORY_PAGE_BYTES
            )
        else:
            async with _owned_pinned_client(request_url, timeout) as new_client:
                body, status_code, _ = await _stream_capped(
                    new_client, request_url, headers, timeout, MAX_DIRECTORY_PAGE_BYTES
                )
    except httpx.TimeoutException as e:
        raise AdagentsTimeoutError(parsed.netloc, timeout) from e
    except httpx.RequestError as e:
        raise AdagentsValidationError(f"Failed to fetch agent-publishers directory: {e}") from e

    if status_code == 404:
        # Per adcp#4828, a directory that has not indexed this agent
        # answers 404. Surface as an empty result so callers don't need
        # to special-case the exception path for "no edges" — the
        # protocol is intentionally permissive here.
        return AgentAuthorizationsDirectoryResult(
            agent_url=agent_url,
            directory_indexed_at=None,
            publishers=[],
            next_cursor=None,
        )

    if status_code != 200:
        raise AdagentsValidationError(f"Agent-publishers directory returned HTTP {status_code}")

    try:
        data = json.loads(body)
    except json.JSONDecodeError as e:
        raise AdagentsValidationError(
            f"Invalid JSON in agent-publishers directory response: {str(e)[:200]}"
        ) from e

    try:
        return AgentAuthorizationsDirectoryResult.model_validate(data)
    except Exception as e:  # pydantic.ValidationError + any coercion failure
        raise AdagentsValidationError(
            f"Agent-publishers directory response failed schema validation: {e}"
        ) from e

def filter_revoked_selectors(
    selectors: list[dict[str, Any]],
    revoked_domains: set[str],
) -> list[dict[str, Any]]:
    """Strip selectors whose ``publisher_domain`` is revoked.

    Apply this AFTER the compact-form fan-out so each remaining selector
    addresses exactly one publisher, then drop any whose domain is in
    ``revoked_domains``. Revocation takes precedence over every other
    listing of that domain in the file (selectors, top-level properties,
    etc.) per adcp#4504.
    """
    if not revoked_domains:
        return selectors
    return [s for s in selectors if s.get("publisher_domain") not in revoked_domains]

def format_is_supported(
    requested: str | FormatId | Mapping[str, Any],
    supported: str | FormatId | Mapping[str, Any],
    *,
    default_agent_url: str = CANONICAL_CREATIVE_AGENT_URL,
) -> bool:
    """Return true when ``requested`` is acceptable for ``supported``.

    This is intentionally stricter than :func:`formats_are_equivalent`.
    A broad supported format such as ``display_image`` accepts a specific
    request such as ``display_image`` 300x250, but a fixed supported product
    format requires the request to provide and match every fixed parameter
    (``width``, ``height``, and ``duration_ms``).
    """
    req = upgrade_legacy_format_id(requested, default_agent_url=default_agent_url)
    sup = upgrade_legacy_format_id(supported, default_agent_url=default_agent_url)
    if not formats_are_equivalent(req, sup, default_agent_url=default_agent_url):
        return False

    for field in ("width", "height", "duration_ms"):
        supported_value = getattr(sup, field)
        if supported_value is None:
            continue
        if getattr(req, field) != supported_value:
            return False
    return True

def formats_are_equivalent(
    a: str | FormatId | Mapping[str, Any],
    b: str | FormatId | Mapping[str, Any],
    *,
    default_agent_url: str = CANONICAL_CREATIVE_AGENT_URL,
) -> bool:
    """Return true when two format IDs identify the same canonical family.

    Both inputs are first passed through :func:`upgrade_legacy_format_id`.
    Declared parameters must not conflict, but an omitted parameter on either
    side is treated as unspecified rather than a mismatch. Use
    :func:`format_is_supported` for product/capability gating where a
    supported fixed size or duration requires the request to state that value.
    """
    left = upgrade_legacy_format_id(a, default_agent_url=default_agent_url)
    right = upgrade_legacy_format_id(b, default_agent_url=default_agent_url)
    if canonicalize_agent_url(left.agent_url) != canonicalize_agent_url(right.agent_url):
        return False
    if left.id != right.id:
        return False

    for field in ("width", "height", "duration_ms"):
        left_value = getattr(left, field)
        right_value = getattr(right, field)
        if left_value is not None and right_value is not None and left_value != right_value:
            return False
    return True

def generate_webhook_challenge_value() -> str:
    """Generate an opaque random value for a proof-of-control challenge."""

    return f"wch_{secrets.token_urlsafe(32)}"

def generate_webhook_idempotency_key() -> str:
    """Generate a cryptographically random idempotency_key for a webhook event.

    Returns a UUID v4 prefixed with ``whk_`` — matches the example format in
    ``webhooks.mdx`` and stays within the spec's length + charset bounds
    (``^[A-Za-z0-9_.:-]{16,255}$``).

    Publishers generate this once per delivery and reuse it for exact retries.
    A re-emission of the same logical event uses a new delivery key while
    retaining its stable ``notification_id``.
    """
    return f"whk_{uuid.uuid4()}"

def get_adcp_signed_headers_for_webhook(
    headers: dict[str, Any],
    secret: str,
    timestamp: str | int | None,
    payload: dict[str, Any] | AdCPBaseModel,
) -> dict[str, Any]:
    """
    Generate AdCP-compliant signed headers for webhook delivery.

    This function creates a cryptographic signature that proves the webhook
    came from an authorized agent and protects against replay attacks by
    including a timestamp in the signed message.

    The function adds two headers to the provided headers dict:
    - X-AdCP-Signature: HMAC-SHA256 signature in format "sha256=<hex_digest>"
    - X-AdCP-Timestamp: Unix timestamp in seconds

    The signing algorithm:
    1. Constructs message as "{timestamp}.{json_payload}"
    2. JSON-serializes payload with default separators (matches wire format from json= kwarg)
    3. UTF-8 encodes the message
    4. HMAC-SHA256 signs with the shared secret
    5. Hex-encodes and prefixes with "sha256="

    Args:
        headers: Existing headers dictionary to add signature headers to
        secret: Shared secret key for HMAC signing
        timestamp: Unix timestamp in seconds (str or int). If None, uses current time.
        payload: Webhook payload (dict or Pydantic model - will be JSON-serialized)

    Returns:
        The modified headers dictionary with signature headers added

    Examples:
        Sign and send an MCP webhook:
        >>> import time
        >>> from adcp.webhooks import create_mcp_webhook_payload
        >>> from adcp.webhooks import get_adcp_signed_headers_for_webhook
        >>>
        >>> payload = create_mcp_webhook_payload(
        ...     task_id="task_123",
        ...     operation_id="op_123",
        ...     status="completed",
        ...     task_type="create_media_buy",
        ...     result={"media_buy_id": "mb_1"},
        ... )
        >>> headers = {"Content-Type": "application/json"}
        >>> signed_headers = get_adcp_signed_headers_for_webhook(
        ...     headers, secret="my-webhook-secret", timestamp=str(int(time.time())),
        ...     payload=payload,
        ... )
        >>>
        >>> # Send webhook with signed headers
        >>> import httpx
        >>> response = await httpx.post(
        ...     webhook_url,
        ...     json=payload,
        ...     headers=signed_headers
        ... )

        Headers will contain:
        >>> print(signed_headers)
        {
            "Content-Type": "application/json",
            "X-AdCP-Signature": "sha256=a1b2c3...",
            "X-AdCP-Timestamp": "1773185740"
        }
    """
    signature_headers, _body_bytes = _compute_legacy_signature(
        secret=secret, timestamp=timestamp, payload=payload
    )
    headers.update(signature_headers)
    return headers

def get_adcp_version() -> str:
    """Return the AdCP *spec* version (legacy name).

    .. deprecated:: 4.1
        Kept for backwards compatibility with pre-4.1 callers. Prefer
        :func:`get_adcp_spec_version` (spec version) or
        :func:`get_adcp_sdk_version` / :attr:`adcp.__version__` (SDK
        package version) — the split disambiguates what the caller
        actually wants at the call site.
    """
    import warnings

    warnings.warn(
        "get_adcp_version() is deprecated; use get_adcp_spec_version() "
        "for the AdCP spec version or get_adcp_sdk_version() / "
        "adcp.__version__ for the SDK package version.",
        DeprecationWarning,
        stacklevel=2,
    )
    return get_adcp_spec_version()

def get_all_properties(adagents_data: dict[str, Any]) -> list[dict[str, Any]]:
    """Extract all properties from adagents.json data.

    Handles all authorization types: inline_properties, property_ids,
    property_tags, and publisher_properties.

    For ``publisher_properties`` selectors whose target ``publisher_domain``
    is NOT present inline in this file's top-level ``properties[]`` array,
    this function returns no properties for that selector. Federated
    fallback (fetching the child publisher's own adagents.json to resolve
    the selector remotely) is out of scope here and lives in
    :func:`fetch_agent_authorizations_from_directory` and
    :func:`detect_publisher_properties_divergence` from companion PR #752.
    Wire-only authorization checks that assume federated resolution will
    under-authorize against managed-network parent files that only inline
    a subset of their child domains.

    Args:
        adagents_data: Parsed adagents.json data

    Returns:
        List of all properties across all authorized agents, with agent_url added

    Raises:
        AdagentsValidationError: If adagents_data is malformed
    """
    if not isinstance(adagents_data, dict):
        raise AdagentsValidationError("adagents_data must be a dictionary")

    authorized_agents = adagents_data.get("authorized_agents")
    if not isinstance(authorized_agents, list):
        raise AdagentsValidationError("adagents.json must have 'authorized_agents' array")

    top_level_properties = adagents_data.get("properties", [])
    if not isinstance(top_level_properties, list):
        top_level_properties = []

    revoked = _get_revoked_publisher_domains(adagents_data)
    revoked_top_level = [
        p
        for p in top_level_properties
        if not (
            isinstance(p, dict)
            and isinstance(p.get("publisher_domain"), str)
            and p["publisher_domain"] in revoked
        )
    ]

    # Build the domain index once per file — _resolve_agent_properties is
    # called per-agent, and at cafemedia scale (thousands of properties ×
    # multiple agents) rebuilding it inside each call is O(agents × N).
    domain_index = _build_domain_index(revoked_top_level)

    properties = []
    for agent in authorized_agents:
        if not isinstance(agent, dict):
            continue

        agent_url = agent.get("url", "")
        if not agent_url:
            continue

        # revoked_top_level pre-filters revoked domains from the per-domain
        # index, so inline resolution honors revocation transparently.
        agent_properties = _resolve_agent_properties(agent, revoked_top_level, domain_index)

        for prop in agent_properties:
            prop_with_agent = {**prop, "agent_url": agent_url}
            properties.append(prop_with_agent)

    return properties

def get_all_tags(adagents_data: dict[str, Any]) -> set[str]:
    """Extract all unique tags from properties in adagents.json data.

    Args:
        adagents_data: Parsed adagents.json data

    Returns:
        Set of all unique tags across all properties

    Raises:
        AdagentsValidationError: If adagents_data is malformed
    """
    properties = get_all_properties(adagents_data)
    tags = set()

    for prop in properties:
        prop_tags = prop.get("tags", [])
        if isinstance(prop_tags, list):
            for tag in prop_tags:
                if isinstance(tag, str):
                    tags.add(tag)

    return tags

def get_asset_count(format: FormatAssetsInput | Mapping[str, Any]) -> int:
    """Get the count of assets in a format (for display purposes).

    Args:
        format: The Format object

    Returns:
        Number of assets, or 0 if none defined
    """
    return len(get_format_assets(format))

def get_format_assets(format: FormatAssetsInput | Mapping[str, Any]) -> list[FormatAsset]:
    """Get assets from a Format.

    Returns the list of assets from the format's `assets` field.
    Returns empty list if no assets are defined (flexible format with no assets).

    Args:
        format: Any canonical declaration or raw catalog object exposing ``assets``

    Returns:
        List of assets

    Example:
        ```python
        for declaration in product.format_options:
            assets = get_format_assets(declaration)
            print(f"{declaration.format_kind} has {len(assets)} assets")
        ```
    """
    if isinstance(format, Mapping):
        assets = format.get("assets")
        if assets is None and isinstance(format.get("params"), Mapping):
            assets = format["params"].get("slots")
    else:
        assets = getattr(format, "assets", None)
        if assets is None:
            params = getattr(format, "params", None)
            assets = params.get("slots") if isinstance(params, Mapping) else None
    if assets:
        return list(assets)

    return []

def get_individual_assets(format: FormatAssetsInput | Mapping[str, Any]) -> list[FormatAsset]:
    """Get individual assets (not repeatable groups) from a Format.

    Args:
        format: The Format object

    Returns:
        List of individual assets (item_type='individual')
    """
    return [asset for asset in get_format_assets(format) if _get_item_type(asset) == "individual"]

def get_optional_assets(format: FormatAssetsInput | Mapping[str, Any]) -> list[FormatAsset]:
    """Get only optional assets from a Format.

    Note: When using deprecated `assets_required`, this will always return empty
    since assets_required only contained required assets.

    Args:
        format: The Format object

    Returns:
        List of optional assets only

    Example:
        ```python
        optional_assets = get_optional_assets(format)
        print(f"Can optionally provide {len(optional_assets)} additional assets")
        ```
    """
    return [asset for asset in get_format_assets(format) if not _is_required(asset)]

def get_properties_by_agent(adagents_data: dict[str, Any], agent_url: str) -> list[dict[str, Any]]:
    """Get all properties authorized for a specific agent.

    Handles all authorization types per the AdCP specification:
    - inline_properties: Properties defined directly in the agent's properties array
    - property_ids: Filter top-level properties by property_id
    - property_tags: Filter top-level properties by tags
    - publisher_properties: Inline-resolved properties from cross-publisher
      selectors (resolved from the parent file's top-level properties[]
      array per adcp#4827)

    For ``publisher_properties`` selectors whose target ``publisher_domain``
    is NOT present inline in this file's top-level ``properties[]`` array,
    this function returns no properties for that selector. Federated
    fallback (fetching the child publisher's own adagents.json to resolve
    the selector remotely) is out of scope here and lives in
    :func:`fetch_agent_authorizations_from_directory` and
    :func:`detect_publisher_properties_divergence` from companion PR #752.
    Wire-only authorization checks that assume federated resolution will
    under-authorize against managed-network parent files that only inline
    a subset of their child domains.

    Args:
        adagents_data: Parsed adagents.json data
        agent_url: URL of the agent to filter by

    Returns:
        List of properties for the specified agent (empty if agent not found)

    Raises:
        AdagentsValidationError: If adagents_data is malformed
    """
    return _resolve_properties_for_agent(adagents_data, agent_url, permissive_bare_top_level=False)

def get_repeatable_groups(format: FormatAssetsInput | Mapping[str, Any]) -> list[FormatAsset]:
    """Get repeatable asset groups from a Format.

    Args:
        format: The Format object

    Returns:
        List of repeatable asset groups (item_type='repeatable_group')
    """
    return [
        asset for asset in get_format_assets(format) if _get_item_type(asset) == "repeatable_group"
    ]

def get_required_assets(format: FormatAssetsInput | Mapping[str, Any]) -> list[FormatAsset]:
    """Get only required assets from a Format.

    Args:
        format: The Format object

    Returns:
        List of required assets only

    Example:
        ```python
        required_assets = get_required_assets(format)
        print(f"Must provide {len(required_assets)} assets")
        ```
    """
    return [asset for asset in get_format_assets(format) if _is_required(asset)]

def has_assets(format: FormatAssetsInput | Mapping[str, Any]) -> bool:
    """Check if a format has any assets defined.

    Args:
        format: The Format object

    Returns:
        True if format has assets, False otherwise
    """
    return get_asset_count(format) > 0

def identifiers_match(
    property_identifiers: list[dict[str, str]],
    agent_identifiers: list[dict[str, str]],
) -> bool:
    """Check if any property identifier matches agent's authorized identifiers.

    Args:
        property_identifiers: Identifiers from property
            (e.g., [{"type": "domain", "value": "cnn.com"}])
        agent_identifiers: Identifiers from adagents.json

    Returns:
        True if any identifier matches

    Notes:
        - Domain identifiers use AdCP domain matching rules
        - Other identifiers (bundle_id, roku_store_id, etc.) require exact match
    """
    for prop_id in property_identifiers:
        prop_type = prop_id.get("type", "")
        prop_value = prop_id.get("value", "")

        for agent_id in agent_identifiers:
            agent_type = agent_id.get("type", "")
            agent_value = agent_id.get("value", "")

            # Type must match
            if prop_type != agent_type:
                continue

            # Domain identifiers use special matching rules
            if prop_type == "domain":
                if domain_matches(prop_value, agent_value):
                    return True
            else:
                # Other identifier types require exact match
                if prop_value == agent_value:
                    return True

    return False

def normalize_assets_required(assets_required: list[Any]) -> list[FormatAsset]:
    """Convert deprecated assets_required to new assets format.

    .. deprecated:: 3.2.0
        The ``assets_required`` field was removed in ADCP 3.0.0-beta.2.
        This function will be removed in a future version.

    All assets in assets_required are required by definition (that's why they were in
    that array). The new `assets` field has an explicit `required: boolean` to allow
    both required AND optional assets.

    Args:
        assets_required: The deprecated assets_required array

    Returns:
        Normalized assets as Pydantic models with explicit required=True
    """
    warnings.warn(
        "normalize_assets_required() is deprecated. "
        "The assets_required field was removed in ADCP 3.0.0-beta.2. "
        "This function will be removed in a future version.",
        DeprecationWarning,
        stacklevel=2,
    )
    normalized: list[FormatAsset] = []
    for asset in assets_required:
        # Get asset data as dict
        if isinstance(asset, dict):
            asset_dict = asset
        else:
            asset_dict = asset.model_dump() if hasattr(asset, "model_dump") else dict(asset)

        # Map old fields to new schema format
        mapped = {**asset_dict, "required": True}
        # Ensure asset_id is present (map from asset_group_id if needed)
        if "asset_group_id" in mapped and "asset_id" not in mapped:
            mapped["asset_id"] = mapped.pop("asset_group_id")
        # Remove fields that don't exist in the new schema
        for old_field in ("min_count", "max_count", "assets"):
            mapped.pop(old_field, None)
        # Use AssetsModel (individual asset type)
        normalized.append(AssetsModel(**mapped))

    return normalized

def resolve_properties_for_agent(
    adagents_data: dict[str, Any],
    agent_url: str,
    *,
    mode: PropertyResolutionMode = "strict",
) -> list[dict[str, Any]]:
    """Resolve properties for an agent with an explicit strict/permissive mode.

    ``mode="strict"`` is identical to :func:`get_properties_by_agent` and
    only honors schema-conformant authorization selectors plus the historical
    inline ``properties`` legacy shape.

    ``mode="permissive"`` keeps every strict selector behavior unchanged, but
    treats one exact matching bare ``authorized_agents`` entry
    (``{"url": ..., "authorized_for": ...}``) as authorizing the file's
    top-level ``properties[]``. This is for operational binding of
    non-conformant publisher files that list an agent URL without an
    ``authorization_type`` or selector. If the agent is not listed, has any
    explicit or unknown selector field, or has multiple same-URL entries, the
    resolver still returns the strict result.

    Args:
        adagents_data: Parsed adagents.json data
        agent_url: URL of the agent to filter by
        mode: ``"strict"`` for spec-conformant resolution, ``"permissive"``
            to opt into bare-entry top-level property fallback.

    Returns:
        List of properties for the specified agent.

    Raises:
        AdagentsValidationError: If adagents_data is malformed
        ValueError: If mode is not ``"strict"`` or ``"permissive"``
    """
    if mode == "strict":
        return _resolve_properties_for_agent(
            adagents_data,
            agent_url,
            permissive_bare_top_level=False,
        )
    if mode == "permissive":
        return _resolve_properties_for_agent(
            adagents_data,
            agent_url,
            permissive_bare_top_level=True,
        )
    raise ValueError("mode must be 'strict' or 'permissive'")

def sign_legacy_webhook(
    secret: str,
    payload: dict[str, Any] | AdCPBaseModel,
    *,
    timestamp: str | int | None = None,
    headers: dict[str, Any] | None = None,
) -> tuple[dict[str, str], bytes]:
    """Return ``(signed_headers, body_bytes)`` for a legacy HMAC webhook.

    Byte-equality between signature input and HTTP body is guaranteed —
    callers POST ``content=body_bytes`` instead of ``json=payload``, so the
    separator-drift trap that caused silent 401s in every spaced-vs-compact
    interop is structurally impossible here.

    This is a lower-level companion to :func:`deliver` for callers who need
    to own the HTTP transport themselves (custom auth, pre-configured
    ``httpx.AsyncClient``, non-httpx clients). For the one-shot "send a
    webhook" path, prefer :func:`deliver`.

    The returned ``body_bytes`` use compact separators (``","``/``":"``)
    matching the canonical on-wire form pinned by adcontextprotocol/adcp#2478.

    Example:
        >>> signed, body = sign_legacy_webhook("shared-secret", payload)
        >>> headers = {**signed, "Content-Type": "application/json"}
        >>> await client.post(url, content=body, headers=headers)
    """
    signature_headers, body_bytes = _compute_legacy_signature(
        secret=secret, timestamp=timestamp, payload=payload
    )
    if headers is not None:
        merged = {str(k): str(v) for k, v in headers.items()}
        merged.update(signature_headers)
        return merged, body_bytes
    return signature_headers, body_bytes

def sign_webhook(
    *,
    method: str,
    url: str,
    headers: Mapping[str, str],
    body: bytes,
    private_key: PrivateKey,
    key_id: str,
    alg: str,
    created: int | None = None,
    expires_in_seconds: int = DEFAULT_EXPIRES_IN_SECONDS,
    nonce: str | None = None,
    label: str = SIG_LABEL_DEFAULT,
) -> SignedHeaders:
    """Sign an outgoing webhook POST per adcp/webhook-signing/v1.

    ``cover_content_digest=True`` and ``tag=WEBHOOK_TAG`` are pinned. The
    caller attaches ``SignedHeaders.as_dict()`` to the outgoing HTTP request.

    The ``method`` is normally ``"POST"`` for webhook delivery; passed through
    unchanged so callers signing a retried ``PUT`` or variant delivery verb
    are not forced into an extra translation.

    See also:
        :class:`adcp.webhooks.WebhookSender` — higher-level one-call helper
        that builds the payload, signs, and POSTs in a single call. Prefer it
        unless you need to own the HTTP transport yourself.
    """
    return sign_request(
        method=method,
        url=url,
        headers=headers,
        body=body,
        private_key=private_key,
        key_id=key_id,
        alg=alg,
        cover_content_digest=True,
        created=created,
        expires_in_seconds=expires_in_seconds,
        nonce=nonce,
        tag=WEBHOOK_TAG,
        label=label,
        signing_profile_version="3.2",
    )

def to_wire_dict(
    payload: AdCPBaseModel | Task | TaskStatusUpdateEvent | Mapping[str, Any],
) -> dict[str, Any]:
    """Serialize any AdCP webhook payload to a JSON-ready dict.

    Single seam for adopters that accept "any AdCP webhook payload" — a
    sender wrapping :func:`create_a2a_webhook_payload` and
    :func:`create_mcp_webhook_payload` would otherwise have to write
    per-shape dispatch (``isinstance`` checks, ``MessageToDict`` for
    protobuf, ``model_dump`` for Pydantic, passthrough for dict). Brittle:
    a future a2a-sdk that swaps protobuf for a Pydantic façade silently
    changes which branch runs, and adopters duplicate the dispatch in
    every send path. Use this helper instead — the dispatch lives here.

    Behaviour by input shape:

    * a2a ``Task`` / ``TaskStatusUpdateEvent`` (protobuf, a2a-sdk 1.0+) →
      ``MessageToDict(..., preserving_proto_field_name=False)`` so JSON
      keys match the A2A wire spec (camelCase: ``id``, ``contextId``,
      ``artifactId``). Enum values are normalized from the 1.0 protobuf
      form (``TASK_STATE_COMPLETED``, ``ROLE_AGENT``) to the 0.3-spec
      lowercase form (``completed``, ``agent``) so 0.3 buyer receivers
      keep parsing.
    * Any Pydantic model (``McpWebhookPayload``, future Pydantic façades,
      :class:`AdCPBaseModel` subclasses) → ``model_dump(mode="json",
      exclude_none=True)``.
    * ``Mapping`` → coerced to ``dict``. Legacy adopter passthrough for
      callers that build the wire dict by hand.

    Raises:
        TypeError: payload is none of the above.
    """
    if isinstance(payload, (Task, TaskStatusUpdateEvent)):
        data = MessageToDict(payload, preserving_proto_field_name=False)
        _normalize_a2a_task_state_to_v03(data)
        return data
    if hasattr(payload, "model_dump"):
        model = cast(AdCPBaseModel, payload)
        return model.model_dump(mode="json", exclude_none=True)
    if isinstance(payload, Mapping):
        return dict(payload)
    raise TypeError(
        f"Unsupported webhook payload type {type(payload).__name__}: expected "
        "a2a Task / TaskStatusUpdateEvent (protobuf), an AdCP Pydantic model "
        "(e.g. McpWebhookPayload), or a Mapping[str, Any]."
    )

def translate_universal_macros(
    pixel_url: str,
    mapping: MacroMapping,
) -> TranslateUniversalMacrosResult:
    """Translate universal macros in query-parameter values.

    ``ValueMacroMapping`` values are UTF-8 percent-encoded with
    :func:`encode_unreserved`; ``NativeMacroMapping`` values are inserted
    verbatim after the full mapping passes the control-character guard. If a
    parameter value contains any unmapped universal macro, that whole parameter
    is dropped. Query keys, the path, the fragment, and parameters without
    universal macros pass through byte-for-byte. Replacement is single-pass.

    Consent macros supplied through ``ValueMacroMapping`` are translated but
    reported in ``frozen_consent_macros`` because freezing impression-time
    consent at producer time can create a privacy defect. Callers should also
    inspect ``dropped_consent_macros`` and ``suspect_native_values`` before
    publishing a tracker.

    Raises:
        UniversalMacroTranslationError: A native mapping, used or unused,
            contains U+0000-U+001F or U+007F. No URL is emitted.
        TypeError: A mapping entry is not a supported typed mapping model.
    """

    frozen_consent_macros: list[str] = []
    frozen_seen: set[str] = set()
    suspect_native_values: list[str] = []
    suspect_seen: set[str] = set()
    validated_mapping: dict[str, MacroMappingEntry] = {}

    # Validate the entire raw-token trust boundary before doing any URL work.
    # An unused unsafe entry must reject just like an entry present in the URL.
    for macro, entry in mapping.items():
        if isinstance(entry, NativeMacroMapping):
            native = entry.native
            if _has_unsafe_native_character(native):
                raise UniversalMacroTranslationError(macro)
            validated_mapping[macro] = NativeMacroMapping(native=native)
        elif isinstance(entry, ValueMacroMapping):
            value = entry.value
            if macro in _CONSENT_MACROS:
                _append_once(frozen_consent_macros, frozen_seen, macro)
            if _NATIVE_TOKEN_SHAPE.fullmatch(value):
                _append_once(suspect_native_values, suspect_seen, macro)
            validated_mapping[macro] = ValueMacroMapping(value=value)
        else:
            raise TypeError(
                f"mapping entry for {macro!r} must be NativeMacroMapping " "or ValueMacroMapping"
            )

    fragment_index = pixel_url.find("#")
    if fragment_index == -1:
        without_fragment = pixel_url
        fragment = ""
    else:
        without_fragment = pixel_url[:fragment_index]
        fragment = pixel_url[fragment_index:]

    query_index = without_fragment.find("?")
    if query_index == -1:
        return TranslateUniversalMacrosResult(
            url=pixel_url,
            dropped_params=[],
            unmapped_macros=[],
            dropped_consent_macros=[],
            frozen_consent_macros=frozen_consent_macros,
            suspect_native_values=suspect_native_values,
        )

    base = without_fragment[:query_index]
    raw_query = without_fragment[query_index + 1 :]

    dropped_params: list[str] = []
    unmapped_macros: list[str] = []
    unmapped_seen: set[str] = set()
    dropped_consent_macros: list[str] = []
    dropped_consent_seen: set[str] = set()
    output_parts: list[str] = []

    for raw_param in raw_query.split("&"):
        key, separator, value = raw_param.partition("=")
        tokens = _UNIVERSAL_MACRO.findall(value)
        if not tokens:
            output_parts.append(raw_param)
            continue

        missing = [token for token in tokens if token not in validated_mapping]
        if missing:
            dropped_params.append(key)
            for macro in missing:
                _append_once(unmapped_macros, unmapped_seen, macro)
                if macro in _CONSENT_MACROS:
                    _append_once(
                        dropped_consent_macros,
                        dropped_consent_seen,
                        macro,
                    )
            continue

        def replace(match: re.Match[str]) -> str:
            macro = match.group(0)
            entry = validated_mapping[macro]
            if isinstance(entry, NativeMacroMapping):
                return entry.native
            if isinstance(entry, ValueMacroMapping):
                return encode_unreserved(entry.value)
            # The mapping-wide validation above makes this unreachable even
            # for mutable custom Mapping implementations under normal use.
            raise TypeError(f"unsupported mapping entry for {macro!r}")

        translated = _UNIVERSAL_MACRO.sub(replace, value)
        output_parts.append(f"{key}{separator}{translated}")

    new_query = "&".join(output_parts)
    url = f"{base}?{new_query}{fragment}" if new_query else f"{base}{fragment}"
    return TranslateUniversalMacrosResult(
        url=url,
        dropped_params=dropped_params,
        unmapped_macros=unmapped_macros,
        dropped_consent_macros=dropped_consent_macros,
        frozen_consent_macros=frozen_consent_macros,
        suspect_native_values=suspect_native_values,
    )

def upgrade_legacy_format_id(
    value: str | FormatId | Mapping[str, Any],
    *,
    default_agent_url: str = CANONICAL_CREATIVE_AGENT_URL,
) -> FormatId:
    """Return ``value`` as a canonical, parameterized ``FormatId`` when known.

    The current canonical upgrade maps legacy display size IDs such as
    ``display_300x250`` and ``display_300x250_image`` to
    ``display_image`` with ``width=300`` and ``height=250``. Unknown IDs are
    still returned as structured ``FormatId`` values so callers can compare
    them consistently.
    """
    is_bare_legacy_id = isinstance(value, str)
    fid = _coerce_format_id(value, default_agent_url=default_agent_url)
    match = _DISPLAY_SIZE_RE.fullmatch(fid.id)
    if match is None:
        return fid
    default_fid = _coerce_format_id("__default__", default_agent_url=default_agent_url)
    if not is_bare_legacy_id and canonicalize_agent_url(fid.agent_url) != canonicalize_agent_url(
        default_fid.agent_url
    ):
        return fid

    return FormatId.model_validate(
        {
            "agent_url": str(fid.agent_url),
            "id": "display_image",
            "width": int(match.group("width")),
            "height": int(match.group("height")),
            "duration_ms": fid.duration_ms,
        }
    )

def uses_deprecated_assets_field(format: FormatAssetsInput | Mapping[str, Any]) -> bool:
    """Check if format uses deprecated assets_required field.

    .. deprecated:: 3.2.0
        The ``assets_required`` field was removed in ADCP 3.0.0-beta.2.
        This function always returns False and will be removed in a future version.

    Args:
        format: The Format object

    Returns:
        Always False (deprecated field no longer exists)
    """
    warnings.warn(
        "uses_deprecated_assets_field() is deprecated and always returns False. "
        "The assets_required field was removed in ADCP 3.0.0-beta.2. "
        "This function will be removed in a future version.",
        DeprecationWarning,
        stacklevel=2,
    )
    return False

def validate_adagents(adagents: dict[str, Any]) -> None:
    """Validate an adagents.json structure.

    Args:
        adagents: The adagents.json dict

    Raises:
        ValidationError: If validation fails
    """
    authorized_agents = adagents.get("authorized_agents")
    if isinstance(authorized_agents, list):
        for agent in authorized_agents:
            if isinstance(agent, dict):
                validate_agent_authorization(agent)

    revoked = adagents.get("revoked_publisher_domains")
    if revoked is not None:
        if not isinstance(revoked, list):
            raise ValidationError("'revoked_publisher_domains' must be an array")
        for entry in revoked:
            if not isinstance(entry, dict):
                raise ValidationError("revoked_publisher_domains entry must be an object")
            validate_revoked_publisher_domain_entry(entry)

async def validate_adagents_domain(
    publisher_domain: str,
    timeout: float = 10.0,
    user_agent: str = "AdCP-Client/1.0",
    client: httpx.AsyncClient | None = None,
) -> AdAgentsValidationResult:
    """Discover and validate a publisher's adagents.json with provenance.

    Mirrors :func:`fetch_adagents` discovery semantics but returns a
    typed :class:`AdAgentsValidationResult` exposing which path
    produced the data (``discovery_method``) and the manager domain
    used for the RFC 4175 fallback (``manager_domain``), if any.

    Errors are reported on the result rather than raised. A manager
    domain 404 is a terminal failure: ``valid`` is False and
    ``manager_domain`` is recorded for diagnostics.

    .. warning::

        When ``discovery_method == 'ads_txt_managerdomain'`` the data
        came from the manager, not the publisher. Callers wiring this
        into authorization decisions must verify that the source
        publisher is explicitly named in the manager's adagents.json
        (e.g., via ``publisher_properties.publisher_domain`` on the
        relevant authorized_agents entry) before trusting an agent
        claim — otherwise a manager that lists agent A unconditionally
        implicitly authorizes A for every publisher pointing
        MANAGERDOMAIN at the manager.
    """
    try:
        normalized = _validate_publisher_domain(publisher_domain)
    except AdagentsValidationError as e:
        return AdAgentsValidationResult(
            domain=publisher_domain,
            url="",
            errors=[str(e)],
        )

    url = f"https://{normalized}/.well-known/adagents.json"

    try:
        data, discovery, *_ = await _resolve_direct(normalized, timeout, user_agent, client)
        return AdAgentsValidationResult(
            domain=normalized,
            url=url,
            discovery_method=discovery,
            data=data,
            valid=True,
        )
    except AdagentsNotFoundError as direct_error:
        direct_error_msg = str(direct_error)
    except (AdagentsValidationError, AdagentsTimeoutError) as e:
        return AdAgentsValidationResult(
            domain=normalized,
            url=url,
            errors=[str(e)],
        )

    managers = await _fetch_ads_txt_managerdomains(normalized, timeout, user_agent, client)
    if not managers:
        return AdAgentsValidationResult(
            domain=normalized,
            url=url,
            errors=[direct_error_msg],
        )

    manager_domain = managers[-1]

    if manager_domain == normalized:
        return AdAgentsValidationResult(
            domain=normalized,
            url=url,
            errors=[
                direct_error_msg,
                f"ads.txt managerdomain {manager_domain} points back to source publisher",
            ],
        )

    manager_normalized = _ensure_safe_manager_domain(manager_domain)
    if manager_normalized is None:
        return AdAgentsValidationResult(
            domain=normalized,
            url=url,
            errors=[
                direct_error_msg,
                f"ads.txt managerdomain {manager_domain!r} is malformed or "
                "targets a private/reserved address",
            ],
        )

    try:
        manager_data, *_ = await _resolve_direct(
            manager_normalized, timeout, user_agent, client=None
        )
    except AdagentsNotFoundError:
        return AdAgentsValidationResult(
            domain=normalized,
            url=url,
            discovery_method="ads_txt_managerdomain",
            manager_domain=manager_normalized,
            errors=[
                direct_error_msg,
                f"manager domain {manager_normalized} did not serve adagents.json",
            ],
        )
    except (AdagentsValidationError, AdagentsTimeoutError) as e:
        return AdAgentsValidationResult(
            domain=normalized,
            url=url,
            discovery_method="ads_txt_managerdomain",
            manager_domain=manager_normalized,
            errors=[direct_error_msg, str(e)],
        )

    return AdAgentsValidationResult(
        domain=normalized,
        url=url,
        discovery_method="ads_txt_managerdomain",
        manager_domain=manager_normalized,
        data=manager_data,
        valid=True,
    )

def validate_adagents_structure(
    adagents_data: dict[str, Any], *, source_url: str | None = None
) -> AdagentsValidationReport:
    """Structurally validate a parsed adagents.json against the AdCP schema.

    Use this to distinguish a schema-invalid file from a valid file that
    doesn't list a particular agent. :func:`get_properties_by_agent`
    returns ``[]`` for both cases, which makes "publisher hasn't
    authorized us yet" indistinguishable from "publisher's file is
    structurally broken." This helper reports per-entry violations
    against the authoritative ``authorized_agents`` oneOf in the AdCP
    adagents.json schema.

    The two real-world failure modes this catches in production
    publisher files are:

    * **Bare entries** — ``{url, authorized_for}`` with no
      ``authorization_type``. The agent looks listed, but matches no
      schema variant, so the SDK treats the entry as authorizing
      nothing.
    * **Wrong selector for type** — e.g.,
      ``{authorization_type: "property_ids", property_tags: [...]}``,
      where the discriminator and selector array disagree.

    Args:
        adagents_data: Parsed adagents.json (the dict returned by
            :func:`fetch_adagents` or loaded directly from JSON).
        source_url: Final URL that supplied the document. Required when
            validating a catalog containing ``reference_renderer`` because
            ``catalog_role`` is self-declared and does not establish trust.

    Returns:
        :class:`AdagentsValidationReport`. ``schema_valid`` is True only
        when every entry in ``authorized_agents`` satisfies the schema.

    Raises:
        AdagentsValidationError: If ``adagents_data`` is not a dict, or
            ``authorized_agents`` is not a list. These are
            input-shape errors, not per-entry schema violations.

    Notes:
        * URL-reference variants (``authoritative_location`` form) have
          no inline ``authorized_agents`` array. They're reported with
          ``is_reference=True``, ``authorized_agents_count == 0``, and
          ``schema_valid=True``. Callers should follow the redirect
          (e.g., via :func:`fetch_adagents`, which resolves it
          automatically) and re-validate the resolved file.
        * The schema targets AdCP 3.2. Files written against 2.5 (no
          signal_ids / signal_tags variants) will flag those entries as
          ``unknown_authorization_type`` — correct for the 3.2 target,
          but worth knowing if you're validating mixed-version traffic.
        * Selector-array *item* patterns (e.g., the
          ``^[a-zA-Z0-9_-]+$`` constraint on each signal_id) are out of
          scope. This helper validates the discriminator + required
          selector array; it does not deep-validate selector contents.
    """
    if not isinstance(adagents_data, dict):
        raise AdagentsValidationError("adagents_data must be a dictionary")

    authorized_agents = adagents_data.get("authorized_agents")
    if authorized_agents is None:
        # URL-reference variant: file points at an authoritative_location
        # rather than carrying an inline authorized_agents array.
        properties = adagents_data.get("properties", [])
        is_reference = isinstance(adagents_data.get("authoritative_location"), str)
        missing_errors: list[AdagentsEntryError] = []
        if not is_reference:
            missing_errors.append(
                AdagentsEntryError(
                    index=-1,
                    kind="missing_authorized_agents",
                    message="inline adagents.json requires an authorized_agents array",
                )
            )
        return AdagentsValidationReport(
            schema_valid=not missing_errors,
            errors=missing_errors,
            authorized_agents_count=0,
            properties_count=len(properties) if isinstance(properties, list) else 0,
            is_reference=is_reference,
        )

    if not isinstance(authorized_agents, list):
        raise AdagentsValidationError("'authorized_agents' must be an array")

    properties = adagents_data.get("properties", [])
    properties_count = len(properties) if isinstance(properties, list) else 0

    errors: list[AdagentsEntryError] = []

    has_catalog_content = any(
        isinstance(adagents_data.get(field_name), list) and adagents_data[field_name]
        for field_name in _CATALOG_CONTENT_FIELDS
    )
    if len(authorized_agents) == 0 and not has_catalog_content:
        # Beta.3 permits an empty authorization list for catalog-only files,
        # but a file with neither authorization nor catalog content is invalid.
        errors.append(
            AdagentsEntryError(
                index=-1,
                kind="empty_authorized_agents",
                message=(
                    "adagents.json requires at least one authorized agent or a non-empty "
                    "formats, properties, placements, collections, or signals catalog"
                ),
            )
        )

    errors.extend(_reference_renderer_catalog_errors(adagents_data, source_url=source_url))

    for index, entry in enumerate(authorized_agents):
        if not isinstance(entry, dict):
            errors.append(
                AdagentsEntryError(
                    index=index,
                    kind="not_an_object",
                    message=f"authorized_agents[{index}] is not a JSON object",
                )
            )
            continue

        raw_url = entry.get("url")
        url = raw_url if isinstance(raw_url, str) and raw_url else None

        if url is None:
            errors.append(
                AdagentsEntryError(
                    index=index,
                    kind="missing_url",
                    message=f"authorized_agents[{index}] is missing required 'url'",
                )
            )

        authorized_for = entry.get("authorized_for")
        if not isinstance(authorized_for, str) or not authorized_for:
            errors.append(
                AdagentsEntryError(
                    index=index,
                    kind="missing_authorized_for",
                    message=(
                        f"authorized_agents[{index}] is missing required "
                        "'authorized_for' description (string, minLength 1)"
                    ),
                    url=url,
                )
            )

        authorization_type = entry.get("authorization_type")
        if authorization_type is None:
            errors.append(
                AdagentsEntryError(
                    index=index,
                    kind="missing_authorization_type",
                    message=(
                        f"authorized_agents[{index}] is missing required "
                        "'authorization_type' discriminator (expected one of: "
                        f"{', '.join(sorted(_AUTHORIZATION_TYPE_TO_SELECTOR))})"
                    ),
                    url=url,
                )
            )
            continue

        if authorization_type not in _AUTHORIZATION_TYPE_TO_SELECTOR:
            errors.append(
                AdagentsEntryError(
                    index=index,
                    kind="unknown_authorization_type",
                    message=(
                        f"authorized_agents[{index}] has unknown "
                        f"authorization_type={authorization_type!r} "
                        f"(expected one of: "
                        f"{', '.join(sorted(_AUTHORIZATION_TYPE_TO_SELECTOR))})"
                    ),
                    url=url,
                )
            )
            continue

        required_selector = _AUTHORIZATION_TYPE_TO_SELECTOR[authorization_type]
        selector_value = entry.get(required_selector)
        if not isinstance(selector_value, list) or len(selector_value) == 0:
            errors.append(
                AdagentsEntryError(
                    index=index,
                    kind="missing_selector_for_type",
                    message=(
                        f"authorized_agents[{index}] has "
                        f"authorization_type={authorization_type!r} but is "
                        f"missing required non-empty {required_selector!r} array"
                    ),
                    url=url,
                )
            )

    return AdagentsValidationReport(
        schema_valid=not errors,
        errors=errors,
        authorized_agents_count=len(authorized_agents),
        properties_count=properties_count,
    )

def validate_agent_authorization(agent: dict[str, Any]) -> None:
    """Validate agent authorization discriminated union.

    AdCP v2.4.0+ uses discriminated unions with authorization_type discriminator:
    - authorization_type: "property_ids" requires property_ids
    - authorization_type: "property_tags" requires property_tags
    - authorization_type: "inline_properties" requires properties
    - authorization_type: "publisher_properties" requires publisher_properties

    For backward compatibility, also validates the old mutual exclusivity constraint.

    Args:
        agent: An agent dict from adagents.json

    Raises:
        ValidationError: If discriminator or field constraints are violated
    """
    authorization_type = agent.get("authorization_type")
    auth_fields = ["properties", "property_ids", "property_tags", "publisher_properties"]
    present_fields = [field for field in auth_fields if field in agent and agent[field] is not None]

    # If authorization_type discriminator is present, validate discriminated union
    if authorization_type:
        if authorization_type == "property_ids" and "property_ids" not in present_fields:
            raise ValidationError(
                "Agent with authorization_type='property_ids' must have property_ids"
            )
        elif authorization_type == "property_tags" and "property_tags" not in present_fields:
            raise ValidationError(
                "Agent with authorization_type='property_tags' must have property_tags"
            )
        elif authorization_type == "inline_properties" and "properties" not in present_fields:
            raise ValidationError(
                "Agent with authorization_type='inline_properties' must have properties"
            )
        elif (
            authorization_type == "publisher_properties"
            and "publisher_properties" not in present_fields
        ):
            raise ValidationError(
                "Agent with authorization_type='publisher_properties' "
                "must have publisher_properties"
            )
        elif authorization_type not in (
            "property_ids",
            "property_tags",
            "inline_properties",
            "publisher_properties",
        ):
            raise ValidationError(f"Agent has invalid authorization_type: {authorization_type}")

    # Validate mutual exclusivity (for both old and new formats)
    if len(present_fields) > 1:
        raise ValidationError(
            f"Agent authorization cannot have multiple fields: {', '.join(present_fields)}. "
            f"Only one of {', '.join(auth_fields)} is allowed."
        )

    if len(present_fields) == 0:
        raise ValidationError(
            f"Agent authorization must have exactly one of: {', '.join(auth_fields)}."
        )

    # If using publisher_properties, validate each item
    if "publisher_properties" in present_fields:
        for pub_prop in agent["publisher_properties"]:
            validate_publisher_properties_item(pub_prop)

def validate_capabilities(
    handler: Any,
    capabilities: GetAdcpCapabilitiesResponse,
) -> list[str]:
    """Check that a handler implements the methods required by its declared features.

    Compares the features declared in a capabilities response against the handler's
    method implementations. Returns warnings for features that are declared but
    whose corresponding handler methods are not overridden from the base class.

    This is a development-time check — call it at startup to catch misconfigurations.

    Args:
        handler: An ADCPHandler instance (or any object with handler methods).
        capabilities: The capabilities response the handler will serve.

    Returns:
        List of warning strings. Empty if everything is consistent.
    """
    # Late import to avoid circular dependency: server.base imports from adcp.types
    # which may transitively import from this module.
    from adcp.server.base import ADCPHandler

    resolver = FeatureResolver(capabilities)
    warnings: list[str] = []

    for feature, handler_methods in FEATURE_HANDLER_MAP.items():
        if not resolver.supports(feature):
            continue

        for method_name in handler_methods:
            if not hasattr(handler, method_name):
                warnings.append(
                    f"Feature '{feature}' is declared but handler has no " f"'{method_name}' method"
                )
                continue

            # Walk MRO to check if any class between the leaf and ADCPHandler
            # overrides the method (handles mixin / intermediate-class patterns).
            if isinstance(handler, ADCPHandler):
                overridden = any(
                    method_name in cls.__dict__
                    for cls in type(handler).__mro__
                    if cls is not ADCPHandler and not issubclass(ADCPHandler, cls)
                )
                if not overridden:
                    warnings.append(
                        f"Feature '{feature}' is declared but '{method_name}' "
                        f"is not overridden from ADCPHandler"
                    )

    return warnings

def validate_product(product: dict[str, Any]) -> None:
    """Validate a Product object.

    Args:
        product: Product dict

    Raises:
        ValidationError: If validation fails
    """
    if "publisher_properties" in product and product["publisher_properties"]:
        for item in product["publisher_properties"]:
            validate_publisher_properties_item(item)

def validate_publisher_properties_item(item: Any) -> None:
    """Validate a single ``publisher_properties[]`` entry.

    Accepts either a raw ``dict`` (the wire form) or a parsed Pydantic
    model instance (``PublisherPropertySelector1`` / ``…2`` / ``…3``).
    For Pydantic instances the model is coerced via
    ``.model_dump(exclude_none=False)`` and the same checks apply.

    Two XORs are enforced per the publisher-property-selector JSON Schema
    (adcp#4504):

    * Selector XOR: exactly one of ``property_ids`` / ``property_tags``
      is present for ``by_id`` / ``by_tag`` (``all`` requires neither).
    * Publisher XOR: exactly one of ``publisher_domain`` (singular) or
      ``publisher_domains`` (compact array) is present — both or neither
      both fail. ``publisher_domains`` is NOT allowed on
      ``selection_type='by_id'`` since property IDs are publisher-scoped;
      callers wanting per-publisher ID sets must use one entry per
      publisher.

    Why the Pydantic input form matters: ``datamodel-code-generator``
    cannot translate the JSON Schema's
    ``allOf[not[required[both]]] + anyOf[required[either]]`` construct
    into Pydantic field constraints, so the typed surface (selector 1/3
    direct instantiation) is laxer than the schema. Consumers parsing
    via Pydantic should call this helper post-construction to close the
    gap.

    Args:
        item: A single item from publisher_properties array — either a
            ``dict`` or a Pydantic ``BaseModel`` instance.

    Raises:
        ValidationError: If discriminator or field constraints are violated
    """
    if hasattr(item, "model_dump"):
        item = item.model_dump(exclude_none=False)
    if not isinstance(item, dict):
        raise ValidationError(
            "publisher_properties item must be a dict or a Pydantic model "
            f"instance, got {type(item).__name__}"
        )
    selection_type = item.get("selection_type")
    has_property_ids = "property_ids" in item and item["property_ids"] is not None
    has_property_tags = "property_tags" in item and item["property_tags"] is not None
    has_publisher_domain = "publisher_domain" in item and item["publisher_domain"] is not None
    publisher_domains = item.get("publisher_domains")
    has_publisher_domains = publisher_domains is not None

    if selection_type:
        if selection_type == "by_id" and not has_property_ids:
            raise ValidationError(
                "publisher_properties item with selection_type='by_id' must have property_ids"
            )
        elif selection_type == "by_tag" and not has_property_tags:
            raise ValidationError(
                "publisher_properties item with selection_type='by_tag' must have property_tags"
            )
        elif selection_type not in ("all", "by_id", "by_tag"):
            raise ValidationError(
                f"publisher_properties item has invalid selection_type: {selection_type}"
            )

    if has_property_ids and has_property_tags:
        raise ValidationError(
            "publisher_properties item cannot have both property_ids and property_tags. "
            "These fields are mutually exclusive."
        )

    # selection_type='all' carries neither selector array; older callers
    # without the discriminator must still provide one of the two.
    if selection_type not in ("all",) and not has_property_ids and not has_property_tags:
        raise ValidationError(
            "publisher_properties item must have either property_ids or property_tags. "
            "At least one is required."
        )

    if has_publisher_domain and has_publisher_domains:
        raise ValidationError(
            "publisher_properties item cannot have both publisher_domain and "
            "publisher_domains. These fields are mutually exclusive (XOR)."
        )

    if not has_publisher_domain and not has_publisher_domains:
        raise ValidationError(
            "publisher_properties item must have exactly one of publisher_domain "
            "or publisher_domains."
        )

    if has_publisher_domains and selection_type == "by_id":
        # by_id is single-publisher only — property IDs are publisher-scoped,
        # so fanning the same ID set across multiple publishers is meaningless.
        raise ValidationError(
            "publisher_properties item with selection_type='by_id' cannot use "
            "publisher_domains[]; property IDs are publisher-scoped. Use one "
            "entry per publisher with publisher_domain."
        )

    if has_publisher_domains:
        if not isinstance(publisher_domains, list) or len(publisher_domains) == 0:
            raise ValidationError(
                "publisher_properties item publisher_domains must be a non-empty array"
            )
        if any(not isinstance(d, str) or not d for d in publisher_domains):
            raise ValidationError(
                "publisher_properties item publisher_domains entries must be non-empty strings"
            )
        if len(set(publisher_domains)) != len(publisher_domains):
            raise ValidationError(
                "publisher_properties item publisher_domains entries must be unique"
            )

def validate_webhook_challenge_response(
    response: bytes | Mapping[str, Any],
    *,
    challenge: str,
    field: str | None = None,
    url: str | None = None,
) -> str:
    """Validate that a receiver echoed the challenge value.

    Receivers may respond with either ``{"challenge": "<value>"}`` or
    ``{"token": "<value>"}``. The return value is the field that matched.
    """

    try:
        if isinstance(response, bytes):
            decoded = json.loads(response.decode("utf-8"))
        else:
            decoded = dict(response)
    except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError) as exc:
        raise WebhookChallengeError(
            "webhook challenge response must be a JSON object",
            reason="invalid_json",
            field=field,
            url=url,
        ) from exc

    if not isinstance(decoded, Mapping):
        raise WebhookChallengeError(
            "webhook challenge response must be a JSON object",
            reason="invalid_json",
            field=field,
            url=url,
        )

    for key in ("challenge", "token"):
        value = decoded.get(key)
        if value == challenge:
            return key

    if "challenge" in decoded or "token" in decoded:
        reason = "challenge_mismatch"
        message = "webhook challenge response did not echo the expected value"
    else:
        reason = "missing_echo"
        message = "webhook challenge response must include 'challenge' or 'token'"
    raise WebhookChallengeError(message, reason=reason, field=field, url=url)

def verify_agent_authorization(
    adagents_data: dict[str, Any],
    agent_url: str,
    property_type: str | None = None,
    property_identifiers: list[dict[str, str]] | None = None,
) -> bool:
    """Check if agent is authorized for a property.

    Args:
        adagents_data: Parsed adagents.json data
        agent_url: URL of the sales agent to verify
        property_type: Type of property (website, app, etc.) - optional
        property_identifiers: List of identifiers to match - optional

    Returns:
        True if agent is authorized, False otherwise

    Raises:
        AdagentsValidationError: If adagents_data is malformed

    Notes:
        - If property_type/identifiers are None, checks if agent is authorized
          for ANY property on this domain
        - Implements AdCP domain matching rules
        - Agent URLs are matched ignoring protocol and trailing slash
    """
    # Validate structure
    if not isinstance(adagents_data, dict):
        raise AdagentsValidationError("adagents_data must be a dictionary")

    authorized_agents = adagents_data.get("authorized_agents")
    if not isinstance(authorized_agents, list):
        raise AdagentsValidationError("adagents.json must have 'authorized_agents' array")

    # Normalize the agent URL for comparison
    normalized_agent_url = normalize_url(agent_url)

    # Check each authorized agent
    for agent in authorized_agents:
        if not isinstance(agent, dict):
            continue

        agent_url_from_json = agent.get("url", "")
        if not agent_url_from_json:
            continue

        # Match agent URL (protocol-agnostic)
        if normalize_url(agent_url_from_json) != normalized_agent_url:
            continue

        # Found matching agent - now check properties
        properties = agent.get("properties")

        # If properties field is missing or empty, agent is authorized for all properties
        if properties is None or (isinstance(properties, list) and len(properties) == 0):
            return True

        # If no property filters specified, we found the agent - authorized
        if property_type is None and property_identifiers is None:
            return True

        # Check specific property authorization
        if isinstance(properties, list):
            for prop in properties:
                if not isinstance(prop, dict):
                    continue

                # Check property type if specified
                if property_type is not None:
                    prop_type = prop.get("property_type", "")
                    if prop_type != property_type:
                        continue

                # Check identifiers if specified
                if property_identifiers is not None:
                    prop_identifiers = prop.get("identifiers", [])
                    if not isinstance(prop_identifiers, list):
                        continue

                    if identifiers_match(property_identifiers, prop_identifiers):
                        return True
                else:
                    # Property type matched and no identifier check needed
                    return True

    return False

async def verify_agent_for_property(
    publisher_domain: str,
    agent_url: str,
    property_identifiers: list[dict[str, str]],
    property_type: str | None = None,
    timeout: float = 10.0,
    client: httpx.AsyncClient | None = None,
) -> bool:
    """Convenience wrapper to fetch adagents.json and verify authorization in one call.

    Args:
        publisher_domain: Domain hosting the adagents.json file
        agent_url: URL of the sales agent to verify
        property_identifiers: List of identifiers to match
        property_type: Type of property (website, app, etc.) - optional
        timeout: Request timeout in seconds
        client: Optional httpx.AsyncClient for connection pooling

    Returns:
        True if agent is authorized, False otherwise

    Raises:
        AdagentsNotFoundError: If adagents.json not found (404)
        AdagentsValidationError: If JSON is invalid or malformed
        AdagentsTimeoutError: If request times out
    """
    adagents_data = await fetch_adagents(publisher_domain, timeout=timeout, client=client)
    return verify_agent_authorization(
        adagents_data=adagents_data,
        agent_url=agent_url,
        property_type=property_type,
        property_identifiers=property_identifiers,
    )

class ADCPAuthenticationError(ADCPError):
    """Authentication failed (401, 403).

    `is_retryable` defaults to ``False`` (inherited). Per the AdCP 3.0.4 prose
    tightening, `AUTH_REQUIRED` covers two sub-cases: credentials missing
    (correctable — supply credentials and retry) and credentials presented but
    rejected (terminal — re-presenting creates SSO retry-storm patterns).
    Defaulting to non-retryable is the safe biased-toward-the-dangerous-case
    choice; callers handling the missing-credentials case should retry only
    after attaching credentials, not on a timer. The 3.1 line splits this
    into `AUTH_MISSING` and `AUTH_INVALID`.
    """

    def __init__(self, message: str, agent_id: str | None = None, agent_uri: str | None = None):
        """Initialize authentication error."""
        suggestion = (
            "Check that your auth_token is valid and not expired.\n"
            "     Verify auth_type ('bearer' vs 'token') and auth_header are correct.\n"
            "     Some agents (like Optable) require auth_type='bearer' and "
            "auth_header='Authorization'"
        )
        super().__init__(message, agent_id, agent_uri, suggestion)

class ADCPClient:
    """Client for interacting with a single AdCP agent."""

    def __init__(
        self,
        agent_config: AgentConfig,
        webhook_url_template: str | None = None,
        webhook_secret: str | None = None,
        on_activity: Callable[[Activity], None] | None = None,
        webhook_timestamp_tolerance: int = 300,
        capabilities_ttl: float = 3600.0,
        validate_features: bool = False,
        strict_idempotency: bool = False,
        signing: SigningConfig | None = None,
        context_id: str | None = None,
        validation: ValidationHookConfig | None = None,
        force_a2a_version: str | None = None,
        adcp_version: str | None = None,
        server_version: str | None = None,
        legacy_format_converter: LegacyFormatConverter | None = None,
        canonical_format_legacy_resolver: CanonicalFormatLegacyResolver | None = None,
        allow_unauthenticated_webhooks: bool = False,
    ):
        """
        Initialize ADCP client for a single agent.

        Args:
            agent_config: Agent configuration
            webhook_url_template: Template for webhook URLs with {agent_id},
                {task_type}, {operation_id}
            webhook_secret: Shared secret for the deprecated HMAC-SHA256 webhook
                fallback. Configure this only when the registration explicitly
                selected legacy HMAC; conformant public endpoints should use
                :class:`adcp.webhooks.WebhookReceiver` for RFC 9421 verification.
            allow_unauthenticated_webhooks: Explicit compatibility escape for
                accepting unsigned MCP webhooks when ``webhook_secret`` is not
                configured. Defaults to False so public webhook receivers fail
                closed. Only enable this for endpoints that cannot be reached
                from an untrusted network. A2A webhook handling is unaffected.
            on_activity: Callback for activity events
            webhook_timestamp_tolerance: Maximum age (in seconds) for webhook
                timestamps. Webhooks with timestamps older than this or more than
                this far in the future are rejected. Defaults to 300 (5 minutes).
            capabilities_ttl: Time-to-live in seconds for cached capabilities (default: 1 hour)
            validate_features: When True, automatically check that the seller supports
                required features before making task calls (e.g., sync_audiences requires
                audience_targeting). Requires capabilities to have been fetched first.
            strict_idempotency: When True, verify the seller declared
                ``adcp.idempotency.replay_ttl_seconds`` in capabilities before any
                mutating call. Fetches capabilities lazily on first use. Raises
                ``IdempotencyUnsupportedError`` if the declaration is missing —
                sellers that don't declare it provide no retry-safety guarantee
                per AdCP #2315. Defaults to False for backward compatibility.
            signing: Optional RFC 9421 request-signing config. When provided,
                the client automatically attaches ``Signature`` /
                ``Signature-Input`` / ``Content-Digest`` headers to operations
                the seller's ``request_signing`` capability lists in
                ``required_for``, ``warn_for``, or ``supported_for``. The
                seller's ``covers_content_digest`` policy determines whether
                the body is bound to the signature. Generate a key with
                ``adcp-keygen`` and publish the public JWK at your
                ``jwks_uri``. Supported on both A2A and MCP
                (``mcp_transport="streamable_http"``); SSE-transport MCP
                logs a warning and falls through unsigned.
            validation: Schema-driven validation modes for outgoing
                requests and incoming responses against the bundled AdCP
                JSON schemas. Defaults (matching the TS port): requests
                in ``warn`` mode (drift logged but not blocked — partial
                payloads in error-path tests still work) and responses
                in ``strict`` mode (agent drift fails the task).
                ``ADCP_VALIDATION_MODE=strict|warn|off`` overrides both
                sides at call time (matches the TS port); ``ADCP_ENV``
                set to ``production`` / ``prod`` flips only the response
                default to ``warn``. Generic ``ENV`` / ``ENVIRONMENT`` /
                ``PYTHON_ENV`` are deliberately ignored — they collide
                with unrelated tooling. Storyboards and compliance
                runners that want hard-stop enforcement everywhere pass
                ``validation=ValidationHookConfig(requests="strict",
                responses="strict")``; high-throughput callers can set
                either side to ``"off"`` to skip the validator entirely
                with zero overhead.
            context_id: A2A-only. Seed the A2A conversation context. Pass a
                previously-returned ``context_id`` to resume a session
                across process restarts, or a self-assigned UUID to name
                the session with your own correlation key (the ADK server
                honors buyer-proposed ids). If omitted, the server mints
                one on the first message and this client auto-retains it
                for subsequent calls. Read the current value via
                ``client.context_id``; call ``client.reset_context()`` to
                start a fresh conversation. Rule of thumb: one
                ``ADCPClient`` per A2A conversation — if a buyer has
                multiple concurrent briefs with the same agent, construct
                one client per brief rather than sharing.

                For HITL flows that can span a process restart mid-task,
                use ``checkpoint()`` / ``from_checkpoint()`` instead of
                persisting ``context_id`` alone — full resume state is
                both ``context_id`` AND ``active_task_id``.

                Raises ``TypeError`` if passed with a non-A2A protocol.
            force_a2a_version: A2A-only. Pin the **A2A transport
                version** (e.g. ``"0.3"``, ``"1.0"``) by filtering the
                peer's advertised ``supported_interfaces`` to entries
                whose ``protocol_version`` matches. Not for AdCP
                protocol pinning — see ``adcp_version`` for that.
                Intended for tests or for forcing a 0.3-speaking path
                against a dual-advertising peer. Raises
                :class:`ADCPConnectionError` on the first call if no
                advertised interface matches. ``None`` (default) lets
                the SDK's ``ClientFactory`` pick the most capable
                transport the peer supports. Use
                :attr:`a2a_protocol_versions` to probe what a peer
                advertises before pinning.

                Raises ``TypeError`` if passed with a non-A2A protocol.
            adcp_version: AdCP protocol release this client speaks
                (release-precision string, e.g. ``"3.0"``, ``"3.1"``,
                ``"3.1-beta"``). Stripe-style per-instance pin: the
                value is sent as ``adcp_version`` on every outbound
                request and selects creative dialect behavior. ``None``
                (default) resolves
                to the SDK's compile-time pin (``ADCP_VERSION``
                packaged with the wheel). Cross-major pins raise
                :class:`ConfigurationError` at construction; install
                the SDK major that targets your wire version instead.
                Patch-precision strings (``"3.0.1"``) and build
                metadata (``"3.0.1+canary"``) are accepted at construction
                but normalized to release-precision before wire emission
                per the spec — patches and build metadata are not part
                of the negotiation contract. ``get_adcp_version()``
                returns the normalized form.

                Caller-supplied ``adcp_version`` on a per-call params
                dict wins over the constructor pin: the enricher is
                the default, not an override.

                Migration from ``adcp_major_version`` (legacy integer
                wire field): generated request types still expose
                ``adcp_major_version: int | None`` from the pre-#3493
                schema. Both fields will coexist on the wire through
                3.x; servers prefer the new ``adcp_version`` when both
                are present. Stop populating ``adcp_major_version`` on
                request models once your seller advertises 3.1 in
                ``supported_versions``.
            server_version: AdCP wire shape the *seller* speaks. Most
                adopters leave this ``None`` — the SDK assumes a v3
                seller and the seller's
                ``/.well-known/agent-card.json`` is the canonical
                source of truth once a probe-and-cache path lands.

                Pin explicitly when:

                * You're talking to a known-legacy seller (e.g.
                  ``server_version="3.0"``). Canonical discovery results
                  are upgraded for application code and canonical writes
                  are downgraded only through preserved or explicit routes.
                * You want telemetry to attribute outbound traffic to
                  a specific server-side version regardless of what the
                  seller advertises.

                Retrieve the current value via :meth:`get_server_version`.
        """
        self._adcp_version: str = resolve_adcp_version(adcp_version)
        self._server_version: str | None = _resolve_server_version(server_version)
        if type(allow_unauthenticated_webhooks) is not bool:
            raise TypeError("allow_unauthenticated_webhooks must be a bool")

        self.agent_config = agent_config
        self.webhook_url_template = webhook_url_template
        self.webhook_secret = webhook_secret
        self.allow_unauthenticated_webhooks = allow_unauthenticated_webhooks
        self.on_activity = on_activity
        self.webhook_timestamp_tolerance = webhook_timestamp_tolerance
        self.capabilities_ttl = capabilities_ttl
        self.validate_features = validate_features
        self.strict_idempotency = strict_idempotency
        self.signing = signing
        if (
            signing is not None
            and agent_config.agent_uri.startswith("http://")
            and not is_loopback_http_uri(agent_config.agent_uri)
        ):
            raise ValueError(
                "request signing requires an https:// agent_uri for non-loopback hosts; "
                "plain HTTP is only allowed for localhost/loopback development"
            )
        self.legacy_format_converter = legacy_format_converter
        self.canonical_format_legacy_resolver = canonical_format_legacy_resolver
        self._canonical_legacy_routes: dict[
            tuple[str | None, str | None, str], list[dict[str, Any]] | None
        ] = {}

        # Capabilities cache
        self._capabilities: GetAdcpCapabilitiesResponse | None = None
        self._feature_resolver: FeatureResolver | None = None
        self._capabilities_fetched_at: float | None = None
        self._idempotency_capability_verified: bool = False
        # Unique per-instance token so use_idempotency_key scopes to this
        # client and does not bleed to siblings (AdCP #2315 cross-seller risk).
        from uuid import uuid4 as _uuid4

        self._idempotency_client_token: str = _uuid4().hex

        if force_a2a_version is not None and agent_config.protocol != Protocol.A2A:
            raise TypeError(
                f"force_a2a_version is only supported for A2A protocol; got {agent_config.protocol}"
            )

        # Initialize protocol adapter
        self.adapter: ProtocolAdapter
        if agent_config.protocol == Protocol.A2A:
            self.adapter = A2AAdapter(agent_config, force_a2a_version=force_a2a_version)
        elif agent_config.protocol == Protocol.MCP:
            self.adapter = MCPAdapter(agent_config)
        else:
            raise ValueError(f"Unsupported protocol: {agent_config.protocol}")

        self.adapter.idempotency_client_token = self._idempotency_client_token
        if strict_idempotency:
            self.adapter.idempotency_capability_check = self._ensure_idempotency_capability
        if signing is not None:
            self.adapter.signing_request_hook = self._sign_outgoing_request
            self.adapter.signing_capability_check = self._prepare_signing_capabilities
        # Apply schema validation modes (default: requests=warn, responses=strict
        # in dev/test, warn in production — see ``ValidationHookConfig`` docs).
        self.adapter.configure_validation(validation)
        # Auto-inject the per-instance ``adcp_version`` pin into every
        # outbound request envelope. Caller-supplied values on the
        # request object win — the enricher is the default, not an
        # override — so per-call overrides remain available once the
        # generated request types declare the field.
        _pinned_version = self._server_version or self._adcp_version
        self._signing_profile_version = (
            None
            if signing is None
            else signing.signing_profile_version
            or signing_profile_for_adcp_version(_pinned_version)
        )

        def _inject_adcp_version(params: dict[str, Any]) -> dict[str, Any]:
            return {"adcp_version": _pinned_version, **params}

        self.adapter.envelope_enricher = _inject_adcp_version

        if context_id:
            # Empty string is treated as "not provided" — callers using
            # ``context_id=os.getenv("...") or ""`` patterns shouldn't
            # silently seed an empty id on the wire.
            if not isinstance(self.adapter, A2AAdapter):
                raise TypeError(
                    f"context_id is only supported for A2A protocol; got {agent_config.protocol}"
                )
            self.adapter.set_context_id(context_id)

        # Initialize simple API accessor (lazy import to avoid circular dependency)
        from adcp.simple import SimpleAPI

        self.simple = SimpleAPI(self)

    def get_adcp_version(self) -> str:
        """Return the AdCP protocol release this client is pinned to.

        Resolved at construction from the ``adcp_version`` kwarg, with
        fallback to the SDK's compile-time pin (``ADCP_VERSION``
        packaged with the wheel) when the caller didn't pin
        explicitly. Same value across the client's lifetime — the pin
        is per-instance, not per-call.

        See ``__init__``'s ``adcp_version`` parameter for the full
        semantics, including the cross-major fence and dialect selection.
        """
        return self._adcp_version

    def get_server_version(self) -> str | None:
        """Return the seller's AdCP wire-shape version, or ``None``.

        ``None`` means the SDK is assuming a current-major seller
        (the default). Returns a release-precision string
        (``"3.0"``, ``"3.1"``, ``"2.5"``) when the adopter pinned
        via the ``server_version`` constructor arg or — once the
        agent-card probe lands — when the SDK detected the seller's
        version from its agent-card.

        See ``__init__``'s ``server_version`` parameter for negotiated
        canonical/legacy creative behavior.
        """
        return self._server_version

    @property
    def context_id(self) -> str | None:
        """Current A2A conversation context_id.

        Reads the context_id currently associated with this client: the
        value assigned by the A2A server (auto-captured from the most
        recent response) or the one seeded via the constructor or
        ``reset_context()``. Returns ``None`` before the first A2A call
        in a fresh conversation, or for clients on non-A2A protocols —
        reads are lenient across protocols so generic code can probe
        ``if client.context_id: ...`` safely. Writes (constructor kwarg,
        ``reset_context``) raise on non-A2A because the operation has no
        meaning there.

        Not safe for concurrent calls on the same client — the adapter
        mutates this on every response. Rule of thumb: one ADCPClient
        per A2A conversation.

        For simple completed-task resume, persist this value and pass
        it to ``ADCPClient(context_id=...)``. For HITL flows that may
        restart mid-``input-required``, use ``checkpoint()`` /
        ``from_checkpoint()`` — full resume state is both this id AND
        ``active_task_id``.
        """
        if isinstance(self.adapter, A2AAdapter):
            return self.adapter.context_id
        return None

    @property
    def active_task_id(self) -> str | None:
        """A2A task_id the next send must echo to resume the same task.

        Set when the last A2A response was non-terminal
        (``input-required``, ``working``, ``submitted``,
        ``auth-required``). The adapter echoes this id on the next
        outbound message so the server resumes the same task. Clears
        automatically when the task reaches a terminal state.

        Full resume state is *both* ``context_id`` and
        ``active_task_id`` — persist both (or use ``checkpoint()``) to
        survive a process restart mid-HITL without orphaning the task.

        Returns ``None`` for non-A2A clients.
        """
        if isinstance(self.adapter, A2AAdapter):
            return self.adapter.active_task_id
        return None

    @property
    def a2a_protocol_versions(self) -> list[str] | None:
        """A2A ``protocol_version`` strings the peer advertises, sorted.

        Lazily populated after the first operation that fetches the
        peer's ``AgentCard`` (``fetch_capabilities``, ``list_tools``,
        ``get_agent_info``, or any skill-call). Returns ``None`` before
        the card has been fetched so callers can distinguish "not yet
        known" from "peer advertises nothing" (empty list). Returns
        ``None`` for non-A2A clients.

        Useful for probing which wire version a peer speaks — buyers
        running alongside both 0.3-era and 1.0-era agents can use this
        to confirm what they're talking to.
        """
        if isinstance(self.adapter, A2AAdapter):
            return self.adapter.a2a_protocol_versions
        return None

    def reset_context(self, context_id: str | None = None) -> None:
        """Start a new A2A conversation on this client.

        Passing ``None`` (default) clears the current context so the
        server mints a fresh one on the next call. Passing a string uses
        it as the new conversation id — useful for resuming a specific
        prior session or for naming the conversation with your own
        correlation key. Note: some servers (notably ADK) rewrite
        client-supplied ids into their own session format; the client
        auto-adopts the rewritten id on the next response.

        Also clears any active_task_id — starting a new conversation
        discards any in-flight task on the old one.

        Raises ``TypeError`` when called on a non-A2A client.
        """
        if not isinstance(self.adapter, A2AAdapter):
            raise TypeError(
                f"reset_context is only supported for A2A protocol; "
                f"got {self.agent_config.protocol}"
            )
        self.adapter.set_context_id(context_id)

    def checkpoint(self) -> Checkpoint:
        """Return the minimal state needed to resume this A2A session.

        Full resume for HITL / multi-turn flows requires *both*
        ``context_id`` (which conversation) AND ``active_task_id``
        (which in-flight task to echo). Persisting only ``context_id``
        reconnects to the right conversation but orphans the pending
        task server-side — the next send starts a new task under the
        same context, and the original ``input-required`` task is
        abandoned.

        The returned dict also carries ``agent_id`` so a later
        ``from_checkpoint`` call against a different ``AgentConfig``
        fails loudly instead of sending one agent's session ids to
        another.

        Pair with ``ADCPClient.from_checkpoint(agent_config, state)``.

        Returns a fully-populated ``Checkpoint`` on non-A2A clients
        with ``context_id``/``active_task_id`` set to ``None``, so
        generic persist-and-restore code can call this without
        branching on protocol.
        """
        return Checkpoint(
            agent_id=self.agent_config.id,
            context_id=self.context_id,
            active_task_id=self.active_task_id,
        )

    @classmethod
    def from_checkpoint(
        cls,
        agent_config: AgentConfig,
        state: Checkpoint,
        **kwargs: Any,
    ) -> ADCPClient:
        """Rehydrate an ADCPClient from a prior ``checkpoint()``.

        Restores both ``context_id`` and ``active_task_id`` so a process
        restart mid-``input-required`` can resume the same task, not
        orphan it. Accepts the same keyword arguments as ``__init__``
        (signing, strict_idempotency, etc.) — the checkpoint only
        carries session-resume state; operational config is re-supplied
        by the caller.

        Raises ``ValueError`` if the checkpoint's ``agent_id`` doesn't
        match ``agent_config.id`` — a checkpoint minted for Agent A
        must not be restored onto Agent B, or the client will leak
        Agent A's opaque session ids to Agent B on the next message.

        Raises ``TypeError`` on a non-A2A ``agent_config`` if the
        checkpoint carries a non-empty ``context_id`` or
        ``active_task_id`` — session-resume state on a protocol that
        doesn't support it would be silently dropped, masking bugs.
        An empty/absent checkpoint round-trips cleanly on any protocol.
        """
        saved_agent_id = state.get("agent_id") if state else None
        if saved_agent_id and saved_agent_id != agent_config.id:
            raise ValueError(
                f"checkpoint was minted for agent {saved_agent_id!r}, "
                f"cannot restore against {agent_config.id!r}"
            )
        context_id = state.get("context_id") if state else None
        active_task_id = state.get("active_task_id") if state else None
        if active_task_id and agent_config.protocol != Protocol.A2A:
            raise TypeError(
                f"active_task_id in checkpoint is only supported for A2A "
                f"protocol; got {agent_config.protocol}"
            )
        client = cls(agent_config, context_id=context_id, **kwargs)
        if active_task_id and isinstance(client.adapter, A2AAdapter):
            client.adapter._restore_active_task_id(active_task_id)
        return client

    @classmethod
    def from_mcp_client(
        cls,
        client: ClientSession,
        *,
        agent_id: str | None = None,
        validation: ValidationHookConfig | None = None,
        capabilities_ttl: float = 3600.0,
        validate_features: bool = False,
        strict_idempotency: bool = False,
    ) -> ADCPClient:
        """Create an ADCPClient wrapping a pre-connected MCP ClientSession.

        Parity with JS ``AgentClient.fromMCPClient()`` (v5.19.0). The primary
        use case is compliance test fleets that wire a full ``ADCPClient``
        against an in-process MCP server without standing up a loopback HTTP
        server.

        Warning:
            The returned client's ``close()`` and ``async with`` ``__aexit__``
            are **no-ops** — the caller owns the injected session and is
            responsible for closing it. Code that relies on ``async with
            ADCPClient.from_mcp_client(...) as c:`` to clean up the session
            will leak the session.

            Webhook delivery and ``on_activity`` callbacks are **not wired**
            on the in-process path — there is no HTTP transport for the
            seller to call back through. Don't pass these to the factory
            (they're absent from the signature on purpose).

            If the injected session has not been initialized
            (``await session.initialize()``), the first tool call surfaces
            as an opaque MCP protocol error in ``TaskResult.error``. The
            factory does not initialize for you — verify before calling.

        **Session lifecycle:** the caller owns the session — ``close()`` and
        ``async with`` exit on the returned client are no-ops. Use your own
        ``AsyncExitStack`` to scope both the transport and the client::

            import contextlib
            from mcp import ClientSession
            from mcp.shared.memory import create_client_server_memory_streams

            async with contextlib.AsyncExitStack() as stack:
                (c_read, c_write), (s_read, s_write) = await stack.enter_async_context(
                    create_client_server_memory_streams()
                )
                # wire your in-process server to (s_read, s_write) here
                session = await stack.enter_async_context(
                    ClientSession(c_read, c_write)
                )
                await session.initialize()
                # close() is a no-op on injected sessions; no stack.enter_async_context needed.
                adcp_client = ADCPClient.from_mcp_client(session, agent_id="test-seller")
                result = await adcp_client.get_products(GetProductsRequest(...))

        Note:
            Request signing is not supported on the injected-session path —
            the signing hook is wired into the HTTP transport layer that is
            bypassed here. ``signing=`` is intentionally absent from this
            factory's parameters.

        Args:
            client: A pre-connected ``mcp.ClientSession`` whose
                ``initialize()`` has already been awaited.
            agent_id: Identifier for the wrapped agent used in log messages
                and error objects. Defaults to a unique ``in-process-XXXXXXXX``
                token; set this explicitly when running multiple in-process
                agents concurrently so log lines are distinguishable.
            validation: Schema-validation modes (same as ``__init__``).
            strict_idempotency: Verify seller declared idempotency support
                before each mutating call (same as ``__init__``).
            validate_features: Gate tool calls on fetched capability
                declarations (same as ``__init__``).
            capabilities_ttl: TTL for the capability cache in seconds
                (same as ``__init__``).

        Returns:
            A fully configured ``ADCPClient`` backed by the injected session.
        """
        effective_id = agent_id if agent_id is not None else f"in-process-{uuid4().hex[:8]}"
        config = AgentConfig(
            id=effective_id,
            # RFC 2606 .invalid TLD — passes the http:// validator, guaranteed
            # not to route to a real host. Self-documenting in error messages.
            agent_uri="http://in-process.invalid",
            protocol=Protocol.MCP,
        )
        instance = cls(
            config,
            validation=validation,
            strict_idempotency=strict_idempotency,
            validate_features=validate_features,
            capabilities_ttl=capabilities_ttl,
        )
        if not isinstance(instance.adapter, MCPAdapter):
            raise RuntimeError(  # pragma: no cover
                f"from_mcp_client: expected MCPAdapter but got {type(instance.adapter).__name__}"
            )
        instance.adapter._inject_session(client)
        return instance

    async def _ensure_idempotency_capability(self) -> None:
        """Verify the seller positively declares idempotency support in capabilities.

        Called before every mutating request when ``strict_idempotency=True``.
        Fetches capabilities on first invocation; subsequent calls are no-ops
        once the declaration has been observed. Raises
        ``IdempotencyUnsupportedError`` when ``adcp.idempotency`` is missing,
        declares ``supported=False`` (seller does not dedupe — naive retry
        would double-process), or declares ``supported=True`` without a
        ``replay_ttl_seconds`` window.

        Sets ``_idempotency_capability_verified = True`` BEFORE calling
        ``fetch_capabilities`` so any recursive dispatch through the adapter
        terminates (``get_adcp_capabilities`` is non-mutating, so it would
        short-circuit anyway — but this guard protects against future refactors
        that might add it to the mutating set).
        """
        from adcp.exceptions import IdempotencyUnsupportedError

        if self._idempotency_capability_verified:
            return

        self._idempotency_capability_verified = True
        try:
            caps = await self.fetch_capabilities()
            adcp_info = getattr(caps, "adcp", None)
            idempotency_info = getattr(adcp_info, "idempotency", None) if adcp_info else None

            if idempotency_info is None:
                raise IdempotencyUnsupportedError(
                    agent_id=self.agent_config.id,
                    agent_uri=self.agent_config.agent_uri,
                    reason="seller did not declare adcp.idempotency",
                )

            supported = getattr(idempotency_info, "supported", None)
            if supported is False:
                raise IdempotencyUnsupportedError(
                    agent_id=self.agent_config.id,
                    agent_uri=self.agent_config.agent_uri,
                    reason="seller declared adcp.idempotency.supported=false",
                )

            ttl = getattr(idempotency_info, "replay_ttl_seconds", None)
            if ttl is None:
                raise IdempotencyUnsupportedError(
                    agent_id=self.agent_config.id,
                    agent_uri=self.agent_config.agent_uri,
                    reason=(
                        "seller declared adcp.idempotency.supported=true but omitted "
                        "replay_ttl_seconds"
                    ),
                )
        except Exception:
            self._idempotency_capability_verified = False
            raise

    async def _prepare_signing_capabilities(self) -> None:
        """Populate signing policy before a transport writer sends a request."""
        await self.fetch_capabilities()

    @staticmethod
    def _mcp_operation_from_request(request: httpx.Request) -> str | None:
        """Extract one MCP ``tools/call`` name from its JSON-RPC body."""
        try:
            payload = json.loads(request.content)
        except (json.JSONDecodeError, TypeError, UnicodeDecodeError):
            return None
        if not isinstance(payload, dict) or payload.get("method") != "tools/call":
            return None
        params = payload.get("params")
        if not isinstance(params, dict):
            return None
        name = params.get("name")
        return name if isinstance(name, str) and name else None

    async def _sign_outgoing_request(self, request: httpx.Request) -> None:
        """httpx request event hook that attaches RFC 9421 signature headers.

        Installed on the protocol adapter's httpx client when a
        ``SigningConfig`` was passed to ``ADCPClient``. Consults the
        seller's advertised ``request_signing`` capability and signs only
        the operations the seller listed in ``required_for``, ``warn_for``,
        or ``supported_for`` — other requests (including the agent-card
        fetch and ``get_adcp_capabilities`` itself) pass through unsigned.
        The ``covers_content_digest`` tri-state determines whether the
        body is bound to the signature.
        """
        if self.signing is None:
            return
        mcp_operation = self._mcp_operation_from_request(request)
        operation = mcp_operation or _signing_current_operation.get()
        # Unset ContextVar → out-of-band call (agent-card fetch, session
        # initialize, etc). Skip without fetching capabilities.
        #
        # get_adcp_capabilities → bootstrap carve-out: signing it would
        # require capabilities we don't have yet, and if a pathological
        # seller listed this op in its own required_for we'd recurse.
        # Keep this check narrow — only operations strictly required to
        # *obtain* capabilities belong here. Today that's just
        # get_adcp_capabilities. A future adapter that adds another
        # capabilities-precondition op MUST extend this guard.
        if operation is None or operation == "get_adcp_capabilities":
            return

        if mcp_operation is not None:
            # MCP's httpx hook runs in a writer task whose ContextVar snapshot
            # was captured when the session connected. The adapter prefetches
            # capabilities in the caller task before enqueueing tools/call;
            # fetching here would deadlock the same writer stream.
            caps = self._capabilities
            if caps is None:
                raise RuntimeError(
                    "MCP request signing policy was not prefetched before tools/call"
                )
        else:
            # A2A's event hook runs in the caller task. Retain this fallback
            # for direct/custom transports, while the bundled adapter also
            # prefetches so normal hooks stay network-free.
            caps = self._capabilities or await self.fetch_capabilities()
        req_signing = getattr(caps, "request_signing", None)

        # Detect and surface a malformed seller config: supported=False is
        # "signatures are ignored", but populating required_for alongside
        # it is contradictory. The classifier correctly skips (matches
        # verifier behavior) but the silent downgrade hides a config bug
        # that will bite pilots.
        if (
            req_signing is not None
            and not req_signing.supported
            and (req_signing.required_for or req_signing.warn_for)
        ):
            logger.warning(
                "Seller %s advertises request_signing.supported=false but "
                "populates required_for/warn_for — treating as unsupported "
                "per spec. Verify the seller's capability advertisement.",
                self.agent_config.id,
            )

        decision = operation_needs_signing(req_signing, operation)
        if decision == "skip":
            return

        covers_policy: str | None = None
        if req_signing is not None and req_signing.covers_content_digest is not None:
            covers_policy = req_signing.covers_content_digest.value
        if covers_policy == "forbidden":
            cover_digest = False
        elif covers_policy == "required":
            cover_digest = True
        else:
            # "either" or absent — signer's choice; default stricter.
            cover_digest = True

        body = request.content
        signing_profile_version = self._signing_profile_version
        if signing_profile_version is None:  # pragma: no cover - constructor invariant
            raise RuntimeError("request signing profile was not initialized")
        signed = sign_request(
            method=request.method,
            url=str(request.url),
            headers=dict(request.headers),
            body=body,
            private_key=self.signing.private_key,
            key_id=self.signing.key_id,
            alg=self.signing.alg,
            cover_content_digest=cover_digest,
            tag=self.signing.tag,
            signing_profile_version=signing_profile_version,
        )
        # pop-then-set ensures our signed values are authoritative even if
        # another hook or earlier layer added a same-named header. httpx
        # headers are a case-insensitive MultiDict, so a naive assignment
        # could leave a duplicate value in a different case.
        for header_name, header_value in signed.as_dict().items():
            request.headers.pop(header_name, None)
            request.headers[header_name] = header_value

    def get_webhook_url(self, task_type: str, operation_id: str) -> str:
        """Generate webhook URL for a task."""
        if not self.webhook_url_template:
            raise ValueError("webhook_url_template not configured")

        return self.webhook_url_template.format(
            agent_id=self.agent_config.id,
            task_type=task_type,
            operation_id=operation_id,
        )

    def _emit_activity(self, activity: Activity) -> None:
        """Emit activity event."""
        if self.on_activity:
            self.on_activity(activity)

    @contextlib.contextmanager
    def use_idempotency_key(self, key: str) -> Iterator[str]:
        """Pin an ``idempotency_key`` for the next mutating call on THIS client.

        Use when you've persisted a key (e.g., in a buyer-side database) and
        want the SDK to send that exact key on resume or retry across process
        restarts. The key is validated against ``^[A-Za-z0-9_.:-]{16,255}$`` on
        entry; a ``ValueError`` is raised for malformed keys.

        Scope rules:

        * **Single-use within scope.** The first mutating call inside the
          ``with`` block consumes the pinned key; a second mutating call falls
          through to a fresh UUID. This protects against ``asyncio.gather``
          siblings accidentally sharing the key (which would trigger
          ``IDEMPOTENCY_CONFLICT`` or silently duplicate work). If you need to
          retry, wrap each attempt in its own ``with`` block.
        * **Client-scoped.** The pinned key applies only to calls on THIS
          client. A mutating call on a sibling ``ADCPClient`` inside the same
          ``with`` block generates a fresh key and emits a ``UserWarning`` —
          keys must be unique per (seller, request) pair (AdCP #2315).
        * **No nesting.** Nested ``use_idempotency_key`` on the same client
          raises ``RuntimeError``.

        Example::

            with client.use_idempotency_key(campaign.stored_key):
                result = await client.create_media_buy(request)
        """
        from adcp import _idempotency

        _idempotency.validate_key(key)
        token = self._idempotency_client_token
        if token in _idempotency._scoped_keys:
            raise RuntimeError(
                "use_idempotency_key is already active on this client; "
                "nested usage is not supported."
            )
        _idempotency._scoped_keys[token] = key
        try:
            yield key
        finally:
            _idempotency._scoped_keys.pop(token, None)

    # ========================================================================
    # Capability Validation
    # ========================================================================

    @property
    def capabilities(self) -> GetAdcpCapabilitiesResponse | None:
        """Return cached capabilities, or None if not yet fetched."""
        return self._capabilities

    @property
    def feature_resolver(self) -> FeatureResolver | None:
        """Return the FeatureResolver for cached capabilities, or None."""
        return self._feature_resolver

    async def fetch_capabilities(self) -> GetAdcpCapabilitiesResponse:
        """Fetch capabilities, using cache if still valid.

        Returns:
            The seller's capabilities response.
        """
        if self._capabilities is not None and self._capabilities_fetched_at is not None:
            elapsed = time.monotonic() - self._capabilities_fetched_at
            if elapsed < self.capabilities_ttl:
                return self._capabilities

        return await self.refresh_capabilities()

    async def refresh_capabilities(self) -> GetAdcpCapabilitiesResponse:
        """Fetch capabilities from the seller, bypassing cache.

        On strict-schema validation failure the raw response is inspected with
        ``looks_like_v3_capabilities``: if the agent is structurally v3-shaped,
        a wire-shape bug is surfaced loudly with the original validation error
        rather than silently downgrading to v2 (the v2 fallback would then ask
        for v2.5 schemas, which aren't shipped — one missing field would
        cascade into "AdCP schema data for version v2.5 not found"). Genuinely
        non-v3 responses still fall through to the transport-error path.

        Returns:
            The seller's capabilities response.

        Raises:
            ADCPError: On transport failure, or when the response is
                v3-shaped but fails schema validation. The error message
                explicitly references v3 in the latter case so the underlying
                wire-shape bug doesn't get blamed on a v2.5-schema cascade.
        """
        result = await self.get_adcp_capabilities(GetAdcpCapabilitiesRequest())
        if result.success and result.data is not None:
            self._capabilities = result.data
            self._feature_resolver = FeatureResolver(result.data)
            self._capabilities_fetched_at = time.monotonic()
            return self._capabilities

        # The typed call discards the raw payload on parse failure (only the
        # error string survives). Distinguish parse-failure (worth shape-
        # checking) from transport-failure (no data ever arrived) by the
        # error prefix produced by ProtocolAdapter._parse_response. Only on
        # parse-failure do we re-fetch the raw dict from the adapter to
        # inspect its shape; transport failures fall straight through to
        # the original error path.
        raw_data: Any = None
        is_parse_failure = result.error is not None and result.error.startswith(
            "Failed to parse response:"
        )
        if is_parse_failure:
            raw_result = await self.adapter.get_adcp_capabilities(
                GetAdcpCapabilitiesRequest().model_dump(mode="json", exclude_none=True)
            )
            raw_data = raw_result.data
            if isinstance(raw_data, list) and len(raw_data) == 1 and isinstance(raw_data[0], dict):
                # MCP content array — unwrap a single-item content envelope
                # so the heuristic sees the same shape the parser would.
                raw_data = raw_data[0]

        if looks_like_v3_capabilities(raw_data):
            logger.warning(
                "[AdCP] Agent %r returned a get_adcp_capabilities response that "
                "failed validation, but the response is structurally v3-shaped. "
                "The agent has a wire-shape bug — that's the thing to fix. "
                "(has_error=%s, has_data=%s)",
                self.agent_config.id,
                bool(result.error),
                raw_data is not None,
            )
            raise ADCPError(
                f"v3 capabilities response from agent {self.agent_config.id!r} "
                f"failed schema validation: {result.error or result.message}. "
                f"The response is structurally v3-shaped (carries `adcp`, "
                f"`supported_protocols`, or a v3 protocol block) — fix the "
                f"agent's wire shape rather than downgrading to v2.",
                agent_id=self.agent_config.id,
                agent_uri=self.agent_config.agent_uri,
            )

        raise ADCPError(
            f"Failed to fetch capabilities: {result.error or result.message}",
            agent_id=self.agent_config.id,
            agent_uri=self.agent_config.agent_uri,
        )

    def _ensure_resolver(self) -> FeatureResolver:
        """Return the FeatureResolver, raising if capabilities haven't been fetched."""
        if self._feature_resolver is None:
            raise ADCPError(
                "Cannot check feature support: capabilities have not been fetched. "
                "Call fetch_capabilities() first.",
                agent_id=self.agent_config.id,
                agent_uri=self.agent_config.agent_uri,
            )
        return self._feature_resolver

    def supports(self, feature: str) -> bool:
        """Check if the seller supports a feature.

        Supports multiple feature namespaces:
        - Protocol support: ``supports("media_buy")`` checks ``supported_protocols``
        - Extension support: ``supports("ext:scope3")`` checks ``extensions_supported``
        - Targeting: ``supports("targeting.geo_countries")`` checks
          ``media_buy.execution.targeting``
        - Media buy features: ``supports("audience_targeting")`` checks
          ``media_buy.features``
        - Signals features: ``supports("catalog_signals")`` checks
          ``signals.features``

        Args:
            feature: Feature identifier to check.

        Returns:
            True if the seller declares the feature as supported.

        Raises:
            ADCPError: If capabilities have not been fetched yet.
        """
        return self._ensure_resolver().supports(feature)

    def require(self, *features: str) -> None:
        """Assert that the seller supports all listed features.

        Args:
            *features: Feature identifiers to require.

        Raises:
            ADCPFeatureUnsupportedError: If any features are not supported.
            ADCPError: If capabilities have not been fetched yet.
        """
        self._ensure_resolver().require(
            *features,
            agent_id=self.agent_config.id,
            agent_uri=self.agent_config.agent_uri,
        )

    def _validate_task_features(self, task_name: str) -> None:
        """Check feature requirements for a task if validate_features is enabled.

        Returns without checking if validate_features is False or capabilities
        haven't been fetched yet (logs a warning in the latter case).
        """
        if not self.validate_features:
            return
        if self._feature_resolver is None:
            logger.warning(
                "validate_features is enabled but capabilities have not been fetched. "
                "Call fetch_capabilities() to enable feature validation."
            )
            return
        required_feature = TASK_FEATURE_MAP.get(task_name)
        if required_feature is None:
            return
        self.require(required_feature)

    def _remember_canonical_product_routes(self, products: list[Any]) -> None:
        """Retain bounded same-client canonical-to-legacy routes from discovery."""

        for product in products:
            product_id = getattr(product, "product_id", None)
            for declaration in getattr(product, "format_options", None) or []:
                option_id = getattr(declaration, "format_option_id", None)
                if not option_id:
                    continue
                refs = getattr(declaration, "legacy_format_refs", ())
                if not refs:
                    continue
                route = [ref.model_dump(mode="json") for ref in refs]
                publisher = getattr(declaration, "publisher_domain", None)
                self._canonical_legacy_routes[(product_id, publisher, option_id)] = route
                global_key = (None, publisher, option_id)
                existing = self._canonical_legacy_routes.get(global_key)
                if global_key not in self._canonical_legacy_routes:
                    self._canonical_legacy_routes[global_key] = route
                elif existing != route:
                    self._canonical_legacy_routes[global_key] = None

    def _creative_dialect(
        self, request: Any = None, *, legacy_projection_available: bool = False
    ) -> CreativeDialect:
        """Resolve this call's wire dialect from version, capability, and request evidence."""

        request_version = None
        if isinstance(request, dict):
            request_version = request.get("adcp_version")
        elif request is not None:
            request_version = getattr(request, "adcp_version", None)
        version = request_version or self._server_version or self._adcp_version
        if not str(version).startswith("3."):
            return CreativeDialect.LEGACY
        return resolve_creative_dialect(
            version,
            capabilities=self._capabilities,
            request=request,
            legacy_projection_available=legacy_projection_available,
        )

    def _callback_creative_dialect(self, result: Any) -> CreativeDialect:
        """Resolve callbacks from payload evidence, canonical-first when neutral."""

        try:
            return self._creative_dialect(result)
        except CreativeDialectError:
            return CreativeDialect.CANONICAL

    def _legacy_refs_for_option(
        self,
        option: dict[str, Any],
        *,
        product_id: str | None,
        field: str,
    ) -> list[dict[str, Any]]:
        option_id = option.get("format_option_id")
        publisher = option.get("publisher_domain")
        if not isinstance(option_id, str):
            raise CanonicalFormatLegacyResolutionError(f"{field} has no format_option_id")
        route = self._canonical_legacy_routes.get((product_id, publisher, option_id))
        if route is None:
            route = self._canonical_legacy_routes.get((None, publisher, option_id))
        if route is None:
            raise CanonicalFormatLegacyResolutionError(
                f"no discovered legacy route for {field}; rediscover the product or configure "
                "canonical_format_legacy_resolver"
            )
        return [dict(ref) for ref in route]

    def _legacy_refs_for_declaration(
        self,
        declaration: dict[str, Any],
        *,
        product_id: str | None,
        field: str,
    ) -> list[dict[str, Any]]:
        option_id = declaration.get("format_option_id")
        if isinstance(option_id, str):
            try:
                return self._legacy_refs_for_option(
                    declaration,
                    product_id=product_id,
                    field=field,
                )
            except CanonicalFormatLegacyResolutionError:
                pass
        canonical = Format.model_validate(declaration)
        return [
            ref.model_dump(mode="json")
            for ref in resolve_legacy_format_refs(
                canonical,
                resolver=self.canonical_format_legacy_resolver,
                product_id=product_id,
                field=field,
            )
        ]

    def _downgrade_package(self, package: dict[str, Any], *, field: str) -> None:
        product_id = package.get("product_id")
        product_id = product_id if isinstance(product_id, str) else None
        refs = package.pop("format_option_refs", None)
        if refs:
            legacy: list[dict[str, Any]] = []
            for index, option in enumerate(refs):
                legacy.extend(
                    self._legacy_refs_for_option(
                        option,
                        product_id=product_id,
                        field=f"{field}.format_option_refs[{index}]",
                    )
                )
            package["format_ids"] = legacy
        elif package.get("format_kind") is not None:
            declaration = {
                "format_kind": package.pop("format_kind"),
                "params": package.pop("params", None) or {},
            }
            package["format_ids"] = self._legacy_refs_for_declaration(
                declaration,
                product_id=product_id,
                field=f"{field}.format_kind",
            )
        else:
            package.pop("params", None)

        for index, creative in enumerate(package.get("creatives") or []):
            self._downgrade_creative(
                creative,
                field=f"{field}.creatives[{index}]",
                product_id=product_id,
            )

    def _downgrade_creative(
        self,
        creative: dict[str, Any],
        *,
        field: str,
        product_id: str | None = None,
    ) -> None:
        option = creative.pop("format_option_ref", None)
        if option:
            refs = self._legacy_refs_for_option(
                option, product_id=product_id, field=f"{field}.format_option_ref"
            )
        else:
            declaration = {
                "format_kind": creative.pop("format_kind"),
                "params": creative.pop("params", None) or {},
            }
            refs = self._legacy_refs_for_declaration(
                declaration, product_id=product_id, field=field
            )
        if len(refs) != 1:
            raise CanonicalFormatLegacyResolutionError(
                f"{field} resolves to {len(refs)} legacy formats; a creative requires exactly one"
            )
        creative["format_id"] = refs[0]

    def _prepare_creative_params(self, request: BaseModel) -> dict[str, Any]:
        """Serialize canonical input and deterministically adapt legacy peers."""

        params = request.model_dump(mode="json", exclude_none=True)
        if strip_legacy_creative_identity(params) != params:
            raise ValueError(
                "primary creative methods reject legacy format identity; use the explicit "
                "*_legacy method"
            )
        if self._creative_dialect(request) is CreativeDialect.CANONICAL:
            return params
        for collection in ("packages", "new_packages"):
            for index, package in enumerate(params.get(collection) or []):
                self._downgrade_package(package, field=f"{collection}[{index}]")
        for index, creative in enumerate(params.get("creatives") or []):
            self._downgrade_creative(creative, field=f"creatives[{index}]")
        return params

    def _canonicalize_get_products_result(
        self,
        raw_result: TaskResult[Any],
    ) -> TaskResult[GetProductsResponse]:
        """Parse the wire shape, project products, and enforce the primary boundary."""

        # Canonical responses must be parsed before the legacy compatibility
        # model. The generated legacy ProductFormatDeclaration intentionally
        # lacks canonical ``format_kind`` and ``params`` fields, so parsing a
        # canonical-only response through it first irreversibly discards the
        # declaration before ``project_legacy_product`` can inspect it.
        canonical_result: TaskResult[GetProductsResponse] = self.adapter._parse_response(
            raw_result, GetProductsResponse
        )
        if not raw_result.success or raw_result.data is None:
            return canonical_result
        if canonical_result.success and canonical_result.data is not None:
            direct_products = list(canonical_result.data.products or [])
            self._remember_canonical_product_routes(direct_products)
            metadata = dict(canonical_result.metadata or {})
            metadata["projection"] = {"diagnostics": []}
            canonical_result.metadata = metadata
            return canonical_result

        legacy_result: TaskResult[Any] = self.adapter._parse_response(
            raw_result, LegacyGetProductsResponse
        )
        if not legacy_result.success or legacy_result.data is None:
            return cast(TaskResult[GetProductsResponse], legacy_result)

        body = legacy_result.data.model_dump(mode="json")
        canonical_products: list[Product] = []
        diagnostics: list[dict[str, Any]] = []
        portable_errors = list(body.get("errors") or [])
        for raw_product in body.get("products") or []:
            projected = project_legacy_product(
                raw_product,
                legacy_format_converter=self.legacy_format_converter,
            )
            if projected.product is not None:
                canonical_products.append(projected.product)
            for diagnostic in projected.diagnostics:
                dumped = diagnostic.model_dump()
                diagnostics.append(dumped)
                details = dumped["error"]["details"]
                portable_errors.append(
                    {
                        "code": diagnostic.code,
                        "message": (
                            "Product has no format declaration representable on the "
                            "canonical-only surface"
                            if diagnostic.code == "CANONICAL_PRODUCT_FORMATS_UNAVAILABLE"
                            else "Legacy creative format could not be projected to a "
                            "canonical declaration"
                        ),
                        "field": diagnostic.field,
                        "recovery": "correctable",
                        "source": "sdk",
                        "sdk_id": dumped["sdk_id"],
                        "details": details,
                    }
                )

        body["products"] = canonical_products
        if portable_errors:
            body["errors"] = portable_errors
        try:
            canonical = GetProductsResponse.model_validate(body)
        except ValueError as exc:
            return TaskResult[GetProductsResponse](
                status=TaskStatus.FAILED,
                success=False,
                error=f"Failed to project canonical get_products response: {exc}",
                message=legacy_result.message,
                metadata=legacy_result.metadata,
                debug_info=legacy_result.debug_info,
                idempotency_key=legacy_result.idempotency_key,
                replayed=legacy_result.replayed,
            )

        self._remember_canonical_product_routes(canonical_products)
        metadata = dict(legacy_result.metadata or {})
        metadata["projection"] = {"diagnostics": diagnostics}
        return TaskResult[GetProductsResponse](
            status=legacy_result.status,
            data=canonical,
            message=legacy_result.message,
            success=True,
            error=legacy_result.error,
            metadata=metadata,
            debug_info=legacy_result.debug_info,
            idempotency_key=legacy_result.idempotency_key,
            replayed=legacy_result.replayed,
        )

    def _canonicalize_format_read_result(
        self,
        raw_result: TaskResult[Any],
        *,
        legacy_type: Any,
        canonical_type: Any,
        collection: str,
        require_format: bool,
    ) -> TaskResult[Any]:
        """Upgrade legacy creative rows through the same projector used by discovery."""

        legacy_result = self.adapter._parse_response(raw_result, legacy_type)
        if not legacy_result.success or legacy_result.data is None:
            return legacy_result
        body = legacy_result.data.model_dump(mode="json")
        converted: list[dict[str, Any]] = []
        diagnostics: list[dict[str, Any]] = []
        errors = list(body.get("errors") or [])
        for index, raw_item in enumerate(body.get(collection) or []):
            item = dict(raw_item)
            legacy_id = item.pop("format_id", None)
            if item.get("format_kind") is None and legacy_id is not None:
                projection = project_legacy_format_id(
                    legacy_id,
                    product_id=str(item.get("media_buy_id") or ""),
                    field=f"{collection}[{index}].format_id",
                    legacy_format_converter=self.legacy_format_converter,
                )
                if projection.declaration is not None:
                    item["format_kind"] = projection.declaration.format_kind.value
                elif projection.diagnostic is not None:
                    diagnostic = projection.diagnostic.model_dump()
                    diagnostics.append(diagnostic)
                    errors.append(
                        {
                            "code": "FORMAT_PROJECTION_FAILED",
                            "message": "Legacy creative format could not be projected.",
                            "field": f"{collection}[{index}].format_id",
                            "severity": "warning",
                            "source": "sdk",
                            "details": diagnostic["error"]["details"],
                        }
                    )
            if require_format and item.get("format_kind") is None:
                continue
            converted.append(item)
        body[collection] = converted
        if errors:
            body["errors"] = errors
        try:
            canonical = canonical_type.model_validate(body)
        except ValueError as exc:
            return TaskResult[Any](
                status=TaskStatus.FAILED,
                success=False,
                error=f"Failed to project canonical {collection} response: {exc}",
                message=legacy_result.message,
                metadata=legacy_result.metadata,
            )
        metadata = dict(legacy_result.metadata or {})
        metadata["projection"] = {"diagnostics": diagnostics}
        return TaskResult[Any](
            status=legacy_result.status,
            data=canonical,
            message=legacy_result.message,
            success=True,
            metadata=metadata,
            debug_info=legacy_result.debug_info,
            idempotency_key=legacy_result.idempotency_key,
            replayed=legacy_result.replayed,
        )

    def _canonicalize_lifecycle_result(
        self,
        raw_result: TaskResult[Any],
        *,
        legacy_type: Any,
        canonical_type: Any,
    ) -> TaskResult[Any]:
        """Upgrade legacy package/creative selectors on lifecycle responses."""

        legacy_result = self.adapter._parse_response(raw_result, legacy_type)
        if not legacy_result.success or legacy_result.data is None:
            return legacy_result
        try:
            body = normalize_legacy_creative_request(
                legacy_result.data.model_dump(mode="json"),
                legacy_format_converter=self.legacy_format_converter,
            )
            canonical = TypeAdapter(canonical_type).validate_python(body)
        except (LegacyCreativeProjectionError, ValueError) as exc:
            return TaskResult[Any](
                status=TaskStatus.FAILED,
                success=False,
                error=f"Failed to project canonical lifecycle response: {exc}",
                message=legacy_result.message,
                metadata=legacy_result.metadata,
            )
        return TaskResult[Any](
            status=legacy_result.status,
            data=canonical,
            message=legacy_result.message,
            success=True,
            metadata=legacy_result.metadata,
            debug_info=legacy_result.debug_info,
            idempotency_key=legacy_result.idempotency_key,
            replayed=legacy_result.replayed,
        )

    async def _execute_typed_task(
        self,
        task_type: str,
        request: BaseModel,
        response_type: type[BaseModel] | Any,
    ) -> TaskResult[Any]:
        """Execute and parse one typed AdCP task with activity events."""
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)
        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type=task_type,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )
        method = getattr(self.adapter, task_type)
        raw_result = await method(params)
        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type=task_type,
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )
        return self.adapter._parse_response(raw_result, response_type)

    async def list_products(self, request: ListProductsRequest) -> TaskResult[ListProductsResponse]:
        """List products using the AdCP 3.2 compact discovery lifecycle."""
        return cast(
            TaskResult[ListProductsResponse],
            await self._execute_typed_task("list_products", request, ListProductsResponse),
        )

    async def request_proposals(
        self, request: RequestProposalsRequest
    ) -> TaskResult[RequestProposalsResponse]:
        """Request seller proposals for selected products."""
        return cast(
            TaskResult[RequestProposalsResponse],
            await self._execute_typed_task("request_proposals", request, RequestProposalsResponse),
        )

    async def refine_proposals(
        self, request: RefineProposalsRequest
    ) -> TaskResult[RefineProposalsResponse]:
        """Refine one or more seller proposals."""
        return cast(
            TaskResult[RefineProposalsResponse],
            await self._execute_typed_task("refine_proposals", request, RefineProposalsResponse),
        )

    async def decline_proposals(
        self, request: DeclineProposalsRequest
    ) -> TaskResult[DeclineProposalsResponse]:
        """Decline one or more seller proposals."""
        return cast(
            TaskResult[DeclineProposalsResponse],
            await self._execute_typed_task("decline_proposals", request, DeclineProposalsResponse),
        )

    async def buy_products(self, request: BuyProductsRequest) -> TaskResult[BuyProductsResponse]:
        """Commit a direct product purchase."""
        return cast(
            TaskResult[BuyProductsResponse],
            await self._execute_typed_task("buy_products", request, BuyProductsResponse),
        )

    async def accept_proposal(
        self, request: AcceptProposalRequest
    ) -> TaskResult[AcceptProposalResponse]:
        """Accept a seller proposal and create its media buy."""
        return cast(
            TaskResult[AcceptProposalResponse],
            await self._execute_typed_task("accept_proposal", request, AcceptProposalResponse),
        )

    async def control_media_buy(
        self, request: ControlMediaBuyRequest
    ) -> TaskResult[ControlMediaBuyResponse]:
        """Apply lifecycle controls to an existing media buy."""
        return cast(
            TaskResult[ControlMediaBuyResponse],
            await self._execute_typed_task("control_media_buy", request, ControlMediaBuyResponse),
        )

    async def get_products(
        self,
        request: GetProductsRequest,
        fetch_previews: bool = False,
        preview_output_format: str = "url",
        creative_agent_client: ADCPClient | None = None,
    ) -> TaskResult[GetProductsResponse]:
        """
        Get advertising products.

        Args:
            request: Request parameters
            fetch_previews: If True, generate preview URLs for each product's formats
                (uses batch API for 5-10x performance improvement)
            preview_output_format: "url" for iframe URLs (default), "html" for direct
                embedding (2-3x faster, no iframe overhead)
            creative_agent_client: Client for creative agent (required if
                fetch_previews=True)

        Returns:
            TaskResult containing GetProductsResponse with optional preview URLs in metadata

        Raises:
            ValueError: If fetch_previews=True but creative_agent_client is not provided
        """
        if fetch_previews and not creative_agent_client:
            raise ValueError("creative_agent_client is required when fetch_previews=True")

        self._creative_dialect(request, legacy_projection_available=True)
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_products",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_products(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_products",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        result = self._canonicalize_get_products_result(raw_result)

        if (
            fetch_previews
            and result.success
            and result.data
            and result.data.products
            and creative_agent_client
        ):
            from adcp.utils.preview_cache import add_preview_urls_to_products

            products_with_previews = await add_preview_urls_to_products(
                result.data.products,
                creative_agent_client,
                use_batch=True,
                output_format=preview_output_format,
            )
            result.metadata = result.metadata or {}
            result.metadata["products_with_previews"] = products_with_previews

        return result

    async def get_products_legacy(
        self,
        request: LegacyGetProductsRequest,
    ) -> TaskResult[LegacyGetProductsResponse]:
        """Return the raw AdCP 3.x product wire shape for migration tooling."""

        self._warn_legacy_creative_api("get_products_legacy")
        raw_result = await self.adapter.get_products(request.model_dump(mode="json"))
        return self.adapter._parse_response(raw_result, LegacyGetProductsResponse)

    @staticmethod
    def _warn_legacy_creative_api(method: str) -> None:
        warnings.warn(
            f"{method} exposes legacy named-format identity and will be removed with AdCP 4.0",
            DeprecationWarning,
            stacklevel=3,
        )

    async def list_creative_formats_legacy(
        self,
        request: ListCreativeFormatsRequest,
        fetch_previews: bool = False,
        preview_output_format: str = "url",
    ) -> TaskResult[ListCreativeFormatsResponse]:
        """
        List supported creative formats.

        Args:
            request: Request parameters
            fetch_previews: If True, generate preview URLs for each format using
                sample manifests (uses batch API for 5-10x performance improvement)
            preview_output_format: "url" for iframe URLs (default), "html" for direct
                embedding (2-3x faster, no iframe overhead)

        Returns:
            TaskResult containing ListCreativeFormatsResponse with optional preview URLs in metadata
        """
        self._warn_legacy_creative_api("list_creative_formats_legacy")
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_creative_formats",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.list_creative_formats(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_creative_formats",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        result: TaskResult[ListCreativeFormatsResponse] = self.adapter._parse_response(
            raw_result, ListCreativeFormatsResponse
        )

        if fetch_previews and result.success and result.data:
            from adcp.utils.preview_cache import add_preview_urls_to_formats

            formats_with_previews = await add_preview_urls_to_formats(
                result.data.formats,
                self,
                use_batch=True,
                output_format=preview_output_format,
            )
            result.metadata = result.metadata or {}
            result.metadata["formats_with_previews"] = formats_with_previews

        return result

    async def create_media_buy_legacy(
        self, request: LegacyCreateMediaBuyRequest
    ) -> TaskResult[LegacyCreateMediaBuyResponse]:
        """Execute create_media_buy without the canonical application boundary."""

        self._warn_legacy_creative_api("create_media_buy_legacy")
        raw = await self.adapter.create_media_buy(
            request.model_dump(mode="json", exclude_none=True)
        )
        return self.adapter._parse_response(raw, LegacyCreateMediaBuyResponse)

    async def update_media_buy_legacy(
        self, request: LegacyUpdateMediaBuyRequest
    ) -> TaskResult[LegacyUpdateMediaBuyResponse]:
        """Execute update_media_buy without the canonical application boundary."""

        self._warn_legacy_creative_api("update_media_buy_legacy")
        raw = await self.adapter.update_media_buy(
            request.model_dump(mode="json", exclude_none=True)
        )
        return self.adapter._parse_response(raw, LegacyUpdateMediaBuyResponse)

    async def sync_creatives_legacy(
        self, request: LegacySyncCreativesRequest
    ) -> TaskResult[LegacySyncCreativesResponse]:
        """Execute sync_creatives without the canonical application boundary."""

        self._warn_legacy_creative_api("sync_creatives_legacy")
        raw = await self.adapter.sync_creatives(request.model_dump(mode="json", exclude_none=True))
        return self.adapter._parse_response(raw, LegacySyncCreativesResponse)

    async def list_creatives_legacy(
        self, request: LegacyListCreativesRequest
    ) -> TaskResult[LegacyListCreativesResponse]:
        """Return raw creative rows carrying legacy format identity."""

        self._warn_legacy_creative_api("list_creatives_legacy")
        raw = await self.adapter.list_creatives(request.model_dump(mode="json", exclude_none=True))
        return self.adapter._parse_response(raw, LegacyListCreativesResponse)

    async def get_media_buys_legacy(
        self, request: GetMediaBuysRequest
    ) -> TaskResult[LegacyGetMediaBuysResponse]:
        """Return raw media-buy rows carrying legacy format identity."""

        self._warn_legacy_creative_api("get_media_buys_legacy")
        raw = await self.adapter.get_media_buys(request.model_dump(mode="json", exclude_none=True))
        return self.adapter._parse_response(raw, LegacyGetMediaBuysResponse)

    async def get_media_buy_delivery_legacy(
        self, request: GetMediaBuyDeliveryRequest
    ) -> TaskResult[LegacyGetMediaBuyDeliveryResponse]:
        """Return raw media-buy delivery carrying legacy format identity."""

        self._warn_legacy_creative_api("get_media_buy_delivery_legacy")
        raw = await self.adapter.get_media_buy_delivery(
            request.model_dump(mode="json", exclude_none=True)
        )
        return self.adapter._parse_response(raw, LegacyGetMediaBuyDeliveryResponse)

    async def get_creative_delivery_legacy(
        self, request: GetCreativeDeliveryRequest
    ) -> TaskResult[LegacyGetCreativeDeliveryResponse]:
        """Return raw creative delivery carrying legacy format identity."""

        self._warn_legacy_creative_api("get_creative_delivery_legacy")
        raw = await self.adapter.get_creative_delivery(
            request.model_dump(mode="json", exclude_none=True)
        )
        return self.adapter._parse_response(raw, LegacyGetCreativeDeliveryResponse)

    async def preview_creative_legacy(
        self,
        request: LegacyPreviewCreativeRequest,
    ) -> TaskResult[LegacyPreviewCreativeResponse]:
        """
        Generate preview of a creative manifest.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing PreviewCreativeResponse with preview URLs
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="preview_creative",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.preview_creative(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="preview_creative",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        self._warn_legacy_creative_api("preview_creative_legacy")
        return self.adapter._parse_response(raw_result, LegacyPreviewCreativeResponse)

    async def sync_creatives(
        self,
        request: SyncCreativesRequest,
    ) -> TaskResult[SyncCreativesResponse]:
        """
        Sync Creatives.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing SyncCreativesResponse
        """
        dialect = self._creative_dialect(request, legacy_projection_available=True)
        operation_id = create_operation_id()
        params = self._prepare_creative_params(request)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_creatives",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.sync_creatives(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_creatives",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        if dialect is CreativeDialect.LEGACY:
            return cast(
                TaskResult[SyncCreativesResponse],
                self._canonicalize_lifecycle_result(
                    raw_result,
                    legacy_type=LegacySyncCreativesResponse,
                    canonical_type=SyncCreativesResponse,
                ),
            )
        return self.adapter._parse_response(raw_result, SyncCreativesResponse)

    async def list_creatives(
        self,
        request: ListCreativesRequest,
    ) -> TaskResult[ListCreativesResponse]:
        """
        List Creatives.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing ListCreativesResponse
        """
        dialect = self._creative_dialect(request, legacy_projection_available=True)
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_creatives",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.list_creatives(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_creatives",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        if dialect is CreativeDialect.CANONICAL:
            return self.adapter._parse_response(raw_result, ListCreativesResponse)
        return cast(
            TaskResult[ListCreativesResponse],
            self._canonicalize_format_read_result(
                raw_result,
                legacy_type=LegacyListCreativesResponse,
                canonical_type=ListCreativesResponse,
                collection="creatives",
                require_format=True,
            ),
        )

    async def get_media_buy_delivery(
        self,
        request: GetMediaBuyDeliveryRequest,
    ) -> TaskResult[GetMediaBuyDeliveryResponse]:
        """
        Get Media Buy Delivery.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing GetMediaBuyDeliveryResponse
        """
        dialect = self._creative_dialect(request, legacy_projection_available=True)
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_media_buy_delivery",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_media_buy_delivery(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_media_buy_delivery",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        if dialect is CreativeDialect.LEGACY:
            return cast(
                TaskResult[GetMediaBuyDeliveryResponse],
                self._canonicalize_lifecycle_result(
                    raw_result,
                    legacy_type=LegacyGetMediaBuyDeliveryResponse,
                    canonical_type=GetMediaBuyDeliveryResponse,
                ),
            )
        return self.adapter._parse_response(raw_result, GetMediaBuyDeliveryResponse)

    async def get_media_buys(
        self,
        request: GetMediaBuysRequest,
    ) -> TaskResult[GetMediaBuysResponse]:
        """
        Get Media Buys.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing GetMediaBuysResponse
        """
        dialect = self._creative_dialect(request, legacy_projection_available=True)
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)
        if params.get("include_webhook_activity") is False:
            params.pop("include_webhook_activity")
        if params.get("webhook_activity_limit") == 50:
            params.pop("webhook_activity_limit")

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_media_buys",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_media_buys(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_media_buys",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        if dialect is CreativeDialect.LEGACY:
            return cast(
                TaskResult[GetMediaBuysResponse],
                self._canonicalize_lifecycle_result(
                    raw_result,
                    legacy_type=LegacyGetMediaBuysResponse,
                    canonical_type=GetMediaBuysResponse,
                ),
            )
        return self.adapter._parse_response(raw_result, GetMediaBuysResponse)

    async def get_signals(
        self,
        request: GetSignalsRequest,
    ) -> TaskResult[GetSignalsResponse]:
        """
        Get Signals.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing GetSignalsResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_signals",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_signals(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_signals",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetSignalsResponse)

    async def activate_signal(
        self,
        request: ActivateSignalRequest,
    ) -> TaskResult[ActivateSignalResponse]:
        """
        Activate Signal.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing ActivateSignalResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="activate_signal",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.activate_signal(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="activate_signal",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ActivateSignalResponse)

    async def provide_performance_feedback(
        self,
        request: ProvidePerformanceFeedbackRequest,
    ) -> TaskResult[ProvidePerformanceFeedbackResponse]:
        """
        Provide Performance Feedback.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing ProvidePerformanceFeedbackResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="provide_performance_feedback",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.provide_performance_feedback(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="provide_performance_feedback",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ProvidePerformanceFeedbackResponse)

    async def create_media_buy(
        self,
        request: CreateMediaBuyRequest,
    ) -> TaskResult[CreateMediaBuyResponse]:
        """
        Create a new media buy reservation.

        Requests the agent to reserve inventory for a campaign. The agent returns a
        media_buy_id that tracks this reservation and can be used for updates.

        Args:
            request: Media buy creation parameters including:
                - brand: Brand reference; resolved from brand.json or the registry at execution
                - packages: List of package requests specifying desired inventory
                - publisher_properties: Target properties for ad placement
                - budget: Optional budget constraints
                - start_date/end_date: Campaign flight dates

        Returns:
            TaskResult containing CreateMediaBuyResponse with:
                - media_buy_id: Unique identifier for this reservation
                - status: Current state of the media buy
                - packages: Confirmed package details
                - Additional platform-specific metadata

        Example:
            >>> from adcp import ADCPClient, CreateMediaBuyRequest, BrandReference
            >>> client = ADCPClient(agent_config)
            >>> request = CreateMediaBuyRequest(
            ...     brand=BrandReference(domain="acme.com"),
            ...     packages=[package_request],
            ...     publisher_properties=properties,
            ... )
            >>> result = await client.create_media_buy(request)
            >>> if result.success:
            ...     media_buy_id = result.data.media_buy_id
        """
        dialect = self._creative_dialect(request, legacy_projection_available=True)
        operation_id = create_operation_id()
        params = self._prepare_creative_params(request)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="create_media_buy",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.create_media_buy(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="create_media_buy",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        if dialect is CreativeDialect.LEGACY:
            return cast(
                TaskResult[CreateMediaBuyResponse],
                self._canonicalize_lifecycle_result(
                    raw_result,
                    legacy_type=LegacyCreateMediaBuyResponse,
                    canonical_type=CreateMediaBuyResponse,
                ),
            )
        return self.adapter._parse_response(raw_result, CreateMediaBuyResponse)

    async def update_media_buy(
        self,
        request: UpdateMediaBuyRequest,
    ) -> TaskResult[UpdateMediaBuyResponse]:
        """
        Update an existing media buy reservation.

        Modifies a previously created media buy by updating packages or publisher
        properties. The update operation uses discriminated unions to specify what
        to change - either package details or targeting properties.

        Args:
            request: Media buy update parameters including:
                - media_buy_id: Identifier from create_media_buy response
                - updates: Discriminated union specifying update type:
                    * UpdateMediaBuyPackagesRequest: Modify package selections
                    * UpdateMediaBuyPropertiesRequest: Change targeting properties

        Returns:
            TaskResult containing UpdateMediaBuyResponse with:
                - media_buy_id: The updated media buy identifier
                - status: Updated state of the media buy
                - packages: Updated package configurations
                - Additional platform-specific metadata

        Example:
            >>> from adcp import ADCPClient, UpdateMediaBuyPackagesRequest
            >>> client = ADCPClient(agent_config)
            >>> request = UpdateMediaBuyPackagesRequest(
            ...     media_buy_id="mb_123",
            ...     packages=[updated_package]
            ... )
            >>> result = await client.update_media_buy(request)
            >>> if result.success:
            ...     updated_packages = result.data.packages
        """
        dialect = self._creative_dialect(request, legacy_projection_available=True)
        operation_id = create_operation_id()
        params = self._prepare_creative_params(request)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="update_media_buy",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.update_media_buy(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="update_media_buy",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        if dialect is CreativeDialect.LEGACY:
            return cast(
                TaskResult[UpdateMediaBuyResponse],
                self._canonicalize_lifecycle_result(
                    raw_result,
                    legacy_type=LegacyUpdateMediaBuyResponse,
                    canonical_type=UpdateMediaBuyResponse,
                ),
            )
        return self.adapter._parse_response(raw_result, UpdateMediaBuyResponse)

    async def build_creative_legacy(
        self,
        request: LegacyBuildCreativeRequest,
    ) -> TaskResult[LegacyBuildCreativeResponse]:
        """
        Generate production-ready creative assets.

        Requests the creative agent to build final deliverable assets in the target
        format (e.g., VAST, DAAST, HTML5). This is typically called after previewing
        and approving a creative manifest.

        Args:
            request: Creative build parameters including:
                - manifest: Creative manifest with brand info and content
                - target_format_id: Desired output format identifier
                - inputs: Optional user-provided inputs for template variables
                - deployment: Platform or agent deployment configuration

        Returns:
            TaskResult containing BuildCreativeResponse with:
                - assets: Production-ready creative files (URLs or inline content)
                - format_id: The generated format identifier
                - manifest: The creative manifest used for generation
                - metadata: Additional platform-specific details

        Example:
            >>> from adcp import ADCPClient, LegacyBuildCreativeRequest
            >>> client = ADCPClient(agent_config)
            >>> request = LegacyBuildCreativeRequest(
            ...     manifest=creative_manifest,
            ...     target_format_id="vast_2.0",
            ...     inputs={"duration": 30}
            ... )
            >>> result = await client.build_creative_legacy(request)
            >>> if result.success:
            ...     vast_url = result.data.assets[0].url
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="build_creative",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.build_creative(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="build_creative",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        self._warn_legacy_creative_api("build_creative_legacy")
        return self.adapter._parse_response(raw_result, LegacyBuildCreativeResponse)

    async def list_accounts(
        self,
        request: ListAccountsRequest,
    ) -> TaskResult[ListAccountsResponse]:
        """
        List Accounts.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing ListAccountsResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_accounts",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.list_accounts(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_accounts",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ListAccountsResponse)

    async def sync_accounts(
        self,
        request: SyncAccountsRequest,
    ) -> TaskResult[SyncAccountsResponse]:
        """
        Sync Accounts.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing SyncAccountsResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_accounts",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.sync_accounts(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_accounts",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, SyncAccountsResponse)

    async def get_account_financials(
        self,
        request: GetAccountFinancialsRequest,
    ) -> TaskResult[GetAccountFinancialsResponse]:
        """
        Get Account Financials.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing GetAccountFinancialsResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_account_financials",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_account_financials(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_account_financials",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetAccountFinancialsResponse)

    async def report_usage(
        self,
        request: ReportUsageRequest,
    ) -> TaskResult[ReportUsageResponse]:
        """
        Report Usage.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing ReportUsageResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="report_usage",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.report_usage(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="report_usage",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ReportUsageResponse)

    async def log_event(
        self,
        request: LogEventRequest,
    ) -> TaskResult[LogEventResponse]:
        """
        Log Event.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing LogEventResponse
        """
        self._validate_task_features("log_event")
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="log_event",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.log_event(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="log_event",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, LogEventResponse)

    async def sync_event_sources(
        self,
        request: SyncEventSourcesRequest,
    ) -> TaskResult[SyncEventSourcesResponse]:
        """
        Sync Event Sources.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing SyncEventSourcesResponse
        """
        self._validate_task_features("sync_event_sources")
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_event_sources",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.sync_event_sources(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_event_sources",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, SyncEventSourcesResponse)

    async def sync_audiences(
        self,
        request: SyncAudiencesRequest,
    ) -> TaskResult[SyncAudiencesResponse]:
        """
        Sync Audiences.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing SyncAudiencesResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_audiences",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.sync_audiences(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_audiences",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, SyncAudiencesResponse)

    async def sync_catalogs(
        self,
        request: SyncCatalogsRequest,
    ) -> TaskResult[SyncCatalogsResponse]:
        """
        Sync Catalogs.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing SyncCatalogsResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_catalogs",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.sync_catalogs(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_catalogs",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, SyncCatalogsResponse)

    async def get_creative_delivery(
        self,
        request: GetCreativeDeliveryRequest,
    ) -> TaskResult[GetCreativeDeliveryResponse]:
        """
        Get Creative Delivery.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing GetCreativeDeliveryResponse
        """
        self._creative_dialect(request, legacy_projection_available=True)
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_creative_delivery",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_creative_delivery(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_creative_delivery",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        if (
            self._creative_dialect(request, legacy_projection_available=True)
            is CreativeDialect.CANONICAL
        ):
            return self.adapter._parse_response(raw_result, GetCreativeDeliveryResponse)
        return cast(
            TaskResult[GetCreativeDeliveryResponse],
            self._canonicalize_format_read_result(
                raw_result,
                legacy_type=LegacyGetCreativeDeliveryResponse,
                canonical_type=GetCreativeDeliveryResponse,
                collection="creatives",
                require_format=False,
            ),
        )

    async def list_transformers(
        self,
        request: ListTransformersRequest,
    ) -> TaskResult[ListTransformersResponse]:
        """
        List Creative Transformers.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing ListTransformersResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_transformers",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.list_transformers(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_transformers",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ListTransformersResponse)

    # ========================================================================
    # V3 Protocol Methods - Protocol Discovery
    # ========================================================================

    async def get_adcp_capabilities(
        self,
        request: GetAdcpCapabilitiesRequest,
    ) -> TaskResult[GetAdcpCapabilitiesResponse]:
        """
        Get AdCP capabilities from the agent.

        Queries the agent's supported AdCP features, protocol versions, and
        domain-specific capabilities (media_buy, signals, sponsored_intelligence).

        Args:
            request: Request parameters including optional protocol filters

        Returns:
            TaskResult containing GetAdcpCapabilitiesResponse with:
                - adcp: Core protocol version information
                - supported_protocols: List of supported domain protocols
                - media_buy: Media buy capabilities (if supported)
                - sponsored_intelligence: SI capabilities (if supported)
                - signals: Signals capabilities (if supported)
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_adcp_capabilities",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_adcp_capabilities(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_adcp_capabilities",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetAdcpCapabilitiesResponse)

    async def sync_agent_notification_configs(
        self,
        request: SyncAgentNotificationConfigsRequest,
    ) -> TaskResult[SyncAgentNotificationConfigsResponse]:
        """Replace the caller-scoped agent notification subscriber set."""
        return cast(
            TaskResult[SyncAgentNotificationConfigsResponse],
            await self._execute_typed_task(
                "sync_agent_notification_configs",
                request,
                SyncAgentNotificationConfigsResponse,
            ),
        )

    async def get_task_status(
        self,
        request: GetTaskStatusRequest,
    ) -> TaskResult[GetTaskStatusResponse]:
        """
        Get Task Status.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing GetTaskStatusResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_task_status",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_task_status(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_task_status",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetTaskStatusResponse)

    async def list_tasks(
        self,
        request: ListTasksRequest,
    ) -> TaskResult[ListTasksResponse]:
        """
        List Tasks.

        Args:
            request: Request parameters

        Returns:
            TaskResult containing ListTasksResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_tasks",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.list_tasks(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_tasks",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ListTasksResponse)

    # ========================================================================
    # V3 Protocol Methods - Content Standards
    # ========================================================================

    async def create_content_standards(
        self,
        request: CreateContentStandardsRequest,
    ) -> TaskResult[CreateContentStandardsResponse]:
        """
        Create a new content standards configuration.

        Defines acceptable content contexts for ad placement using natural
        language policy and optional calibration exemplars.

        Args:
            request: Request parameters including policy and scope

        Returns:
            TaskResult containing CreateContentStandardsResponse with standards_id
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="create_content_standards",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.create_content_standards(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="create_content_standards",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, CreateContentStandardsResponse)

    async def get_content_standards(
        self,
        request: GetContentStandardsRequest,
    ) -> TaskResult[GetContentStandardsResponse]:
        """
        Get a content standards configuration by ID.

        Args:
            request: Request parameters including standards_id

        Returns:
            TaskResult containing GetContentStandardsResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_content_standards",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_content_standards(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_content_standards",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetContentStandardsResponse)

    async def list_content_standards(
        self,
        request: ListContentStandardsRequest,
    ) -> TaskResult[ListContentStandardsResponse]:
        """
        List content standards configurations.

        Args:
            request: Request parameters including optional filters

        Returns:
            TaskResult containing ListContentStandardsResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_content_standards",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.list_content_standards(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_content_standards",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ListContentStandardsResponse)

    async def update_content_standards(
        self,
        request: UpdateContentStandardsRequest,
    ) -> TaskResult[UpdateContentStandardsResponse]:
        """
        Update a content standards configuration.

        Args:
            request: Request parameters including standards_id and updates

        Returns:
            TaskResult containing UpdateContentStandardsResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="update_content_standards",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.update_content_standards(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="update_content_standards",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, UpdateContentStandardsResponse)

    async def calibrate_content(
        self,
        request: CalibrateContentRequest,
    ) -> TaskResult[CalibrateContentResponse]:
        """
        Calibrate content against standards.

        Evaluates content (artifact or URL) against configured standards to
        determine suitability for ad placement.

        Args:
            request: Request parameters including content to evaluate

        Returns:
            TaskResult containing CalibrateContentResponse with verdict
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="calibrate_content",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.calibrate_content(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="calibrate_content",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, CalibrateContentResponse)

    async def validate_content_delivery(
        self,
        request: ValidateContentDeliveryRequest,
    ) -> TaskResult[ValidateContentDeliveryResponse]:
        """
        Validate content delivery against standards.

        Validates that ad delivery records comply with content standards.

        Args:
            request: Request parameters including delivery records

        Returns:
            TaskResult containing ValidateContentDeliveryResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="validate_content_delivery",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.validate_content_delivery(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="validate_content_delivery",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ValidateContentDeliveryResponse)

    async def get_media_buy_artifacts(
        self,
        request: GetMediaBuyArtifactsRequest,
    ) -> TaskResult[GetMediaBuyArtifactsResponse]:
        """
        Get artifacts associated with a media buy.

        Retrieves content artifacts where ads were delivered for a media buy.

        Args:
            request: Request parameters including media_buy_id

        Returns:
            TaskResult containing GetMediaBuyArtifactsResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_media_buy_artifacts",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_media_buy_artifacts(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_media_buy_artifacts",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetMediaBuyArtifactsResponse)

    # ========================================================================
    # V3 Protocol Methods - Sponsored Intelligence
    # ========================================================================

    async def si_get_offering(
        self,
        request: SiGetOfferingRequest,
    ) -> TaskResult[SiGetOfferingResponse]:
        """
        Get sponsored intelligence offering.

        Retrieves product/service offerings that can be presented in a
        sponsored intelligence session.

        Args:
            request: Request parameters including brand context

        Returns:
            TaskResult containing SiGetOfferingResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="si_get_offering",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.si_get_offering(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="si_get_offering",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, SiGetOfferingResponse)

    async def si_initiate_session(
        self,
        request: SiInitiateSessionRequest,
    ) -> TaskResult[SiInitiateSessionResponse]:
        """
        Initiate a sponsored intelligence session.

        Starts a conversational brand experience session with a user.

        Args:
            request: Request parameters including identity and context

        Returns:
            TaskResult containing SiInitiateSessionResponse with session_id
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="si_initiate_session",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.si_initiate_session(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="si_initiate_session",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, SiInitiateSessionResponse)

    async def si_send_message(
        self,
        request: SiSendMessageRequest,
    ) -> TaskResult[SiSendMessageResponse]:
        """
        Send a message in a sponsored intelligence session.

        Continues the conversation in an active SI session.

        Args:
            request: Request parameters including session_id and message

        Returns:
            TaskResult containing SiSendMessageResponse with brand response
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="si_send_message",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.si_send_message(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="si_send_message",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, SiSendMessageResponse)

    async def si_terminate_session(
        self,
        request: SiTerminateSessionRequest,
    ) -> TaskResult[SiTerminateSessionResponse]:
        """
        Terminate a sponsored intelligence session.

        Ends an active SI session, optionally with follow-up actions.

        Args:
            request: Request parameters including session_id and termination context

        Returns:
            TaskResult containing SiTerminateSessionResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="si_terminate_session",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.si_terminate_session(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="si_terminate_session",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, SiTerminateSessionResponse)

    # ========================================================================
    # V3 Governance Methods
    # ========================================================================

    async def get_creative_features(
        self,
        request: GetCreativeFeaturesRequest,
    ) -> TaskResult[GetCreativeFeaturesResponse]:
        """Evaluate governance features for a creative manifest."""
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_creative_features",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_creative_features(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_creative_features",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetCreativeFeaturesResponse)

    async def sync_plans(
        self,
        request: SyncPlansRequest,
    ) -> TaskResult[SyncPlansResponse]:
        """Sync campaign governance plans to the governance agent."""
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_plans",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.sync_plans(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_plans",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, SyncPlansResponse)

    async def check_governance(
        self,
        request: CheckGovernanceRequest,
    ) -> TaskResult[CheckGovernanceResponse]:
        """Check a proposed or committed action against campaign governance."""
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="check_governance",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.check_governance(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="check_governance",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, CheckGovernanceResponse)

    async def report_plan_outcome(
        self,
        request: ReportPlanOutcomeRequest,
    ) -> TaskResult[ReportPlanOutcomeResponse]:
        """Report the outcome of a governed action to the governance agent."""
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="report_plan_outcome",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.report_plan_outcome(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="report_plan_outcome",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ReportPlanOutcomeResponse)

    async def report_plan_adjustment(
        self,
        request: ReportPlanAdjustmentRequest,
    ) -> TaskResult[ReportPlanAdjustmentResponse]:
        """Report or review an adjustment to a governed plan outcome."""
        return cast(
            TaskResult[ReportPlanAdjustmentResponse],
            await self._execute_typed_task(
                "report_plan_adjustment", request, ReportPlanAdjustmentResponse
            ),
        )

    async def get_plan_audit_logs(
        self,
        request: GetPlanAuditLogsRequest,
    ) -> TaskResult[GetPlanAuditLogsResponse]:
        """Retrieve governance state and audit logs for one or more plans."""
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_plan_audit_logs",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_plan_audit_logs(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_plan_audit_logs",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetPlanAuditLogsResponse)

    async def create_property_list(
        self,
        request: CreatePropertyListRequest,
    ) -> TaskResult[CreatePropertyListResponse]:
        """
        Create a property list for governance filtering.

        Property lists define dynamic sets of properties based on filters,
        brand manifests, and feature requirements.

        Args:
            request: Request parameters for creating the property list

        Returns:
            TaskResult containing CreatePropertyListResponse with list_id
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="create_property_list",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.create_property_list(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="create_property_list",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, CreatePropertyListResponse)

    async def get_property_list(
        self,
        request: GetPropertyListRequest,
    ) -> TaskResult[GetPropertyListResponse]:
        """
        Get a property list with optional resolution.

        When resolve=true, returns the list of resolved property identifiers.
        Use this to get the actual properties that match the list's filters.

        Args:
            request: Request parameters including list_id and resolve flag

        Returns:
            TaskResult containing GetPropertyListResponse with identifiers
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_property_list",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_property_list(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_property_list",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetPropertyListResponse)

    async def list_property_lists(
        self,
        request: ListPropertyListsRequest,
    ) -> TaskResult[ListPropertyListsResponse]:
        """
        List property lists owned by a principal.

        Retrieves metadata for all property lists, optionally filtered
        by principal or pagination parameters.

        Args:
            request: Request parameters with optional filtering

        Returns:
            TaskResult containing ListPropertyListsResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_property_lists",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.list_property_lists(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_property_lists",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ListPropertyListsResponse)

    async def update_property_list(
        self,
        request: UpdatePropertyListRequest,
    ) -> TaskResult[UpdatePropertyListResponse]:
        """
        Update a property list.

        Modifies the filters, brand manifest, or other parameters
        of an existing property list.

        Args:
            request: Request parameters with list_id and updates

        Returns:
            TaskResult containing UpdatePropertyListResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="update_property_list",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.update_property_list(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="update_property_list",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, UpdatePropertyListResponse)

    async def delete_property_list(
        self,
        request: DeletePropertyListRequest,
    ) -> TaskResult[DeletePropertyListResponse]:
        """
        Delete a property list.

        Removes a property list. Any active subscriptions to this list
        will be terminated.

        Args:
            request: Request parameters with list_id

        Returns:
            TaskResult containing DeletePropertyListResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="delete_property_list",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.delete_property_list(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="delete_property_list",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, DeletePropertyListResponse)

    # ========================================================================
    # V3 Protocol Methods - Governance (Collection Lists)
    # ========================================================================

    async def create_collection_list(
        self,
        request: CreateCollectionListRequest,
    ) -> TaskResult[CreateCollectionListResponse]:
        """Create a collection list for governance filtering.

        Collection lists define dynamic sets of collections (properties, segments, etc.)
        that can be referenced by authorization rules and audience scoping.

        Args:
            request: Request parameters for creating the collection list

        Returns:
            TaskResult containing CreateCollectionListResponse with list_id
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="create_collection_list",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.create_collection_list(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="create_collection_list",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, CreateCollectionListResponse)

    async def get_collection_list(
        self,
        request: GetCollectionListRequest,
    ) -> TaskResult[GetCollectionListResponse]:
        """Get a collection list with optional resolution.

        When resolve=true, returns the resolved members of the collection list.

        Args:
            request: Request parameters including list_id and resolve flag

        Returns:
            TaskResult containing GetCollectionListResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_collection_list",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_collection_list(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_collection_list",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetCollectionListResponse)

    async def list_collection_lists(
        self,
        request: ListCollectionListsRequest,
    ) -> TaskResult[ListCollectionListsResponse]:
        """List collection lists owned by a principal.

        Args:
            request: Request parameters with optional filtering

        Returns:
            TaskResult containing ListCollectionListsResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_collection_lists",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.list_collection_lists(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="list_collection_lists",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ListCollectionListsResponse)

    async def update_collection_list(
        self,
        request: UpdateCollectionListRequest,
    ) -> TaskResult[UpdateCollectionListResponse]:
        """Update a collection list.

        Args:
            request: Request parameters with list_id and updates

        Returns:
            TaskResult containing UpdateCollectionListResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="update_collection_list",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.update_collection_list(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="update_collection_list",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, UpdateCollectionListResponse)

    async def delete_collection_list(
        self,
        request: DeleteCollectionListRequest,
    ) -> TaskResult[DeleteCollectionListResponse]:
        """Delete a collection list.

        Args:
            request: Request parameters with list_id

        Returns:
            TaskResult containing DeleteCollectionListResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="delete_collection_list",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.delete_collection_list(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="delete_collection_list",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, DeleteCollectionListResponse)

    # ========================================================================
    # V3 Protocol Methods - Governance (Sync Governance)
    # ========================================================================

    async def sync_governance(
        self,
        request: SyncGovernanceRequest,
    ) -> TaskResult[SyncGovernanceResponse]:
        """Sync governance agents attached to an account.

        Attach, detach, or replace the set of governance agents that must be
        consulted for plan approval on an account.

        Args:
            request: Request parameters with account and governance agents

        Returns:
            TaskResult containing SyncGovernanceResponse
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_governance",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.sync_governance(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="sync_governance",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, SyncGovernanceResponse)

    # ========================================================================
    # V3 Protocol Methods - Temporal Matching Protocol (TMP)
    # ========================================================================

    async def context_match(
        self,
        request: ContextMatchRequest,
    ) -> TaskResult[ContextMatchResponse]:
        """Match ad context to buyer packages.

        Evaluates contextual signals for a publisher placement against the
        buyer's active packages and returns matching offers.

        Args:
            request: Context match request with placement, property, and
                optional artifact refs, context signals, and geo data.

        Returns:
            TaskResult containing ContextMatchResponse with offers.
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True, by_alias=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="context_match",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.context_match(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="context_match",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ContextMatchResponse)

    async def identity_match(
        self,
        request: IdentityMatchRequest,
    ) -> TaskResult[IdentityMatchResponse]:
        """Match user identity for package eligibility.

        Evaluates a user identity token against all active packages for
        frequency capping and personalization.

        Args:
            request: Identity match request with user_token, uid_type,
                and package_ids.

        Returns:
            TaskResult containing IdentityMatchResponse with eligible_package_ids.
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True, by_alias=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="identity_match",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.identity_match(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="identity_match",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, IdentityMatchResponse)

    # ========================================================================
    # V3 Protocol Methods - Brand Rights
    # ========================================================================

    async def get_brand_identity(
        self,
        request: GetBrandIdentityRequest,
    ) -> TaskResult[GetBrandIdentityResponse]:
        """Get brand identity information.

        Retrieves brand identity data including logos, colors, fonts,
        voice synthesis config, and rights availability.

        Args:
            request: Request with brand_id and optional fields filter.

        Returns:
            TaskResult containing GetBrandIdentityResponse.
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_brand_identity",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_brand_identity(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_brand_identity",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetBrandIdentityResponse)

    async def get_rights(
        self,
        request: GetRightsRequest,
    ) -> TaskResult[GetRightsResponse]:
        """Get available rights for licensing.

        Searches for rights offerings using natural language query and
        filters by type, uses, countries, and buyer compatibility.

        Args:
            request: Request with query, uses, and optional filters.

        Returns:
            TaskResult containing GetRightsResponse with matched rights.
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_rights",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.get_rights(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="get_rights",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, GetRightsResponse)

    async def acquire_rights(
        self,
        request: AcquireRightsRequest,
    ) -> TaskResult[AcquireRightsResponse]:
        """Acquire rights for brand content usage.

        Binding contractual request to license rights for a campaign.
        Returns credentials for generating rights-cleared content.

        Args:
            request: Request with rights_id, pricing_option_id, buyer,
                campaign, and revocation_webhook.

        Returns:
            TaskResult containing AcquireRightsResponse (acquired,
            pending_approval, rejected, or error).
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="acquire_rights",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.acquire_rights(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="acquire_rights",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, AcquireRightsResponse)

    async def update_rights(
        self,
        request: UpdateRightsRequest,
    ) -> TaskResult[UpdateRightsResponse]:
        """Update terms of an existing rights acquisition.

        Modifies a previously acquired rights record — typically to extend
        the ``end_date``, raise the ``impression_cap``, pause/unpause via
        ``paused``, or swap to a compatible ``pricing_option_id``. Partial
        update: pass only the fields you want to change.

        Failure modes (surface as ``TaskResult`` with ``success=False``):

        * Acquisition is expired or revoked — the seller rejects the update
          outright; mint a fresh ``acquire_rights`` instead.
        * ``pricing_option_id`` swap to an incompatible option — rejected;
          the new option's terms must be a strict superset / compatible
          with the original acquisition.
        * No partial-state mutations on rejection: the acquisition remains
          at its prior state when any field fails validation.

        Args:
            request: Request with ``rights_id`` and at least one mutable
                field (``end_date``, ``impression_cap``, ``paused``, or
                ``pricing_option_id``).

        Returns:
            TaskResult containing UpdateRightsResponse (updated or error).
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="update_rights",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.update_rights(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="update_rights",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, UpdateRightsResponse)

    async def validate_input(self, request: Any) -> TaskResult[Any]:
        """Validate creative input against a format declaration."""
        from adcp.types import _generated as gen

        params = request.model_dump(mode="json", exclude_none=True)
        raw_result = await self.adapter.validate_input(params)
        return self.adapter._parse_response(raw_result, gen.ValidateInputResponse)

    async def verify_brand_claim(self, request: Any) -> TaskResult[Any]:
        """Verify a single brand claim."""
        from adcp.types import _generated as gen

        params = request.model_dump(mode="json", exclude_none=True)
        raw_result = await self.adapter.verify_brand_claim(params)
        return self.adapter._parse_response(raw_result, gen.VerifyBrandClaimResponse)

    async def verify_brand_claims(self, request: Any) -> TaskResult[Any]:
        """Verify multiple brand claims."""
        from adcp.types import _generated as gen

        params = request.model_dump(mode="json", exclude_none=True)
        raw_result = await self.adapter.verify_brand_claims(params)
        return self.adapter._parse_response(raw_result, gen.VerifyBrandClaimsResponseBulk)

    # ========================================================================
    # V3 Protocol Methods - Compliance
    # ========================================================================

    async def comply_test_controller(
        self,
        request: ComplyTestControllerRequest,
    ) -> TaskResult[ComplyTestControllerResponse]:
        """Compliance test controller for sandbox testing.

        Enables sellers to simulate state transitions and delivery data
        in a sandbox environment for compliance testing.

        Args:
            request: Request specifying scenario and parameters.

        Returns:
            TaskResult containing ComplyTestControllerResponse.
        """
        operation_id = create_operation_id()
        params = request.model_dump(mode="json", exclude_none=True)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_REQUEST,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="comply_test_controller",
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        raw_result = await self.adapter.comply_test_controller(params)

        self._emit_activity(
            Activity(
                type=ActivityType.PROTOCOL_RESPONSE,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type="comply_test_controller",
                status=raw_result.status,
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
        )

        return self.adapter._parse_response(raw_result, ComplyTestControllerResponse)

    async def execute_task(self, task_name: str, request: BaseModel) -> TaskResult[Any]:
        """Execute a standard task through the canonical primary API map."""

        legacy_only_tasks = {
            "build_creative",
            "list_creative_formats",
            "preview_creative",
        }
        if task_name in legacy_only_tasks:
            raise ValueError(
                f"{task_name} is legacy-only; use execute_task_legacy() or the "
                "corresponding *_legacy method"
            )
        serialized = request.model_dump(mode="json", exclude_none=True)
        if strip_legacy_creative_identity(serialized) != serialized:
            raise ValueError(
                f"{task_name} request contains legacy creative identity; generic execute_task "
                "is canonical-only"
            )
        method = getattr(self, task_name, None)
        if method is None or not callable(method) or task_name.endswith("_legacy"):
            raise ValueError(f"Unknown canonical AdCP task: {task_name}")
        return cast(TaskResult[Any], await method(request))

    async def execute_task_legacy(self, task_name: str, request: BaseModel) -> TaskResult[Any]:
        """Execute an explicitly raw creative task for migration tooling."""

        methods: dict[str, Callable[[Any], Any]] = {
            "build_creative": self.build_creative_legacy,
            "create_media_buy": self.create_media_buy_legacy,
            "get_creative_delivery": self.get_creative_delivery_legacy,
            "get_media_buy_delivery": self.get_media_buy_delivery_legacy,
            "get_media_buys": self.get_media_buys_legacy,
            "get_products": self.get_products_legacy,
            "list_creative_formats": self.list_creative_formats_legacy,
            "list_creatives": self.list_creatives_legacy,
            "preview_creative": self.preview_creative_legacy,
            "sync_creatives": self.sync_creatives_legacy,
            "update_media_buy": self.update_media_buy_legacy,
        }
        method = methods.get(task_name)
        if method is None:
            raise ValueError(
                f"No explicit legacy task adapter for {task_name!r}; "
                "use the raw protocol adapter for conformance tooling"
            )
        return cast(TaskResult[Any], await method(request))

    async def list_tools(self) -> list[str]:
        """
        List available tools from the agent.

        Returns:
            List of tool names
        """
        return await self.adapter.list_tools()

    async def get_info(self) -> dict[str, Any]:
        """
        Get agent information including AdCP extension metadata.

        Returns agent card information including:
        - Agent name, description, version
        - Protocol type (mcp or a2a)
        - AdCP version (from extensions.adcp.adcp_version)
        - Supported protocols (from extensions.adcp.protocols_supported)
        - Available tools/skills

        Returns:
            Dictionary with agent metadata
        """
        return await self.adapter.get_agent_info()

    async def close(self) -> None:
        """Close the adapter and clean up resources."""
        if hasattr(self.adapter, "close"):
            logger.debug(f"Closing adapter for agent {self.agent_config.id}")
            await self.adapter.close()

    async def close_mcp_session(self, session_id: str | None = None) -> None:
        """Explicitly terminate a stateful MCP Streamable HTTP session.

        This sends ``DELETE`` to the configured MCP endpoint with the
        ``Mcp-Session-Id`` header. When ``session_id`` is omitted, the
        SDK-managed current session is closed. It is only valid for MCP
        agents using ``mcp_transport="streamable_http"``.
        """
        if not isinstance(self.adapter, MCPAdapter):
            raise TypeError(
                "close_mcp_session is only supported for MCP clients; "
                f"got {self.agent_config.protocol}"
            )
        await self.adapter.close_mcp_session(session_id)

    async def __aenter__(self) -> ADCPClient:
        """Async context manager entry."""
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        """Async context manager exit."""
        await self.close()

    def _verify_webhook_signature(
        self,
        payload: dict[str, Any],
        signature: str,
        timestamp: str,
        raw_body: bytes | str | None = None,
    ) -> bool:
        """
        Verify HMAC-SHA256 signature of webhook payload.

        The verification algorithm matches get_adcp_signed_headers_for_webhook:
        1. Constructs message as "{timestamp}.{raw_http_body_bytes}"
        2. HMAC-SHA256 signs with the shared secret
        3. Compares against the provided signature (with "sha256=" prefix stripped)
           using constant-time comparison.

        Per AdCP spec (adcontextprotocol/adcp#2478): verifiers MUST use the raw
        HTTP body bytes captured before any JSON parse; they SHOULD NOT
        re-serialize a parsed payload to reconstruct the signed bytes, because
        re-serialization silently fails against signers whose output differs in
        separator choice, key order, unicode escapes, or number formatting —
        masking signer bugs the verifier should surface. Callers that genuinely
        cannot capture raw bytes MUST fail closed.

        This implementation therefore rejects verification attempts that don't
        supply ``raw_body``. Capture it from your framework's pre-parse hook
        (FastAPI ``Request.body()``, Flask ``request.get_data(cache=True)``,
        aiohttp ``Request.read()``, Express ``express.raw()``).

        Args:
            payload: Parsed webhook payload dict (not used for signing; kept
                for signature parity with callers, but verification derives
                solely from ``raw_body``).
            signature: Signature to verify (with or without "sha256=" prefix)
            timestamp: Unix timestamp in seconds from X-AdCP-Timestamp header
            raw_body: Raw HTTP request body bytes as received on the wire,
                captured before any JSON parse. Required.

        Returns:
            True if signature is valid, False otherwise (including when
            ``raw_body`` is missing — fails closed per spec).
        """
        if not self.webhook_secret:
            logger.error("Webhook signature verification failed: no webhook_secret configured")
            return False

        # Fail closed per adcontextprotocol/adcp#2478: verifiers that cannot
        # capture raw bytes MUST reject, surfacing the infrastructure gap
        # rather than silently reconstructing a signed body that may diverge
        # from the bytes the signer actually hashed.
        if raw_body is None:
            logger.error(
                "Webhook signature verification failed: raw_body is required. "
                "Capture the raw HTTP body pre-parse and pass it to "
                "handle_webhook(raw_body=...). See "
                "https://adcontextprotocol.org/docs/building/implementation/security"
                "#legacy-hmac-sha256-fallback-deprecated-removed-in-40"
            )
            return False

        # Reject stale or future timestamps to prevent replay attacks
        try:
            ts = int(timestamp)
        except (ValueError, TypeError):
            return False
        now = int(time.time())
        if abs(now - ts) > self.webhook_timestamp_tolerance:
            return False

        # Strip "sha256=" prefix if present
        if signature.startswith("sha256="):
            signature = signature[7:]

        payload_str = raw_body.decode("utf-8") if isinstance(raw_body, bytes) else raw_body

        # Construct signed message: timestamp.payload
        signed_message = f"{timestamp}.{payload_str}"

        # Generate expected signature
        expected_signature = hmac.new(
            self.webhook_secret.encode("utf-8"), signed_message.encode("utf-8"), hashlib.sha256
        ).hexdigest()

        return hmac.compare_digest(signature, expected_signature)

    def _parse_webhook_result(
        self,
        task_id: str,
        task_type: str,
        operation_id: str,
        status: GeneratedTaskStatus,
        result: Any,
        timestamp: datetime | str,
        message: str | None,
        context_id: str | None,
        *,
        preserve_legacy_identity: bool = False,
    ) -> TaskResult[AdcpAsyncResponseData]:
        """
        Parse webhook data into typed TaskResult based on task_type.

        Args:
            task_id: Unique identifier for this task
            task_type: Task type from application routing (e.g., "get_products")
            operation_id: Operation identifier from application routing
            status: Current task status
            result: Task-specific payload (AdCP response data)
            timestamp: ISO 8601 timestamp when webhook was generated
            message: Human-readable summary of task state
            context_id: Session/conversation identifier

        Returns:
            TaskResult with task-specific typed response data

        Note:
            This method works with both MCP and A2A protocols by accepting
            protocol-agnostic parameters rather than protocol-specific objects.
        """
        from adcp.utils.response_parser import parse_json_or_text

        # Map task types to their response types (using string literals, not enum)
        # Note: Some response types are Union types (e.g., ActivateSignalResponse = Success | Error)
        response_type_map: dict[str, type[BaseModel] | Any] = {
            # Core operations
            "get_products": GetProductsResponse,
            "list_products": ListProductsResponse,
            "request_proposals": RequestProposalsResponse,
            "refine_proposals": RefineProposalsResponse,
            "decline_proposals": DeclineProposalsResponse,
            "buy_products": BuyProductsResponse,
            "accept_proposal": AcceptProposalResponse,
            "control_media_buy": ControlMediaBuyResponse,
            "list_creative_formats": ListCreativeFormatsResponse,
            "sync_creatives": SyncCreativesResponse,
            "list_creatives": ListCreativesResponse,
            "build_creative": LegacyBuildCreativeResponse,
            "preview_creative": LegacyPreviewCreativeResponse,
            "create_media_buy": CreateMediaBuyResponse,
            "update_media_buy": UpdateMediaBuyResponse,
            "get_media_buy_delivery": GetMediaBuyDeliveryResponse,
            "get_media_buys": GetMediaBuysResponse,
            "get_signals": GetSignalsResponse,
            "activate_signal": ActivateSignalResponse,
            "provide_performance_feedback": ProvidePerformanceFeedbackResponse,
            "report_usage": ReportUsageResponse,
            "get_account_financials": GetAccountFinancialsResponse,
            "list_accounts": ListAccountsResponse,
            "sync_accounts": SyncAccountsResponse,
            "log_event": LogEventResponse,
            "sync_event_sources": SyncEventSourcesResponse,
            "sync_audiences": SyncAudiencesResponse,
            "sync_catalogs": SyncCatalogsResponse,
            "get_creative_delivery": GetCreativeDeliveryResponse,
            # V3 Protocol Discovery
            "get_adcp_capabilities": GetAdcpCapabilitiesResponse,
            "sync_agent_notification_configs": SyncAgentNotificationConfigsResponse,
            # V3 Content Standards
            "create_content_standards": CreateContentStandardsResponse,
            "get_content_standards": GetContentStandardsResponse,
            "list_content_standards": ListContentStandardsResponse,
            "update_content_standards": UpdateContentStandardsResponse,
            "calibrate_content": CalibrateContentResponse,
            "validate_content_delivery": ValidateContentDeliveryResponse,
            "get_media_buy_artifacts": GetMediaBuyArtifactsResponse,
            # V3 Sponsored Intelligence
            "si_get_offering": SiGetOfferingResponse,
            "si_initiate_session": SiInitiateSessionResponse,
            "si_send_message": SiSendMessageResponse,
            "si_terminate_session": SiTerminateSessionResponse,
            # V3 Governance
            "get_creative_features": GetCreativeFeaturesResponse,
            "sync_plans": SyncPlansResponse,
            "check_governance": CheckGovernanceResponse,
            "report_plan_outcome": ReportPlanOutcomeResponse,
            "report_plan_adjustment": ReportPlanAdjustmentResponse,
            "get_plan_audit_logs": GetPlanAuditLogsResponse,
            "create_property_list": CreatePropertyListResponse,
            "get_property_list": GetPropertyListResponse,
            "list_property_lists": ListPropertyListsResponse,
            "update_property_list": UpdatePropertyListResponse,
            "delete_property_list": DeletePropertyListResponse,
            # TMP
            "context_match": ContextMatchResponse,
            "identity_match": IdentityMatchResponse,
            # Brand Rights
            "get_brand_identity": GetBrandIdentityResponse,
            "get_rights": GetRightsResponse,
            "acquire_rights": AcquireRightsResponse,
            "update_rights": UpdateRightsResponse,
            # Compliance
            "comply_test_controller": ComplyTestControllerResponse,
        }

        if preserve_legacy_identity:
            response_type_map.update(
                {
                    "get_products": LegacyGetProductsResponse,
                    "list_creative_formats": ListCreativeFormatsResponse,
                    "sync_creatives": LegacySyncCreativesResponse,
                    "list_creatives": LegacyListCreativesResponse,
                    "build_creative": LegacyBuildCreativeResponse,
                    "preview_creative": LegacyPreviewCreativeResponse,
                    "create_media_buy": LegacyCreateMediaBuyResponse,
                    "update_media_buy": LegacyUpdateMediaBuyResponse,
                    "get_media_buy_delivery": LegacyGetMediaBuyDeliveryResponse,
                    "get_media_buys": LegacyGetMediaBuysResponse,
                    "get_creative_delivery": LegacyGetCreativeDeliveryResponse,
                }
            )

        # Handle completed tasks with result parsing
        if status == GeneratedTaskStatus.completed and result is not None:
            if task_type == "get_products" and not preserve_legacy_identity:
                projected = self._canonicalize_get_products_result(
                    TaskResult[Any](
                        status=TaskStatus.COMPLETED,
                        data=result,
                        success=True,
                        message=message,
                        metadata={
                            "task_id": task_id,
                            "operation_id": operation_id,
                            "timestamp": timestamp,
                            "message": message,
                        },
                    )
                )
                return cast(TaskResult[AdcpAsyncResponseData], projected)
            legacy_lifecycle_types = {
                "create_media_buy": (LegacyCreateMediaBuyResponse, CreateMediaBuyResponse),
                "update_media_buy": (LegacyUpdateMediaBuyResponse, UpdateMediaBuyResponse),
                "sync_creatives": (LegacySyncCreativesResponse, SyncCreativesResponse),
                "get_media_buys": (LegacyGetMediaBuysResponse, GetMediaBuysResponse),
                "get_media_buy_delivery": (
                    LegacyGetMediaBuyDeliveryResponse,
                    GetMediaBuyDeliveryResponse,
                ),
            }
            lifecycle_types = legacy_lifecycle_types.get(task_type)
            if (
                not preserve_legacy_identity
                and lifecycle_types is not None
                and self._callback_creative_dialect(result) is CreativeDialect.LEGACY
            ):
                projected = self._canonicalize_lifecycle_result(
                    TaskResult[Any](
                        status=TaskStatus.COMPLETED,
                        data=result,
                        success=True,
                        message=message,
                    ),
                    legacy_type=lifecycle_types[0],
                    canonical_type=lifecycle_types[1],
                )
                return cast(TaskResult[AdcpAsyncResponseData], projected)
            if (
                not preserve_legacy_identity
                and task_type == "list_creatives"
                and self._callback_creative_dialect(result) is CreativeDialect.LEGACY
            ):
                projected = self._canonicalize_format_read_result(
                    TaskResult[Any](
                        status=TaskStatus.COMPLETED,
                        data=result,
                        success=True,
                        message=message,
                    ),
                    legacy_type=LegacyListCreativesResponse,
                    canonical_type=ListCreativesResponse,
                    collection="creatives",
                    require_format=True,
                )
                return cast(TaskResult[AdcpAsyncResponseData], projected)
            if (
                not preserve_legacy_identity
                and task_type == "get_creative_delivery"
                and self._callback_creative_dialect(result) is CreativeDialect.LEGACY
            ):
                projected = self._canonicalize_format_read_result(
                    TaskResult[Any](
                        status=TaskStatus.COMPLETED,
                        data=result,
                        success=True,
                        message=message,
                    ),
                    legacy_type=LegacyGetCreativeDeliveryResponse,
                    canonical_type=GetCreativeDeliveryResponse,
                    collection="creatives",
                    require_format=False,
                )
                return cast(TaskResult[AdcpAsyncResponseData], projected)
            response_type = response_type_map.get(task_type)
            if response_type:
                try:
                    parsed_result: Any = parse_json_or_text(result, response_type)
                    return TaskResult[AdcpAsyncResponseData](
                        status=TaskStatus.COMPLETED,
                        data=parsed_result,
                        success=True,
                        metadata={
                            "task_id": task_id,
                            "operation_id": operation_id,
                            "timestamp": timestamp,
                            "message": message,
                        },
                    )
                except ValueError as e:
                    logger.warning(f"Failed to parse webhook result: {e}")
                    # Fall through to untyped result

        # Handle failed, input-required, or unparseable results
        # Convert status to core TaskStatus enum
        status_map = {
            GeneratedTaskStatus.completed: TaskStatus.COMPLETED,
            GeneratedTaskStatus.submitted: TaskStatus.SUBMITTED,
            GeneratedTaskStatus.working: TaskStatus.WORKING,
            GeneratedTaskStatus.failed: TaskStatus.FAILED,
            GeneratedTaskStatus.input_required: TaskStatus.NEEDS_INPUT,
        }
        task_status = status_map.get(status, TaskStatus.FAILED)

        # Extract error message from result.errors if present
        error_message: str | None = None
        if result is not None and hasattr(result, "errors"):
            errors = getattr(result, "errors", None)
            if errors and len(errors) > 0:
                first_error = errors[0]
                if hasattr(first_error, "message"):
                    error_message = first_error.message

        return TaskResult[AdcpAsyncResponseData](
            status=task_status,
            data=result,
            success=status == GeneratedTaskStatus.completed,
            error=error_message,
            metadata={
                "task_id": task_id,
                "operation_id": operation_id,
                "timestamp": timestamp,
                "message": message,
                "context_id": context_id,
            },
        )

    async def _handle_mcp_webhook(
        self,
        payload: dict[str, Any],
        task_type: str,
        operation_id: str,
        signature: str | None,
        timestamp: str | None = None,
        raw_body: bytes | str | None = None,
        preserve_legacy_identity: bool = False,
    ) -> TaskResult[AdcpAsyncResponseData]:
        """
        Handle MCP webhook delivered via HTTP POST.

        Args:
            payload: Webhook payload dict
            task_type: Task type from application routing
            operation_id: Operation identifier from application routing
            signature: HMAC-SHA256 signature from X-AdCP-Signature. Required
                when ``webhook_secret`` configures the deprecated HMAC fallback.
            timestamp: Unix timestamp from X-AdCP-Timestamp. Required with the
                deprecated HMAC fallback.
            raw_body: Raw HTTP request body. Required with the deprecated HMAC
                fallback so the authenticated bytes are the bytes processed.

        Returns:
            TaskResult with parsed task-specific response data

        Raises:
            ADCPWebhookSignatureError: If signature verification fails
            ValidationError: If payload doesn't match McpWebhookPayload schema
        """
        from adcp.types.generated_poc.core.mcp_webhook_payload import McpWebhookPayload

        # Signed MCP webhooks are the secure default. Receiving without a
        # verifier requires an explicit compatibility opt-in so a missing
        # secret cannot silently turn a public endpoint into an unauthenticated
        # callback receiver.
        if self.webhook_secret:
            if not signature or not timestamp:
                raise ADCPWebhookSignatureError(
                    "Webhook signature and timestamp headers are required"
                )
            if not self._verify_webhook_signature(payload, signature, timestamp, raw_body):
                logger.warning(
                    f"Webhook signature verification failed for agent {self.agent_config.id}"
                )
                raise ADCPWebhookSignatureError("Invalid webhook signature")
            if raw_body is None:  # Defensive type narrowing; verifier rejects this above.
                raise ADCPWebhookSignatureError("Signed webhook raw body is required")
            try:
                authenticated_payload = json.loads(raw_body)
            except (TypeError, ValueError, UnicodeDecodeError) as exc:
                raise ADCPWebhookSignatureError("Invalid signed webhook body") from exc
            if not isinstance(authenticated_payload, dict):
                raise ADCPWebhookSignatureError("Signed webhook body must be a JSON object")
            # Process the bytes that were authenticated, not a separately
            # supplied parsed object that middleware could have transformed.
            payload = cast(dict[str, Any], authenticated_payload)
        elif self.allow_unauthenticated_webhooks is not True:
            raise ADCPWebhookSignatureError(
                "MCP webhook cannot be authenticated because webhook_secret is not configured; "
                "use WebhookReceiver for RFC 9421 callbacks, configure a shared secret only for "
                "an explicitly selected legacy HMAC registration, or set "
                "allow_unauthenticated_webhooks=True only for receivers isolated from "
                "untrusted networks"
            )

        # Select the canonical/legacy surface from the body only after signed
        # callbacks have been replaced by their authenticated raw bytes.
        payload_task_type = payload.get("task_type")
        if not preserve_legacy_identity and payload_task_type in _LEGACY_ONLY_CREATIVE_TASKS:
            raise ValueError(
                f"{payload_task_type} webhook payloads carry legacy creative identity; use "
                "handle_webhook_legacy()"
            )

        # Validate and parse MCP webhook payload
        webhook = McpWebhookPayload.model_validate(payload)
        authenticated_task_type = webhook.task_type.value
        authenticated_operation_id = webhook.operation_id

        if preserve_legacy_identity:
            if authenticated_task_type not in _LEGACY_CREATIVE_TASKS:
                raise ValueError(
                    f"{authenticated_task_type} is not a legacy-only callback; use "
                    "handle_webhook()"
                )

        # Emit activity for monitoring
        self._emit_activity(
            Activity(
                type=ActivityType.WEBHOOK_RECEIVED,
                operation_id=authenticated_operation_id,
                agent_id=self.agent_config.id,
                task_type=authenticated_task_type,
                timestamp=datetime.now(timezone.utc).isoformat(),
                metadata={
                    "task_id": webhook.task_id,
                    "status": webhook.status.value,
                    "protocol": "mcp",
                },
            )
        )

        # Extract fields and parse result
        return self._parse_webhook_result(
            task_id=webhook.task_id,
            task_type=authenticated_task_type,
            operation_id=authenticated_operation_id,
            status=webhook.status,
            result=webhook.result,
            timestamp=webhook.timestamp,
            message=webhook.message,
            context_id=webhook.context_id,
            preserve_legacy_identity=preserve_legacy_identity,
        )

    async def _handle_a2a_webhook(
        self,
        payload: Task | TaskStatusUpdateEvent,
        task_type: str,
        operation_id: str,
        preserve_legacy_identity: bool = False,
    ) -> TaskResult[AdcpAsyncResponseData]:
        """
        Handle A2A webhook delivered through Task or TaskStatusUpdateEvent.

        Per A2A specification:
        - Terminated statuses (completed, failed): Payload is Task with artifacts[].parts[]
        - Intermediate statuses (working, input-required, submitted):
        Payload is TaskStatusUpdateEvent with status.message.parts[]

        Args:
            payload: A2A Task or TaskStatusUpdateEvent object
            task_type: Task type from application routing
            operation_id: Operation identifier from application routing

        Returns:
            TaskResult with parsed task-specific response data

        Note:
            Signature verification is NOT applicable for A2A webhooks
            as they arrive through authenticated A2A connections, not HTTP.
        """
        from a2a import types as _pb
        from google.protobuf.json_format import MessageToDict as _MessageToDict

        def _a2a_part_data_dict(part: _pb.Part) -> Any:
            if part.WhichOneof("content") != "data":
                return None
            return _MessageToDict(part.data)

        def _a2a_part_text(part: _pb.Part) -> str | None:
            if part.WhichOneof("content") != "text":
                return None
            return part.text

        def _a2a_state_to_string(state_value: int) -> str:
            """Map ``TaskState`` int → spec string (``TASK_STATE_COMPLETED`` → ``completed``)."""
            name = _pb.TaskState.Name(state_value)
            if name.startswith("TASK_STATE_"):
                return name[len("TASK_STATE_") :].lower().replace("_", "-")
            return name.lower()

        def _a2a_timestamp(ts: Any) -> datetime | str:
            """Convert a proto Timestamp (or string) to datetime/ISO string."""
            if ts is None:
                return datetime.now(timezone.utc)
            if isinstance(ts, str):
                return ts or datetime.now(timezone.utc)
            try:
                return cast(datetime, ts.ToDatetime().replace(tzinfo=timezone.utc))
            except AttributeError:
                return datetime.now(timezone.utc)

        adcp_data: Any = None
        text_message: str | None = None
        task_id: str
        context_id: str | None
        status_state: str
        timestamp: datetime | str

        # Type detection and extraction based on payload type
        if isinstance(payload, TaskStatusUpdateEvent):
            task_id = payload.task_id
            context_id = payload.context_id or None
            has_status = payload.HasField("status")
            status_state = _a2a_state_to_string(payload.status.state) if has_status else "failed"
            timestamp = (
                _a2a_timestamp(payload.status.timestamp)
                if has_status and payload.status.HasField("timestamp")
                else datetime.now(timezone.utc)
            )

            if has_status and payload.status.HasField("message") and payload.status.message.parts:
                data_parts = [
                    d
                    for d in (_a2a_part_data_dict(p) for p in payload.status.message.parts)
                    if d is not None
                ]
                if data_parts:
                    adcp_data = data_parts[-1]
                    if isinstance(adcp_data, dict) and "response" in adcp_data:
                        adcp_data = adcp_data["response"]

                for part in payload.status.message.parts:
                    text = _a2a_part_text(part)
                    if text is not None:
                        text_message = text
                        break

        else:
            task_id = payload.id
            context_id = payload.context_id or None
            has_status = payload.HasField("status")
            status_state = _a2a_state_to_string(payload.status.state) if has_status else "failed"
            timestamp = (
                _a2a_timestamp(payload.status.timestamp)
                if has_status and payload.status.HasField("timestamp")
                else datetime.now(timezone.utc)
            )

            if payload.artifacts:
                target_artifact = payload.artifacts[-1]
                if target_artifact.parts:
                    data_parts = [
                        d
                        for d in (_a2a_part_data_dict(p) for p in target_artifact.parts)
                        if d is not None
                    ]
                    if data_parts:
                        adcp_data = data_parts[-1]
                        if isinstance(adcp_data, dict) and "response" in adcp_data:
                            adcp_data = adcp_data["response"]

                    for part in target_artifact.parts:
                        text = _a2a_part_text(part)
                        if text is not None:
                            text_message = text
                            break

        # Map A2A status.state to GeneratedTaskStatus enum
        status_map = {
            "completed": GeneratedTaskStatus.completed,
            "submitted": GeneratedTaskStatus.submitted,
            "working": GeneratedTaskStatus.working,
            "failed": GeneratedTaskStatus.failed,
            "input-required": GeneratedTaskStatus.input_required,
            "input_required": GeneratedTaskStatus.input_required,  # Handle both formats
        }
        mapped_status = status_map.get(status_state, GeneratedTaskStatus.failed)

        # Emit activity for monitoring
        self._emit_activity(
            Activity(
                type=ActivityType.WEBHOOK_RECEIVED,
                operation_id=operation_id,
                agent_id=self.agent_config.id,
                task_type=task_type,
                timestamp=datetime.now(timezone.utc).isoformat(),
                metadata={
                    "task_id": task_id,
                    "protocol": "a2a",
                    "payload_type": (
                        "TaskStatusUpdateEvent"
                        if isinstance(payload, TaskStatusUpdateEvent)
                        else "Task"
                    ),
                },
            )
        )

        # Parse and return typed result by passing extracted fields directly
        return self._parse_webhook_result(
            task_id=task_id,
            task_type=task_type,
            operation_id=operation_id,
            status=mapped_status,
            result=adcp_data,
            timestamp=timestamp,
            message=text_message,
            context_id=context_id,
            preserve_legacy_identity=preserve_legacy_identity,
        )

    async def handle_webhook(
        self,
        payload: dict[str, Any] | Task | TaskStatusUpdateEvent,
        task_type: str,
        operation_id: str,
        signature: str | None = None,
        timestamp: str | None = None,
        raw_body: bytes | str | None = None,
    ) -> TaskResult[AdcpAsyncResponseData]:
        """
        Handle incoming webhook and return typed result.

        This method provides a unified interface for handling webhooks from both
        MCP and A2A protocols:

        - MCP Webhooks: HTTP POST with dict payload; the deprecated HMAC fallback
          requires a signature, timestamp, and raw body
        - A2A Webhooks: Task or TaskStatusUpdateEvent objects based on status

        The method automatically detects the protocol type and routes to the
        appropriate handler. Both protocols return a consistent TaskResult
        structure with typed AdCP response data.

        Args:
            payload: Webhook payload - one of:
                - dict[str, Any]: MCP webhook payload from HTTP POST
                - Task: A2A webhook for terminated statuses (completed, failed)
                - TaskStatusUpdateEvent: A2A webhook for intermediate statuses
                  (working, input-required, submitted)
            task_type: Task type from application routing for A2A callbacks. For
                MCP callbacks, the validated payload's authenticated ``task_type``
                controls parsing and activity correlation.
            operation_id: Operation identifier from application routing. For MCP
                callbacks, the authenticated payload value controls correlation
                when present; this argument is a compatibility fallback for old
                payloads that omit it.
            signature: HMAC-SHA256 signature from X-AdCP-Signature. Required when
                ``webhook_secret`` configures the deprecated HMAC fallback and
                ignored for A2A callbacks.
            timestamp: Unix timestamp from X-AdCP-Timestamp. Required with the
                deprecated HMAC fallback and ignored for A2A callbacks.
            raw_body: Raw HTTP request body captured before JSON parsing. Required
                with the deprecated HMAC fallback and ignored for A2A callbacks.

        Returns:
            TaskResult with parsed task-specific response data. The structure
            is identical regardless of protocol.

        Raises:
            ADCPWebhookSignatureError: If MCP signature verification fails
            ValidationError: If MCP payload doesn't match WebhookPayload schema

        Note:
            AdCP-conformant public MCP endpoints should use
            :class:`adcp.webhooks.WebhookReceiver`, which verifies RFC 9421,
            deduplicates retries, and parses the authenticated body. This method's
            HMAC mode exists only for explicitly selected legacy registrations.

        Examples:
            MCP webhook (HTTP endpoint):
            >>> @app.post("/webhook/{task_type}/{agent_id}/{operation_id}")
            >>> async def webhook_handler(task_type: str, operation_id: str, request: Request):
            >>>     raw_body = await request.body()
            >>>     payload = json.loads(raw_body)
            >>>     signature = request.headers.get("X-AdCP-Signature")
            >>>     timestamp = request.headers.get("X-AdCP-Timestamp")
            >>>     result = await client.handle_webhook(
            >>>         payload, task_type, operation_id, signature, timestamp,
            >>>         raw_body=raw_body,
            >>>     )
            >>>     if result.success:
            >>>         print(f"Task completed: {result.data}")

            A2A webhook with Task (terminated status):
            >>> async def on_task_completed(task: Task):
            >>>     # Extract task_type and operation_id from your app's task tracking
            >>>     task_type = your_task_registry.get_type(task.id)
            >>>     operation_id = your_task_registry.get_operation_id(task.id)
            >>>     result = await client.handle_webhook(
            >>>         task, task_type, operation_id
            >>>     )
            >>>     if result.success:
            >>>         print(f"Task completed: {result.data}")

            A2A webhook with TaskStatusUpdateEvent (intermediate status):
            >>> async def on_task_update(event: TaskStatusUpdateEvent):
            >>>     # Extract task_type and operation_id from your app's task tracking
            >>>     task_type = your_task_registry.get_type(event.task_id)
            >>>     operation_id = your_task_registry.get_operation_id(event.task_id)
            >>>     result = await client.handle_webhook(
            >>>         event, task_type, operation_id
            >>>     )
            >>>     if result.status == GeneratedTaskStatus.working:
            >>>         print(f"Task still working: {result.metadata.get('message')}")
        """
        if (
            isinstance(payload, (Task, TaskStatusUpdateEvent))
            and task_type in _LEGACY_ONLY_CREATIVE_TASKS
        ):
            raise ValueError(
                f"{task_type} webhook payloads carry legacy creative identity; use "
                "handle_webhook_legacy()"
            )
        result = await self._dispatch_webhook(
            payload,
            task_type,
            operation_id,
            signature,
            timestamp,
            raw_body,
            preserve_legacy_identity=False,
        )
        sanitized = strip_legacy_creative_identity(result.model_dump(mode="python"))
        return TaskResult[AdcpAsyncResponseData].model_validate(sanitized)

    async def handle_webhook_legacy(
        self,
        payload: dict[str, Any] | Task | TaskStatusUpdateEvent,
        task_type: str,
        operation_id: str,
        signature: str | None = None,
        timestamp: str | None = None,
        raw_body: bytes | str | None = None,
    ) -> TaskResult[AdcpAsyncResponseData]:
        """Parse a callback for a task whose protocol shape is explicitly legacy-only."""

        if isinstance(payload, (Task, TaskStatusUpdateEvent)) and task_type not in (
            _LEGACY_CREATIVE_TASKS
        ):
            raise ValueError(f"{task_type} is not a legacy-only callback; use handle_webhook()")
        self._warn_legacy_creative_api("handle_webhook_legacy")
        return await self._dispatch_webhook(
            payload,
            task_type,
            operation_id,
            signature,
            timestamp,
            raw_body,
            preserve_legacy_identity=True,
        )

    async def _dispatch_webhook(
        self,
        payload: dict[str, Any] | Task | TaskStatusUpdateEvent,
        task_type: str,
        operation_id: str,
        signature: str | None,
        timestamp: str | None,
        raw_body: bytes | str | None,
        *,
        preserve_legacy_identity: bool,
    ) -> TaskResult[AdcpAsyncResponseData]:
        """Route a callback after the public canonical/legacy boundary is selected."""

        # Detect protocol type and route to appropriate handler
        if isinstance(payload, (Task, TaskStatusUpdateEvent)):
            # A2A webhook (Task or TaskStatusUpdateEvent)
            return await self._handle_a2a_webhook(
                payload,
                task_type,
                operation_id,
                preserve_legacy_identity,
            )
        else:
            # MCP webhook (dict payload)
            return await self._handle_mcp_webhook(
                payload,
                task_type,
                operation_id,
                signature,
                timestamp,
                raw_body,
                preserve_legacy_identity,
            )

class ADCPConnectionError(ADCPError):
    """Connection to agent failed."""

    def __init__(self, message: str, agent_id: str | None = None, agent_uri: str | None = None):
        """Initialize connection error."""
        suggestion = (
            "Check that the agent URI is correct and the agent is running.\n"
            "     Try testing with: python -m adcp test --config <agent-id>"
        )
        super().__init__(message, agent_id, agent_uri, suggestion)

    @property
    def is_retryable(self) -> bool:
        return True

class ADCPError(Exception):
    """Base exception for all AdCP client errors."""

    def __init__(
        self,
        message: str,
        agent_id: str | None = None,
        agent_uri: str | None = None,
        suggestion: str | None = None,
    ):
        """Initialize exception with context."""
        self.message = message
        self.agent_id = agent_id
        self.agent_uri = agent_uri
        self.suggestion = suggestion

        full_message = message
        if agent_id:
            full_message = f"[Agent: {agent_id}] {full_message}"
        if agent_uri:
            full_message = f"{full_message}\n  URI: {agent_uri}"
        if suggestion:
            full_message = f"{full_message}\n  Suggestion: {suggestion}"

        super().__init__(full_message)

    @property
    def is_retryable(self) -> bool:
        """Whether this error is safe to retry."""
        return False

class ADCPFeatureUnsupportedError(ADCPError):
    """Seller does not support one or more required features."""

    def __init__(
        self,
        unsupported_features: list[str],
        declared_features: list[str] | None = None,
        agent_id: str | None = None,
        agent_uri: str | None = None,
    ):
        """Initialize feature unsupported error.

        Args:
            unsupported_features: Features that are not supported.
            declared_features: Features the seller does declare.
            agent_id: Optional agent ID for context.
            agent_uri: Optional agent URI for context.
        """
        self.unsupported_features = unsupported_features
        self.declared_features = declared_features or []

        missing = ", ".join(unsupported_features)
        message = f"Seller does not support: {missing}"

        suggestion = None
        if self.declared_features:
            declared = ", ".join(sorted(self.declared_features))
            suggestion = f"Declared features: {declared}"

        super().__init__(message, agent_id, agent_uri, suggestion)

class ADCPMultiAgentClient:
    """Client for managing multiple AdCP agents."""

    def __init__(
        self,
        agents: list[AgentConfig],
        webhook_url_template: str | None = None,
        webhook_secret: str | None = None,
        on_activity: Callable[[Activity], None] | None = None,
        handlers: dict[str, Callable[..., Any]] | None = None,
        signing: SigningConfig | None = None,
        adcp_version: str | dict[str, str] | None = None,
        legacy_format_converter: LegacyFormatConverter | None = None,
        canonical_format_legacy_resolver: CanonicalFormatLegacyResolver | None = None,
        allow_unauthenticated_webhooks: bool | Mapping[str, bool] = False,
    ):
        """
        Initialize multi-agent client.

        Args:
            agents: List of agent configurations
            webhook_url_template: Template for webhook URLs
            webhook_secret: Shared secret for the deprecated HMAC-SHA256 webhook
                fallback. Configure only for registrations that explicitly
                selected legacy HMAC; use ``WebhookReceiver`` for RFC 9421.
            on_activity: Callback for activity events
            handlers: Task completion handlers
            signing: Optional RFC 9421 signing config forwarded to every
                per-agent ADCPClient. The same identity signs traffic to
                all agents. See ADCPClient.__init__ for details.
            allow_unauthenticated_webhooks: Explicit compatibility escape.
                A mapping scopes the opt-in by agent ID; omitted IDs remain
                protected. A uniform True is accepted only for a single-agent
                collection. Defaults to False.
            adcp_version: AdCP protocol release pin. Three forms:

                - ``None`` (default): every per-agent ADCPClient resolves
                  the SDK's compile-time pin.
                - ``str`` (e.g. ``"3.1"``): every agent uses this pin.
                - ``dict[str, str]`` (e.g.
                  ``{"seller_a": "3.0", "seller_b": "3.1"}``): per-agent
                  override map keyed by ``agent.id``. Agents missing
                  from the map fall back to the SDK default — useful
                  for holdco/multi-tenant operators where one seller is
                  ahead of the others on the upgrade cadence.

                See ADCPClient.__init__ for per-instance semantics.
                Cross-major pins raise ConfigurationError at construction.
        """
        agent_ids = {agent.id for agent in agents}
        if isinstance(allow_unauthenticated_webhooks, Mapping):
            unknown_ids = set(allow_unauthenticated_webhooks) - agent_ids
            if unknown_ids:
                unknown = ", ".join(sorted(unknown_ids))
                raise ValueError(
                    "allow_unauthenticated_webhooks contains unknown agent IDs: " + unknown
                )
            if any(type(value) is not bool for value in allow_unauthenticated_webhooks.values()):
                raise TypeError("allow_unauthenticated_webhooks mapping values must be bools")
            per_agent_unauthenticated = dict(allow_unauthenticated_webhooks)
        else:
            if type(allow_unauthenticated_webhooks) is not bool:
                raise TypeError(
                    "allow_unauthenticated_webhooks must be a bool or mapping of agent IDs to bools"
                )
            if allow_unauthenticated_webhooks is True and len(agents) > 1:
                raise ValueError(
                    "allow_unauthenticated_webhooks=True cannot be applied to multiple agents; "
                    "pass a mapping keyed by the isolated agent IDs"
                )
            per_agent_unauthenticated = {
                agent.id: allow_unauthenticated_webhooks for agent in agents
            }

        # Per-agent map → resolve each pin individually for the dict form;
        # otherwise use the uniform pin for all agents.
        if isinstance(adcp_version, dict):
            self._adcp_version: str | None = None  # mixed pins
            self._per_agent_versions: dict[str, str] = {
                agent_id: resolve_adcp_version(pin) for agent_id, pin in adcp_version.items()
            }
            default_pin = resolve_adcp_version(None)
            self.agents = {
                agent.id: ADCPClient(
                    agent,
                    webhook_url_template=webhook_url_template,
                    webhook_secret=webhook_secret,
                    on_activity=on_activity,
                    signing=signing,
                    adcp_version=self._per_agent_versions.get(agent.id, default_pin),
                    legacy_format_converter=legacy_format_converter,
                    canonical_format_legacy_resolver=canonical_format_legacy_resolver,
                    allow_unauthenticated_webhooks=per_agent_unauthenticated.get(agent.id, False),
                )
                for agent in agents
            }
        else:
            self._adcp_version = resolve_adcp_version(adcp_version)
            self._per_agent_versions = {}
            self.agents = {
                agent.id: ADCPClient(
                    agent,
                    webhook_url_template=webhook_url_template,
                    webhook_secret=webhook_secret,
                    on_activity=on_activity,
                    signing=signing,
                    adcp_version=self._adcp_version,
                    legacy_format_converter=legacy_format_converter,
                    canonical_format_legacy_resolver=canonical_format_legacy_resolver,
                    allow_unauthenticated_webhooks=per_agent_unauthenticated.get(agent.id, False),
                )
                for agent in agents
            }
        self.handlers = handlers or {}

    def get_adcp_version(self) -> str:
        """Return the AdCP protocol release pin for this multi-client.

        Returns the uniform pin when all agents share one. Raises
        :class:`ValueError` when agents have heterogeneous pins (the
        ``dict[str, str]`` constructor form) — in that case, query
        the per-agent pin via ``multi.agent(agent_id).get_adcp_version()``.
        """
        if self._adcp_version is not None:
            return self._adcp_version
        # Heterogeneous: surface uniformly if all agents agree at runtime.
        versions = {client.get_adcp_version() for client in self.agents.values()}
        if len(versions) == 1:
            return next(iter(versions))
        raise ValueError(
            "Multi-agent client has heterogeneous adcp_version pins; "
            "use multi.agent(agent_id).get_adcp_version() to read per-agent. "
            f"Pins by agent: { {a: c.get_adcp_version() for a, c in self.agents.items()} }"
        )

    def agent(self, agent_id: str) -> ADCPClient:
        """Get client for specific agent."""
        if agent_id not in self.agents:
            raise ValueError(f"Agent not found: {agent_id}")
        return self.agents[agent_id]

    @property
    def agent_ids(self) -> list[str]:
        """Get list of agent IDs."""
        return list(self.agents.keys())

    async def close(self) -> None:
        """Close all agent clients and clean up resources."""
        import asyncio

        logger.debug("Closing all agent clients in multi-agent client")
        close_tasks = [client.close() for client in self.agents.values()]
        await asyncio.gather(*close_tasks, return_exceptions=True)

    async def __aenter__(self) -> ADCPMultiAgentClient:
        """Async context manager entry."""
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        """Async context manager exit."""
        await self.close()

    async def get_products(
        self,
        request: GetProductsRequest,
    ) -> list[TaskResult[GetProductsResponse]]:
        """
        Execute get_products across all agents in parallel.

        Args:
            request: Request parameters

        Returns:
            List of TaskResults containing GetProductsResponse for each agent
        """
        import asyncio

        tasks = [agent.get_products(request) for agent in self.agents.values()]
        return await asyncio.gather(*tasks)

    async def list_products(
        self, request: ListProductsRequest
    ) -> list[TaskResult[ListProductsResponse]]:
        """Execute compact product discovery across all agents."""
        import asyncio

        return await asyncio.gather(
            *(agent.list_products(request) for agent in self.agents.values())
        )

    async def request_proposals(
        self, request: RequestProposalsRequest
    ) -> list[TaskResult[RequestProposalsResponse]]:
        """Request proposals from all agents."""
        import asyncio

        return await asyncio.gather(
            *(agent.request_proposals(request) for agent in self.agents.values())
        )

    async def refine_proposals(
        self, request: RefineProposalsRequest
    ) -> list[TaskResult[RefineProposalsResponse]]:
        """Refine proposals across all agents."""
        import asyncio

        return await asyncio.gather(
            *(agent.refine_proposals(request) for agent in self.agents.values())
        )

    async def decline_proposals(
        self, request: DeclineProposalsRequest
    ) -> list[TaskResult[DeclineProposalsResponse]]:
        """Decline proposals across all agents."""
        import asyncio

        return await asyncio.gather(
            *(agent.decline_proposals(request) for agent in self.agents.values())
        )

    async def buy_products(
        self, request: BuyProductsRequest
    ) -> list[TaskResult[BuyProductsResponse]]:
        """Commit direct purchases across all agents."""
        import asyncio

        return await asyncio.gather(
            *(agent.buy_products(request) for agent in self.agents.values())
        )

    async def accept_proposal(
        self, request: AcceptProposalRequest
    ) -> list[TaskResult[AcceptProposalResponse]]:
        """Accept proposals across all agents."""
        import asyncio

        return await asyncio.gather(
            *(agent.accept_proposal(request) for agent in self.agents.values())
        )

    async def control_media_buy(
        self, request: ControlMediaBuyRequest
    ) -> list[TaskResult[ControlMediaBuyResponse]]:
        """Control media buys across all agents."""
        import asyncio

        return await asyncio.gather(
            *(agent.control_media_buy(request) for agent in self.agents.values())
        )

    async def get_products_legacy(
        self, request: LegacyGetProductsRequest
    ) -> list[TaskResult[LegacyGetProductsResponse]]:
        """Execute the explicit raw discovery adapter across all agents."""

        import asyncio

        return await asyncio.gather(
            *(agent.get_products_legacy(request) for agent in self.agents.values())
        )

    async def execute_task(self, task_name: str, request: BaseModel) -> list[TaskResult[Any]]:
        """Execute a canonical task across all configured agents."""

        import asyncio

        return await asyncio.gather(
            *(agent.execute_task(task_name, request) for agent in self.agents.values())
        )

    async def execute_task_legacy(
        self, task_name: str, request: BaseModel
    ) -> list[TaskResult[Any]]:
        """Execute an explicit raw task adapter across all configured agents."""

        import asyncio

        return await asyncio.gather(
            *(agent.execute_task_legacy(task_name, request) for agent in self.agents.values())
        )

    @classmethod
    def from_env(cls) -> ADCPMultiAgentClient:
        """Create client from environment variables."""
        agents_json = os.getenv("ADCP_AGENTS")
        if not agents_json:
            raise ValueError("ADCP_AGENTS environment variable not set")

        agents_data = json.loads(agents_json)
        agents = [AgentConfig(**agent) for agent in agents_data]

        return cls(
            agents=agents,
            webhook_url_template=os.getenv("WEBHOOK_URL_TEMPLATE"),
            webhook_secret=os.getenv("WEBHOOK_SECRET"),
        )

class ADCPProtocolError(ADCPError):
    """Protocol-level error (malformed response, unexpected format)."""

    def __init__(self, message: str, agent_id: str | None = None, protocol: str | None = None):
        """Initialize protocol error."""
        suggestion = (
            f"The agent returned an unexpected {protocol} response format."
            if protocol
            else "Unexpected response format."
        )
        suggestion += "\n     Enable debug mode to see the full request/response."
        super().__init__(message, agent_id, None, suggestion)

class ADCPSigningRequiredError(ADCPError):
    """Raised when an operation in the seller's ``request_signing.required_for``
    is called without a ``SigningConfig`` on the client.

    Signing a ``required_for`` operation is mandatory — sending it unsigned
    would produce a ``request_signature_required`` rejection from the seller.
    Raising locally before the wire call saves a round-trip and gives the
    caller a clear, actionable error.
    """

    def __init__(
        self,
        operation: str,
        agent_id: str | None = None,
        agent_uri: str | None = None,
    ):
        self.operation = operation
        message = (
            f"Operation {operation!r} is in the seller's request_signing.required_for "
            f"list; signing is mandatory but no SigningConfig was provided"
        )
        suggestion = (
            "Pass signing=SigningConfig(private_key=..., key_id=...) when "
            "constructing ADCPClient. See adcp-keygen for key generation."
        )
        super().__init__(message, agent_id, agent_uri, suggestion)

class ADCPTaskError(ADCPError):
    """A task returned an ADCP error response.

    Provides structured access to the error objects from the response,
    including error codes for programmatic handling.
    """

    def __init__(
        self,
        operation: str,
        errors: list[Any],
        agent_id: str | None = None,
    ):
        """Initialize task error.

        Args:
            operation: The task that failed (e.g., "create_media_buy")
            errors: List of ADCP Error objects from the response
            agent_id: Optional agent ID for context
        """
        self.operation = operation
        self.errors = errors
        self.error_codes = [e.code for e in errors if hasattr(e, "code") and e.code]

        message = f"{operation} failed"
        if errors:
            first_msg = getattr(errors[0], "message", str(errors[0]))
            message = f"{operation} failed: {first_msg}"
            if len(errors) > 1:
                message += f" (+{len(errors) - 1} more)"

        super().__init__(message, agent_id=agent_id)

    @property
    def is_retryable(self) -> bool:
        """True if any error code is transient (RATE_LIMITED, etc.)."""
        from adcp.server.helpers import TRANSIENT_CODES

        return bool(TRANSIENT_CODES & set(self.error_codes))

class ADCPTimeoutError(ADCPError):
    """Request timed out."""

    def __init__(
        self,
        message: str,
        agent_id: str | None = None,
        agent_uri: str | None = None,
        timeout: float | None = None,
    ):
        """Initialize timeout error."""
        suggestion = (
            f"The request took longer than {timeout}s." if timeout else "The request timed out."
        )
        suggestion += "\n     Try increasing the timeout value or check if the agent is overloaded."
        super().__init__(message, agent_id, agent_uri, suggestion)

    @property
    def is_retryable(self) -> bool:
        return True

class ADCPToolNotFoundError(ADCPError):
    """Requested tool not found on agent."""

    def __init__(
        self, tool_name: str, agent_id: str | None = None, available_tools: list[str] | None = None
    ):
        """Initialize tool not found error."""
        message = f"Tool '{tool_name}' not found on agent"
        suggestion = "List available tools with: python -m adcp list-tools --config <agent-id>"
        if available_tools:
            tools_list = ", ".join(available_tools[:5])
            if len(available_tools) > 5:
                tools_list += f", ... ({len(available_tools)} total)"
            suggestion = f"Available tools: {tools_list}"
        super().__init__(message, agent_id, None, suggestion)

class ADCPWebhookError(ADCPError):
    """Webhook handling error."""

class ADCPWebhookSignatureError(ADCPWebhookError):
    """Webhook signature verification failed."""

    def __init__(self, message: str = "Invalid webhook signature", agent_id: str | None = None):
        """Initialize webhook signature error."""
        suggestion = (
            "Verify that the webhook_secret matches the secret configured on the agent.\n"
            "     Webhook signatures use HMAC-SHA256 for security."
        )
        super().__init__(message, agent_id, None, suggestion)

class AcceptProposalRequest(AdCPBaseModel):
    model_config = ConfigDict(
        extra='forbid',
    )
    adcp_version: version_envelope.AdcpVersion | None = None
    adcp_major_version: version_envelope.AdcpMajorVersion | None = None
    idempotency_key: Annotated[
        str, Field(max_length=255, min_length=16, pattern='^[A-Za-z0-9_.:-]{16,255}$')
    ]
    account: canonical_account_ref.CanonicalAccountReference
    proposal_id: Annotated[str, Field(min_length=1)]
    proposal_terms_digest: Annotated[
        str,
        Field(
            description='terms_digest from the committed proposal. The seller MUST atomically verify both ID and digest before acceptance.',
            pattern='^sha256:[A-Za-z0-9_-]{43}$',
        ),
    ]
    total_budget: Annotated[
        TotalBudget | None,
        Field(
            description='Execution amount when the committed proposal defines scalable percentages or constraints rather than a fixed total.'
        ),
    ] = None
    daily_budget_cap: Annotated[
        float | None,
        Field(
            description="Optional hard aggregate daily spend ceiling applied when the committed proposal is accepted. It constrains execution without changing the proposal's negotiated pricing.",
            ge=0.0,
        ),
    ] = None
    budget_cap_timezone: Annotated[
        str | None,
        Field(
            description='Optional shared IANA cap-day timezone override. Requires buyer_timezone_override support. When omitted, budget_capping.timezone_basis selects Account.timezone or fixed_timezone.',
            min_length=1,
        ),
    ] = None
    io_acceptance: IoAcceptance | None = None
    purchase_order_ref: Annotated[str | None, Field(max_length=255, min_length=1)] = None
    governance_context: Annotated[str | None, Field(max_length=4096, min_length=1)] = None
    push_notification_config: push_notification_config_1.PushNotificationConfig | None = None
    reporting_webhook: Annotated[
        reporting_webhook_1.ReportingWebhook | None,
        Field(
            description='Optional reporting delivery configuration established atomically when the proposal is accepted. This is execution metadata and does not alter the accepted commercial terms digest.'
        ),
    ] = None
    opportunity: Annotated[
        Opportunity | None,
        Field(
            description='Optional planning-cycle closure. Success infers closed with accepted_with_seller when status is omitted. If the proposal carries opportunity_id, a supplied ID MUST match; the accepted proposal preserves that association.'
        ),
    ] = None
    context: context_1.ContextObject | None = None
    ext: ext_1.ExtensionObject | None = None

class AcceptedLoss(StrEnum):
    feed_version_not_atomic = 'feed_version_not_atomic'
    pricing_version_not_atomic = 'pricing_version_not_atomic'
    mutation_idempotency_not_guaranteed = 'mutation_idempotency_not_guaranteed'

class AccountAuthorization(AdCPBaseModel):
    model_config = ConfigDict(
        extra='allow',
    )
    allowed_tasks: Annotated[
        list[AllowedTask],
        Field(
            description='Canonical snake_case task names the caller may invoke against this account (for example get_media_buys, buy_products, accept_proposal, control_media_buy, or sync_creatives). Absence of a task from this list means not permitted and returns SCOPE_INSUFFICIENT. Compact 3.2 tools are authorized by their own names; grants for deprecated get_products/create_media_buy/update_media_buy do not silently transfer across aliases.'
        ),
    ]
    field_scopes: Annotated[
        dict[str, list[str]] | None,
        Field(
            description='Optional per-task allowlist of request fields the caller may set. Keys are task names (which MUST also appear in allowed_tasks). Values are arrays of top-level request-field paths permitted for that task. When a task appears in field_scopes, requests to that task with any field outside the allowlist MUST be rejected with FIELD_NOT_PERMITTED. Compact product tools use their own task names and top-level fields such as criteria and refinements. Implicit framing fields are always permitted and do NOT need to appear in the allowlist — they identify the resource or shape the call rather than mutating business state. Tasks absent from field_scopes have no field-level restriction beyond what the task schema already enforces.'
        ),
    ] = None
    scope_name: Annotated[
        Literal['attestation_verifier'] | ScopeName | None,
        Field(
            description='Optional named scope identifier. When present, callers and the vendor agent can reason about the grant by name rather than by enumerating allowed_tasks and field_scopes. Modeled as a discriminated union so code generators produce a literal type for the standard scope(s) and a distinct type for agent-defined values — this prevents a typo of `attestation_verifier` from being silently accepted as a custom scope. Agent-defined scope names MUST use a `custom:` prefix to avoid collision with future standard scopes. The prefix is protocol-neutral: a signals agent, a governance agent, or a creative agent defines custom scopes the same way a media-buy sales agent does.'
        ),
    ] = None
    read_only: Annotated[
        bool | None,
        Field(
            description='Convenience flag. When true, the caller is permitted only non-mutating tasks. Sellers MUST reject any mutation from a read-only caller with READ_ONLY_SCOPE. For the AdCP 3.2 product split, read-only permits list_products but rejects request_proposals, refine_proposals, and decline_proposals; a legacy get_products call is permitted only when the seller can guarantee the selected arm is a synchronous side-effect-free read. Sellers MAY omit this field; omission is equivalent to `false`. Callers MUST NOT infer read-only from `allowed_tasks` alone — the seller MUST set this explicitly when it applies.'
        ),
    ] = False

class AccountReference(RootModel[AccountReference1 | AccountReference2]):
    root: Annotated[
        AccountReference1 | AccountReference2,
        Field(
            description="Reference to an advertiser account by seller-assigned ID or natural key. account_id is the seller/storefront handle returned by list_accounts. The natural key is brand + operator + optional operator_unit.id + optional fixed currency + optional buyer-selected account timezone + sandbox; it describes the advertiser object the seller resolves or provisions. After a capability-gated identity rekey, the former natural key is tombstoned and returns ACCOUNT_MOVED to authorized callers rather than resolving or provisioning another account. brand.countries qualifies the advertiser's geographic identity. operator_unit identifies the operator-owned buying seat or business unit and is distinct from the seller account. operator_unit.name is display metadata and does not participate in identity. Use account_id when the seller or upstream platform owns the canonical account namespace; use the natural key for buyer-declared accounts (require_operator_auth: false).",
            examples=[
                {'account_id': 'acc_acme_001'},
                {'brand': {'domain': 'acme-corp.com'}, 'operator': 'acme-corp.com'},
                {
                    'brand': {'domain': 'nova-brands.com', 'brand_id': 'spark'},
                    'operator': 'pinnacle-media.com',
                },
                {
                    'brand': {'domain': 'nova-athletics.example', 'countries': ['NL']},
                    'operator': 'nova-athletics.example',
                    'operator_unit': {'id': '234284238', 'name': 'Nova EMEA'},
                    'currency': 'EUR',
                    'timezone': 'Europe/Amsterdam',
                },
                {
                    'brand': {'domain': 'acme-corp.com'},
                    'operator': 'acme-corp.com',
                    'sandbox': True,
                },
            ],
            title='Account Reference',
        ),
    ]
    def __getattr__(self, name: str) -> Any:
        """Proxy attribute access to the wrapped type."""
        if name.startswith('_'):
            raise AttributeError(name)
        return getattr(self.root, name)

class AccountReference1(AdCPBaseModel):
    model_config = ConfigDict(
        extra='forbid',
    )
    account_id: Annotated[
        str,
        Field(
            description="Seller-assigned account identifier. For upstream-managed account namespaces, this value comes from list_accounts; for seller-defined namespaces without a list_accounts surface, it is supplied out-of-band. Buyer-declared account sellers MAY echo account_id from sync_accounts as an internal handle, but they MUST continue accepting the account's current natural-key AccountRef on subsequent calls. A former key tombstoned by identity reconciliation returns ACCOUNT_MOVED to authorized callers."
        ),
    ]

class AccountReference2(AdCPBaseModel):
    model_config = ConfigDict(
        extra='forbid',
    )
    brand: Annotated[
        brand_ref.BrandReference, Field(description='Brand reference identifying the advertiser')
    ]
    operator: Annotated[
        str,
        Field(
            description="Domain of the entity operating on the brand's behalf. When the brand operates directly, this is the brand's domain.",
            pattern='^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)*$',
        ),
    ]
    operator_unit: Annotated[
        operator_unit_1.OperatorUnit | None,
        Field(
            description='Optional operator-owned business unit, agency seat, or platform account. Only id participates in the natural account key; name is mutable display metadata.'
        ),
    ] = None
    currency: Annotated[
        str | None,
        Field(
            description="Immutable ISO 4217 transaction currency when the seller's advertiser object is currency-bound. When present, this is part of the natural account key and media buys on the account MUST use it. Omit when currency is selected independently per media buy.",
            pattern='^[A-Z]{3}$',
        ),
    ] = None
    timezone: Annotated[
        str | None,
        Field(
            description='Immutable account timezone. Include it in the natural key when get_adcp_capabilities.account.timezone declares account_fixed with buyer_selected; omit it for seller_fixed or seller_assigned accounts.',
            min_length=1,
        ),
    ] = None
    sandbox: Annotated[
        bool | None,
        Field(
            description='When true, references the sandbox account for this brand/operator pair. Defaults to false (production account).'
        ),
    ] = False

class AccountScope(StrEnum):
    operator = 'operator'
    brand = 'brand'
    operator_brand = 'operator_brand'
    agent = 'agent'

class AccountWithAuthorization(Account):
    authorization: Annotated[
        account_authorization.AccountAuthorization | None,
        Field(
            description="Optional. The caller's scope grant against this account. Vendor agents of any type (media-buy, signals, governance, creative, brand) that support scope introspection SHOULD populate this so callers can preempt SCOPE_INSUFFICIENT / FIELD_NOT_PERMITTED errors rather than discovering scope by trial and error. Media-buy sales agents claiming the `attestation_verifier` standard scope MUST populate it. Absence means the vendor agent does not advertise introspectable scope for this account — callers MUST NOT infer access from absence, and fall back to error-driven discovery via the RBAC error codes."
        ),
    ] = None

class AcquireRightsRequest(AdcpVersionEnvelope):
    model_config = ConfigDict(
        extra='allow',
    )
    governance_context: Annotated[
        str | None,
        Field(
            description='Opaque intent authorization for this rights commitment. Required when governance applies to the resolved account.',
            max_length=4096,
            min_length=1,
            pattern='^[\\x20-\\x7E]+$',
        ),
    ] = None
    rights_id: Annotated[
        str, Field(description='Rights offering identifier from get_rights response')
    ]
    pricing_option_id: Annotated[
        str, Field(description='Selected pricing option from the rights offering')
    ]
    buyer: Annotated[brand_ref.BrandReference, Field(description="The buyer's brand identity")]
    account: Annotated[
        account_ref.AccountReference | None,
        Field(
            description='Account context for this acquisition. Used by the brand agent to resolve any governance agent previously bound for this brand+operator pair via sync_governance. When both an inline governance_context token and a bound governance agent are present, the token is verified against that resolved relationship. An agent advertising adcp.governance_enforcement for acquire_rights MUST resolve an account and returns ACCOUNT_REQUIRED when neither the request nor an existing resource supplies one; it MUST NOT infer that a missing token means ungoverned. Legacy non-claiming agents may continue to treat omission of both fields as ungoverned during 3.x. Pass a natural key (brand, operator, optional sandbox) or a seller-assigned account_id from list_accounts.'
        ),
    ] = None
    campaign: Annotated[Campaign, Field(description='Campaign details for rights clearance')]
    revocation_webhook: Annotated[
        push_notification_config_1.PushNotificationConfig,
        Field(
            description='Webhook for rights revocation notifications. If the rights holder needs to revoke rights (talent scandal, contract violation, etc.), they POST a revocation-notification to this URL. The buyer is responsible for stopping creative delivery upon receipt.'
        ),
    ]
    push_notification_config: Annotated[
        push_notification_config_1.PushNotificationConfig | None,
        Field(
            description='Webhook for async status updates if the acquisition requires approval. The rights agent sends a webhook notification when the status transitions to acquired or rejected.'
        ),
    ] = None
    idempotency_key: Annotated[
        str,
        Field(
            description='Client-generated key for safe retries. Resubmitting with the same key returns the original response rather than creating a duplicate acquisition. MUST be unique per (seller, request) pair to prevent cross-seller correlation. Use a fresh UUID v4 for each request.',
            max_length=255,
            min_length=16,
            pattern='^[A-Za-z0-9_.:-]{16,255}$',
        ),
    ]
    context: context_1.ContextObject | None = None
    ext: ext_1.ExtensionObject | None = None

class AcquireRightsResponse1(AdcpVersionEnvelope):
    model_config = ConfigDict(extra='allow')
    rights_id: str
    rights_status: Literal['acquired'] = 'acquired'
    brand_id: str
    terms: rights_terms_1.RightsTerms
    generation_credentials: list[generation_credential_1.GenerationCredential]
    restrictions: list[str] | None = None
    disclosure: Disclosure | None = None
    approval_webhook: push_notification_config_1.PushNotificationConfig | None = None
    usage_reporting_url: AnyUrl | None = None
    rights_constraint: Any
    context: context_1.ContextObject | None = None
    ext: ext_1.ExtensionObject | None = None

class AcquireRightsResponse1(AdcpVersionEnvelope):
    model_config = ConfigDict(extra='allow')
    rights_id: str
    rights_status: Literal['acquired'] = 'acquired'
    brand_id: str
    terms: rights_terms_1.RightsTerms
    generation_credentials: list[generation_credential_1.GenerationCredential]
    restrictions: list[str] | None = None
    disclosure: Disclosure | None = None
    approval_webhook: push_notification_config_1.PushNotificationConfig | None = None
    usage_reporting_url: AnyUrl | None = None
    rights_constraint: Any
    context: context_1.ContextObject | None = None
    ext: ext_1.ExtensionObject | None = None

class AcquireRightsResponse2(AdcpVersionEnvelope):
    model_config = ConfigDict(extra='allow')
    rights_id: str
    rights_status: Literal['pending_approval'] = 'pending_approval'
    brand_id: str
    detail: str | None = None
    estimated_response_time: str | None = None
    context: context_1.ContextObject | None = None
    ext: ext_1.ExtensionObject | None = None

class AcquireRightsResponse3(AdcpVersionEnvelope):
    model_config = ConfigDict(extra='allow')
    rights_id: str
    rights_status: Literal['rejected'] = 'rejected'
    brand_id: str
    reason: str
    suggestions: list[str] | None = None
    context: context_1.ContextObject | None = None
    ext: ext_1.ExtensionObject | None = None

class AcquireRightsResponse4(AdcpVersionEnvelope):
    model_config = ConfigDict(extra='allow')
    errors: Annotated[list[error_1.Error], Field(min_length=1)]
    context: context_1.ContextObject | None = None
    ext: ext_1.ExtensionObject | None = None

class ActivateSignalRequest(AdcpVersionEnvelope):
    model_config = ConfigDict(
        extra='allow',
    )
    action: Annotated[
        Action | None,
        Field(
            description="Whether to activate or deactivate the signal. Deactivating removes the segment from downstream platforms, required when campaigns end to comply with data governance policies (GDPR, CCPA). Defaults to 'activate' when omitted."
        ),
    ] = Action.activate
    signal_agent_segment_id: Annotated[
        str,
        Field(
            description='Opaque activation handle returned in the signal_agent_segment_id field of each get_signals response entry. Pass this string verbatim — do not pass the signal_id object.'
        ),
    ]
    destinations: Annotated[
        list[destination.Destination],
        Field(
            description='Target destination(s) for activation. If the authenticated caller matches one of these destinations, activation keys will be included in the response.',
            min_length=1,
        ),
    ]
    pricing_option_id: Annotated[
        str | None,
        Field(
            description="The pricing option selected from the signal's pricing_options in the get_signals response. Required when the signal has pricing options. Records the buyer's pricing commitment at activation time; pass this same value in report_usage for billing verification."
        ),
    ] = None
    governance_context: Annotated[
        str | None,
        Field(
            description='Opaque authorization context returned by an approved check_governance decision for this signal activation. Required when the account has a registered governance agent; signal agents MUST reject governed activations that omit a valid context. Conditions and denied decisions never produce this value.',
            max_length=4096,
            min_length=1,
            pattern='^[\\x20-\\x7E]+$',
        ),
    ] = None
    account: Annotated[
        account_ref.AccountReference | None,
        Field(
            description='Account for this activation. Associates with a commercial relationship established via sync_accounts.'
        ),
    ] = None
    idempotency_key: Annotated[
        str,
        Field(
            description='Client-generated unique key for this request. Prevents duplicate activations on retries. MUST be unique per (seller, request) pair to prevent cross-seller correlation. Use a fresh UUID v4 for each request.',
            max_length=255,
            min_length=16,
            pattern='^[A-Za-z0-9_.:-]{16,255}$',
        ),
    ]
    context: context_1.ContextObject | None = None
    ext: ext_1.ExtensionObject | None = None

class ActivateSignalResponse1(AdcpVersionEnvelope):
    model_config = ConfigDict(extra='allow')
    deployments: list[deployment_1.Deployment]
    sandbox: bool | None = None
    context: context_1.ContextObject | None = None
    ext: ext_1.ExtensionObject | None = None

class ActivateSignalResponse1(AdcpVersionEnvelope):
    model_config = ConfigDict(extra='allow')
    deployments: list[deployment_1.Deployment]
    sandbox: bool | None = None
    context: context_1.ContextObject | None = None
    ext: ext_1.ExtensionObject | None = None

class ActivateSignalResponse2(AdcpVersionEnvelope):
    model_config = ConfigDict(extra='allow')
    errors: Annotated[list[error_1.Error], Field(min_length=1)]
    context: context_1.ContextObject | None = None
    ext: ext_1.ExtensionObject | None = None

class ActivationKey1(AdCPBaseModel):
    model_config = ConfigDict(
        extra='allow',
    )
    type: Annotated[Literal['segment_id'], Field(description='Segment ID based targeting')] = 'segment_id'
    segment_id: Annotated[
        str,
        Field(description='The platform-specific segment identifier to use in campaign targeting'),
    ]

class ActivationKey1(AdCPBaseModel):
    model_config = ConfigDict(
        extra='allow',
    )
    type: Annotated[Literal['segment_id'], Field(description='Segment ID based targeting')] = 'segment_id'
    segment_id: Annotated[
        str,
        Field(description='The platform-specific segment identifier to use in campaign targeting'),
    ]

class ActivationKey2(AdCPBaseModel):
    model_config = ConfigDict(
        extra='allow',
    )
    type: Annotated[Literal['key_value'], Field(description='Key-value pair based targeting')] = 'key_value'
    key: Annotated[str, Field(description='The targeting parameter key')]
    value: Annotated[str, Field(description='The targeting parameter value')]

class ActivationKey2(AdCPBaseModel):
    model_config = ConfigDict(
        extra='allow',
    )
    type: Annotated[Literal['key_value'], Field(description='Key-value pair based targeting')] = 'key_value'
    key: Annotated[str, Field(description='The targeting parameter key')]
    value: Annotated[str, Field(description='The targeting parameter value')]

@dataclass
class AdAgentsValidationResult:
    """Result of discovering and validating a publisher's adagents.json.

    ``discovery_method`` records which path produced ``data``:
    ``direct`` for ``/.well-known/adagents.json`` on the publisher,
    ``authoritative_location`` for a URL-reference redirect, and
    ``ads_txt_managerdomain`` for the one-hop ads.txt MANAGERDOMAIN
    fallback (RFC 4175). ``manager_domain`` is set only on the
    managerdomain path.
    """

    domain: str
    url: str
    discovery_method: DiscoveryMethod = "direct"
    manager_domain: str | None = None
    data: dict[str, Any] | None = None
    valid: bool = False
    errors: list[str] = field(default_factory=list)

class AdagentsAccessBlockedError(AdagentsValidationError):
    """adagents.json fetch blocked by publisher-side bot management (403, cf-mitigated: challenge).

    Only surfaces in direct-fetch workflows (``fetch_adagents``). SDK callers that
    use ``fetch_agent_authorizations`` avoid this entirely — the AAO directory crawler
    handles publisher fetches and serves cached results without exposing the SDK to
    publisher-side bot management.

    If you need to catch this specifically without catching all
    ``AdagentsValidationError``s, use ``except AdagentsAccessBlockedError``.
    """

    def __init__(self, publisher_domain: str):
        """Initialize bot-management blocked error."""
        self.publisher_domain = publisher_domain
        message = (
            f"adagents.json blocked by bot management for {publisher_domain} "
            f"(HTTP 403, cf-mitigated: challenge)"
        )
        suggestion = (
            "The publisher's origin blocked this request with a Cloudflare bot management\n"
            "     challenge. This only affects direct adagents.json fetches (fetch_adagents).\n"
            "\n"
            "     To unblock local debugging:\n"
            "     - Retry with a browser-like User-Agent via the user_agent= parameter, e.g.\n"
            '       user_agent="Mozilla/5.0"\n'
            "     - Or call fetch_agent_authorizations() to query the AAO directory instead,\n"
            "       which bypasses publisher-side bot management entirely."
        )
        super().__init__(message, None, None, suggestion)

@dataclass(frozen=True)
class AdagentsCacheEntry:
    """Conditional-refresh cache state for an adagents.json URL.

    Pass an entry into :func:`fetch_adagents_with_cache` to send
    ``If-None-Match`` (preferred) and ``If-Modified-Since`` validators
    on the next fetch. A 304 from the publisher is treated as a
    successful cache-lifetime refresh — the ``body`` is returned
    unchanged with refreshed timing, per the adcp#4504 fetch contract.
    """

    body: dict[str, Any]
    etag: str | None = None
    last_modified: str | None = None

@dataclass(frozen=True)
class AdagentsEntryError:
    """A single schema violation found in an adagents.json file.

    ``kind`` is a stable string literal callers can branch on (e.g.,
    distinguish a publisher who shipped bare entries from one who picked
    an unknown authorization_type). ``message`` is developer-facing and
    its wording may change between releases — pattern-match on ``kind``
    when surfacing publisher-facing diagnostics.

    For file-level errors (e.g., an empty authorization and catalog file)
    ``index`` is ``-1`` and ``url`` is ``None``.
    """

    index: int
    kind: EntryErrorKind
    message: str
    url: str | None = None

@dataclass(frozen=True)
class AdagentsFetchResult:
    """Result of a fetch, including refreshed cache validators.

    ``not_modified`` is True when the server returned 304 and ``data``
    came from the supplied cache entry. ``etag`` / ``last_modified`` are
    the validators to persist for the next fetch — on 304 they come
    from the 304 response headers if present, falling back to the
    supplied entry's values.
    """

    data: dict[str, Any]
    discovery_method: DiscoveryMethod
    etag: str | None = None
    last_modified: str | None = None
    not_modified: bool = False

class AdagentsNotFoundError(AdagentsValidationError):
    """adagents.json file not found (404)."""

    def __init__(self, publisher_domain: str):
        """Initialize not found error."""
        message = f"adagents.json not found for domain: {publisher_domain}"
        suggestion = (
            "Verify that the publisher has deployed adagents.json to:\n"
            f"     https://{publisher_domain}/.well-known/adagents.json"
        )
        super().__init__(message, None, None, suggestion)

class AdagentsTimeoutError(AdagentsValidationError):
    """Request for adagents.json timed out."""

    def __init__(self, publisher_domain: str, timeout: float):
        """Initialize timeout error."""
        message = f"Request to fetch adagents.json timed out after {timeout}s"
        suggestion = (
            "The publisher's server may be slow or unresponsive.\n"
            "     Try increasing the timeout value or check the domain is correct."
        )
        super().__init__(message, None, None, suggestion)

class AdagentsValidationError(ADCPError):
    """Base error for adagents.json validation issues."""

@dataclass(frozen=True)
class AdagentsValidationReport:
    """Result of structurally validating a parsed adagents.json.

    Distinguishes the two failure modes that
    :func:`get_properties_by_agent` collapses into an empty list:
    a schema-invalid file (``schema_valid`` is False, ``errors`` populated)
    versus a valid file that simply doesn't list the caller's agent.

    ``authorized_agents_count`` and ``properties_count`` reflect the
    array lengths as observed in the input — they are reported regardless
    of ``schema_valid`` so callers can show "0 agents listed" diagnostics
    on partially-broken files.

    ``is_reference`` is True for the URL-reference variant of the schema
    (an ``authoritative_location`` pointer with no inline
    ``authorized_agents`` array). Callers that received a report with
    ``is_reference=True`` should follow the redirect (e.g., via
    :func:`fetch_adagents`) and validate the resolved file. This flag
    lets callers distinguish a legitimate URL-reference file from an inline
    catalog-only file. AdCP 3.2 permits ``authorized_agents: []`` when at least
    one of ``formats``, ``properties``, ``placements``, ``collections``, or
    ``signals`` is non-empty; that catalog content grants no sales authority.
    """

    schema_valid: bool
    errors: list[AdagentsEntryError]
    authorized_agents_count: int
    properties_count: int
    is_reference: bool = False

class AdvertiserIndustry(StrEnum):
    automotive = 'automotive'
    automotive_electric_vehicles = 'automotive.electric_vehicles'
    automotive_parts_accessories = 'automotive.parts_accessories'
    automotive_luxury = 'automotive.luxury'
    beauty_cosmetics = 'beauty_cosmetics'
    beauty_cosmetics_skincare = 'beauty_cosmetics.skincare'
    beauty_cosmetics_fragrance = 'beauty_cosmetics.fragrance'
    beauty_cosmetics_haircare = 'beauty_cosmetics.haircare'
    cannabis = 'cannabis'
    cpg = 'cpg'
    cpg_personal_care = 'cpg.personal_care'
    cpg_household = 'cpg.household'
    dating = 'dating'
    education = 'education'
    education_higher_education = 'education.higher_education'
    education_online_learning = 'education.online_learning'
    education_k12 = 'education.k12'
    energy_utilities = 'energy_utilities'
    energy_utilities_renewable = 'energy_utilities.renewable'
    fashion_apparel = 'fashion_apparel'
    fashion_apparel_luxury = 'fashion_apparel.luxury'
    fashion_apparel_sportswear = 'fashion_apparel.sportswear'
    finance = 'finance'
    finance_banking = 'finance.banking'
    finance_insurance = 'finance.insurance'
    finance_investment = 'finance.investment'
    finance_cryptocurrency = 'finance.cryptocurrency'
    food_beverage = 'food_beverage'
    food_beverage_alcohol = 'food_beverage.alcohol'
    food_beverage_restaurants = 'food_beverage.restaurants'
    food_beverage_packaged_goods = 'food_beverage.packaged_goods'
    gambling_betting = 'gambling_betting'
    gambling_betting_sports_betting = 'gambling_betting.sports_betting'
    gambling_betting_casino = 'gambling_betting.casino'
    gaming = 'gaming'
    gaming_mobile = 'gaming.mobile'
    gaming_console_pc = 'gaming.console_pc'
    gaming_esports = 'gaming.esports'
    government_nonprofit = 'government_nonprofit'
    government_nonprofit_political = 'government_nonprofit.political'
    government_nonprofit_charity = 'government_nonprofit.charity'
    healthcare = 'healthcare'
    healthcare_pharmaceutical = 'healthcare.pharmaceutical'
    healthcare_medical_devices = 'healthcare.medical_devices'
    healthcare_wellness = 'healthcare.wellness'
    home_garden = 'home_garden'
    home_garden_furniture = 'home_garden.furniture'
    home_garden_home_improvement = 'home_garden.home_improvement'
    media_entertainment = 'media_entertainment'
    media_entertainment_podcasts = 'media_entertainment.podcasts'
    media_entertainment_music = 'media_entertainment.music'
    media_entertainment_film_tv = 'media_entertainment.film_tv'
    media_entertainment_publishing = 'media_entertainment.publishing'
    media_entertainment_live_events = 'media_entertainment.live_events'
    pets = 'pets'
    professional_services = 'professional_services'
    professional_services_legal = 'professional_services.legal'
    professional_services_consulting = 'professional_services.consulting'
    real_estate = 'real_estate'
    real_estate_residential = 'real_estate.residential'
    real_estate_commercial = 'real_estate.commercial'
    recruitment_hr = 'recruitment_hr'
    retail = 'retail'
    retail_ecommerce = 'retail.ecommerce'
    retail_department_stores = 'retail.department_stores'
    sports_fitness = 'sports_fitness'
    sports_fitness_equipment = 'sports_fitness.equipment'
    sports_fitness_teams_leagues = 'sports_fitness.teams_leagues'
    technology = 'technology'
    technology_software = 'technology.software'
    technology_hardware = 'technology.hardware'
    technology_ai_ml = 'technology.ai_ml'
    telecom = 'telecom'
    telecom_mobile_carriers = 'telecom.mobile_carriers'
    telecom_internet_providers = 'telecom.internet_providers'
    transportation_logistics = 'transportation_logistics'
    travel_hospitality = 'travel_hospitality'
    travel_hospitality_airlines = 'travel_hospitality.airlines'
    travel_hospitality_hotels = 'travel_hospitality.hotels'
    travel_hospitality_cruise = 'travel_hospitality.cruise'
    travel_hospitality_tourism = 'travel_hospitality.tourism'