fix(bulk): close review round-4's 6 blocking gaps plus 11 smaller findings

Blocking:
- Stop skipping scene entities in the operations-mode conflict gate: a
  scene dispatch really does cascade into its configured entities
  (scene.turn_on), unlike selector mode's aggregate-root question, which
  is why that skip existed there in the first place. The two now answer
  their own, different questions instead of sharing one wrong exemption.
- Split bulk_selector's empty-result gate into three causes instead of
  two (empty aggregate / wrong-domain aggregate / hidden), each gated on
  "nothing in the match set is hidden" -- a hidden entity elsewhere in
  the same area no longer gets masked by an unrelated visible aggregate
  expanding to the wrong domain.
- Fix the selector-only-parameter worked example rendering a duplicate
  'action' key when action is itself the offending parameter.
- Cap the group-conflict message's member list at 10 (+N more) instead
  of enumerating an instance-sized membership list in a fail-closed
  error whose remedy sentence is the entire value of the response.
- Cover both _GroupConflict message branches with tests that would catch
  the two fields being swapped (previously only the unlisted-members
  branch was ever exercised).

Smaller:
- Bind topology fetch-failure labels directly to their named locals
  instead of a position-zipped tuple, so a future reordered gather()
  can't silently mislabel which registry failed.
- _operation_common needs hash=False, not just compare=True: the frozen
  dataclass claimed to be hashable (mypy agreed) but raised at the first
  real hash() call, since a dict has none.
- Give the args-too-deeply-nested fail-closed branch its own error code
  (POLICY_ARGS_TOO_DEEPLY_NESTED) instead of reusing POLICY_LOAD_FAILED,
  which already means a corrupt tool_policy.json -- a different failure
  with a different, caller-shaped remedy.
- Log the dynamic pending-entry removal at INFO; it was the only silent
  state transition on the approval queue, and a late Approve click on it
  used to log as an "attacker probing tokens" WARNING for ordinary use.
- Restore actionable suggestions on the group-safety check's own
  states-fetch failure, lost when the isinstance(states, list) guard was
  removed -- otherwise the agent sees a raw client string with no hint
  that the safety check itself is what failed.
- Deep-copy operation parameters (both the per-row and resolution-level
  copies) instead of a shallow dict() that only stops top-level rebinding
  and still shares nested values like an rgb_color list.
- Fix an inverted test docstring describing the guard it tests backwards.
- Assert expires_in_seconds is absent from the dynamic pending-error body,
  not just that its message/suggestions text omits a countdown.
- Extend the topology multi-failure test to all five registries instead
  of two, closing the remaining label-typo blind spot.
- E2E test: use safe_call_tool for cleanup (so a cleanup failure can't
  mask a real assertion failure in finally), and discover real lights
  dynamically instead of hardcoding demo-platform entity IDs.

Skipped: a CodeRabbit suggestion to replace the established
MCPAssertions.call_tool_failure helper (used across 13 e2e files) with a
manual safe_call_tool call in the new E2E test -- no functional benefit,
and goes against the repo's own convention. Also skipped narrowing
ServiceTools' client: Any typing (untested seam around a dict-shaped
get_states() response) -- reviewer confirmed this is not a live gap, and
the fix would mean a broader typing change across ServiceTools,
register_service_tools, and registry.py.

10722 unit tests pass (up from 10716 by exactly the 6 new tests added
here); the 26 remaining failures are the same pre-existing,
Windows-platform-only failures already present before this change.

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

Author: swissmo

src/ha_mcp/errors.py

@@ -100,6 +100,16 @@ class ErrorCode(StrEnum):
     USER_DENIED = "USER_DENIED"
     POLICY_LOAD_FAILED = "POLICY_LOAD_FAILED"
 
+    # Distinct from POLICY_LOAD_FAILED above: the policy file itself loaded
+    # fine, but this call's own arguments were too deeply nested to
+    # evaluate against it safely (see normalize_stringified_containers'
+    # RecursionError handling in middleware.py). A different failure with a
+    # different, caller-shaped remedy ("reduce the nesting depth and
+    # retry") -- sharing POLICY_LOAD_FAILED would make the two
+    # indistinguishable to anything grouping on the code (a dashboard, an
+    # operator grepping logs after "my policy broke").
+    POLICY_ARGS_TOO_DEEPLY_NESTED = "POLICY_ARGS_TOO_DEEPLY_NESTED"
+
     # Read Only Mode (discussion #1569). A write operation was blocked
     # because the server-wide Read Only Mode toggle is on.
     READ_ONLY_MODE = "READ_ONLY_MODE"

src/ha_mcp/policy/middleware.py

@@ -154,7 +154,7 @@ async def on_call_tool(
             )
             raise_tool_error(
                 create_error_response(
-                    ErrorCode.POLICY_LOAD_FAILED,
+                    ErrorCode.POLICY_ARGS_TOO_DEEPLY_NESTED,
                     "This call's arguments are nested too deeply to "
                     "evaluate safely against the security policy.",
                     suggestions=[
@@ -287,6 +287,23 @@ def _finalize_timed_out_pending(
         """
         if dynamic_targets:
             self._queue.remove(pending.token)
+            # INFO, matching the reissue log below: silently removing this
+            # is otherwise the only state transition in this file that
+            # leaves no trace, on a queue whose whole purpose is auditable
+            # human approval. Without it, a user who clicks Approve a
+            # moment after this call's wait window closed -- an ordinary
+            # sequence, not an attack -- lands on
+            # ApprovalQueue.approve's "unknown token" WARNING, which is
+            # explicitly documented there as meaning a UI bug, a stale
+            # tab, OR an attacker probing tokens. This line is what lets
+            # an operator tell those apart.
+            logger.info(
+                "policy middleware: dynamic pending token %s removed after "
+                "its single-use wait window closed for tool=%s; a late "
+                "Approve click on it will log as an unknown token",
+                pending.token,
+                name,
+            )
             return pending
         if self._queue.get(pending.token) is None:
             old_token = pending.token

src/ha_mcp/tools/bulk_selector.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import logging
 from collections.abc import Mapping
 from dataclasses import dataclass, field
@@ -121,18 +122,25 @@ class BulkSelectorResolution:
     expanded_group_ids: tuple[str, ...]
     hidden_entity_count: int
     warnings: tuple[str, ...] = field(default_factory=tuple)
-    # repr=False only (compare stays at its True default): `_operation_common`
-    # can carry a lock or alarm code (selector mode's `parameters` argument,
-    # e.g. `lock.open` with a keypad code), so the default dataclass __repr__
-    # would otherwise write it to logs/tracebacks anywhere a resolution is
-    # printed or logged unredacted. It is NOT compare-excludable, though: it
-    # holds `action`, so two resolutions over the identical entity set but
-    # OPPOSITE actions must not compare equal, and `compare=False` would make
-    # them -- there is no dispatch-irrelevant reading of this field despite
-    # its name. The trade-off this accepts (unchanged from before repr=False
-    # was added) is that the frozen dataclass stays unhashable, since a plain
-    # `dict` has no __hash__; nothing hashes a resolution today.
-    _operation_common: dict[str, Any] = field(default_factory=dict, repr=False)
+    # repr=False: `_operation_common` can carry a lock or alarm code
+    # (selector mode's `parameters` argument, e.g. `lock.open` with a
+    # keypad code), so the default dataclass __repr__ would otherwise write
+    # it to logs/tracebacks anywhere a resolution is printed or logged
+    # unredacted. It stays compare=True (the default): it holds `action`,
+    # so two resolutions over the identical entity set but OPPOSITE actions
+    # must not compare equal, and `compare=False` would make them -- there
+    # is no dispatch-irrelevant reading of this field despite its name.
+    # hash=False, though: `@dataclass(frozen=True)` generates a real
+    # __hash__ from every compare=True field, and a plain `dict` has none
+    # -- left at its default, `hash(resolution)` would type-check (mypy
+    # sees a real __hash__, isinstance(r, Hashable) is True) and then raise
+    # `TypeError: unhashable type: 'dict'` at runtime the first time
+    # anything actually hashes one. hash=False excludes just this field
+    # from __hash__ while keeping it in __eq__, so the class is genuinely
+    # hashable (hash/eq stay consistent) instead of merely claiming to be.
+    _operation_common: dict[str, Any] = field(
+        default_factory=dict, repr=False, hash=False
+    )
 
     @property
     def operations(self) -> list[dict[str, Any]]:
@@ -149,10 +157,14 @@ def operations(self) -> list[dict[str, Any]]:
             row = {"entity_id": entity_id, **self._operation_common}
             if "parameters" in row:
                 # `**self._operation_common` only shallow-copies: every row
-                # would otherwise share the SAME "parameters" dict object,
-                # so an in-place mutation of one row's parameters (e.g. by
-                # the dispatcher) would silently leak into every other row.
-                row["parameters"] = dict(row["parameters"])
+                # would otherwise share the SAME "parameters" dict object
+                # (and the same nested values inside it, e.g. an
+                # `rgb_color` list), so an in-place mutation of one row's
+                # parameters -- even a nested one, like
+                # `params["rgb_color"][0] = 0` -- would otherwise leak into
+                # every other row. Deep, not shallow: a `dict(...)` copy
+                # would only stop top-level key rebinding.
+                row["parameters"] = copy.deepcopy(row["parameters"])
             operations.append(row)
         return operations
 
@@ -402,15 +414,6 @@ def _validate_selector(selector: Mapping[str, Any], action: str) -> _ValidatedSe
 # ACTUAL registry that failed ("the device registry fetch failed") instead
 # of a generic "a topology fetch failed" that gives an operator nothing to
 # search HA's own logs for.
-_TOPOLOGY_FETCH_LABELS = (
-    "states",
-    "entity registry",
-    "device registry",
-    "area registry",
-    "floor registry",
-)
-
-
 async def _load_topology(client: Any) -> _Topology:
     """Load the HA state and registry views used by one resolution.
 
@@ -440,9 +443,26 @@ async def _load_topology(client: Any) -> _Topology:
         return_exceptions=True,
     )
     states, entities, devices, areas, floors = results
+    # Labels paired with their named local, not zipped against `results` by
+    # position: a `zip(strict=True)` against a separate label tuple only
+    # catches a LENGTH change on reorder, not the reorder itself -- if a
+    # future edit swaps two `gather()` lines and correctly updates the
+    # unpack above to match, a position-based label tuple would silently
+    # keep naming the OLD order, confidently blaming the wrong registry in
+    # the WARNING below during exactly the outage this logging exists to
+    # diagnose. Binding each label directly to its already-correct local
+    # makes a mislabel impossible without an edit that is visibly wrong on
+    # this line itself.
+    labeled_results = (
+        ("states", states),
+        ("entity registry", entities),
+        ("device registry", devices),
+        ("area registry", areas),
+        ("floor registry", floors),
+    )
     failures = [
         (label, result)
-        for label, result in zip(_TOPOLOGY_FETCH_LABELS, results, strict=True)
+        for label, result in labeled_results
         if isinstance(result, BaseException)
     ]
     if failures:
@@ -474,7 +494,22 @@ def _require_domain_known(domain: str, states: Mapping[str, Any]) -> None:
 
 def _all_excluded_or_hidden_message(*, excluded_count: int, hidden_count: int) -> str:
     """Build the final empty-result message with a visibility-safe count
-    breakdown (never entity IDs) when available."""
+    breakdown (never entity IDs) when available.
+
+    Only ever called once ``resolve_bulk_selector``'s own empty-aggregate
+    and wrong-domain gates have both already ruled themselves out (see the
+    ``not directly_hidden`` guards above this function's one call site), so
+    by construction at least one count here is non-zero -- refuse the
+    zero/zero case outright rather than let a future regression upstream
+    silently render a confident "excluded or hidden" claim with no
+    evidence behind it.
+    """
+    if not excluded_count and not hidden_count:
+        raise AssertionError(
+            "_all_excluded_or_hidden_message called with both counts zero -- "
+            "resolve_bulk_selector's empty-result gates should have already "
+            "raised a more specific error for this case"
+        )
     counts = []
     if excluded_count:
         counts.append(f"{excluded_count} excluded")
@@ -728,22 +763,37 @@ async def resolve_bulk_selector(
     selected_leaves = {
         entity_id for entity_id in expanded_leaves if entity_id.startswith(f"{domain}.")
     }
-    if expanded_leaves and not selected_leaves:
+    # Both branches below are gated on `not directly_hidden`: `candidate_roots`
+    # already subtracts `hidden`, so if ANY matching root is hidden, some of
+    # the "wrong domain" or "empty aggregate" evidence below could really be
+    # explained by that hidden entity having been the one that mattered --
+    # e.g. one visible root expands to a different domain while a SEPARATE,
+    # hidden root would have matched directly. `directly_hidden` is the more
+    # actionable fact in that case, so neither branch fires and both counts
+    # fall through together into `_all_excluded_or_hidden_message` below.
+    # `directly_hidden` empty additionally guarantees `candidate_roots` is
+    # non-empty here (matching_roots was already confirmed non-empty above,
+    # and nothing in it is hidden), so neither branch needs its own
+    # `candidate_roots` check.
+    if not directly_hidden and not expanded_leaves:
+        # A matched aggregate (or every one of several) has literally no
+        # members -- `normalize_member_entity_ids` returned `[]`, not
+        # `None`, so it WAS admitted as a root, but its own expansion
+        # contributed nothing. Distinct from "wrong domain" below: no leaf
+        # of ANY domain resulted, so there is nothing to blame on a
+        # different domain either.
+        raise BulkSelectorValidationError(
+            f"No entities of domain '{domain}' exist in the selected area(s) -- "
+            "the matched aggregate(s) have no members"
+        )
+    if not directly_hidden and expanded_leaves and not selected_leaves:
         # A fourth, distinct empty-result cause: matching_roots was
         # non-empty (a non-scene aggregate of a DIFFERENT domain qualified
         # as a root, e.g. a `group.living_room` whose members are all
         # `switch.*`), so expansion ran and produced leaves, but none of
         # them were the target domain -- neither exclusion nor visibility
         # ever entered into it. Caught here, before either subtraction
         # below, so it can never be misreported as "excluded or hidden".
-        #
-        # Gated on `expanded_leaves` (not just `not selected_leaves`):
-        # `candidate_roots` already subtracts `hidden` above, so when EVERY
-        # matching root in the area is hidden, `candidate_roots` -- and
-        # therefore `expanded_leaves` -- is itself empty, and this branch
-        # must NOT fire for that case: those entities are real, are in the
-        # area, and are simply hidden, which is exactly what the
-        # `_all_excluded_or_hidden_message` branch below exists to report.
         raise BulkSelectorValidationError(
             f"No entities of domain '{domain}' exist in the selected area(s) -- "
             "a matched aggregate's members are all a different domain"
@@ -770,13 +820,16 @@ async def resolve_bulk_selector(
         "validate_first": validate_first,
     }
     if parameters is not None:
-        # Copy, not the caller's own object by reference: the per-row copy
-        # in BulkSelectorResolution.operations protects each DISPATCH row
-        # from cross-row mutation, but does nothing about the resolution's
-        # own stored copy -- without this, a caller that still holds
-        # `parameters` and mutates it after this call returns would silently
-        # rewrite the "frozen" resolution's own payload too.
-        operation_common["parameters"] = dict(parameters)
+        # Deep copy, not the caller's own object by reference (and not a
+        # shallow `dict(...)`, which stops only top-level rebinding and
+        # still shares nested values like an `rgb_color` list): the
+        # per-row copy in BulkSelectorResolution.operations protects each
+        # DISPATCH row from cross-row mutation, but does nothing about the
+        # resolution's own stored copy -- without this, a caller that
+        # still holds `parameters` and mutates it (at any depth) after
+        # this call returns would silently rewrite the "frozen"
+        # resolution's own payload too.
+        operation_common["parameters"] = copy.deepcopy(parameters)
     if timeout_seconds is not None:
         operation_common["timeout_seconds"] = timeout_seconds
     excluded_and_hidden = effective_excluded & hidden