fix(sdks/python): propagate worker registration failures

frontend/docs/content/docs/reference/changelog/python.mdx

@@ -6,6 +6,12 @@ description: "Release notes for every Hatchet Python SDK release, newest first."
 
 {/* AUTOGENERATED — do not edit. Run `task sync-changelog` to regenerate from sdks/python/CHANGELOG.md */}
 
+## v1.43.3 - 2026-10-27
+
+### Fixed
+
+- Propagate worker registration failures without a secondary cleanup error, and fail the worker when its action listener exits unexpectedly so supervisors can restart it.
+
 ## v1.43.2 - 2026-10-27
 
 ### Changed

sdks/python/CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to Hatchet's Python SDK will be documented in this changelog
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.43.3] - 2026-10-27
+
+### Fixed
+
+- Propagate worker registration failures without a secondary cleanup error, and fail the worker when its action listener exits unexpectedly so supervisors can restart it.
+
 ## [1.43.2] - 2026-10-27
 
 ### Changed

sdks/python/hatchet_sdk/worker/action_listener_process.py

@@ -119,6 +119,7 @@ def __init__(
         self.listener: ActionListener | None = None
         self.killing = False
         self.action_loop_task: asyncio.Task[None] | None = None
+        self.blocked_main_loop: asyncio.Task[None] | None = None
         self.event_send_loop_task: asyncio.Task[None] | None = None
         self._stop_event_task: asyncio.Task[None] | None = None
         self.running_step_runs: dict[str, float] = {}
@@ -361,8 +362,7 @@ async def stop_health_server(self) -> None:
 
     async def start(self, retry_attempt: int = 0) -> None:
         if retry_attempt > 5:
-            logger.error("could not start action listener")
-            return
+            raise RuntimeError("could not start action listener")
 
         logger.debug(f"starting action listener: {self.name}")
 
@@ -383,7 +383,7 @@ async def start(self, retry_attempt: int = 0) -> None:
             self.worker_id_queue.put(self.listener.worker_id)
         except grpc.RpcError:
             logger.exception("could not start action listener")
-            return
+            raise
 
         # Start both loops as background tasks
         self.action_loop_task = asyncio.create_task(self.start_action_loop())
@@ -662,7 +662,11 @@ async def run() -> None:
             shared_slot_usage_by_pool=shared_slot_usage_by_pool,
         )
         await process.start_health_server()
-        await process.start()
+        try:
+            await process.start()
+        except BaseException:
+            await process.stop_health_server()
+            raise
 
         # Stay alive until the parent worker puts STOP_LOOP in the event_queue.
         # That happens only after all active task runs have finished and their

sdks/python/hatchet_sdk/worker/worker.py

@@ -686,7 +686,15 @@ async def _check_listener_health(self) -> None:
                     not self._action_listener_process
                     or not self._action_listener_process.is_alive()
                 ):
-                    logger.debug("child action listener process killed...")
+                    exitcode = (
+                        self._action_listener_process.exitcode
+                        if self._action_listener_process is not None
+                        else None
+                    )
+                    self._aio_start_exception = RuntimeError(
+                        f"action listener process exited unexpectedly (exit code: {exitcode})"
+                    )
+                    logger.error(str(self._aio_start_exception))
                     self._status = WorkerStatus.UNHEALTHY
                     if self._loop:
                         self._loop.create_task(self.exit_gracefully())

sdks/python/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "hatchet-sdk"
-version = "1.43.2"
+version = "1.43.3"
 description = "This is the official Python SDK for Hatchet, a distributed, fault-tolerant task queue. The SDK allows you to easily integrate Hatchet's task scheduling and workflow orchestration capabilities into your Python applications."
 readme = "README.md"
 license = { text = "MIT" }

sdks/python/tests/unit/test_worker_registration_failure.py

@@ -0,0 +1,221 @@
+import asyncio
+import multiprocessing
+from typing import cast
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import grpc
+import pytest
+
+from hatchet_sdk.worker.action_listener_process import (
+    WorkerActionListenerProcess,
+    worker_action_listener_process,
+)
+from hatchet_sdk.worker.worker import Worker, WorkerStatus
+
+LISTENER_MODULE = "hatchet_sdk.worker.action_listener_process"
+WORKER_MODULE = "hatchet_sdk.worker.worker"
+
+
+def _registration_error(
+    status: grpc.StatusCode = grpc.StatusCode.INTERNAL,
+) -> grpc.aio.AioRpcError:
+    return grpc.aio.AioRpcError(
+        status,
+        initial_metadata=grpc.aio.Metadata(),
+        trailing_metadata=grpc.aio.Metadata(),
+        details="registration failed",
+    )
+
+
+@pytest.mark.parametrize(
+    "status",
+    [
+        grpc.StatusCode.INTERNAL,
+        grpc.StatusCode.UNAVAILABLE,
+        grpc.StatusCode.UNAUTHENTICATED,
+    ],
+)
+async def test_registration_failure_propagates_without_starting_tasks(
+    status: grpc.StatusCode,
+) -> None:
+    config = MagicMock()
+    config.healthcheck.enabled = False
+    with patch(f"{LISTENER_MODULE}.Client"):
+        process = WorkerActionListenerProcess(
+            name="test-worker",
+            actions=["test-action"],
+            slot_config={"default": 1},
+            config=config,
+            action_queue=MagicMock(),
+            event_queue=MagicMock(),
+            handle_kill=False,
+            debug=False,
+            labels=[],
+            worker_id_queue=MagicMock(),
+            stop_event=MagicMock(),
+        )
+
+    error = _registration_error(status)
+    dispatcher = MagicMock()
+    dispatcher.get_action_listener = AsyncMock(side_effect=error)
+    with patch(f"{LISTENER_MODULE}.DispatcherClient", return_value=dispatcher):
+        with pytest.raises(grpc.aio.AioRpcError) as caught:
+            await process.start()
+
+    assert caught.value is error
+    assert process.listener is None
+    assert process.action_loop_task is None
+    assert process.event_send_loop_task is None
+    assert process.blocked_main_loop is None
+    assert process._stop_event_task is None
+    cast(MagicMock, process.worker_id_queue).put.assert_not_called()
+    process.finalize_listener_cleanup()
+
+
+def test_registration_failure_closes_health_server() -> None:
+    process = MagicMock()
+    process.start_health_server = AsyncMock()
+    error = _registration_error()
+    process.start = AsyncMock(side_effect=error)
+    process.stop_health_server = AsyncMock()
+
+    with patch(f"{LISTENER_MODULE}.WorkerActionListenerProcess", return_value=process):
+        with pytest.raises(grpc.aio.AioRpcError) as caught:
+            worker_action_listener_process(
+                name="test-worker",
+                actions=["test-action"],
+                slot_config={"default": 1},
+                config=MagicMock(),
+                action_queue=MagicMock(),
+                event_queue=MagicMock(),
+                handle_kill=False,
+                debug=False,
+                labels=[],
+                worker_id_queue=MagicMock(),
+                stop_event=MagicMock(),
+            )
+
+    assert caught.value is error
+    process.start_health_server.assert_awaited_once()
+    process.stop_health_server.assert_awaited_once()
+
+
+def _failed_registration_subprocess() -> None:
+    config = MagicMock()
+    config.healthcheck.enabled = False
+    dispatcher = MagicMock()
+    dispatcher.get_action_listener = AsyncMock(side_effect=_registration_error())
+    with (
+        patch(f"{LISTENER_MODULE}.Client"),
+        patch(f"{LISTENER_MODULE}.DispatcherClient", return_value=dispatcher),
+    ):
+        worker_action_listener_process(
+            name="test-worker",
+            actions=["test-action"],
+            slot_config={"default": 1},
+            config=config,
+            action_queue=MagicMock(),
+            event_queue=MagicMock(),
+            handle_kill=False,
+            debug=False,
+            labels=[],
+            worker_id_queue=MagicMock(),
+            stop_event=MagicMock(),
+        )
+
+
+def test_registration_failure_exits_subprocess_with_original_error(
+    capfd: pytest.CaptureFixture[str],
+) -> None:
+    process = multiprocessing.get_context("spawn").Process(
+        target=_failed_registration_subprocess
+    )
+    process.start()
+    try:
+        process.join(timeout=10)
+        assert process.exitcode == 1
+        output = capfd.readouterr()
+        assert "registration failed" in output.err
+        assert "AttributeError" not in output.err
+    finally:
+        if process.is_alive():
+            process.kill()
+            process.join(timeout=5)
+        process.close()
+
+
+@pytest.mark.parametrize("handle_kill", [False, True])
+@pytest.mark.parametrize("listener_exitcode", [0, 1, -9])
+def test_worker_start_propagates_listener_death_after_cleanup(
+    handle_kill: bool, listener_exitcode: int
+) -> None:
+    worker = _make_worker(handle_kill)
+    listener = MagicMock()
+    listener.is_alive.return_value = False
+    listener.exitcode = listener_exitcode
+    worker._action_listener_process = listener
+
+    try:
+        with patch.object(worker, "_aio_start", new=lambda: _monitor(worker)):
+            with pytest.raises(
+                RuntimeError, match="action listener process exited unexpectedly"
+            ):
+                worker.start()
+        assert worker.status == WorkerStatus.UNHEALTHY
+        cast(MagicMock, worker._action_runner).exit_gracefully.assert_awaited_once()
+        cast(MagicMock, worker._action_runner).cleanup.assert_called_once()
+        cast(AsyncMock, worker._cleanup_lifespan).assert_awaited_once()
+    finally:
+        _cleanup_worker(worker)
+
+
+def test_task_limit_shutdown_exits_successfully() -> None:
+    worker = _make_worker(handle_kill=True)
+    worker._config.terminate_worker_after_num_tasks = 1
+    listener = MagicMock()
+    listener.is_alive.return_value = True
+    worker._action_listener_process = listener
+    try:
+        with (
+            patch.object(worker, "_aio_start", new=lambda: _monitor(worker)),
+            patch(f"{WORKER_MODULE}.task_count", value=1),
+        ):
+            with pytest.raises(SystemExit) as caught:
+                worker.start()
+        assert caught.value.code == 0
+        assert worker._aio_start_exception is None
+        cast(MagicMock, worker._action_runner).exit_gracefully.assert_awaited_once()
+    finally:
+        _cleanup_worker(worker)
+
+
+def _make_worker(handle_kill: bool) -> Worker:
+    config = MagicMock()
+    config.apply_namespace.return_value = "test-worker"
+    config.healthcheck.enabled = False
+    config.terminate_worker_after_num_tasks = None
+    with (
+        patch(f"{WORKER_MODULE}.Client"),
+        patch.object(Worker, "_setup_signal_handlers"),
+    ):
+        worker = Worker("test-worker", config, {"default": 1}, handle_kill=handle_kill)
+    worker._pause_task_assignment = AsyncMock()  # type: ignore[method-assign]
+    worker._cleanup_lifespan = AsyncMock()  # type: ignore[method-assign]
+    runner = MagicMock()
+    runner.exit_gracefully = AsyncMock()
+    worker._action_runner = runner
+    return worker
+
+
+async def _monitor(worker: Worker) -> None:
+    worker._action_listener_health_check = asyncio.create_task(
+        worker._check_listener_health()
+    )
+    await worker._action_listener_health_check
+
+
+def _cleanup_worker(worker: Worker) -> None:
+    worker._close_queues()
+    worker._worker_id_queue.close()
+    if worker._loop is not None:
+        worker._loop.close()