fix: retry Supervisor add-on calls on job-group collisions (#2041)

* fix: retry Supervisor add-on calls on job-group collisions

Supervisor serialises jobs per add-on job group and rejects a state-changing
call outright with "Another job is running for job group addon_<slug>"
whenever a still-settling job (a watchdog restart, a prior start/stop, a store
reload) still holds that group. ha_manage_addon surfaced that as
SERVICE_CALL_FAILED with "Check Home Assistant connection and Supervisor
availability", which is neither accurate nor actionable: nothing is wrong with
the connection and the holder clears within seconds.

Retry the call inside one bounded wall-clock window with backoff. The window is
total rather than per-attempt, so a persistently blocked group cannot stretch a
call into N * timeout. The collision is matched as a case-insensitive substring
on Supervisor's own error text and every other failure raises on the first
attempt, so this never masks a real error - the same containment the test
runtime's dev-addon refresh already relies on for this collision.

This also removes a real source of e2e flakiness: the add-on lifecycle tests
drive stop/start/restart against a live Supervisor, so a sibling add-on job in
flight could fail the run with this error.

* fix(addons): anchor the collision marker and surface stuck-group guidance

Review follow-ups on the job-group retry:

- Anchor the marker on "another job is running for job group". Supervisor
  emits that phrasing from JobGroup.acquire for the transient per-add-on
  collision, but the job-level JobConcurrency.REJECT path raises a bare
  "Another job is running" for long OS/data-disk operations that must not
  ride a seconds-scale retry. No behaviour change for today's call sites
  (all add-on/store scoped), but the helper is generic.
- Give up with a job-specific ToolError instead of falling through to the
  generic handler, which attached "Check Home Assistant connection and
  Supervisor availability" - the exact misleading suggestion this PR exists
  to remove, on the one path where it matters most. The give-up now logs at
  WARNING with attempt count and elapsed time, and carries the stuck-job
  guidance plus Supervisor's original error text.
- Correct the comment: a genuine error arriving mid-retry raises immediately,
  not "on the first attempt". Drop the confusing N * timeout aside.
- Document the retry in the docstring (the call can block for the window) and
  fix the stale Returns line: every failure raises, it never returns an error
  dict.
- Tests: assert the real backoff schedule and its cap, pin the exact attempt
  count at the window bound, cover a real error mid-retry, the bare
  job-level rejection, timeout forwarding across retries, and that the
  give-up preserves Supervisor's text while dropping the connectivity hint.

---------

Co-authored-by: kingpanther13 <kingpanther13@users.noreply.github.com>

Author: kingpanther13

src/ha_mcp/tools/tools_addons.py

@@ -210,6 +210,19 @@ def _merge_options(base: dict, override: dict) -> dict:
     return merged
 
 
+# Supervisor's per-job-group rejection, matched case-insensitively, plus the
+# bounded window and backoff used to ride it out. See _supervisor_api_call.
+# The "for job group" tail is load-bearing: JobGroup.acquire raises
+# "Another job is running for job group <name>" for the transient per-add-on
+# collision, while the job-level JobConcurrency.REJECT path raises a bare
+# "Another job is running" for long operations (OS update, data-disk wipe)
+# that must NOT be retried on this schedule.
+_JOB_COLLISION_MARKER = "another job is running for job group"
+_JOB_COLLISION_RETRY_WINDOW = 60.0
+_JOB_COLLISION_RETRY_INITIAL_DELAY = 1.0
+_JOB_COLLISION_RETRY_MAX_DELAY = 5.0
+
+
 async def _supervisor_api_call(
     client: HomeAssistantClient,
     endpoint: str,
@@ -228,8 +241,13 @@ async def _supervisor_api_call(
         data: Optional request body data
         timeout: Optional timeout override
 
+    A transient Supervisor job-group collision is retried with backoff, so a
+    contended add-on endpoint can block for up to
+    ``_JOB_COLLISION_RETRY_WINDOW`` before returning or raising.
+
     Returns:
-        The "result" field from a successful response, or an error dict.
+        ``{"success": True, "result": ...}``. Every failure raises — this
+        never returns an error dict.
     """
     try:
         kwargs: dict[str, Any] = {"endpoint": endpoint, "method": method}
@@ -252,16 +270,78 @@ async def _supervisor_api_call(
         # ``{"success": False, "error": ...}``; re-raise it as the same
         # ``HomeAssistantCommandError`` the dedicated send_command used to raise
         # so the classifier below maps schema/not-found/etc. identically.
-        result = await client.send_websocket_message(
-            {"type": "supervisor/api", "_wait_timeout": wait_timeout, **kwargs}
-        )
+        #
+        # Supervisor serialises jobs per add-on job group and rejects a
+        # state-changing call outright while a still-settling job (a watchdog
+        # restart, a prior start/stop, a store reload) holds that group. The
+        # rejection happens before the job body runs, so nothing was applied
+        # and retrying cannot double-execute. The holder clears within seconds
+        # and the caller can do nothing useful with the failure, so ride it out
+        # inside one bounded window. The window is total rather than
+        # per-attempt, so a wedged group gives up once the window elapses
+        # instead of after some fixed retry count. Any other failure raises
+        # immediately, without retrying.
+        deadline = time.monotonic() + _JOB_COLLISION_RETRY_WINDOW
+        delay = _JOB_COLLISION_RETRY_INITIAL_DELAY
+        attempts = 0
+        while True:
+            attempts += 1
+            result = await client.send_websocket_message(
+                {"type": "supervisor/api", "_wait_timeout": wait_timeout, **kwargs}
+            )
+
+            if result.get("success"):
+                return {"success": True, "result": result.get("result", {})}
+
+            error_text = str(
+                result.get("error", f"Supervisor API call failed: {endpoint}")
+            )
+            if _JOB_COLLISION_MARKER not in error_text.lower():
+                raise HomeAssistantCommandError(error_text)
+
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                # A group still held after the whole window is wedged, not
+                # settling. Raise a ToolError here (the guard below re-raises
+                # it untouched) so the caller gets guidance about the stuck
+                # job instead of the generic connectivity suggestion the
+                # exception handler attaches to every other failure.
+                waited = _JOB_COLLISION_RETRY_WINDOW - remaining
+                logger.warning(
+                    "Supervisor job group still busy on %s after %.0fs "
+                    "(%d attempts); giving up: %s",
+                    endpoint,
+                    waited,
+                    attempts,
+                    error_text,
+                )
+                raise_tool_error(
+                    create_error_response(
+                        ErrorCode.SERVICE_CALL_FAILED,
+                        error_text,
+                        context={
+                            "endpoint": endpoint,
+                            "attempts": attempts,
+                            "waited_seconds": round(waited, 1),
+                        },
+                        suggestions=[
+                            "Another job has held this add-on's job group for "
+                            f"over {_JOB_COLLISION_RETRY_WINDOW:.0f}s — check "
+                            "Supervisor logs for a stuck or long-running job",
+                            "Retry once the in-flight add-on operation "
+                            "(install, update, restart or backup) finishes",
+                        ],
+                    )
+                )
 
-        if not result.get("success"):
-            raise HomeAssistantCommandError(
-                str(result.get("error", f"Supervisor API call failed: {endpoint}"))
+            logger.info(
+                "Supervisor job-group collision on %s; retrying in %.1fs (%s)",
+                endpoint,
+                delay,
+                error_text,
             )
-
-        return {"success": True, "result": result.get("result", {})}
+            await asyncio.sleep(min(delay, remaining))
+            delay = min(delay * 2, _JOB_COLLISION_RETRY_MAX_DELAY)
 
     except ToolError:
         raise

tests/src/unit/test_tools_addons.py

@@ -3768,6 +3768,191 @@ async def test_no_timeout_keeps_default_local_wait(self):
         assert "timeout" not in message
 
 
+class TestSupervisorJobGroupCollisionRetry:
+    """Supervisor serialises jobs per add-on job group.
+
+    A state-changing call lands on "Another job is running for job group
+    addon_<slug>" whenever a still-settling job holds that group. The holder
+    clears within seconds, so the call is retried inside one bounded window
+    instead of surfacing a failure the caller can do nothing with. Every other
+    failure must raise immediately, without retrying.
+    """
+
+    @staticmethod
+    def _collision(slug: str = "a0d7b954_appdaemon") -> dict:
+        return {
+            "success": False,
+            "error": f"Another job is running for job group addon_{slug}",
+        }
+
+    @pytest.mark.asyncio
+    async def test_job_group_collision_is_retried_until_it_clears(self):
+        """The transient collision is retried and the eventual success returned."""
+        from ha_mcp.tools.tools_addons import _supervisor_api_call
+
+        client = _make_mock_client()
+        client.send_websocket_message = AsyncMock(
+            side_effect=[
+                self._collision(),
+                self._collision(),
+                {"success": True, "result": {"state": "stopped"}},
+            ]
+        )
+
+        sleep = AsyncMock()
+        with patch("ha_mcp.tools.tools_addons.asyncio.sleep", new=sleep):
+            result = await _supervisor_api_call(
+                client, "/addons/a0d7b954_appdaemon/stop", method="POST"
+            )
+
+        assert result == {"success": True, "result": {"state": "stopped"}}
+        assert client.send_websocket_message.await_count == 3
+        # Backoff doubles from the initial delay; a flat or runaway schedule
+        # would still pass an outcome-only assertion.
+        assert [c.args[0] for c in sleep.await_args_list] == [1.0, 2.0]
+
+    @pytest.mark.asyncio
+    async def test_backoff_is_capped(self):
+        """Delay doubles but never exceeds the max — an uncapped schedule would
+        overshoot the window on a long-held group."""
+        from ha_mcp.tools.tools_addons import _supervisor_api_call
+
+        client = _make_mock_client()
+        client.send_websocket_message = AsyncMock(
+            side_effect=[self._collision()] * 5 + [{"success": True, "result": {}}]
+        )
+
+        sleep = AsyncMock()
+        with patch("ha_mcp.tools.tools_addons.asyncio.sleep", new=sleep):
+            await _supervisor_api_call(
+                client, "/addons/a0d7b954_appdaemon/stop", method="POST"
+            )
+
+        assert [c.args[0] for c in sleep.await_args_list] == [1.0, 2.0, 4.0, 5.0, 5.0]
+
+    @pytest.mark.asyncio
+    async def test_other_failures_raise_without_retrying(self):
+        """A non-collision failure must not be retried — it would mask the error."""
+        from ha_mcp.tools.tools_addons import _supervisor_api_call
+
+        client = _make_mock_client()
+        client.send_websocket_message = AsyncMock(
+            return_value={"success": False, "error": "Addon is not installed"}
+        )
+
+        with (
+            patch("ha_mcp.tools.tools_addons.asyncio.sleep", new=AsyncMock()),
+            pytest.raises(ToolError),
+        ):
+            await _supervisor_api_call(
+                client, "/addons/a0d7b954_appdaemon/stop", method="POST"
+            )
+
+        assert client.send_websocket_message.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_real_error_after_a_collision_raises_immediately(self):
+        """ "Raises immediately" is per-failure, not just on the literal first
+        attempt: a genuine error arriving mid-retry must not be ridden out."""
+        from ha_mcp.tools.tools_addons import _supervisor_api_call
+
+        client = _make_mock_client()
+        client.send_websocket_message = AsyncMock(
+            side_effect=[
+                self._collision(),
+                {"success": False, "error": "Addon is not installed"},
+                {"success": True, "result": {}},
+            ]
+        )
+
+        with (
+            patch("ha_mcp.tools.tools_addons.asyncio.sleep", new=AsyncMock()),
+            pytest.raises(ToolError) as exc_info,
+        ):
+            await _supervisor_api_call(
+                client, "/addons/a0d7b954_appdaemon/stop", method="POST"
+            )
+
+        assert client.send_websocket_message.await_count == 2
+        assert "not installed" in str(exc_info.value).lower()
+
+    @pytest.mark.asyncio
+    async def test_bare_job_level_rejection_is_not_retried(self):
+        """Supervisor's job-level REJECT ("Another job is running", no group
+        suffix) guards long OS/data-disk operations and must not ride this
+        seconds-scale schedule."""
+        from ha_mcp.tools.tools_addons import _supervisor_api_call
+
+        client = _make_mock_client()
+        client.send_websocket_message = AsyncMock(
+            return_value={"success": False, "error": "Another job is running"}
+        )
+
+        with (
+            patch("ha_mcp.tools.tools_addons.asyncio.sleep", new=AsyncMock()),
+            pytest.raises(ToolError),
+        ):
+            await _supervisor_api_call(client, "/os/update", method="POST")
+
+        assert client.send_websocket_message.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_retry_forwards_timeout_on_every_attempt(self):
+        """The Supervisor-side timeout and the extended local wait must ride
+        every retried attempt, not just the first."""
+        from ha_mcp.tools.tools_addons import _supervisor_api_call
+
+        client = _make_mock_client()
+        client.send_websocket_message = AsyncMock(
+            side_effect=[self._collision(), {"success": True, "result": {}}]
+        )
+
+        with patch("ha_mcp.tools.tools_addons.asyncio.sleep", new=AsyncMock()):
+            await _supervisor_api_call(
+                client,
+                "/addons/a0d7b954_appdaemon/restart",
+                method="POST",
+                timeout=120,
+            )
+
+        for call in client.send_websocket_message.await_args_list:
+            assert call.args[0]["timeout"] == 120
+            assert call.args[0]["_wait_timeout"] == 135.0
+
+    @pytest.mark.asyncio
+    async def test_retry_window_is_bounded(self):
+        """A group that never clears gives up rather than retrying forever, and
+        surfaces the stuck-job guidance instead of the generic connectivity
+        suggestion the exception handler attaches to every other failure."""
+        from ha_mcp.tools.tools_addons import _supervisor_api_call
+
+        client = _make_mock_client()
+        client.send_websocket_message = AsyncMock(return_value=self._collision())
+
+        # monotonic() is called once for the deadline, then once per failed
+        # attempt; 20s ticks exhaust the 60s window on the third attempt.
+        ticks = iter([0.0] + [float(i) * 20.0 for i in range(1, 40)])
+        with (
+            patch("ha_mcp.tools.tools_addons.asyncio.sleep", new=AsyncMock()),
+            patch(
+                "ha_mcp.tools.tools_addons.time.monotonic",
+                side_effect=lambda: next(ticks),
+            ),
+            pytest.raises(ToolError) as exc_info,
+        ):
+            await _supervisor_api_call(
+                client, "/addons/a0d7b954_appdaemon/stop", method="POST"
+            )
+
+        assert client.send_websocket_message.await_count == 3
+        payload = json.loads(str(exc_info.value))["error"]
+        # Supervisor's own text survives the give-up path...
+        assert "another job is running for job group" in payload["message"].lower()
+        # ...and the guidance names the stuck job, not the connection.
+        assert any("Supervisor logs" in s for s in payload["suggestions"])
+        assert not any("connection" in s.lower() for s in payload["suggestions"])
+
+
 class TestManageAddonActionMode:
     """Lifecycle (action) mode: install/start/stop/etc. via Supervisor."""