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

A comprehensive review flagged two blocking issues plus a long list of
correctness, error-routing, and coverage gaps in the structural-selector
feature. Fixed:

Blocking:
- Root admission for aggregate expansion was not domain-constrained: a
  scene assigned to a selected area was admitted as a root and expanded,
  pulling entities from anywhere in the house into the dispatch (HA's
  scene.entity_id attribute lists controlled targets, not structural
  members). Scenes are now excluded from aggregate-root admission on
  architectural grounds (not an integration-identity heuristic).
- A dynamic selector call's approval could be consumed by a DIFFERENT
  invocation than the one that created it -- via a race (a concurrent
  call's find() observing "approved" before the original waiter resumes)
  or a later identical call reusing a stale approval within
  approval_ttl_minutes, dispatching against re-resolved topology. Dynamic
  entries are now bound to their creating invocation only.

Also fixed:
- A wildcard rule path with a literal tail (args.*.domain) was
  over-classified as needing resolved-operations data by the previous
  fix, regressing legitimate conditional selector rules back into an
  unconditional gate. Narrowed to paths that can actually reach the
  operations list.
- Stringified JSON arguments (Claude Desktop stdio sends these) bypassed
  policy evaluation entirely: PolicyMiddleware reads raw
  context.message.arguments before the tool's own Pydantic coercion runs,
  so a rule targeting a nested field silently never matched. Normalized
  once, before evaluation and hashing.
- _load_topology used a bare asyncio.gather; the first exception left the
  other four awaitables orphaned. Now mirrors tools_areas.py's
  return_exceptions=True + explicit re-raise pattern.
- Registry/visibility infrastructure failures were raised as the same
  exception class as caller input errors, reaching the agent as
  VALIDATION_FAILED and inviting a pointless selector rewrite against an
  HA outage. Split into BulkSelectorInfrastructureError, routed through
  create_connection_error.
- An unknown selector.domain (e.g. "lights") passed validation and
  resolved to nothing, blaming exclusions/areas that were correct. Domain
  existence is now checked against loaded states, and the empty-result
  message distinguishes "no entities of domain X" from "all matches were
  excluded or hidden".
- VALIDATION_FAILED had no DEFAULT_SUGGESTIONS entry, so ~14 selector
  failure modes returned no actionable guidance.
- _parse_bulk_operations lost its security rationale comments
  (include_input=False; why malformed rows are preserved) during an
  earlier extraction.

Plus: NamedTuple returns instead of bare tuples, BulkSelectorResolution's
operations as a computed property (was a separately-tracked list that
could desync from resolved_entity_ids), _SELECTOR_KEYS derived from the
TypedDict instead of a parallel literal, a logger for the ~14 selector
failure modes, dispatch failures now attach the resolution to their error
context, dangling-member/cycle errors name their aggregate and parameter,
and the "exactly one of operations/selector" and "selector-only
parameter" validation messages now name the actual offender instead of a
collapsed generic message.

Test coverage added: scene exclusion, aggregate exclusion (only leaf
exclusion was tested before), unknown domain, domain-not-in-area vs
all-excluded-or-hidden, whitespace-padded IDs, orphaned-task-free
topology fetch failure, visibility-infrastructure routing, the wildcard
over-classification, stringified-argument policy bypass (pure-function
and end-to-end), the concurrent-approval race, and a real non-dry-run E2E
dispatch that verifies the excluded entity's live state never moves (the
only prior E2E test asserted the dry-run preview, never a real dispatch
or real post-dispatch state).

Deliberately deferred (documented in PR comment, not silently dropped):
a handful of narrower test-coverage gaps (exact MAX_SELECTOR_ENTITIES=100
boundary, individual selector-only-parameter-rejection cases, malformed
JSON body edge case) and a request to expose a hidden-vs-excluded
breakdown finer than the current warnings message.

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

Author: swissmo

src/ha_mcp/errors.py

@@ -210,6 +210,10 @@ class ErrorCode(StrEnum):
         "Check documentation for required fields",
         "Ensure all required parameters are provided",
     ],
+    ErrorCode.VALIDATION_FAILED: [
+        "Check the parameter values and format against the tool documentation",
+        "Review the message and details fields for the specific constraint that failed",
+    ],
     ErrorCode.VALIDATION_INVALID_JSON: [
         "Ensure the parameter is valid JSON",
         "Check for syntax errors in JSON",

src/ha_mcp/policy/evaluator.py

@@ -6,6 +6,7 @@
 from enum import StrEnum
 from typing import Any
 
+from ..tools.util_helpers import _loads_if_json_container_str
 from .model import Policy, Predicate, Rule
 
 logger = logging.getLogger(__name__)
@@ -16,6 +17,56 @@ class Verdict(StrEnum):
     REQUIRE_APPROVAL = "require_approval"
 
 
+def normalize_stringified_containers(value: Any) -> Any:
+    """Recursively parse JSON-encoded object/array strings into real containers.
+
+    Some MCP client stacks (Claude Desktop stdio among them — see
+    ``tools/util_helpers.py``'s ``JSON_STRING_COERCION``) pass model-emitted
+    stringified objects through unrepaired, e.g. sending
+    ``{"selector": "{\\"domain\\": \\"light\\"}"}`` instead of a nested
+    object. Pydantic's ``JSON_STRING_COERCION`` ``BeforeValidator`` repairs
+    this, but only when the tool's own parameter validation runs — INSIDE
+    ``call_next``, after policy evaluation. ``iter_path_values`` only
+    descends into ``dict``/``list``, so a still-stringified value makes a
+    predicate targeting a nested field (``args.selector.domain``,
+    ``args.operations.*.entity_id``) silently yield nothing: no rule
+    matches, and — for ``ha_bulk_control`` — ``args.get("selector")`` still
+    sees a truthy string, so the selector fail-safe's own dynamic-call
+    detection still fires, but nothing inside it can be inspected either.
+    Applying the same repair here, once, before evaluation and hashing,
+    closes that gap for both plain predicate rules and the fail-safe, and
+    makes ``compute_args_hash`` key on the same logical value regardless of
+    which wire shape the client sent.
+
+    Malformed JSON that merely looks like a container is left as the raw
+    string (not raised): policy evaluation is not the place to surface a
+    JSON syntax error — the tool's own validation does that, with a
+    properly attributed parameter name.
+    """
+    if isinstance(value, str):
+        try:
+            return _loads_if_json_container_str(value)
+        except ValueError:
+            return value
+    if isinstance(value, dict):
+        return {key: normalize_stringified_containers(v) for key, v in value.items()}
+    if isinstance(value, list):
+        return [normalize_stringified_containers(v) for v in value]
+    return value
+
+
+def has_dynamic_selector_targets(name: str, args: dict[str, Any]) -> bool:
+    """Return whether identical arguments can resolve to different targets later.
+
+    Single source of truth for the ``ha_bulk_control`` selector-mode predicate:
+    both ``PolicyMiddleware`` (approval-sharing/remembering gates) and
+    ``evaluate()`` below (the operations fail-safe) must agree on exactly
+    which calls are "dynamic", or a future rename/extension of this check in
+    one place silently stops applying to the other.
+    """
+    return name == "ha_bulk_control" and args.get("selector") is not None
+
+
 def iter_path_values(args: dict[str, Any], path: str) -> Iterator[Any]:
     """Yield every value the dotted path resolves to.
 
@@ -216,14 +267,10 @@ def evaluate(tool_name: str, args: dict[str, Any], policy: Policy) -> Verdict:
     # 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", "*")
-            and _rule_needs_resolved_operations(rule)
-            for rule in policy.rules
-        )
+    if has_dynamic_selector_targets(tool_name, args) 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

@@ -15,7 +15,13 @@
 from ..renamed_tools import current_tool_name
 from ..tools.helpers import raise_tool_error, safe_progress
 from .approval_queue import ApprovalQueue, PendingApproval, compute_args_hash
-from .evaluator import Verdict, evaluate, find_matching_rule
+from .evaluator import (
+    Verdict,
+    evaluate,
+    find_matching_rule,
+    has_dynamic_selector_targets,
+    normalize_stringified_containers,
+)
 from .model import Policy, Rule
 
 logger = logging.getLogger(__name__)
@@ -64,11 +70,6 @@ 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."""
 
@@ -115,7 +116,15 @@ async def on_call_tool(
         # name, and ``evaluate`` returns ALLOW when nothing matches, so a gate
         # reading a stale name lets the call through.
         name = current_tool_name(context.message.name)
-        args = context.message.arguments or {}
+        # Normalize stringified JSON containers (a client like Claude Desktop
+        # stdio can send a nested parameter, e.g. `selector`, as a JSON
+        # string rather than an object) before any evaluation/hashing below
+        # — the tool's own Pydantic coercion happens only later, inside
+        # call_next, which is too late for a predicate path or the selector
+        # fail-safe to see the real structure. This only affects the local
+        # copy used for gating; `context` itself is untouched, so the actual
+        # tool call still receives the client's original wire shape.
+        args = normalize_stringified_containers(context.message.arguments or {})
 
         if _passes_ungated(name, args):
             return await call_next(context)
@@ -125,15 +134,37 @@ 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)
+        dynamic_targets = has_dynamic_selector_targets(name, args)
         remember_minutes = (
             0 if dynamic_targets else rule.remember_minutes if rule else 0
         )
 
         if not dynamic_targets and self._queue.is_remembered(name, args_hash):
             return await call_next(context)
 
-        existing = self._queue.find(name, args_hash)
+        # A dynamic selector call must never consume an entry it did not
+        # itself create and is not itself still waiting on. Two reachable
+        # ways an unguarded lookup here breaks the "approve once, dispatch
+        # once, against fresh topology" invariant: (1) a race -- decide()
+        # flips a waiter's own PendingApproval.decision before the waiter's
+        # task is rescheduled, so a second, concurrent call's find() can
+        # observe "approved" and consume-and-dispatch before the original
+        # waiter wakes and does the same from its own reference, executing
+        # the click twice; (2) a later, non-racy call -- the original call
+        # times out and is told to re-call, the user approves afterwards,
+        # and ANY later identical call within approval_ttl_minutes claims
+        # that stale approval and dispatches against topology re-resolved
+        # at that later moment -- exactly the reuse-across-a-time-gap this
+        # PR disables `remember_minutes` to prevent, just via a different
+        # mechanism. Binding a dynamic entry to its creating invocation
+        # closes both: only the call that created a pending entry ever
+        # observes it (via its own `pending` reference after its own wait,
+        # below), so every other call -- concurrent or a later retry --
+        # unconditionally mints its own independent entry and wait window.
+        # The accepted cost is that a retry never silently rides an earlier
+        # approval: each blocked call gets its own approval row, and only
+        # approving the row for the CURRENTLY-blocked call has any effect.
+        existing = None if dynamic_targets else self._queue.find(name, args_hash)
         if existing and existing.decision == "approved":
             self._queue.consume_and_maybe_remember(
                 existing,