fix: report per-item failures in bulk operation status instead of aborting the batch (#2076)

Started as the residual-dead-code sweep from #2043's final review; reviewing that sweep surfaced real defects in the operation-status path, fixed here as well.

Fixes (user-facing):
* ha_get_operation_status with a list of IDs re-raised the first failed/timed-out/not-found operation's error, discarding the status of every other operation and making the failed/timeout aggregation unreachable. Per-item errors now become structured entries in detailed_results — preserving the full structured payload including top-level context keys — with a not_found bucket in the summary and a top-level success key per the repo-wide response contract.
* The bulk path polled each pending operation serially for the full 10s single-op timeout while the docstring promised "a short internal timeout"; it now polls all operations concurrently under the caller's timeout_seconds window.
* cleanup_expired_operations never reclaimed TIMEOUT-status operations (get_operation() flips an expired PENDING op to TIMEOUT in place on the read path), so polled-after-timeout operations accumulated for the process lifetime. TIMEOUT now shares FAILED's 60s TTL, anchored on completion_time, and the sweep-marked path keeps the same terminal minute as the read path so a poll after expiry reports timeout rather than not_found regardless of which path noticed first. The overflow trim considers all terminal statuses — never in-flight PENDING — so max_operations is enforceable except against a backlog of purely in-flight PENDING operations, which expire on their own timeouts.
* completion_percentage is now the terminal fraction (completed + failed + not_found over total) so it agrees with all_complete; success_rate keeps the successful fraction. The ha_get_operation_status docstring starts with an approved verb (Get) with the locale source baseline regenerated.
* get_domain_handler required a dot in its argument while both control paths pass a bare domain, so every per-domain valid_actions table was unreachable and the default handler's on/off/toggle applied everywhere — climate's heat/cool/set were rejected upfront as invalid actions (closes #2090). The lookup now accepts an entity ID or a bare domain, the tables were made purely additive and resolvable (short on/off/toggle actions restored where their services exist; unmappable entries mapped — cover stop/set, media next/previous, lock open, alarm arm_*/disarm, climate heat_cool — or trimmed), and a new e2e pin drives every table action through the resolver against the live service list.

Dead-code residuals from #2043 (internal):
* settings_ui/locales/es.json: drop the orphaned advanced.entity_search_limit.* keys (es.json merged via #2055 mid-review; the help string still named ha_search_entities).
* tools/util_helpers.py: remove strip_internal_fields (its only production caller was the server bridge deleted in #2043); its rationale moves into public_fields' docstring.
* client/websocket_listener.py: drop the unread self.settings assignment and its import.
* tests/src/unit/test_websocket_listener.py: fixture no longer builds the stats dict deleted in #2043.
* utils/operation_manager.py: remove OperationStatus.CANCELLED (unproducible since cancel_operation went; nothing ever read it).

Tests: new unit pins for the cleanup TTLs (terminal anchor, discriminating overflow trim), the bulk summary contract (completion_percentage semantics, preserved per-item payload), and the shared wait window (a pending operation flipping to completed inside the window, red under a 0s snapshot); real per-item assertions in the bulk-status e2e test via MCPAssertions; the mixed-entities e2e block in test_network_errors made live end to end (the bulk response has no top-level success key, so the gate now mirrors assert_mcp_success's bulk indicator; the action string is "on" — "turn_on" was rejected for every entity; value checks pin that the valid entities succeed and both fabricated entities fail as per-item entries; a 5s status window avoids dead-heating the 10s call wrapper); projection pin for the area-only branch's public_fields strip; flip task joined via asyncio.gather per CodeQL.

Author: kingpanther13

src/ha_mcp/client/websocket_listener.py

@@ -9,7 +9,6 @@
 import logging
 from typing import Any
 
-from ..config import get_global_settings
 from ..utils.operation_manager import get_operation_manager, update_pending_operations
 from .websocket_client import HomeAssistantWebSocketClient, get_websocket_client
 
@@ -21,7 +20,6 @@ class WebSocketListenerService:
 
     def __init__(self) -> None:
         """Initialize the WebSocket listener service."""
-        self.settings = get_global_settings()
         self.operation_manager = get_operation_manager()
         self.websocket_client: HomeAssistantWebSocketClient | None = None
         self.listener_task: asyncio.Task | None = None

src/ha_mcp/tools/device_control.py

@@ -301,33 +301,50 @@ def _resolve_service_name(
         action: str,
         parameters: dict[str, Any] | None,
     ) -> tuple[str, dict[str, Any] | None]:
-        service_mapping = {
-            "on": "turn_on",
-            "off": "turn_off",
-            "toggle": "toggle",
-            "open": "open_cover" if domain == "cover" else "turn_on",
-            "close": "close_cover" if domain == "cover" else "turn_off",
-            "set": "turn_on" if domain == "light" else "set_temperature",
-        }
-
-        service_name = service_mapping.get(action, action)
+        if domain == "climate" and action in ("heat", "cool", "auto", "heat_cool"):
+            parameters = dict(parameters or {})
+            parameters["hvac_mode"] = action
+            return "set_hvac_mode", parameters
 
-        if domain == "climate":
-            if action in ["heat", "cool", "auto"]:
-                service_name = "set_hvac_mode"
-                if not parameters:
-                    parameters = {}
-                parameters["hvac_mode"] = action
-            elif action == "set":
-                service_name = "set_temperature"
+        service_name = self._SERVICE_OVERRIDES.get(
+            (domain, action), self._GENERIC_SERVICE_MAP.get(action, action)
+        )
+        return service_name, parameters
 
-        elif domain == "media_player":
-            if action in ["play", "pause", "stop"]:
-                service_name = f"media_{action}"
-            elif action == "set":
-                service_name = "volume_set"
+    # Domain-specific action-to-service mappings; anything not listed
+    # falls through to _GENERIC_SERVICE_MAP, then to the action verbatim.
+    # tests/src/e2e/workflows/device_control/test_action_service_resolution.py
+    # pins every DOMAIN_HANDLERS table entry through this resolver against
+    # a live instance's service list.
+    _SERVICE_OVERRIDES: ClassVar[dict[tuple[str, str], str]] = {
+        ("cover", "open"): "open_cover",
+        ("cover", "close"): "close_cover",
+        ("cover", "stop"): "stop_cover",
+        ("cover", "set"): "set_cover_position",
+        ("light", "set"): "turn_on",
+        # lock.open exists; the generic mapping's turn_on does not.
+        ("lock", "open"): "open",
+        ("climate", "set"): "set_temperature",
+        ("media_player", "play"): "media_play",
+        ("media_player", "pause"): "media_pause",
+        ("media_player", "stop"): "media_stop",
+        ("media_player", "next"): "media_next_track",
+        ("media_player", "previous"): "media_previous_track",
+        ("media_player", "set"): "volume_set",
+        ("alarm_control_panel", "arm_home"): "alarm_arm_home",
+        ("alarm_control_panel", "arm_away"): "alarm_arm_away",
+        ("alarm_control_panel", "arm_night"): "alarm_arm_night",
+        ("alarm_control_panel", "disarm"): "alarm_disarm",
+    }
 
-        return service_name, parameters
+    _GENERIC_SERVICE_MAP: ClassVar[dict[str, str]] = {
+        "on": "turn_on",
+        "off": "turn_off",
+        "toggle": "toggle",
+        "open": "turn_on",
+        "close": "turn_off",
+        "set": "set_temperature",
+    }
 
     _DOMAIN_PARAMS: ClassVar[dict[str, list[str]]] = {
         "light": ["brightness", "color_temp_kelvin", "rgb_color", "effect"],
@@ -1274,13 +1291,20 @@ def _build_bulk_response(
         return response
 
     async def get_bulk_operation_status(
-        self, operation_ids: list[str]
+        self, operation_ids: list[str], timeout_seconds: int = 10
     ) -> dict[str, Any]:
         """
         Check status of multiple operations.
 
+        Polls all operations concurrently under one shared
+        ``timeout_seconds`` window, so the wall time is bounded by the
+        window rather than growing with the batch size. Per-item failures
+        become structured entries in ``detailed_results`` rather than
+        aborting the batch.
+
         Args:
             operation_ids: List of operation IDs to check
+            timeout_seconds: Wait window applied to every operation
 
         Returns:
             Status summary for all operations
@@ -1296,26 +1320,69 @@ async def get_bulk_operation_status(
                 )
             )
 
-        # Check all operations
-        statuses = []
-        for op_id in operation_ids:
-            status = await self.get_device_operation_status(op_id)
-            statuses.append(status)
+        # Check all operations concurrently under one shared wait window.
+        # Per-item failures must not abort the batch: get_device_operation_status
+        # raises ToolError for failed / timed-out / not-found operations, so each
+        # one is caught and folded back into detailed_results as a structured
+        # entry — otherwise the first bad operation would discard the status of
+        # every other one. The whole parsed error payload is preserved (context
+        # keys like entity_id / duration_ms sit at its top level), with the
+        # batch "status" field layered on for the summary counts.
+        error_code_to_status = {
+            ErrorCode.SERVICE_CALL_FAILED.value: "failed",
+            ErrorCode.TIMEOUT_OPERATION.value: "timeout",
+            ErrorCode.RESOURCE_NOT_FOUND.value: "not_found",
+        }
+
+        async def check_one(op_id: str) -> dict[str, Any]:
+            try:
+                return await self.get_device_operation_status(
+                    op_id, timeout_seconds=timeout_seconds
+                )
+            except ToolError as e:
+                try:
+                    err = json.loads(str(e))
+                except ValueError:
+                    err = {"success": False, "error": {"message": str(e)}}
+                error_info = err.get("error") or {}
+                return {
+                    **err,
+                    "operation_id": op_id,
+                    "status": error_code_to_status.get(
+                        error_info.get("code", ""), "failed"
+                    ),
+                }
+
+        statuses = list(
+            await asyncio.gather(*(check_one(op_id) for op_id in operation_ids))
+        )
 
         # Summarize results
         completed = len([s for s in statuses if s.get("status") == "completed"])
         failed = len([s for s in statuses if s.get("status") in ["failed", "timeout"]])
+        not_found = len([s for s in statuses if s.get("status") == "not_found"])
         pending = len([s for s in statuses if s.get("status") == "pending"])
 
+        # The batch call itself succeeded; per-item failures live inside
+        # detailed_results (batch-item pattern). Top-level ``success`` is
+        # the repo-wide response contract every tool return carries.
         return {
+            "success": True,
             "total_operations": len(operation_ids),
             "completed": completed,
             "failed": failed,
+            "not_found": not_found,
             "pending": pending,
             "all_complete": pending == 0,
             "summary": {
                 "success_rate": f"{completed}/{len(operation_ids)}",
-                "completion_percentage": (completed / len(operation_ids)) * 100,
+                # Terminal fraction (nothing left in flight) — consistent
+                # with all_complete; success_rate carries the successful
+                # fraction separately.
+                "completion_percentage": (
+                    (completed + failed + not_found) / len(operation_ids)
+                )
+                * 100,
             },
             "detailed_results": statuses,
             "recommendations": (
@@ -1324,7 +1391,7 @@ async def get_bulk_operation_status(
                     "Check failed operations for specific error messages",
                     "Retry failed operations with different parameters if needed",
                 ]
-                if pending > 0 or failed > 0
+                if pending > 0 or failed > 0 or not_found > 0
                 else ["All operations completed successfully!"]
             ),
         }

src/ha_mcp/tools/tools_service.py

@@ -1385,13 +1385,15 @@ async def ha_get_operation_status(
         timeout_seconds: int = 10,
     ) -> dict[str, Any]:
         """
-        Check status of one or more device operations with real-time WebSocket verification.
+        Get the status of one or more device operations with real-time WebSocket verification.
 
         Pass a single operation_id string to check one operation, or a list of IDs
         to check multiple operations at once (bulk status).
 
-        The timeout_seconds parameter applies to single-operation checks only.
-        Bulk checks poll each operation individually with a short internal timeout.
+        The timeout_seconds wait window bounds both modes. Bulk checks poll
+        all operations concurrently under one shared window and report
+        per-item failures inside detailed_results instead of aborting the
+        batch.
 
         Use this to track operations initiated by ha_bulk_control or ha_call_service.
         For current entity states, use ha_get_state instead.
@@ -1401,7 +1403,7 @@ async def ha_get_operation_status(
             # before the body runs, so operation_id is already the final shape.
             if isinstance(operation_id, list):
                 result = await self._device_tools.get_bulk_operation_status(
-                    operation_ids=operation_id
+                    operation_ids=operation_id, timeout_seconds=timeout_seconds
                 )
                 return cast(dict[str, Any], result)
             result = await self._device_tools.get_device_operation_status(

src/ha_mcp/tools/util_helpers.py

@@ -77,43 +77,14 @@ def summarize_theme_listing(raw_themes: dict[str, Any]) -> dict[str, Any]:
     }
 
 
-def strip_internal_fields(obj: Any, _seen: set[int] | None = None) -> Any:
-    """Remove leading-underscore keys from ``obj`` and any nested dicts
-    or lists in place.
+def public_fields(d: dict[str, Any]) -> dict[str, Any]:
+    """Return a shallow copy of ``d`` with leading-underscore keys removed.
 
     The ha-mcp tool layer enriches entity / area dicts with internal
     fields like ``_hidden_by`` and ``_aliases`` so downstream branches
     can rank without re-querying the entity registry. Those keys must
     not leak through public tool returns: this helper centralises the
     convention so individual call sites don't have to remember to strip.
-
-    Mutates in place and returns the same reference for chaining. Cycle
-    guard via ``_seen`` (id-tracked) keeps the recursion safe if a
-    future caller ever feeds it a non-tree structure — JSON payloads
-    don't, but the helper is now a generic utility (importable from
-    ``server.py``) so the protection is cheap insurance.
-    """
-    if _seen is None:
-        _seen = set()
-    obj_id = id(obj)
-    if obj_id in _seen:
-        return obj
-    if isinstance(obj, dict):
-        _seen.add(obj_id)
-        for key in [k for k in obj if isinstance(k, str) and k.startswith("_")]:
-            obj.pop(key, None)
-        for value in obj.values():
-            strip_internal_fields(value, _seen)
-    elif isinstance(obj, list):
-        _seen.add(obj_id)
-        for item in obj:
-            strip_internal_fields(item, _seen)
-    return obj
-
-
-def public_fields(d: dict[str, Any]) -> dict[str, Any]:
-    """Return a shallow copy of ``d`` with leading-underscore keys
-    removed. Non-mutating counterpart to :func:`strip_internal_fields`.
     Shallow only — list/dict values are shared with the source, so a
     later mutation of those values would propagate.
     """