refactor(adagents): address review follow-ups on #752

- Dedupe collected entries by publisher_domain before fan-out so a
  hostile directory can't amplify 1 publisher into a 20-concurrent
  burst against one host. (Medium-severity security finding.)
- Pagination loop guards: detect repeated next_cursor and cap at
  MAX_DIRECTORY_PAGES=1000 pages; raise AdagentsValidationError on
  either violation.
- Emit a one-shot logging warning when the entire sample comes back
  without property_ids[] -- adopters notice they're running in
  count-only mode rather than full set-diff.
- Document the falsy property_id filter behavior in a code comment.
- Cite the spec wire-format precedent for include= repeated-key form.

Bot review: #752

Author: bokelley

src/adcp/adagents.py

@@ -11,6 +11,7 @@
 import asyncio
 import ipaddress
 import json
+import logging
 import re
 import socket
 from dataclasses import dataclass, field
@@ -25,6 +26,8 @@
 from adcp.types.base import AdCPBaseModel
 from adcp.validation import ValidationError, validate_adagents
 
+logger = logging.getLogger(__name__)
+
 DiscoveryMethod = Literal["direct", "authoritative_location", "ads_txt_managerdomain"]
 
 
@@ -1836,6 +1839,11 @@ class AgentAuthorizationsDirectoryResult(AdCPBaseModel):
 # page is a small envelope; pagination handles bulk responses.
 MAX_DIRECTORY_PAGE_BYTES = 5 * 1024 * 1024
 
+# Hard cap on directory pagination iterations. A misbehaving directory that
+# never returns an empty next_cursor would otherwise hang the sweep
+# indefinitely; this cap fail-closes the loop.
+MAX_DIRECTORY_PAGES = 1000
+
 
 async def fetch_agent_authorizations_from_directory(
     agent_url: str,
@@ -1914,6 +1922,9 @@ async def fetch_agent_authorizations_from_directory(
     if since is not None:
         query_pairs.append(("since", since))
     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:
@@ -2054,6 +2065,8 @@ async def detect_publisher_properties_divergence(
     try:
         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,
@@ -2063,13 +2076,51 @@ async def detect_publisher_properties_divergence(
                 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)
 
@@ -2080,6 +2131,11 @@ async def _probe(entry: DirectoryPublisherEntry) -> PublisherDivergence | None:
                         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")
                     }

tests/test_adagents.py

@@ -4106,3 +4106,103 @@ async def releaser():
 
         assert peak <= 4
         assert peak >= 1  # sanity: probes actually ran
+
+    async def test_divergence_dedupes_collected_by_publisher_domain(self, monkeypatch):
+        """Hostile directory: 5 rows all publisher_domain=victim → 1 fetch."""
+        from adcp import adagents as adagents_mod
+        from adcp.adagents import detect_publisher_properties_divergence
+
+        # 5 entries all pointing at the same victim host. A naive
+        # implementation would fan out 5 concurrent fetches against
+        # victim.example; the dedupe path must collapse to a single probe.
+        publishers = [
+            self._entry("victim.example", properties_authorized=1, property_ids=["p1"])
+            for _ in range(5)
+        ]
+        handler = self._directory_handler(publishers)
+
+        call_count = 0
+
+        async def fake_fetch_adagents(domain, timeout=10.0, client=None):
+            nonlocal call_count
+            call_count += 1
+            return {}
+
+        def fake_get_properties_by_agent(data, agent_url):
+            return [{"property_id": "p1"}]
+
+        monkeypatch.setattr(adagents_mod, "fetch_adagents", fake_fetch_adagents)
+        monkeypatch.setattr(adagents_mod, "get_properties_by_agent", fake_get_properties_by_agent)
+
+        async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+            await detect_publisher_properties_divergence(
+                "https://agent.example.com/",
+                directory_url="https://aao.example.com",
+                client=client,
+            )
+
+        assert call_count == 1
+
+    async def test_divergence_aborts_on_repeated_cursor(self, monkeypatch):
+        """Misbehaving directory returns the same next_cursor forever → raise."""
+        from adcp.adagents import detect_publisher_properties_divergence
+
+        # Each response includes a next_cursor that never advances. The
+        # page-walk loop must detect the repeat and raise rather than
+        # spin until OOM.
+        def handler(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(
+                200,
+                json={
+                    "agent_url": "https://agent.example.com/",
+                    "directory_indexed_at": "2026-05-20T12:00:00Z",
+                    "publishers": [],
+                    "next_cursor": "stuck",
+                },
+            )
+
+        async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+            with pytest.raises(AdagentsValidationError, match="cursor 'stuck' repeated"):
+                await detect_publisher_properties_divergence(
+                    "https://agent.example.com/",
+                    directory_url="https://aao.example.com",
+                    sample_size=None,  # full sweep, so we actually walk pages
+                    client=client,
+                )
+
+    async def test_divergence_warns_on_count_only_mode(self, monkeypatch, caplog):
+        """No entry has property_ids → one-shot warning fires."""
+        import logging as _logging
+
+        from adcp import adagents as adagents_mod
+        from adcp.adagents import detect_publisher_properties_divergence
+
+        # No property_ids on any row → directory is in count-only mode.
+        publishers = [
+            self._entry("a.example", properties_authorized=1),
+            self._entry("b.example", properties_authorized=2),
+        ]
+        handler = self._directory_handler(publishers)
+
+        async def fake_fetch_adagents(domain, timeout=10.0, client=None):
+            return {}
+
+        def fake_get_properties_by_agent(data, agent_url):
+            return [{"property_id": "p1"}]
+
+        monkeypatch.setattr(adagents_mod, "fetch_adagents", fake_fetch_adagents)
+        monkeypatch.setattr(adagents_mod, "get_properties_by_agent", fake_get_properties_by_agent)
+
+        with caplog.at_level(_logging.WARNING, logger="adcp.adagents"):
+            async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+                await detect_publisher_properties_divergence(
+                    "https://agent.example.com/",
+                    directory_url="https://aao.example.com",
+                    client=client,
+                )
+
+        count_only_warnings = [
+            r for r in caplog.records if "count-only divergence detection" in r.getMessage()
+        ]
+        assert len(count_only_warnings) == 1
+        assert "https://aao.example.com" in count_only_warnings[0].getMessage()