fix(worker): exit when connection fails after max retries

livekit-agents/livekit/agents/worker.py

@@ -827,6 +827,15 @@ async def _load_task() -> None:
                 )
                 tasks.append(self._conn_task)
 
+                # propagate a terminal connection failure out of run()
+                def _on_conn_task_done(task: asyncio.Task[None]) -> None:
+                    if task.cancelled() or self._close_future is None or self._close_future.done():
+                        return
+                    if (exc := task.exception()) is not None:
+                        self._close_future.set_exception(exc)
+
+                self._conn_task.add_done_callback(_on_conn_task_done)
+
             self.emit("worker_started")
 
         await self._close_future
@@ -991,7 +1000,9 @@ async def aclose(self) -> None:
         async with self._lock:
             if self._closed:
                 if self._close_future is not None:
-                    await self._close_future
+                    # _close_future may hold run()'s error; keep aclose() a no-op
+                    with contextlib.suppress(Exception):
+                        await self._close_future
                 return
 
             self._closed = True

tests/test_worker_connection.py

@@ -0,0 +1,52 @@
+"""Connection-failure propagation tests.
+
+When the worker's connection task exhausts ``max_retry`` it raises, and that
+failure must surface out of ``AgentServer.run()`` instead of leaving the worker
+hanging forever (https://github.com/livekit/agents/issues/6083).
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from livekit.agents.job import JobContext
+from livekit.agents.worker import AgentServer
+
+pytestmark = pytest.mark.unit
+
+
+def _make_server() -> AgentServer:
+    server = AgentServer(
+        ws_url="ws://127.0.0.1:1",  # unreachable: connection refused
+        api_key="devkey",
+        api_secret="devsecret",
+        max_retry=0,
+        num_idle_processes=0,
+    )
+
+    @server.rtc_session()
+    async def _entry(ctx: JobContext) -> None:
+        pass
+
+    server._simulation = True  # skip binding the health HTTP server
+    return server
+
+
+async def test_run_raises_when_connection_exhausts_retries() -> None:
+    server = _make_server()
+
+    fake_pool = MagicMock()
+    fake_pool.start = AsyncMock()
+    fake_pool.aclose = AsyncMock()
+    fake_pool.processes = []
+
+    with patch("livekit.agents.ipc.proc_pool.ProcPool", return_value=fake_pool):
+        try:
+            with pytest.raises(RuntimeError, match="failed to connect"):
+                await asyncio.wait_for(server.run(devmode=True), timeout=10)
+        finally:
+            await server.aclose()
+            await server.aclose()  # repeated aclose() stays a no-op