fix(bulk): detect nested group conflicts and skip the states fetch for single-entity batches

Two nitpicks against the group-safety check, both verified against current
code:

1. _reject_operations_group_member_conflicts always fetched states before
   checking whether the batch even had 2+ distinct entities, even though a
   single-entity batch can never contain a group/member conflict. Moved the
   entity_ids computation and the size check ahead of the fetch, so the
   overwhelming majority of operations-mode calls -- which never touch a
   group at all -- no longer pay for a states round-trip they can't use.

2. _find_group_member_conflicts only checked each targeted entity's DIRECT
   members, missing a conflict reachable through a nested aggregate (an
   outer Zone group containing an inner Room group, with the batch
   targeting the outer group and a leaf that only belongs to the inner
   one) -- exactly as unsafe as targeting the inner group and that leaf
   directly, since the outer group's own service call still cascades
   through the inner one. Added a small cycle-safe transitive-membership
   walk (_expand_membership_transitively) scoped to this check, rather than
   reusing bulk_selector's _expand_entity, which raises on a dangling
   member or cycle -- the wrong contract for a best-effort batch scan.

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

Author: swissmo

src/ha_mcp/tools/tools_service.py

@@ -183,12 +183,49 @@ def _parse_bulk_operations(operations: Any) -> list[Any]:
     return operations_list
 
 
+def _expand_membership_transitively(
+    entity_id: str, states_by_id: dict[str, Any], visited: set[str]
+) -> set[str]:
+    """Return every entity reachable from ``entity_id`` via nested
+    group/aggregate membership -- direct members and, recursively, THEIR
+    members too (an outer Zone containing an inner Room group, say).
+
+    ``visited`` guards against a membership cycle; an entity already seen
+    on this walk is treated as having no further members instead of
+    recursing forever. Best-effort, not authoritative: an unknown or
+    stateless entity simply contributes no members, mirroring
+    ``_find_group_member_conflicts``'s own tolerance for a batch that
+    references something ``states`` doesn't have -- this is a safety-net
+    scan over the small set of entities in one batch, not the selector
+    resolver's own graph-integrity validation (``bulk_selector._expand_entity``),
+    which is right to raise on exactly those cases when it is actually
+    resolving a dispatch set.
+    """
+    if entity_id in visited:
+        return set()
+    visited.add(entity_id)
+    state = states_by_id.get(entity_id)
+    if state is None:
+        return set()
+    members = normalize_member_entity_ids(state.get("attributes"))
+    if not members:
+        return set()
+    expanded = set(members)
+    for member_id in members:
+        expanded.update(
+            _expand_membership_transitively(member_id, states_by_id, visited)
+        )
+    return expanded
+
+
 def _find_group_member_conflicts(
-    operations: list[Any], states: list[Any]
+    entity_ids: set[str], states: list[Any]
 ) -> dict[str, list[str]]:
     """Return ``{group_entity_id: [conflicting_member_ids]}`` for every
     group/aggregate entity this batch targets alongside one or more of its
-    own members.
+    own members -- direct or nested (a batch targeting an outer group and a
+    leaf reachable only through an inner group it contains is exactly as
+    unsafe as targeting the inner group and that leaf directly).
 
     Deliberately does not distinguish same-action duplicates from opposing
     ones (e.g. group "on" plus an explicit member "off"): Home Assistant
@@ -197,27 +234,19 @@ def _find_group_member_conflicts(
     against that fan-out, not a safe override, and gets flagged exactly
     like a same-action one.
     """
-    entity_ids = {
-        op["entity_id"]
-        for op in operations
-        if isinstance(op, dict) and isinstance(op.get("entity_id"), str)
-    }
-    if len(entity_ids) < 2:
-        return {}
     states_by_id = {
         state["entity_id"]: state
         for state in states
         if isinstance(state, dict) and isinstance(state.get("entity_id"), str)
     }
     conflicts: dict[str, list[str]] = {}
     for entity_id in sorted(entity_ids):
-        state = states_by_id.get(entity_id)
-        if state is None:
-            continue
-        members = normalize_member_entity_ids(state.get("attributes"))
-        if not members:
+        transitive_members = _expand_membership_transitively(
+            entity_id, states_by_id, set()
+        )
+        if not transitive_members:
             continue
-        overlap = sorted((entity_ids & set(members)) - {entity_id})
+        overlap = sorted((entity_ids & transitive_members) - {entity_id})
         if overlap:
             conflicts[entity_id] = overlap
     return conflicts
@@ -242,7 +271,18 @@ async def _reject_operations_group_member_conflicts(
     Fails closed on its own states-fetch failure too: an unverifiable batch
     is not a verified-safe one, and this is the same infrastructure-failure
     stance ``bulk_selector.py`` takes for the analogous selector-mode read.
+    A single-entity (or empty/all-malformed) batch cannot contain a
+    group/member conflict by construction, so the states fetch is skipped
+    entirely for it -- the overwhelming majority of operations-mode calls
+    never touch a group at all and should not pay for this check.
     """
+    entity_ids = {
+        op["entity_id"]
+        for op in operations
+        if isinstance(op, dict) and isinstance(op.get("entity_id"), str)
+    }
+    if len(entity_ids) < 2:
+        return
     try:
         states = await client.get_states()
     except ToolError:
@@ -264,7 +304,7 @@ async def _reject_operations_group_member_conflicts(
                 ],
             )
         )
-    conflicts = _find_group_member_conflicts(operations, states)
+    conflicts = _find_group_member_conflicts(entity_ids, states)
     if not conflicts:
         return
     detail = "; ".join(

tests/src/unit/test_ha_bulk_control_selector.py

@@ -242,6 +242,29 @@ async def test_existing_operations_call_shape_remains_supported() -> None:
     )
 
 
+@pytest.mark.asyncio
+async def test_operations_mode_single_entity_skips_states_fetch() -> None:
+    """A single distinct entity_id can never contain a group/member
+    conflict -- there's no second row to overlap with -- so the
+    group-safety check must not fetch states for it at all. The
+    overwhelming majority of operations-mode calls target one entity (or
+    several rows for the SAME entity) and never touch a group, and
+    shouldn't pay for a states round-trip this check can never use.
+    """
+    device_tools = MagicMock()
+    device_tools.bulk_device_control = AsyncMock(return_value={"success": True})
+    client = MagicMock()
+    client.get_states = AsyncMock(return_value=[])
+    tools = ServiceTools(client, device_tools)
+
+    result = await tools.ha_bulk_control(
+        [{"entity_id": "light.one", "action": "off"}], False
+    )
+
+    assert result == {"success": True}
+    client.get_states.assert_not_awaited()
+
+
 def _light_state(entity_id: str, members: list[str] | None = None) -> dict:
     """A minimal HA state dict; ``members`` sets the group-membership
     attribute Home Assistant (and ha_mcp.utils.entity_membership) reads as
@@ -396,14 +419,119 @@ async def test_operations_mode_rejects_opposing_action_group_member_conflict() -
     device_tools.bulk_device_control.assert_not_awaited()
 
 
+@pytest.mark.asyncio
+async def test_operations_mode_rejects_nested_group_conflict() -> None:
+    """A conflict must be caught through a NESTED aggregate, not just a
+    direct member -- an outer Zone group containing an inner Room group,
+    with the batch targeting the outer group and a leaf that only belongs
+    to the inner one. Checking only ``light.outer``'s own direct
+    ``entity_id`` attribute (``["light.inner"]``) would miss this: the
+    outer group's own service call still cascades through the inner group
+    down to the leaf, exactly as unsafe as targeting the inner group and
+    that leaf directly.
+    """
+    device_tools = MagicMock()
+    device_tools.bulk_device_control = AsyncMock(return_value={"success": True})
+    client = MagicMock()
+    client.get_states = AsyncMock(
+        return_value=[
+            _light_state("light.outer", ["light.inner"]),
+            _light_state("light.inner", ["light.leaf_a", "light.leaf_b"]),
+            _light_state("light.leaf_a"),
+            _light_state("light.leaf_b"),
+        ]
+    )
+    tools = ServiceTools(client, device_tools)
+    operations = [
+        {"entity_id": "light.outer", "action": "off"},
+        {"entity_id": "light.leaf_a", "action": "off"},
+    ]
+
+    with pytest.raises(ToolError, match="group/aggregate entity"):
+        await tools.ha_bulk_control(operations, False)
+
+    device_tools.bulk_device_control.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_operations_mode_allows_unrelated_nested_groups() -> None:
+    """Two independent group hierarchies with no shared membership must
+    not cross-contaminate -- the transitive walk for one group must not
+    leak into flagging an entity that only belongs to a completely
+    different group's tree.
+    """
+    device_tools = MagicMock()
+    device_tools.bulk_device_control = AsyncMock(return_value={"success": True})
+    client = MagicMock()
+    client.get_states = AsyncMock(
+        return_value=[
+            _light_state("light.outer_a", ["light.inner_a"]),
+            _light_state("light.inner_a", ["light.leaf_a"]),
+            _light_state("light.leaf_a"),
+            _light_state("light.outer_b", ["light.inner_b"]),
+            _light_state("light.inner_b", ["light.leaf_b"]),
+            _light_state("light.leaf_b"),
+        ]
+    )
+    tools = ServiceTools(client, device_tools)
+    operations = [
+        {"entity_id": "light.outer_a", "action": "off"},
+        {"entity_id": "light.leaf_b", "action": "off"},
+    ]
+
+    result = await tools.ha_bulk_control(operations, False)
+
+    assert result == {"success": True}
+    device_tools.bulk_device_control.assert_awaited_once_with(
+        operations=operations, parallel=False, ctx=None
+    )
+
+
+@pytest.mark.asyncio
+async def test_operations_mode_cyclic_membership_does_not_hang() -> None:
+    """A membership cycle (A lists B as a member, B lists A back) must not
+    hang or crash the group-safety check -- unlike ``bulk_selector``'s own
+    selector-mode expansion (which is resolving an authoritative dispatch
+    set and is right to raise on a cycle), this is a best-effort safety
+    scan over one batch and should degrade gracefully instead.
+    """
+    device_tools = MagicMock()
+    device_tools.bulk_device_control = AsyncMock(return_value={"success": True})
+    client = MagicMock()
+    client.get_states = AsyncMock(
+        return_value=[
+            _light_state("light.a", ["light.b"]),
+            _light_state("light.b", ["light.a"]),
+            _light_state("light.c"),
+        ]
+    )
+    tools = ServiceTools(client, device_tools)
+    operations = [
+        {"entity_id": "light.a", "action": "off"},
+        {"entity_id": "light.c", "action": "off"},
+    ]
+
+    result = await tools.ha_bulk_control(operations, False)
+
+    assert result == {"success": True}
+    device_tools.bulk_device_control.assert_awaited_once_with(
+        operations=operations, parallel=False, ctx=None
+    )
+
+
 @pytest.mark.asyncio
 async def test_operations_mode_group_safety_check_fails_closed_on_states_error() -> (
     None
 ):
     """A states-fetch failure during the group-safety check must not
     silently skip the check and dispatch anyway: an unverifiable batch is
     not a verified-safe one, matching the fail-closed stance
-    bulk_selector.py takes for the analogous selector-mode read."""
+    bulk_selector.py takes for the analogous selector-mode read.
+
+    Two distinct entities: a single-entity batch short-circuits before the
+    states fetch (see test_operations_mode_single_entity_skips_states_fetch),
+    so it would never reach this code path.
+    """
     device_tools = MagicMock()
     device_tools.bulk_device_control = AsyncMock(return_value={"success": True})
     client = MagicMock()
@@ -412,7 +540,11 @@ async def test_operations_mode_group_safety_check_fails_closed_on_states_error()
 
     with pytest.raises(ToolError):
         await tools.ha_bulk_control(
-            [{"entity_id": "light.one", "action": "off"}], False
+            [
+                {"entity_id": "light.one", "action": "off"},
+                {"entity_id": "light.two", "action": "off"},
+            ],
+            False,
         )
 
     device_tools.bulk_device_control.assert_not_awaited()
@@ -433,7 +565,11 @@ async def test_operations_mode_group_safety_check_fails_closed_on_malformed_stat
 
     with pytest.raises(ToolError, match="Home Assistant"):
         await tools.ha_bulk_control(
-            [{"entity_id": "light.one", "action": "off"}], False
+            [
+                {"entity_id": "light.one", "action": "off"},
+                {"entity_id": "light.two", "action": "off"},
+            ],
+            False,
         )
 
     device_tools.bulk_device_control.assert_not_awaited()
@@ -463,7 +599,11 @@ async def test_operations_mode_group_safety_check_propagates_tool_error_unchange
 
     with pytest.raises(ToolError) as exc_info:
         await tools.ha_bulk_control(
-            [{"entity_id": "light.one", "action": "off"}], False
+            [
+                {"entity_id": "light.one", "action": "off"},
+                {"entity_id": "light.two", "action": "off"},
+            ],
+            False,
         )
 
     assert str(exc_info.value) == str(original)