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

- Cite adcp#4827 in _resolve_publisher_property_selectors docstring.
- isinstance(list) guard on property_tags / property_ids in both
  publisher_properties resolution and the pre-existing property_tags /
  property_ids authorization branches — string-iterating-as-tags is now
  caught and resolves to [].
- Lift domain index build out of per-agent loop; _build_domain_index
  module helper used by both get_all_properties and get_properties_by_agent,
  index passed through _resolve_agent_properties → _resolve_publisher_
  property_selectors. O(N+M) instead of O(agents × N).
- Document "no federated fallback" on get_properties_by_agent and
  get_all_properties docstrings; point at PR #752's federated helpers.
- Cafemedia scale test wall-clock budget 2.0s → 5.0s for shared-runner
  margin.

Bot review: #750

Author: bokelley

src/adcp/adagents.py

@@ -1207,12 +1207,16 @@ async def verify_agent_for_property(
 def _resolve_agent_properties(
     agent: dict[str, Any],
     top_level_properties: list[dict[str, Any]],
+    domain_index: dict[str, list[dict[str, Any]]],
 ) -> list[dict[str, Any]]:
     """Resolve properties for a single agent entry based on its authorization_type.
 
     Args:
         agent: An authorized_agents entry
         top_level_properties: The top-level properties array from adagents.json
+        domain_index: Pre-built ``publisher_domain → [property, ...]`` index
+            over ``top_level_properties`` (built once per file by the caller
+            via :func:`_build_domain_index`).
 
     Returns:
         List of resolved property dicts for this agent
@@ -1230,7 +1234,10 @@ def _resolve_agent_properties(
 
     # Handle property_ids (filter top-level properties by property_id)
     if authorization_type == "property_ids":
-        authorized_ids = set(agent.get("property_ids", []))
+        raw_ids = agent.get("property_ids")
+        if not isinstance(raw_ids, list):
+            return []
+        authorized_ids = {i for i in raw_ids if isinstance(i, str)}
         return [
             p
             for p in top_level_properties
@@ -1239,7 +1246,10 @@ def _resolve_agent_properties(
 
     # Handle property_tags (filter top-level properties by tags)
     if authorization_type == "property_tags":
-        authorized_tags = {t for t in agent.get("property_tags", []) if isinstance(t, str)}
+        raw_tags = agent.get("property_tags")
+        if not isinstance(raw_tags, list):
+            return []
+        authorized_tags = {t for t in raw_tags if isinstance(t, str)}
         return [
             p
             for p in top_level_properties
@@ -1260,20 +1270,45 @@ def _resolve_agent_properties(
         selectors = _fanout_publisher_properties(
             [p for p in publisher_props if isinstance(p, dict)]
         )
-        return _resolve_publisher_property_selectors(selectors, top_level_properties)
+        return _resolve_publisher_property_selectors(selectors, domain_index)
 
     return []
 
 
+def _build_domain_index(
+    properties: list[dict[str, Any]],
+) -> dict[str, list[dict[str, Any]]]:
+    """Build a ``publisher_domain → [property, ...]`` index.
+
+    O(N) up-front cost; reused across every selector for that file so the
+    per-selector resolution cost drops from O(properties) to O(1) lookup
+    plus O(matches) filtering. Malformed entries (non-dict, missing or
+    non-string ``publisher_domain``) are skipped.
+    """
+    domain_index: dict[str, list[dict[str, Any]]] = {}
+    for prop in properties:
+        if not isinstance(prop, dict):
+            continue
+        domain = prop.get("publisher_domain")
+        if not isinstance(domain, str) or not domain:
+            continue
+        domain_index.setdefault(domain, []).append(prop)
+    return domain_index
+
+
 def _resolve_publisher_property_selectors(
     selectors: list[dict[str, Any]],
-    top_level_properties: list[dict[str, Any]],
+    domain_index: dict[str, list[dict[str, Any]]],
 ) -> list[dict[str, Any]]:
     """Resolve fanned-out publisher_properties selectors against inline data.
 
+    Resolves selectors per adcp#4827 §Resolution-paths (inline path
+    against the parent file's top-level properties[] indexed by
+    publisher_domain).
+
     For each selector (one per publisher_domain), look up the matching
-    properties in ``top_level_properties`` by ``publisher_domain`` and
-    apply the selector's ``selection_type``:
+    properties in ``domain_index`` by ``publisher_domain`` and apply the
+    selector's ``selection_type``:
 
     - ``"all"``: every property under that domain
     - ``"by_tag"``: properties whose ``tags`` intersect ``property_tags``
@@ -1289,25 +1324,16 @@ def _resolve_publisher_property_selectors(
     federated fallback (fetching the publisher's own adagents.json) is
     out of scope for this resolver and lives in companion helpers.
 
+    ``domain_index`` is built once per file by :func:`_build_domain_index`
+    and reused across every agent's selectors in that file.
+
     Results are deduplicated by ``(publisher_domain, property_id)``.
     Raises :class:`AdagentsValidationError` if any matching property is
     missing the required ``property_id`` field (fail-fast per CLAUDE.md).
     """
     if not selectors:
         return []
 
-    # Build domain → [property, ...] index once. O(N) up-front trades
-    # against O(selectors × properties) per-domain scans, which blows
-    # up at cafemedia scale (6,843 properties × thousands of selectors).
-    domain_index: dict[str, list[dict[str, Any]]] = {}
-    for prop in top_level_properties:
-        if not isinstance(prop, dict):
-            continue
-        domain = prop.get("publisher_domain")
-        if not isinstance(domain, str) or not domain:
-            continue
-        domain_index.setdefault(domain, []).append(prop)
-
     resolved: list[dict[str, Any]] = []
     seen: set[tuple[str, str]] = set()
     for selector in selectors:
@@ -1323,7 +1349,10 @@ def _resolve_publisher_property_selectors(
         if selection_type == "all":
             matched = list(candidates)
         elif selection_type == "by_tag":
-            wanted_tags = {t for t in selector.get("property_tags", []) or [] if isinstance(t, str)}
+            raw_tags = selector.get("property_tags")
+            if not isinstance(raw_tags, list):
+                continue
+            wanted_tags = {t for t in raw_tags if isinstance(t, str)}
             if not wanted_tags:
                 continue
             matched = [
@@ -1332,7 +1361,10 @@ def _resolve_publisher_property_selectors(
                 if {t for t in p.get("tags", []) or [] if isinstance(t, str)} & wanted_tags
             ]
         elif selection_type == "by_id":
-            wanted_ids = {i for i in selector.get("property_ids", []) or [] if isinstance(i, str)}
+            raw_ids = selector.get("property_ids")
+            if not isinstance(raw_ids, list):
+                continue
+            wanted_ids = {i for i in raw_ids if isinstance(i, str)}
             if not wanted_ids:
                 continue
             matched = [p for p in candidates if p.get("property_id") in wanted_ids]
@@ -1433,6 +1465,17 @@ def get_all_properties(adagents_data: dict[str, Any]) -> list[dict[str, Any]]:
     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
 
@@ -1464,6 +1507,11 @@ def get_all_properties(adagents_data: dict[str, Any]) -> list[dict[str, Any]]:
         )
     ]
 
+    # 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):
@@ -1475,7 +1523,7 @@ def get_all_properties(adagents_data: dict[str, Any]) -> list[dict[str, Any]]:
 
         # 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)
+        agent_properties = _resolve_agent_properties(agent, revoked_top_level, domain_index)
 
         for prop in agent_properties:
             prop_with_agent = {**prop, "agent_url": agent_url}
@@ -1520,6 +1568,17 @@ def get_properties_by_agent(adagents_data: dict[str, Any], agent_url: str) -> li
       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
@@ -1554,6 +1613,8 @@ def get_properties_by_agent(adagents_data: dict[str, Any], agent_url: str) -> li
 
     normalized_agent_url = normalize_url(agent_url)
 
+    domain_index = _build_domain_index(revoked_top_level)
+
     for agent in authorized_agents:
         if not isinstance(agent, dict):
             continue
@@ -1567,7 +1628,7 @@ def get_properties_by_agent(adagents_data: dict[str, Any], agent_url: str) -> li
 
         # revoked_top_level pre-filters revoked domains from the per-domain
         # index, so inline resolution honors revocation transparently.
-        resolved = _resolve_agent_properties(agent, revoked_top_level)
+        resolved = _resolve_agent_properties(agent, revoked_top_level, domain_index)
         return resolved
 
     return []

tests/test_adagents.py

@@ -1759,7 +1759,7 @@ def test_property_missing_property_id_raises(self):
             get_properties_by_agent(adagents_data, "https://agent1.example.com")
 
     def test_get_properties_by_agent_cafemedia_scale(self):
-        """6,843 inline properties across 6,800 child domains under one publisher_domains[] selector.
+        """6,843 inline properties across 6,800 child domains under one publisher_domains[].
 
         Wall-clock-bounded to catch O(N×M) regressions in the resolver.
         At this scale a naive per-domain linear scan over the property list
@@ -1815,10 +1815,152 @@ def test_get_properties_by_agent_cafemedia_scale(self):
         result = get_properties_by_agent(adagents_data, "https://agent1.example.com")
         elapsed = time.perf_counter() - start
 
-        assert elapsed < 2.0, f"resolution took {elapsed:.2f}s (>= 2.0s budget)"
+        assert elapsed < 5.0, f"resolution took {elapsed:.2f}s (>= 5.0s budget)"
         assert len(result) == 6843
         assert {p["publisher_domain"] for p in result} == set(child_domains)
 
+    def test_malformed_property_tags_value_resolves_empty(self):
+        """publisher_properties selector with property_tags as a STRING resolves to [].
+
+        Without the isinstance(list) guard, ``property_tags: "ctv"`` iterates
+        char-by-char and matches properties tagged ``"c"``/``"t"``/``"v"``.
+        The resolver must fail-closed on malformed input.
+        """
+        adagents_data = {
+            "properties": [
+                {
+                    "property_id": "p1",
+                    "publisher_domain": "site1.example",
+                    "property_type": "website",
+                    "name": "Site 1",
+                    "identifiers": [{"type": "domain", "value": "site1.example"}],
+                    "tags": ["c"],
+                },
+                {
+                    "property_id": "p2",
+                    "publisher_domain": "site1.example",
+                    "property_type": "website",
+                    "name": "Site 2",
+                    "identifiers": [{"type": "domain", "value": "site2.example"}],
+                    "tags": ["t"],
+                },
+                {
+                    "property_id": "p3",
+                    "publisher_domain": "site1.example",
+                    "property_type": "website",
+                    "name": "Site 3",
+                    "identifiers": [{"type": "domain", "value": "site3.example"}],
+                    "tags": ["v"],
+                },
+            ],
+            "authorized_agents": [
+                {
+                    "url": "https://agent1.example.com",
+                    "authorization_type": "publisher_properties",
+                    "authorized_for": "Test",
+                    "publisher_properties": [
+                        {
+                            "publisher_domain": "site1.example",
+                            "selection_type": "by_tag",
+                            "property_tags": "ctv",  # malformed: string, not list
+                        },
+                    ],
+                },
+            ],
+        }
+
+        result = get_properties_by_agent(adagents_data, "https://agent1.example.com")
+        assert result == []
+
+    def test_get_all_properties_builds_index_once(self):
+        """_build_domain_index runs once per file, not once per agent.
+
+        At N agents × M properties, rebuilding the index inside
+        _resolve_agent_properties is O(agents × M). This test patches the
+        helper with a counter and asserts a single invocation across a
+        file with multiple publisher_properties agents.
+        """
+        from unittest.mock import patch
+
+        from adcp import adagents as adagents_module
+
+        adagents_data = {
+            "properties": [
+                {
+                    "property_id": "p1",
+                    "publisher_domain": "site1.example",
+                    "property_type": "website",
+                    "name": "Site 1",
+                    "identifiers": [{"type": "domain", "value": "site1.example"}],
+                    "tags": ["managed"],
+                },
+                {
+                    "property_id": "p2",
+                    "publisher_domain": "site2.example",
+                    "property_type": "website",
+                    "name": "Site 2",
+                    "identifiers": [{"type": "domain", "value": "site2.example"}],
+                    "tags": ["managed"],
+                },
+            ],
+            "authorized_agents": [
+                {
+                    "url": "https://agent1.example.com",
+                    "authorization_type": "publisher_properties",
+                    "authorized_for": "A",
+                    "publisher_properties": [
+                        {
+                            "publisher_domain": "site1.example",
+                            "selection_type": "all",
+                        },
+                    ],
+                },
+                {
+                    "url": "https://agent2.example.com",
+                    "authorization_type": "publisher_properties",
+                    "authorized_for": "B",
+                    "publisher_properties": [
+                        {
+                            "publisher_domain": "site2.example",
+                            "selection_type": "all",
+                        },
+                    ],
+                },
+                {
+                    "url": "https://agent3.example.com",
+                    "authorization_type": "publisher_properties",
+                    "authorized_for": "C",
+                    "publisher_properties": [
+                        {
+                            "publisher_domain": "site1.example",
+                            "selection_type": "by_tag",
+                            "property_tags": ["managed"],
+                        },
+                    ],
+                },
+            ],
+        }
+
+        original = adagents_module._build_domain_index
+        with patch.object(
+            adagents_module,
+            "_build_domain_index",
+            side_effect=original,
+        ) as spy:
+            result = get_all_properties(adagents_data)
+
+        assert spy.call_count == 1, (
+            f"_build_domain_index called {spy.call_count} times; "
+            "expected exactly once per get_all_properties invocation"
+        )
+        # Sanity: index actually reused — all three agents resolved.
+        agent_urls = {p["agent_url"] for p in result}
+        assert agent_urls == {
+            "https://agent1.example.com",
+            "https://agent2.example.com",
+            "https://agent3.example.com",
+        }
+
     def test_get_properties_by_agent_protocol_agnostic(self):
         """Should match agent URL regardless of protocol."""
         adagents_data = {