fix(bulk): address selector review feedback

Author: swissmo

src/ha_mcp/policy/middleware.py

@@ -64,6 +64,11 @@ def _passes_ungated(name: str, args: dict[str, Any]) -> bool:
     return name in PROXY_META_TOOLS or _is_approval_management(name, args)
 
 
+def _has_dynamic_selector_targets(name: str, args: dict[str, Any]) -> bool:
+    """Return whether identical arguments can resolve to different targets later."""
+    return name == "ha_bulk_control" and args.get("selector") is not None
+
+
 class PolicyMiddleware(Middleware):
     """Gate tool calls against a Policy, blocking with progress heartbeats."""
 
@@ -120,15 +125,19 @@ async def on_call_tool(
 
         rule = find_matching_rule(name, args, policy)
         args_hash = compute_args_hash(args)
+        dynamic_targets = _has_dynamic_selector_targets(name, args)
+        remember_minutes = (
+            0 if dynamic_targets else rule.remember_minutes if rule else 0
+        )
 
-        if self._queue.is_remembered(name, args_hash):
+        if not dynamic_targets and self._queue.is_remembered(name, args_hash):
             return await call_next(context)
 
         existing = self._queue.find(name, args_hash)
         if existing and existing.decision == "approved":
             self._queue.consume_and_maybe_remember(
                 existing,
-                remember_minutes=rule.remember_minutes if rule else 0,
+                remember_minutes=remember_minutes,
             )
             return await call_next(context)
         if existing and existing.decision == "denied":
@@ -155,7 +164,7 @@ async def on_call_tool(
         if pending.decision == "approved":
             self._queue.consume_and_maybe_remember(
                 pending,
-                remember_minutes=rule.remember_minutes if rule else 0,
+                remember_minutes=remember_minutes,
             )
             return await call_next(context)
         if pending.decision == "denied":

src/ha_mcp/tools/bulk_selector.py

@@ -96,9 +96,20 @@ def _expand_entity(
     *,
     path: tuple[str, ...],
     expanded_groups: set[str],
+    cache: dict[str, tuple[frozenset[str], frozenset[str]]],
     parameter: str,
 ) -> set[str]:
     """Expand generic membership recursively and reject incomplete graph walks."""
+    if entity_id in path:
+        cycle_start = path.index(entity_id)
+        cycle = " -> ".join((*path[cycle_start:], entity_id))
+        raise BulkSelectorValidationError(
+            f"Aggregate membership cycle detected: {cycle}"
+        )
+    if cached := cache.get(entity_id):
+        cached_leaves, cached_groups = cached
+        expanded_groups.update(cached_groups)
+        return set(cached_leaves)
     state = states.get(entity_id)
     if state is None:
         raise BulkSelectorValidationError(
@@ -107,27 +118,29 @@ def _expand_entity(
         )
     members = normalize_member_entity_ids(state.get("attributes"))
     if members is None:
+        cache[entity_id] = (frozenset({entity_id}), frozenset())
         return {entity_id}
-    if entity_id in path:
-        cycle_start = path.index(entity_id)
-        cycle = " -> ".join((*path[cycle_start:], entity_id))
-        raise BulkSelectorValidationError(
-            f"Aggregate membership cycle detected: {cycle}"
-        )
     expanded_groups.add(entity_id)
     if not members:
+        cache[entity_id] = (frozenset(), frozenset({entity_id}))
         return set()
     leaves: set[str] = set()
+    nested_groups: set[str] = {entity_id}
     for member_id in members:
+        member_groups: set[str] = set()
         leaves.update(
             _expand_entity(
                 member_id,
                 states,
                 path=(*path, entity_id),
-                expanded_groups=expanded_groups,
+                expanded_groups=member_groups,
+                cache=cache,
                 parameter=parameter,
             )
         )
+        nested_groups.update(member_groups)
+    expanded_groups.update(nested_groups)
+    cache[entity_id] = (frozenset(leaves), frozenset(nested_groups))
     return leaves
 
 
@@ -149,7 +162,7 @@ def _validate_selector(
             parameter="selector.domain",
         )
     domain = domain.strip().lower()
-    normalized_action = action.strip() if isinstance(action, str) else ""
+    normalized_action = action.strip().lower() if isinstance(action, str) else ""
     valid_actions = get_domain_handler(domain).get(
         "valid_actions", ["on", "off", "toggle"]
     )
@@ -214,6 +227,7 @@ def _expand_roots(
     roots: list[str],
     states: Mapping[str, Mapping[str, Any]],
     expanded_groups: set[str],
+    cache: dict[str, tuple[frozenset[str], frozenset[str]]],
     parameter: str,
 ) -> set[str]:
     """Expand a list of aggregate or leaf roots into a deduplicated leaf set."""
@@ -225,6 +239,7 @@ def _expand_roots(
                 states,
                 path=(),
                 expanded_groups=expanded_groups,
+                cache=cache,
                 parameter=parameter,
             )
         )
@@ -315,21 +330,24 @@ async def resolve_bulk_selector(
     hidden = await _load_hidden_entities(
         client, entity_result, states_result, device_result, entity_registry
     )
-    candidate_roots = sorted(
+    matching_roots = {
         entity_id
         for entity_id in states
-        if entity_id not in hidden
-        and entity_id.startswith(f"{domain}.")
+        if entity_id.startswith(f"{domain}.")
         and _entity_area_id(entity_id, entity_registry, device_areas) in selected_areas
-    )
+    }
+    directly_hidden = matching_roots & hidden
+    candidate_roots = sorted(matching_roots - hidden)
     expanded_groups: set[str] = set()
+    expansion_cache: dict[str, tuple[frozenset[str], frozenset[str]]] = {}
     selected_leaves = _expand_roots(
-        candidate_roots, states, expanded_groups, "selector"
+        candidate_roots, states, expanded_groups, expansion_cache, "selector"
     )
     excluded_leaves = _expand_roots(
         excluded_roots,
         states,
         expanded_groups,
+        expansion_cache,
         "selector.exclude_entity_ids",
     )
     selected_leaves = {
@@ -367,5 +385,7 @@ async def resolve_bulk_selector(
         excluded_entity_ids=sorted(effective_excluded - hidden),
         selected_area_ids=sorted(selected_areas),
         expanded_group_ids=sorted(expanded_groups - hidden),
-        hidden_entity_count=len(hidden_selected | (effective_excluded & hidden)),
+        hidden_entity_count=len(
+            directly_hidden | hidden_selected | (effective_excluded & hidden)
+        ),
     )

src/ha_mcp/tools/tools_service.py

@@ -1657,6 +1657,10 @@ async def ha_bulk_control(
                 )
                 if not isinstance(parsed_selector, dict):
                     raise BulkSelectorValidationError("selector must be a JSON object")
+            except (ValueError, BulkSelectorValidationError) as exc:
+                parameter = getattr(exc, "parameter", "selector")
+                raise_tool_error(create_validation_error(str(exc), parameter=parameter))
+            try:
                 resolution = await resolve_bulk_selector(
                     self._client,
                     parsed_selector,
@@ -1665,9 +1669,15 @@ async def ha_bulk_control(
                     timeout_seconds=timeout_seconds,
                     validate_first=validate_first,
                 )
-            except (ValueError, BulkSelectorValidationError) as exc:
+            except BulkSelectorValidationError as exc:
                 parameter = getattr(exc, "parameter", "selector")
                 raise_tool_error(create_validation_error(str(exc), parameter=parameter))
+            except Exception as exc:
+                exception_to_structured_error(
+                    exc,
+                    context={"operation": "resolve bulk selector"},
+                )
+                raise  # unreachable: exception_to_structured_error always raises
             if dry_run:
                 return {
                     "success": True,

tests/src/e2e/workflows/core/test_bulk.py

@@ -13,7 +13,13 @@
 
 import pytest
 
-from ...utilities.assertions import assert_mcp_success, parse_mcp_result, safe_call_tool
+from ...utilities.assertions import (
+    MCPAssertions,
+    assert_mcp_success,
+    parse_mcp_result,
+    safe_call_tool,
+)
+from ...utilities.wait_helpers import wait_for_entity_state
 
 logger = logging.getLogger(__name__)
 
@@ -46,56 +52,78 @@ def _extract_bulk_boolean_entity_id(data: dict) -> str | None:
 class TestBulkControl:
     """Test ha_bulk_control tool functionality."""
 
-    async def test_selector_dry_run_resolves_area_and_exclusion(
-        self, mcp_client, cleanup_tracker
-    ):
+    async def test_selector_dry_run_resolves_area_and_exclusion(self, mcp_client):
         """Preview exact area leaves after applying an entity exclusion."""
         suffix = uuid4().hex[:8]
-        area_result = await mcp_client.call_tool(
-            "ha_set_area_or_floor",
-            {"kind": "area", "name": f"Bulk selector {suffix}"},
-        )
-        area_data = assert_mcp_success(area_result, "Create selector area")
-        area_id = area_data["area_id"]
-        cleanup_tracker.track("area", area_id)
-
-        entity_ids = []
-        for label in ("included", "excluded"):
-            create_result = await mcp_client.call_tool(
-                "ha_config_set_helper",
-                {
-                    "helper_type": "input_boolean",
-                    "name": f"Bulk selector {label} {suffix}",
-                },
-            )
-            create_data = assert_mcp_success(create_result, "Create selector helper")
-            entity_id = _extract_bulk_boolean_entity_id(create_data)
-            assert entity_id, f"Missing helper entity_id: {create_data}"
-            cleanup_tracker.track("input_boolean", entity_id)
-            entity_ids.append(entity_id)
-            assign_result = await mcp_client.call_tool(
-                "ha_set_entity", {"entity_id": entity_id, "area_id": area_id}
+        area_id: str | None = None
+        entity_ids: list[str] = []
+        try:
+            area_result = await mcp_client.call_tool(
+                "ha_set_area_or_floor",
+                {"kind": "area", "name": f"Bulk selector {suffix}"},
             )
-            assert_mcp_success(assign_result, "Assign selector helper to area")
+            area_data = assert_mcp_success(area_result, "Create selector area")
+            area_id = area_data["area_id"]
+
+            for label in ("included", "excluded"):
+                create_result = await mcp_client.call_tool(
+                    "ha_config_set_helper",
+                    {
+                        "helper_type": "input_boolean",
+                        "name": f"Bulk selector {label} {suffix}",
+                    },
+                )
+                create_data = assert_mcp_success(
+                    create_result, "Create selector helper"
+                )
+                entity_id = _extract_bulk_boolean_entity_id(create_data)
+                assert entity_id, f"Missing helper entity_id: {create_data}"
+                entity_ids.append(entity_id)
+                assign_result = await mcp_client.call_tool(
+                    "ha_set_entity", {"entity_id": entity_id, "area_id": area_id}
+                )
+                assert_mcp_success(assign_result, "Assign selector helper to area")
 
-        result = await mcp_client.call_tool(
-            "ha_bulk_control",
-            {
-                "selector": {
-                    "domain": "input_boolean",
-                    "area_ids": [area_id],
-                    "exclude_entity_ids": [entity_ids[1]],
-                },
-                "action": "off",
-                "dry_run": True,
-            },
-        )
-        data = assert_mcp_success(result, "Preview structural bulk selection")
+            for entity_id in entity_ids:
+                assert await wait_for_entity_state(mcp_client, entity_id, "off"), (
+                    f"Selector helper {entity_id} was not registered in time"
+                )
 
-        assert data["dry_run"] is True
-        assert data["dispatched"] is False
-        assert data["resolution"]["resolved_entity_ids"] == [entity_ids[0]]
-        assert data["resolution"]["excluded_entity_ids"] == [entity_ids[1]]
+            async with MCPAssertions(mcp_client) as mcp:
+                data = await mcp.call_tool_success(
+                    "ha_bulk_control",
+                    {
+                        "selector": {
+                            "domain": "input_boolean",
+                            "area_ids": [area_id],
+                            "exclude_entity_ids": [entity_ids[1]],
+                        },
+                        "action": "off",
+                        "dry_run": True,
+                    },
+                )
+
+            assert data["dry_run"] is True
+            assert data["dispatched"] is False
+            assert data["resolution"]["resolved_entity_ids"] == [entity_ids[0]]
+            assert data["resolution"]["excluded_entity_ids"] == [entity_ids[1]]
+        finally:
+            for entity_id in entity_ids:
+                await safe_call_tool(
+                    mcp_client,
+                    "ha_remove_helpers_integrations",
+                    {
+                        "helper_type": "input_boolean",
+                        "target": entity_id,
+                        "confirm": True,
+                    },
+                )
+            if area_id is not None:
+                await safe_call_tool(
+                    mcp_client,
+                    "ha_remove_area_or_floor",
+                    {"kind": "area", "id": area_id},
+                )
 
     async def test_bulk_turn_on_single_light(self, mcp_client, test_light_entity):
         """Test bulk_control with a single light entity."""

tests/src/e2e/workflows/groups/test_lifecycle.py

@@ -95,10 +95,14 @@ async def test_group_basic_lifecycle(self, mcp_client, cleanup_tracker):
                 },
             )
             search_record = next(
-                entity
-                for entity in search_data["entities"]
-                if entity["entity_id"] == f"group.{object_id}"
-            )
+                (
+                    entity
+                    for entity in search_data["entities"]
+                    if entity["entity_id"] == f"group.{object_id}"
+                ),
+                None,
+            )
+            assert search_record is not None, search_data
             assert search_record["is_group"] is True
             assert search_record["member_entity_ids"] == [
                 "light.bed_light",

tests/src/unit/policy/test_evaluator.py

@@ -416,7 +416,7 @@ def test_operations_rule_force_gates_selector(self):
                         Predicate(
                             path="args.operations.*.entity_id",
                             op="regex",
-                            value=r"^lock\\.",
+                            value=r"^lock\.",
                         )
                     ],
                 )
@@ -456,7 +456,7 @@ def test_operations_calls_keep_normal_predicate_semantics(self):
                         Predicate(
                             path="args.operations.*.entity_id",
                             op="regex",
-                            value=r"^lock\\.",
+                            value=r"^lock\.",
                         )
                     ],
                 )
@@ -471,3 +471,11 @@ def test_operations_calls_keep_normal_predicate_semantics(self):
             )
             == Verdict.ALLOW
         )
+        assert (
+            evaluate(
+                "ha_bulk_control",
+                {"operations": [{"entity_id": "lock.front", "action": "lock"}]},
+                policy,
+            )
+            == Verdict.REQUIRE_APPROVAL
+        )