Module adcp.adagents

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

async def fetch_adagents(
    publisher_domain: str,
    timeout: float = 10.0,
    user_agent: str = "AdCP-Client/1.0",
    client: httpx.AsyncClient | None = None,
    *,
    validate_structure: bool = True,
    transport_factory: AdagentsTransportFactory | 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. IAB 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 ads.txt fetch follows up to five HTTPS redirects within a single timeout,
    permitting same-root redirects followed by one terminal off-root hop. Manager
    delegation remains 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 caller-owned client for the initial publisher request.
            The caller controls its transport and lifecycle. Redirect targets
            and manager manifests use fresh SDK-owned client contexts.
        validate_structure: Validate authorization entry structure (default True).
            False permits bare entries without relaxing fetch security checks.
        transport_factory: Optional factory for SDK-owned hops, receiving the URL
            and timeout. Defaults to an IP-pinned, proxy-disabled client. Custom
            factories must supply their own transport-level rebinding protection.

    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``.
        AdagentsHTTPError: On other unsuccessful HTTP statuses; carries
            ``status_code`` and ``url`` and subclasses 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.

        Strict authorization structure validation is enabled by default.
        Pass ``validate_structure=False`` to fetch bare-entry documents, then
        inspect :func:`validate_adagents_structure` and explicitly choose a
        :func:`resolve_properties_for_agent` mode. This opt-out preserves JSON,
        endpoint, renderer-origin, SSRF and body-size checks.
    """
    publisher_domain = _validate_publisher_domain(publisher_domain)

    try:
        data, *_ = await _resolve_direct(
            publisher_domain,
            timeout,
            user_agent,
            client,
            validate_structure=validate_structure,
            transport_factory=transport_factory,
        )
        return data
    except AdagentsNotFoundError:
        manager_data = await _try_managerdomain_fallback(
            publisher_domain,
            timeout,
            user_agent,
            client,
            validate_structure=validate_structure,
            transport_factory=transport_factory,
        )
        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,
    *,
    validate_structure: bool = True,
    transport_factory: AdagentsTransportFactory | 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.

    Cached bodies undergo the same document and renderer checks as a fresh
    response, including the current ``validate_structure`` policy. Switching
    from an opt-out fetch to a strict refresh therefore validates the cached
    document before returning it; a successful refresh preserves body identity.

    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,
        validate_structure=validate_structure,
        transport_factory=transport_factory,
    )
    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

async def fetch_publisher_signing_pins(
    publisher_domains: list[str] | tuple[str, ...],
    agent_url: str,
) -> dict[str, list[dict[str, Any]] | None]:
    """Fetch current pins for inventory selected from a verifier's own records.

    Calls the uncached fetch path so a second call force-refreshes adagents.json.
    An absent signing_keys field imposes no pin; an empty or malformed pin
    matches nothing. Missing or ambiguous authorization fails closed.
    """
    from adcp.signing.canonical import canonicalize_target_uri

    canonical = canonicalize_target_uri(agent_url)
    pins: dict[str, list[dict[str, Any]] | None] = {}
    for domain in publisher_domains:
        record = await fetch_adagents(domain)
        matches = [
            entry
            for entry in record.get("authorized_agents", [])
            if isinstance(entry, dict)
            and isinstance(entry.get("url"), str)
            and canonicalize_target_uri(entry["url"]) == canonical
        ]
        if len(matches) != 1:
            raise ValueError("publisher must authorize exactly one matching agent")
        entry = matches[0]
        if "signing_keys" not in entry:
            pins[domain] = None
        else:
            keys = entry["signing_keys"]
            pins[domain] = (
                [key for key in keys if isinstance(key, dict)] if isinstance(keys, list) else []
            )
    return pins

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 find_authorized_agent_entries(
    adagents_data: dict[str, Any], agent_url: str
) -> list[dict[str, Any]]:
    """Return all entries listing an agent, independently of property binding.

    A non-empty result distinguishes a listed agent whose selectors resolve to
    no properties from an unlisted agent. Entries are returned in document order
    without copying or modifying them. Malformed sibling entries and URL values
    are skipped. Matching preserves HTTP/HTTPS equivalence and path semantics.

    Raises AdagentsValidationError if the document is not an object with an
    ``authorized_agents`` array.
    """
    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")
    normalized_agent_url = normalize_url(agent_url)
    matches: list[dict[str, Any]] = []
    for agent in authorized_agents:
        if not isinstance(agent, dict):
            continue
        entry_url = agent.get("url")
        if not isinstance(entry_url, str) or not entry_url.strip():
            continue
        try:
            normalized_entry_url = normalize_url(entry_url)
        except ValueError:
            continue
        if normalized_entry_url == normalized_agent_url:
            matches.append(agent)
    return matches

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_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 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_url(url: str) -> str:
    """Return a protocol-agnostic URL key with host case/default ports normalized.

    Paths remain case-sensitive. Query strings and fragments are ignored, and
    trailing path slashes are removed, preserving the existing matching rules.
    Only the default port for the supplied scheme is removed (80 for HTTP,
    443 for HTTPS); non-default ports remain significant.
    """
    parsed = urlparse(url)
    authority = parsed.netloc
    if parsed.hostname is not None:
        host = parsed.hostname.lower()
        if ":" in host:
            host = f"[{host}]"
        port = parsed.port
        if port is not None and (parsed.scheme.lower(), port) not in {("http", 80), ("https", 443)}:
            host += f":{port}"
        userinfo = authority.rsplit("@", 1)[0] + "@" if "@" in authority else ""
        authority = userinfo + host
    return (authority + parsed.path).rstrip("/")

def parse_managerdomains(ads_txt_content: str) -> list[str]:
    """Extract MANAGERDOMAIN= directives from ads.txt content.

    Per IAB ads.txt: only directive-form lines
    (``MANAGERDOMAIN=value``) count — pure comment lines beginning with
    ``#`` are rejected. Order in source is preserved so callers can
    apply last-wins.
    """
    managers: list[str] = []
    for line in ads_txt_content.splitlines():
        stripped = line.lstrip()
        if stripped.startswith("#"):
            continue
        match = _MANAGERDOMAIN_RE.match(line)
        if match:
            managers.append(match.group(1).lower())
    return managers

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'")

async def validate_adagents_domain(
    publisher_domain: str,
    timeout: float = 10.0,
    user_agent: str = "AdCP-Client/1.0",
    client: httpx.AsyncClient | None = None,
    *,
    validate_structure: bool = True,
    transport_factory: AdagentsTransportFactory | 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 ads.txt 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,
            validate_structure=validate_structure,
            transport_factory=transport_factory,
        )
        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, transport_factory=transport_factory
    )
    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,
            validate_structure=validate_structure,
            transport_factory=transport_factory,
        )
    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 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
    """
    for agent in find_authorized_agent_entries(adagents_data, agent_url):
        # 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,
    )

@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)

@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

@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 AgentAuthorizationsDirectoryResult(AdCPBaseModel):
    """Response envelope for ``GET /v1/agents/{agent_url}/publishers``.

    Maps directly to ``schemas/aao/agent-publishers.json`` in the AdCP
    bundle (adcp#4828). The directory is a discovery accelerator — each
    ``publisher_domain`` row tells callers where to look; they SHOULD
    verify the publisher's adagents.json directly before treating an
    authorization as trusted.
    """

    agent_url: str
    directory_indexed_at: datetime | None
    publishers: list[DirectoryPublisherEntry] = Field(default_factory=list)
    next_cursor: str | None = None

class AuthorizationContext:
    """Authorization context for a publisher domain.

    Attributes:
        property_ids: List of property IDs the agent is authorized for
        property_tags: List of property tags the agent is authorized for
        raw_properties: Raw property data from adagents.json
    """

    def __init__(self, properties: list[Any]):
        """Initialize from list of properties.

        Args:
            properties: List of property dictionaries from adagents.json
        """
        self.property_ids: list[str] = []
        self.property_tags: list[str] = []
        self.raw_properties = properties

        # Extract property IDs and tags
        for prop in properties:
            if not isinstance(prop, dict):
                continue

            # Extract property ID (per AdCP v2 schema, the field is "property_id")
            prop_id = prop.get("property_id")
            if prop_id and isinstance(prop_id, str):
                self.property_ids.append(prop_id)

            # Extract tags
            tags = prop.get("tags", [])
            if isinstance(tags, list):
                for tag in tags:
                    if isinstance(tag, str) and tag not in self.property_tags:
                        self.property_tags.append(tag)

    def __repr__(self) -> str:
        return (
            f"AuthorizationContext("
            f"property_ids={self.property_ids}, "
            f"property_tags={self.property_tags})"
        )

class DirectoryPublisherEntry(AdCPBaseModel):
    """One publisher row in an AAO directory inverse-lookup response."""

    publisher_domain: str
    discovery_method: DirectoryDiscoveryMethod
    manager_domain: str | None = None
    properties_authorized: int = Field(ge=0)
    properties_total: int = Field(ge=0)
    signing_keys_pinned: bool | None = None
    status: DirectoryEdgeStatus
    last_verified_at: datetime
    property_ids: list[str] | None = Field(
        default=None,
        description=(
            "Canonical property IDs the agent's selectors resolve to under "
            "this publisher. Present iff the request was made with "
            "include=['properties'] AND the directory server supports it "
            "(per adcp#4894). None signals count-only mode for downstream "
            "consumers."
        ),
    )

class PublisherDivergence(AdCPBaseModel):
    """Divergence record for a single publisher domain.

    ``missing_in_inline``: property IDs the federated fetch found in the
    publisher's own adagents.json that the directory did not surface
    (publisher has properties the directory doesn't know about yet).

    ``missing_in_federated``: property IDs the directory claims the agent
    is authorized for but the publisher's own adagents.json does not
    include (stale directory entry or publisher revocation).

    Both fields are None in count-only fallback mode (directory did
    not return ``property_ids[]``). In count-only mode, count-equality
    does NOT guarantee set-equality — same-count substitutions are
    undetectable. Use ``?include=properties`` (adcp#4894) on directories
    that support it for full set-diff precision.

    ``child_fetch_error`` is non-None when the publisher's adagents.json
    could not be fetched or parsed; other fields carry no meaning.
    """

    publisher_domain: str
    directory_properties_authorized: int = Field(ge=0)
    federated_properties_found: int = Field(ge=0)
    missing_in_inline: list[str] | None = None
    missing_in_federated: list[str] | None = None
    child_fetch_error: str | None = None