fix(adagents): address review on directory wrapper + divergence detector (#752)

Spec conformance and concurrency fixes from PR review:

- Endpoint path corrected: /v1/agents/{agent_url}/publishers (drop /api/).
- Response envelope now spec-strict: read `next_cursor` and `publishers`
  only (no `cursor`/`results` fallbacks); read `directory_indexed_at`
  required on non-empty pages; remove dead `total` field.
- `AgentDirectoryLookup.cursor` renamed to `next_cursor`.
- `AgentPublisherEntry.discovery_method` and `.status` now typed as the
  spec Literal enums (`DiscoveryMethod`, new `PublisherStatus`); a
  missing or out-of-enum value raises AdagentsValidationError instead
  of silently defaulting to `"adagents_authoritative"`/`"authorized"`.
- `detect_publisher_properties_divergence` now caps concurrent child
  probes via `max_concurrency` (default 20) using `asyncio.Semaphore`.
  The `_probe` except clause now also catches `httpx.HTTPError`, `OSError`,
  `ValueError` so a single network failure no longer aborts the gather.
- `sample_size` default flipped from `None` to `200`; pass
  `sample_size=None` to opt into a full sweep.
- `directory_url` default now reads `AAO_PUBLISHER_DIRECTORY_URL` env var
  at call time via a sentinel, falling back to
  `"https://agenticadvertising.org"`.
- Module-local `import asyncio` / `from urllib.parse import quote` moved
  to the top of the file.
- Tests for the directory wrapper and divergence detector rewritten to
  use `respx.mock` with spec-shaped JSON payloads instead of bare
  `MagicMock(httpx.AsyncClient.get)`. New tests cover env-var defaulting,
  `directory_indexed_at` null-on-empty rule, enum rejection, default
  sample-size cap, `max_concurrency` peak-in-flight bound, and a
  selective `httpx.ConnectError` that must not poison sibling probes.
- Public API snapshot regenerated to add the six new symbols plus
  `PublisherStatus`.

(See follow-up spec issue, will be linked) for the `?include=properties`
extension that would enable full set-diff in the divergence detector.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bokelley

src/adcp/__init__.py

@@ -21,6 +21,7 @@
     DivergenceReport,
     EntryErrorKind,
     PublisherDivergence,
+    PublisherStatus,
     detect_publisher_properties_divergence,
     domain_matches,
     fetch_adagents,
@@ -832,6 +833,7 @@ def get_adcp_version() -> str:
     "fetch_agent_authorizations",
     "fetch_agent_authorizations_from_directory",
     "PublisherDivergence",
+    "PublisherStatus",
     "validate_adagents_domain",
     "validate_adagents_structure",
     "verify_agent_authorization",

src/adcp/adagents.py

@@ -8,19 +8,41 @@
 for sales agents to verify they are authorized for specific properties.
 """
 
+import asyncio
 import ipaddress
+import os
 import re
 from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any, Literal
-from urllib.parse import urlparse
+from urllib.parse import quote, urlparse
 
 import httpx
 
 from adcp.exceptions import AdagentsNotFoundError, AdagentsTimeoutError, AdagentsValidationError
 from adcp.validation import ValidationError, validate_adagents
 
-DiscoveryMethod = Literal["direct", "authoritative_location", "ads_txt_managerdomain"]
+DiscoveryMethod = Literal[
+    "direct",
+    "authoritative_location",
+    "adagents_authoritative",
+    "ads_txt_managerdomain",
+]
+
+# AAO directory ``status`` enum per agent-publishers.json schema.
+PublisherStatus = Literal["authorized", "revoked"]
+
+# Sentinel used so ``directory_url=`` defaults are evaluated at call time
+# from ``AAO_PUBLISHER_DIRECTORY_URL`` rather than baked in at import.
+_DIRECTORY_URL_SENTINEL: Any = object()
+_DEFAULT_DIRECTORY_URL = "https://agenticadvertising.org"
+
+
+def _resolve_directory_url(directory_url: Any) -> str:
+    """Resolve directory base URL, consulting env var when the sentinel is passed."""
+    if directory_url is _DIRECTORY_URL_SENTINEL:
+        return os.environ.get("AAO_PUBLISHER_DIRECTORY_URL", _DEFAULT_DIRECTORY_URL)
+    return str(directory_url)
 
 
 # authorization_type discriminator -> required selector field, per the AdCP
@@ -1345,7 +1367,6 @@ async def fetch_agent_authorizations(
         - 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(
@@ -1382,6 +1403,12 @@ async def fetch_authorization_for_domain(
 # ---------------------------------------------------------------------------
 
 
+_VALID_DISCOVERY_METHODS: frozenset[str] = frozenset(
+    {"direct", "authoritative_location", "adagents_authoritative", "ads_txt_managerdomain"}
+)
+_VALID_PUBLISHER_STATUSES: frozenset[str] = frozenset({"authorized", "revoked"})
+
+
 @dataclass
 class AgentPublisherEntry:
     """A single publisher row from the AAO directory's agent-publishers endpoint.
@@ -1393,12 +1420,12 @@ class AgentPublisherEntry:
     """
 
     publisher_domain: str
-    discovery_method: str
+    discovery_method: DiscoveryMethod
     manager_domain: str | None
     properties_authorized: int
     properties_total: int
     signing_keys_pinned: bool
-    status: str
+    status: PublisherStatus
     last_verified_at: str | None
 
 
@@ -1407,22 +1434,23 @@ class AgentDirectoryLookup:
     """Envelope returned by :func:`fetch_agent_authorizations_from_directory`.
 
     ``publishers`` is the list of publisher entries for this page.
-    ``cursor`` is ``None`` when there are no further pages.
-    ``total`` is set when the server returns a total count.
+    ``next_cursor`` is ``None`` when there are no further pages.
+    ``directory_indexed_at`` is the server's index timestamp for this response
+    (``None`` only permitted on empty pages per spec NULL-on-empty rule).
     """
 
     agent_url: str
     publishers: list[AgentPublisherEntry]
-    cursor: str | None = None
-    total: int | None = None
+    directory_indexed_at: str | None = None
+    next_cursor: str | None = None
 
 
 async def fetch_agent_authorizations_from_directory(
     agent_url: str,
-    directory_url: str = "https://agenticadvertising.org",
+    directory_url: Any = _DIRECTORY_URL_SENTINEL,
     *,
     since: datetime | None = None,
-    status: list[str] | None = None,
+    status: list[PublisherStatus] | None = None,
     cursor: str | None = None,
     limit: int = 200,
     timeout: float = 30.0,
@@ -1432,7 +1460,7 @@ async def fetch_agent_authorizations_from_directory(
 
     Inverse of :func:`fetch_agent_authorizations` — instead of pulling from
     individual publisher adagents.json files, this queries the directory's
-    ``GET /api/v1/agents/{agent_url}/publishers`` index.
+    ``GET /v1/agents/{agent_url}/publishers`` index.
 
     Each returned :class:`AgentPublisherEntry` carries the same
     ``discovery_method`` / ``manager_domain`` provenance fields as
@@ -1441,41 +1469,49 @@ async def fetch_agent_authorizations_from_directory(
 
     Args:
         agent_url: The sales agent URL to look up (``%``-encoded in the path).
-        directory_url: Base URL of the AAO directory
-            (default ``"https://agenticadvertising.org"``).
+        directory_url: Base URL of the AAO directory. When omitted, defaults
+            to the ``AAO_PUBLISHER_DIRECTORY_URL`` environment variable, or
+            ``"https://agenticadvertising.org"`` if that is unset. The default
+            is resolved on every call so test harnesses can monkeypatch the
+            environment.
         since: Only return publishers whose authorization was last verified
             after this timestamp.  ``None`` returns all.
         status: Filter by authorization status (default ``["authorized"]``).
             Pass ``["authorized", "revoked"]`` to include revoked entries.
-        cursor: Pagination cursor from a previous call's ``cursor`` field.
+        cursor: Pagination cursor from a previous call's ``next_cursor`` field.
         limit: Maximum entries per page (server-side cap may be lower).
         timeout: Per-request timeout in seconds.
         client: Optional ``httpx.AsyncClient`` for connection pooling.
             The client is **not** closed by this function.
 
     Returns:
         :class:`AgentDirectoryLookup` with ``publishers`` for this page and
-        a ``cursor`` for the next page (``None`` when exhausted).
+        a ``next_cursor`` for the next page (``None`` when exhausted).
+
+    Raises:
+        AdagentsValidationError: If the response envelope is malformed, a
+            row is missing required fields, or a row carries an enum value
+            outside the spec.
 
     Example::
 
         lookup = await fetch_agent_authorizations_from_directory(
             "https://interchange.io",
         )
         print(f"{len(lookup.publishers)} publishers on first page")
-        while lookup.cursor:
+        while lookup.next_cursor:
             lookup = await fetch_agent_authorizations_from_directory(
                 "https://interchange.io",
-                cursor=lookup.cursor,
+                cursor=lookup.next_cursor,
             )
     """
-    from urllib.parse import quote
+    resolved_directory_url = _resolve_directory_url(directory_url)
 
     if status is None:
         status = ["authorized"]
 
     encoded_agent = quote(agent_url, safe="")
-    url = f"{directory_url.rstrip('/')}/api/v1/agents/{encoded_agent}/publishers"
+    url = f"{resolved_directory_url.rstrip('/')}/v1/agents/{encoded_agent}/publishers"
 
     # Build params as a list of (key, value) tuples so multi-value status
     # produces repeated keys (?status=authorized&status=revoked), not a
@@ -1501,33 +1537,68 @@ async def fetch_agent_authorizations_from_directory(
     if not isinstance(data, dict):
         raise AdagentsValidationError(
             f"Directory returned unexpected JSON type {type(data).__name__!r} "
-            f"for /api/v1/agents/{{agent_url}}/publishers"
+            f"for /v1/agents/{{agent_url}}/publishers"
+        )
+
+    raw_rows = data.get("publishers")
+    if raw_rows is None:
+        raise AdagentsValidationError("Directory response is missing required 'publishers' array")
+    if not isinstance(raw_rows, list):
+        raise AdagentsValidationError(
+            f"Directory 'publishers' must be an array, got {type(raw_rows).__name__!r}"
         )
 
     publishers: list[AgentPublisherEntry] = []
-    raw_rows = data.get("publishers") or data.get("results") or []
-    for row in raw_rows:
-        domain = row.get("publisher_domain", "")
-        if not domain:
-            continue  # skip malformed rows missing the required field
+    for index, row in enumerate(raw_rows):
+        if not isinstance(row, dict):
+            raise AdagentsValidationError(f"publishers[{index}] is not a JSON object")
+        domain = row.get("publisher_domain")
+        if not isinstance(domain, str) or not domain:
+            raise AdagentsValidationError(
+                f"publishers[{index}] is missing required 'publisher_domain'"
+            )
+
+        discovery_method = row.get("discovery_method")
+        if discovery_method not in _VALID_DISCOVERY_METHODS:
+            raise AdagentsValidationError(
+                f"publishers[{index}] has invalid discovery_method "
+                f"{discovery_method!r} (expected one of: "
+                f"{', '.join(sorted(_VALID_DISCOVERY_METHODS))})"
+            )
+
+        row_status = row.get("status")
+        if row_status not in _VALID_PUBLISHER_STATUSES:
+            raise AdagentsValidationError(
+                f"publishers[{index}] has invalid status {row_status!r} "
+                f"(expected one of: {', '.join(sorted(_VALID_PUBLISHER_STATUSES))})"
+            )
+
         publishers.append(
             AgentPublisherEntry(
                 publisher_domain=domain,
-                discovery_method=row.get("discovery_method", "adagents_authoritative"),
+                discovery_method=discovery_method,
                 manager_domain=row.get("manager_domain"),
                 properties_authorized=row.get("properties_authorized", 0),
                 properties_total=row.get("properties_total", 0),
                 signing_keys_pinned=bool(row.get("signing_keys_pinned", False)),
-                status=row.get("status", "authorized"),
+                status=row_status,
                 last_verified_at=row.get("last_verified_at"),
             )
         )
 
+    directory_indexed_at = data.get("directory_indexed_at")
+    # Per spec, directory_indexed_at is required except on empty pages where
+    # it may be NULL.  Tolerate missing only when the page is empty.
+    if directory_indexed_at is None and publishers:
+        raise AdagentsValidationError(
+            "Directory response is missing required 'directory_indexed_at' " "on a non-empty page"
+        )
+
     return AgentDirectoryLookup(
         agent_url=agent_url,
         publishers=publishers,
-        cursor=data.get("cursor") or data.get("next_cursor"),
-        total=data.get("total"),
+        directory_indexed_at=directory_indexed_at,
+        next_cursor=data.get("next_cursor"),
     )
 
 
@@ -1574,9 +1645,10 @@ class PublisherDivergence:
 
 async def detect_publisher_properties_divergence(
     agent_url: str,
-    directory_url: str = "https://agenticadvertising.org",
+    directory_url: Any = _DIRECTORY_URL_SENTINEL,
     *,
-    sample_size: int | None = None,
+    sample_size: int | None = 200,
+    max_concurrency: int = 20,
     timeout: float = 30.0,
     client: httpx.AsyncClient | None = None,
 ) -> DivergenceReport:
@@ -1603,18 +1675,24 @@ async def detect_publisher_properties_divergence(
     directory endpoint will enable full set-diff once that parameter is
     deployed.
 
-    **Cost warning — ``sample_size`` is mandatory for large networks.**
+    **Cost — ``sample_size`` defaults to 200; opt in to a full sweep.**
     Running a full sweep against cafemedia's ~6,800 child publishers launches
-    ~6,800 concurrent HTTP fetches.  With a 30 s timeout each, total wall-clock
-    is bounded by the slowest fetch, but server-side rate limits may apply.
-    Pass ``sample_size=N`` to cap the sweep; the sample is taken from the
-    first page of directory results.
+    thousands of HTTP fetches. To guard against accidental fan-out the default
+    caps at 200; pass ``sample_size=None`` to scan every publisher the
+    directory returns. ``max_concurrency`` (default 20) bounds the number of
+    in-flight child fetches via an internal ``asyncio.Semaphore`` regardless
+    of ``sample_size``.
 
     Args:
         agent_url: The agent URL to check authorizations for.
-        directory_url: AAO directory base URL.
-        sample_size: Maximum number of publisher domains to probe.  ``None``
-            sweeps all pages (full network — may be very slow).
+        directory_url: AAO directory base URL. Defaults to the
+            ``AAO_PUBLISHER_DIRECTORY_URL`` env var or
+            ``"https://agenticadvertising.org"``.
+        sample_size: Maximum number of publisher domains to probe (default
+            ``200``). Pass ``None`` to opt into a full sweep across every
+            page of directory results.
+        max_concurrency: Maximum number of concurrent child adagents.json
+            fetches. Defaults to 20.
         timeout: Per-request timeout for both directory and child fetches.
         client: Optional shared ``httpx.AsyncClient``.
 
@@ -1643,7 +1721,10 @@ async def detect_publisher_properties_divergence(
                       f"(dir={entry.directory_properties_authorized}, "
                       f"federated={entry.federated_properties_found})")
     """
-    import asyncio
+    if max_concurrency < 1:
+        raise ValueError("max_concurrency must be at least 1")
+
+    resolved_directory_url = _resolve_directory_url(directory_url)
 
     own_client = client is None
     http = client or httpx.AsyncClient()
@@ -1655,7 +1736,7 @@ async def detect_publisher_properties_divergence(
         while True:
             page = await fetch_agent_authorizations_from_directory(
                 agent_url,
-                directory_url=directory_url,
+                directory_url=resolved_directory_url,
                 cursor=page_cursor,
                 timeout=timeout,
                 client=http,
@@ -1664,31 +1745,39 @@ async def detect_publisher_properties_divergence(
             if sample_size is not None and len(all_entries) >= sample_size:
                 all_entries = all_entries[:sample_size]
                 break
-            page_cursor = page.cursor
+            page_cursor = page.next_cursor
             if not page_cursor:
                 break
 
+        semaphore = asyncio.Semaphore(max_concurrency)
+
         async def _probe(entry: AgentPublisherEntry) -> PublisherDivergence | None:
-            try:
-                data = await fetch_adagents(
-                    entry.publisher_domain, timeout=timeout, client=http
-                )
-                federated_props = get_properties_by_agent(data, agent_url)
-                federated_ids = {
-                    p.get("property_id")
-                    for p in federated_props
-                    if p.get("property_id")
-                }
-            except (AdagentsNotFoundError, AdagentsValidationError, AdagentsTimeoutError) as exc:
-                return PublisherDivergence(
-                    publisher_domain=entry.publisher_domain,
-                    directory_properties_authorized=entry.properties_authorized,
-                    federated_properties_found=0,
-                    # None = count-only mode; IDs unavailable from directory
-                    missing_in_inline=None,
-                    missing_in_federated=None,
-                    child_fetch_error=str(exc),
-                )
+            async with semaphore:
+                try:
+                    data = await fetch_adagents(
+                        entry.publisher_domain, timeout=timeout, client=http
+                    )
+                    federated_props = get_properties_by_agent(data, agent_url)
+                    federated_ids = {
+                        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,
+                        # None = count-only mode; IDs unavailable from directory
+                        missing_in_inline=None,
+                        missing_in_federated=None,
+                        child_fetch_error=str(exc) or type(exc).__name__,
+                    )
 
             fed_count = len(federated_ids)
             # Count-only comparison: directory does not currently return