fix(bulk): address second round of selector review feedback

Codex's re-review of the structural-selector feature (PR #2246) surfaced
four remaining gaps:

- bulk_selector: a cross-domain aggregate (e.g. group.living_room_lights)
  assigned to a selected area was excluded from root candidates before
  membership expansion, so its ungrouped target-domain members were never
  discovered.
- bulk_selector: a device-registry entry missing `id` was silently
  skipped by the strict visibility resolver, letting a device-inherited
  exclusion drop out of the hidden set instead of failing closed (mirrors
  the per-entry validation in visibility/enforcement.py).
- policy/middleware: concurrent identical selector calls shared one
  pending approval via find_or_create, so a single approval click could
  release both waiters and dispatch the selector twice.
- policy/evaluator: the selector fail-safe gated every ha_bulk_control
  selector call whenever any rule named the tool, even when that rule's
  predicates were fully selector-inspectable and simply didn't match —
  defeating a deliberately conditional rule.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: swissmo

src/ha_mcp/policy/evaluator.py

@@ -142,6 +142,25 @@ def find_matching_rule(
     return None
 
 
+def _rule_needs_resolved_operations(rule: Rule) -> bool:
+    """Whether ``rule`` inspects fields only known after selector resolution.
+
+    A selector-mode ``ha_bulk_control`` call carries ``args.selector``, not
+    ``args.operations`` — the leaf targets don't exist yet, they're resolved
+    inside the tool after this middleware runs. A rule predicate path rooted
+    at ``args.operations`` can therefore never match a selector call via
+    ordinary evaluation and needs the fail-safe below. A rule whose
+    predicates only reference selector-inspectable fields (e.g.
+    ``args.selector.domain``) already got a fair, precise match attempt in
+    ``find_matching_rule`` and must not be broadened into an unconditional
+    gate.
+    """
+    return any(
+        p.path == "args.operations" or p.path.startswith("args.operations.")
+        for p in rule.when
+    )
+
+
 def evaluate(tool_name: str, args: dict[str, Any], policy: Policy) -> Verdict:
     if find_matching_rule(tool_name, args, policy) is not None:
         return Verdict.REQUIRE_APPROVAL
@@ -165,12 +184,19 @@ def evaluate(tool_name: str, args: dict[str, Any], policy: Policy) -> Verdict:
         return Verdict.REQUIRE_APPROVAL
     # Structural selectors are resolved inside the tool, after this middleware.
     # A pre-existing rule that inspects args.operations.* therefore cannot inspect
-    # the eventual leaf targets. Fail safe whenever an operator configured any
-    # rule applicable to ha_bulk_control; selector-aware rules still match above.
+    # the eventual leaf targets. Fail safe only for rules that actually depend on
+    # that unresolved data — a rule fully expressed over selector-inspectable
+    # fields (e.g. args.selector.domain) already had its precise shot at matching
+    # above, and broadening it here would defeat a deliberately conditional rule
+    # (a rule scoped to selector.domain == "lock" must not gate a "light" call).
     if (
         tool_name == "ha_bulk_control"
         and args.get("selector") is not None
-        and any(rule.tool_name in ("ha_bulk_control", "*") for rule in policy.rules)
+        and any(
+            rule.tool_name in ("ha_bulk_control", "*")
+            and _rule_needs_resolved_operations(rule)
+            for rule in policy.rules
+        )
     ):
         return Verdict.REQUIRE_APPROVAL
     return Verdict.ALLOW

src/ha_mcp/policy/middleware.py

@@ -146,12 +146,23 @@ async def on_call_tool(
 
         # find_or_create serialises the create — two concurrent calls with
         # the same args_hash share one pending entry, so the user only sees
-        # one approval row and approving it releases every waiter.
-        pending = await self._queue.find_or_create(
-            name,
-            args_hash,
-            args,
-            ttl_minutes=policy.approval_ttl_minutes,
+        # one approval row and approving it releases every waiter. Dynamic
+        # selector calls must NOT share a pending entry: two concurrent
+        # identical calls can still resolve to different (or overlapping)
+        # target sets by the time each one dispatches, so folding them onto
+        # one approval would let a single click authorize more executions
+        # than the user saw. Skip the sharing and always mint a fresh entry.
+        pending = (
+            self._queue.create(
+                name, args_hash, args, ttl_minutes=policy.approval_ttl_minutes
+            )
+            if dynamic_targets
+            else await self._queue.find_or_create(
+                name,
+                args_hash,
+                args,
+                ttl_minutes=policy.approval_ttl_minutes,
+            )
         )
 
         wait = (

src/ha_mcp/tools/bulk_selector.py

@@ -62,6 +62,24 @@ def _registry_rows(result: Any, label: str) -> list[dict[str, Any]]:
     return rows
 
 
+def _validate_device_registry_rows(device_rows: list[dict[str, Any]]) -> None:
+    """Fail closed on a device row missing ``id``.
+
+    ``visibility.resolver._parse_device_registry`` silently skips any device
+    entry without an ``id`` even when the hidden-set resolver runs
+    ``strict=True`` (see ``_load_hidden_entities``), so a malformed row would
+    otherwise let a device-derived exclusion (area/label inherited by its
+    entities) drop out of the hidden set without warning. Mirrors the
+    per-entry validation ``_refresh_hidden_set`` in
+    ``visibility/enforcement.py`` applies before calling the same strict
+    resolver.
+    """
+    if any(not row.get("id") for row in device_rows):
+        raise BulkSelectorValidationError(
+            "Home Assistant device registry returned a malformed entry"
+        )
+
+
 def _string_list(selector: Mapping[str, Any], key: str) -> list[str]:
     """Read one optional exact-ID list without accepting scalar coercion."""
     raw = selector.get(key, [])
@@ -308,6 +326,7 @@ async def resolve_bulk_selector(
         )
     entity_rows = _registry_rows(entity_result, "entity registry")
     device_rows = _registry_rows(device_result, "device registry")
+    _validate_device_registry_rows(device_rows)
     selected_areas = _select_area_ids(
         _registry_rows(area_result, "area registry"),
         _registry_rows(floor_result, "floor registry"),
@@ -330,11 +349,20 @@ async def resolve_bulk_selector(
     hidden = await _load_hidden_entities(
         client, entity_result, states_result, device_result, entity_registry
     )
+    # A root only needs to LIVE in the selected area; it does not need to be
+    # of the target domain itself. A `group`/other-domain aggregate assigned
+    # to the area (e.g. `group.living_room_lights`) is included here too, so
+    # its membership expansion below can surface `light.*` leaves that have
+    # no individual area assignment of their own. The target-domain filter
+    # is re-applied to the expanded leaves further down.
     matching_roots = {
         entity_id
-        for entity_id in states
-        if entity_id.startswith(f"{domain}.")
-        and _entity_area_id(entity_id, entity_registry, device_areas) in selected_areas
+        for entity_id, state in states.items()
+        if _entity_area_id(entity_id, entity_registry, device_areas) in selected_areas
+        and (
+            entity_id.startswith(f"{domain}.")
+            or normalize_member_entity_ids(state.get("attributes")) is not None
+        )
     }
     directly_hidden = matching_roots & hidden
     candidate_roots = sorted(matching_roots - hidden)

tests/src/unit/policy/test_evaluator.py

@@ -479,3 +479,40 @@ def test_operations_calls_keep_normal_predicate_semantics(self):
             )
             == Verdict.REQUIRE_APPROVAL
         )
+
+    def test_nonmatching_selector_only_rule_stays_allowed(self):
+        """A rule fully expressed over selector fields keeps its condition.
+
+        The fail-safe must only widen rules that depend on unresolved
+        ``args.operations`` data. A rule scoped to
+        ``args.selector.domain == "lock"`` already got a precise match
+        attempt in ``find_matching_rule``; a "light" selector call must not
+        be swept into approval just because a same-named rule exists.
+        """
+        policy = Policy(
+            rules=[
+                Rule(
+                    tool_name="ha_bulk_control",
+                    when=[
+                        Predicate(path="args.selector.domain", op="eq", value="lock")
+                    ],
+                )
+            ]
+        )
+
+        assert (
+            evaluate(
+                "ha_bulk_control",
+                {"selector": {"domain": "light", "area_ids": ["salon"]}},
+                policy,
+            )
+            == Verdict.ALLOW
+        )
+        assert (
+            evaluate(
+                "ha_bulk_control",
+                {"selector": {"domain": "lock", "area_ids": ["salon"]}},
+                policy,
+            )
+            == Verdict.REQUIRE_APPROVAL
+        )

tests/src/unit/policy/test_middleware.py

@@ -140,6 +140,43 @@ async def test_dynamic_bulk_selector_approval_is_not_remembered(queue):
     assert not queue.is_remembered("ha_bulk_control", args_hash)
 
 
+@pytest.mark.anyio
+async def test_concurrent_dynamic_selector_calls_get_independent_approvals(queue):
+    """Two concurrent identical selector calls must not share one pending entry.
+
+    Sharing (via ``find_or_create``'s dedup) would let a single approval
+    click release both waiters, so one click could dispatch the selector
+    twice — or against two different resolutions if topology changed
+    between them. Each invocation must mint its own pending approval.
+    """
+    args = {
+        "selector": {"domain": "lock", "area_ids": ["entry"]},
+        "action": "lock",
+    }
+    policy = Policy(rules=[Rule(tool_name="ha_bulk_control")])
+    middleware = PolicyMiddleware(
+        policy_provider=lambda: policy,
+        queue=queue,
+        wait_seconds=0,
+    )
+    call_next = AsyncMock()
+
+    async def call():
+        with pytest.raises(ToolError):
+            await middleware.on_call_tool(
+                make_context("ha_bulk_control", dict(args)), call_next
+            )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(call)
+        tg.start_soon(call)
+
+    pending = queue.list_pending()
+    assert len(pending) == 2
+    assert pending[0].token != pending[1].token
+    call_next.assert_not_awaited()
+
+
 @pytest.mark.anyio
 async def test_pre_approved_entry_consumed_and_call_proceeds(queue):
     pol = Policy(rules=[Rule(tool_name="ha_call_service")])

tests/src/unit/test_bulk_selector.py

@@ -233,6 +233,69 @@ async def test_directly_hidden_matching_root_is_counted(
     assert result.hidden_entity_count == 1
 
 
+@pytest.mark.asyncio
+async def test_cross_domain_aggregate_root_expands_into_target_domain_leaves() -> None:
+    """An area-assigned aggregate of a different domain still expands.
+
+    ``group.living_room_lights`` is not itself a ``light.*`` entity, so it
+    must not be excluded from the root candidates just because its own
+    domain doesn't match the selector. Its ungrouped ``light.*`` members
+    (no individual area assignment) must still be discovered via membership
+    expansion, then filtered to the requested domain.
+    """
+    client = SelectorClient(
+        states=[
+            _state("group.living_room_lights", ["light.ceiling", "light.lamp"]),
+            _state("light.ceiling"),
+            _state("light.lamp"),
+        ],
+        entities=[
+            {"entity_id": "group.living_room_lights", "area_id": "salon"},
+            {"entity_id": "light.ceiling", "area_id": None},
+            {"entity_id": "light.lamp", "area_id": None},
+        ],
+    )
+
+    result = await resolve_bulk_selector(
+        client,
+        {"domain": "light", "area_ids": ["salon"]},
+        action="off",
+        parameters=None,
+        timeout_seconds=None,
+        validate_first=True,
+    )
+
+    assert result.resolved_entity_ids == ["light.ceiling", "light.lamp"]
+
+
+@pytest.mark.asyncio
+async def test_device_registry_entry_missing_id_fails_closed() -> None:
+    """A malformed device row must fail closed, not silently drop a device.
+
+    ``visibility.resolver._parse_device_registry`` skips a device entry
+    without an ``id`` even under the strict resolver, so a well-formed
+    device registry is what keeps a device-derived hidden dimension from
+    silently dropping out. This mirrors the per-entry validation
+    ``_refresh_hidden_set`` in ``visibility/enforcement.py`` applies before
+    calling the same strict resolver.
+    """
+    client = SelectorClient(
+        states=[_state("light.one")],
+        entities=[{"entity_id": "light.one", "area_id": "salon"}],
+        devices=[{"area_id": "salon"}],
+    )
+
+    with pytest.raises(BulkSelectorValidationError, match="device registry"):
+        await resolve_bulk_selector(
+            client,
+            {"domain": "light", "area_ids": ["salon"]},
+            action="off",
+            parameters=None,
+            timeout_seconds=None,
+            validate_first=True,
+        )
+
+
 @pytest.mark.asyncio
 async def test_action_is_normalized_case_insensitively() -> None:
     """Action normalization accepts surrounding whitespace and mixed case."""