fix(bulk): narrow client: Any to HomeAssistantClient, pin deep-copy and example-value fixes

1. Narrow the untyped client seam the deleted malformed-states test used
   to protect. ServiceTools.__init__, _reject_operations_group_member_conflicts,
   register_service_tools, resolve_bulk_selector, and _load_topology now take
   HomeAssistantClient instead of Any -- so a states-shaped response from the
   wrong client (e.g. the websocket client's get_states() -> dict[str, Any],
   vs. the REST client's list[dict[str, Any]]) is a static error again, not
   a silent fail-open on the group-safety gate.

   This did cascade one step, as flagged as a possibility: registry.py's
   ToolsRegistry.__init__(server: Any, ...) was the remaining break in the
   chain (self.client = server.client inferred Any from it). Closed with a
   small Protocol (_ServerLike, matching the existing CustomRouteServer
   pattern in browser_landing.py) instead of importing the concrete
   HomeAssistantSmartMCPServer class, since server.py's own lazy import of
   ToolsRegistry exists specifically to avoid that circular import.

   Narrowing surfaced two real (independent, pre-existing) issues along the
   way, both fixed: _capture_initial_state called get_entity_state(entity_id)
   where entity_id was str | None -- safe today only because the one call
   site's should_wait guard embeds an entity_id is not None check the type
   checker can't see through, now made explicit. And the first Protocol
   draft declared plain (implicitly settable) attributes, which
   HomeAssistantSmartMCPServer's read-only client/device_tools @Property
   definitions don't satisfy -- fixed by declaring them as @Property in the
   Protocol too. Validated with the exact CI mypy command (mypy src/
   custom_components/ homeassistant-addon/ scripts/, 255 files) -- zero new
   errors anywhere else in the codebase.

2. Add test_parameters_copies_are_genuinely_deep_not_shallow: the existing
   parameters-isolation test only reassigned a top-level scalar key, which
   a shallow dict(...) copy already protects against. This mutates a nested
   list value (rgb_color) both externally (caller's dict, after the call)
   and on one already-returned row, checking the resolution's stored copy
   and a fresh .operations re-read. Verified load-bearing by temporarily
   reverting each of the two copy.deepcopy call sites to dict() independently
   and confirming the test fails both times before restoring the fix.

3. Fix the selector-only-parameter example's placeholder rendering as a
   quoted string for every field, including timeout_seconds (a float).
   New _PER_ROW_PARAMETER_EXAMPLE_VALUES gives each field a
   correctly-typed sample (5 for timeout_seconds, {"brightness_pct": 30}
   for parameters, etc.) instead of a bare "...". Pinned by
   test_selector_only_parameter_example_uses_a_correctly_typed_sample.

Full unit suite: 10725 passed (up from 10722 by the tests added here); the
same 26 pre-existing Windows-platform-only failures as every prior baseline
this session remain, unrelated to this change.

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

Author: swissmo

src/ha_mcp/tools/bulk_selector.py

@@ -12,6 +12,7 @@
 
 from pydantic import ConfigDict, Field
 
+from ..client.rest_client import HomeAssistantClient
 from ..utils.domain_handlers import get_domain_handler
 from ..utils.entity_membership import normalize_member_entity_ids
 from ..visibility.resolver import VisibilityDataUnavailable, load_hidden_set
@@ -414,7 +415,7 @@ 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.
-async def _load_topology(client: Any) -> _Topology:
+async def _load_topology(client: HomeAssistantClient) -> _Topology:
     """Load the HA state and registry views used by one resolution.
 
     ``return_exceptions=True`` plus an explicit re-raise guard, mirroring
@@ -643,7 +644,7 @@ async def _load_hidden_entities(
 
 
 async def resolve_bulk_selector(
-    client: Any,
+    client: HomeAssistantClient,
     selector: Mapping[str, Any],
     *,
     action: str,

src/ha_mcp/tools/registry.py

@@ -21,7 +21,9 @@
 import logging
 import pkgutil
 from pathlib import Path
-from typing import Any
+from typing import Any, Protocol
+
+from ..client.rest_client import HomeAssistantClient
 
 logger = logging.getLogger(__name__)
 
@@ -31,6 +33,43 @@
     "backup": "register_backup_tools",
 }
 
+
+class _ServerLike(Protocol):
+    """Structural type for the object ``ToolsRegistry`` reads tool
+    dependencies from -- matches ``HomeAssistantSmartMCPServer`` without
+    importing it, which ``server.py``'s own lazy
+    ``from .tools.registry import ToolsRegistry`` (inside its
+    ``tools_registry`` property) exists specifically to avoid a circular
+    import for.
+
+    Only ``client`` is narrowed to its real type: it is the one attribute
+    every ``register_*_tools(mcp, client, **kwargs)`` plugin function reads
+    ``.get_states()``/``.send_websocket_message()``/etc. from, and a
+    dict-shaped response from the WRONG client implementation (e.g.
+    ``HomeAssistantWebSocketClient.get_states() -> dict[str, Any]``, vs.
+    the REST client's ``list[dict[str, Any]]``) silently fails open
+    downstream instead of raising -- see
+    ``_find_group_member_conflicts`` in ``tools_service.py``.
+    ``mcp``/``smart_tools``/``device_tools`` stay ``Any``, matching their
+    own already-``Any`` typing on ``HomeAssistantSmartMCPServer`` itself.
+
+    Declared as read-only ``@property`` methods, not plain attributes: a
+    bare ``client: HomeAssistantClient`` class-level annotation implies a
+    SETTABLE member, which ``HomeAssistantSmartMCPServer.client`` (a
+    getter-only ``@property``, lazily creating the client on first access)
+    does not satisfy.
+    """
+
+    @property
+    def client(self) -> HomeAssistantClient: ...
+    @property
+    def mcp(self) -> Any: ...
+    @property
+    def smart_tools(self) -> Any: ...
+    @property
+    def device_tools(self) -> Any: ...
+
+
 # Preset module groups for common use cases
 MODULE_PRESETS = {
     "automation": [
@@ -56,9 +95,9 @@ class ToolsRegistry:
     - Comma-separated list: Load specific modules
     """
 
-    def __init__(self, server: Any, enabled_modules: str = "all") -> None:
+    def __init__(self, server: _ServerLike, enabled_modules: str = "all") -> None:
         self.server = server
-        self.client = server.client
+        self.client: HomeAssistantClient = server.client
         self.mcp = server.mcp
         self._enabled_modules = enabled_modules
         # These are now lazily initialized via server properties

src/ha_mcp/tools/tools_service.py

@@ -14,6 +14,7 @@
 from pydantic import ConfigDict, Field, SkipValidation, TypeAdapter, ValidationError
 
 from ..client.rest_client import (
+    HomeAssistantClient,
     HomeAssistantCommandError,
     HomeAssistantCommandNotSent,
     HomeAssistantConnectionError,
@@ -312,7 +313,7 @@ def _find_group_member_conflicts(
 
 
 async def _reject_operations_group_member_conflicts(
-    client: Any, operations: list[Any]
+    client: HomeAssistantClient, operations: list[Any]
 ) -> None:
     """Fail closed when an operations-mode batch targets a group/aggregate
     entity together with one or more of its own members.
@@ -464,6 +465,18 @@ def _selector_only_parameter_offender(
     {"action", "parameters", "timeout_seconds", "validate_first"}
 )
 
+# One representative, correctly-typed sample value per per-row parameter,
+# for the worked example below. A bare "..." placeholder rendered inside a
+# Python dict literal is always a quoted STRING regardless of the field's
+# real type -- for timeout_seconds (a float) that example teaches the
+# model the wrong shape for the exact value it is being told to copy.
+_PER_ROW_PARAMETER_EXAMPLE_VALUES: dict[str, Any] = {
+    "action": "on",
+    "parameters": {"brightness_pct": 30},
+    "timeout_seconds": 5,
+    "validate_first": True,
+}
+
 
 def _selector_only_parameter_message(offending_parameter: str) -> str:
     """Build the remedy for one selector-only parameter used in operations mode.
@@ -483,7 +496,9 @@ def _selector_only_parameter_message(offending_parameter: str) -> str:
         example_row: dict[str, Any] = {"entity_id": "light.kitchen"}
         if offending_parameter != "action":
             example_row["action"] = "on"
-        example_row[offending_parameter] = "..."
+        example_row[offending_parameter] = _PER_ROW_PARAMETER_EXAMPLE_VALUES[
+            offending_parameter
+        ]
         return (
             f"'{offending_parameter}' is a per-operation field in operations "
             f"mode (see BulkControlOperation), not a top-level tool argument "
@@ -746,7 +761,7 @@ def _build_service_suggestions(
 class ServiceTools:
     """Service call and device operation tools for Home Assistant."""
 
-    def __init__(self, client: Any, device_tools: Any) -> None:
+    def __init__(self, client: HomeAssistantClient, device_tools: Any) -> None:
         self._client = client
         self._device_tools = device_tools
 
@@ -910,7 +925,20 @@ def _build_timeout_response(
         return response
 
     async def _capture_initial_state(self, entity_id: str | None) -> str | None:
-        """Capture the current state of an entity before a service call."""
+        """Capture the current state of an entity before a service call.
+
+        ``entity_id`` stays optional in the signature to match the caller's
+        own ``str | None`` (a service call may target zero entities); the
+        one current call site only reaches this when ``should_wait`` (which
+        embeds an ``entity_id is not None`` check among several AND-ed
+        conditions) is true, so ``entity_id`` is always real there in
+        practice -- but that invariant lives in a boolean a few lines away,
+        not in a form the type checker can see through. Narrowing here
+        instead of trusting the caller keeps ``get_entity_state`` (which
+        genuinely requires a ``str``) honestly typed.
+        """
+        if entity_id is None:
+            return None
         try:
             state_data = await self._client.get_entity_state(entity_id)
             return state_data.get("state") if state_data else None
@@ -2271,7 +2299,9 @@ async def ha_call_event(
         }
 
 
-def register_service_tools(mcp: Any, client: Any, **kwargs: Any) -> None:
+def register_service_tools(
+    mcp: Any, client: HomeAssistantClient, **kwargs: Any
+) -> None:
     """Register service call and operation monitoring tools with the MCP server."""
     device_tools = kwargs.get("device_tools")
     if not device_tools:

tests/src/unit/test_bulk_selector.py

@@ -1190,6 +1190,60 @@ async def test_resolution_parameters_survive_caller_mutating_its_own_dict() -> N
     assert result.operations[0]["parameters"]["brightness_pct"] == 50
 
 
+@pytest.mark.asyncio
+async def test_parameters_copies_are_genuinely_deep_not_shallow() -> None:
+    """Both ``parameters`` copies (the resolution's own stored copy, and
+    each dispatch row's own copy) must be deep, not a shallow ``dict(...)``
+    that only stops top-level key REBINDING while still sharing nested
+    mutable values (e.g. an ``rgb_color`` list) by reference.
+
+    ``test_resolution_parameters_survive_caller_mutating_its_own_dict``
+    above only reassigns a top-level scalar key after the call, which a
+    shallow ``dict(...)`` copy already protects against -- it would keep
+    passing even if either ``copy.deepcopy`` call were reverted. This
+    mutates a NESTED list value instead, at both call sites a caller could
+    plausibly touch:
+
+    1. The caller's own dict, after ``resolve_bulk_selector`` returns --
+       pins the resolution-level store-site copy.
+    2. One already-returned dispatch row, then re-reads ``.operations``
+       fresh -- pins the per-row copy. Both rows trace back to the SAME
+       ``_operation_common["parameters"]`` object via ``**`` spread before
+       their own copy is made, so a shallow per-row copy would let row A's
+       mutation corrupt that shared source and leak into a fresh read of
+       row B.
+    """
+    client = SelectorClient(
+        states=[_state("light.a"), _state("light.b")],
+        entities=[
+            {"entity_id": "light.a", "area_id": "salon"},
+            {"entity_id": "light.b", "area_id": "salon"},
+        ],
+    )
+    caller_parameters = {"rgb_color": [255, 0, 0]}
+
+    result = await resolve_bulk_selector(
+        client,
+        {"domain": "light", "area_ids": ["salon"]},
+        action="on",
+        parameters=caller_parameters,
+        timeout_seconds=None,
+        validate_first=True,
+    )
+    assert len(result.operations) == 2, "test needs 2+ rows to prove no cross-row leak"
+
+    # 1. External mutation, after the call, must never reach the stored copy.
+    caller_parameters["rgb_color"][0] = 111
+    for row in result.operations:
+        assert row["parameters"]["rgb_color"] == [255, 0, 0]
+
+    # 2. Mutating one already-returned row must not corrupt the shared
+    # source a FRESH `.operations` read (a new property call) derives from.
+    result.operations[0]["parameters"]["rgb_color"][0] = -1
+    for row in result.operations:  # freshly recomputed, not the mutated list
+        assert row["parameters"]["rgb_color"] == [255, 0, 0]
+
+
 @pytest.mark.asyncio
 async def test_resolution_equality_is_action_sensitive() -> None:
     """Two resolutions over the identical entity set but OPPOSITE actions

tests/src/unit/test_ha_bulk_control_selector.py

@@ -824,6 +824,27 @@ async def test_selector_only_action_example_never_duplicates_the_key() -> None:
     assert "'action': 'on', 'action'" not in message, message
 
 
+@pytest.mark.asyncio
+async def test_selector_only_parameter_example_uses_a_correctly_typed_sample() -> None:
+    """The worked example's placeholder value must match the offending
+    field's real type -- a bare ``"..."`` string, rendered inside a Python
+    dict literal, is always shown quoted regardless of what type the field
+    actually is. For ``timeout_seconds`` (a ``float``) that taught the
+    model to copy a string where a number belongs.
+    """
+    tools = ServiceTools(MagicMock(), MagicMock())
+
+    with pytest.raises(ToolError) as exc_info:
+        await tools.ha_bulk_control(
+            operations=[{"entity_id": "light.one", "action": "off"}],
+            timeout_seconds=5.0,
+        )
+
+    message = json.loads(str(exc_info.value))["error"]["message"]
+    assert "'timeout_seconds': 5" in message, message
+    assert "'timeout_seconds': '...'" not in message, message
+
+
 @pytest.mark.asyncio
 async def test_selector_mode_requires_action() -> None:
     """Selector mode with no ``action`` must fail before any registry read,