[controllers] fix: surface keep worker startup failures

AGENTS.md

@@ -88,6 +88,9 @@ This file defines how coding agents should work in this repository.
 - Hardware probes must clean up vendor libraries after detection (for example, NVML shutdown and ROCm SMI shutdown after init).
 - ROCm/HIP PyTorch builds take precedence over NVML-based CUDA fallback: if `torch.version.hip` is truthy, `_check_cuda()` must not classify the runtime as CUDA or probe NVML.
 - Keep lifecycle state truthful: a reserved starting session must be visible in status as `state="starting"`; a session is removed only after release succeeds; timed-out or failed stops must stay visible with state and error details.
+- Single-GPU `keep()` must not report success until fatal backend startup setup
+  has succeeded. CUDA/ROCm worker startup failures such as `set_device` errors
+  must propagate synchronously so services cannot register false active sessions.
 - Keep service daemon ownership safe: no stop, force-stop, or fallback path may signal a PID unless the auto-start ownership record verifies the running process.
 - Treat custom `job_id` values as reserved from the moment startup begins; duplicate starts must fail before another controller can begin keep-alive work.
 - Keep custom `job_id` validation centralized in `session_config.py`: only `None` means omitted/all-sessions, and custom IDs must be non-empty URL-path-safe strings before any session state changes.

docs/concepts/architecture.md

@@ -40,7 +40,9 @@ CLI args ──▶ GlobalGPUController ──▶ [CudaGPUController rank=0]
    If a later worker fails to start, already-started workers are released before
    the original start error is re-raised.
 3. Each CUDA worker:
-   - Starts a daemon thread that performs intervalled lightweight elementwise batches.
+   - Starts a daemon thread and confirms fatal backend startup setup before
+     `keep()` reports success.
+   - Performs intervalled lightweight elementwise batches after startup.
    - Calls `_monitor_utilization` (by way of NVML) to detect real activity
      before allocating the keep tensor.
    - Allocates a tensor sized by way of `vram_to_keep` only when backoff allows
@@ -88,8 +90,9 @@ Elementwise keep-alive batches:
   `stop_keep` removes a session only after release succeeds. Timed-out sessions
   stay visible as `state="stopping"` until the background release finishes;
   failed releases stay visible as `state="stop_failed"` with `last_error`.
-- Errors inside a worker are logged but do not bring the whole process down;
-  the loop retries after clearing the CUDA cache.
+- Fatal backend startup errors are reported before `keep()` returns. Later
+  runtime errors inside an already-started worker are logged; recoverable
+  allocation failures retry after clearing the device cache.
 
 ## Platform detection
 

docs/guides/python.md

@@ -45,7 +45,9 @@ ctrl.release()
 ```
 
 The controller spins up a daemon thread. Repeated `keep()` calls are idempotent
-and simply warn if the worker is already running.
+and simply warn if the worker is already running. CUDA and ROCm `keep()` calls
+return only after fatal backend startup setup succeeds, so startup failures such
+as device-selection errors are raised before your guarded work begins.
 
 ## Guard multiple GPUs with a single context
 

docs/plans/controller-startup-failures.md

@@ -0,0 +1,46 @@
+# Controller Startup Failures Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Ensure KeepGPU never reports a CUDA or ROCm keep session as started when the worker thread immediately fails during backend startup.
+
+**Architecture:** Add a small startup handshake between `keep()` and the background worker for CUDA and ROCm controllers. The worker should signal either "backend startup is ready" after `torch.cuda.set_device(...)` succeeds or return the startup exception to `keep()` before public APIs report success. This keeps the service layer honest because `GlobalGPUController.keep()` will fail and roll back instead of registering an active session.
+
+**Tech Stack:** Python, pytest, PyTorch controller classes, KeepGPU MCP/JSON-RPC service.
+
+---
+
+## Background
+
+CUDA and ROCm single-GPU controllers currently call `torch.cuda.set_device(...)`
+inside the daemon worker thread. If that backend startup fails, `keep()` has
+already returned and the service can register an active job even though no
+keepalive worker is running.
+
+## Solution
+
+- Add RED tests that reproduce CUDA and ROCm startup failure without requiring a
+  GPU by monkeypatching `torch.cuda.set_device`.
+- Add a JSON-RPC/service regression test showing failed CUDA worker startup must
+  not leave an active session.
+- Implement a minimal startup handshake in CUDA and ROCm controllers.
+- Document the invariant in `AGENTS.md` and API guidance.
+
+## Todo
+
+- [x] Run targeted controller/service baseline.
+- [x] Add failing controller startup tests.
+- [x] Add failing service no-active-session regression test.
+- [x] Verify the new tests fail on current behavior.
+- [x] Implement the startup handshake.
+- [x] Verify focused tests pass.
+- [x] Update docs and `AGENTS.md`.
+- [x] Run broader tests, docs build, pre-commit, and local subagent review.
+- [ ] Open a PR, resolve review comments, squash merge, and clean up the branch.
+
+## Verification
+
+- `PYTHONPATH=$PWD/src pytest tests/cuda_controller tests/rocm_controller tests/mcp/test_server.py -q`
+- `PYTHONPATH=$PWD/src pytest tests -q`
+- `PYTHONPATH=$PWD/src mkdocs build`
+- `pre-commit run --all-files`

docs/reference/api.md

@@ -31,6 +31,11 @@ and busy or unavailable telemetry sleeps before allocating keep tensors or
 running compute. Pass `busy_threshold=-1` only when you intentionally want
 unconditional keepalive compute without utilization backoff.
 
+CUDA and ROCm `keep()` calls wait for fatal backend startup setup to succeed
+before reporting success. Startup failures such as device-selection errors are
+raised synchronously, while normal low-power allocation can still be deferred by
+utilization backoff after startup succeeds.
+
 Single-GPU workload iteration controls must be positive integers. CUDA exposes
 `relu_iterations`; ROCm and Mac M expose `iterations`. Non-integer values raise
 `TypeError`, and non-positive values raise `ValueError` before a worker can

src/keep_gpu/single_gpu_controller/cuda_gpu_controller.py

@@ -98,6 +98,10 @@ def parse_size(text: str) -> int:
     def keep(self) -> None:
         """Launch the background thread that keeps the GPU busy."""
         if self._thread and self._thread.is_alive():
+            if self._stop_evt is not None and self._stop_evt.is_set():
+                raise RuntimeError(
+                    f"rank {self.rank}: previous keep thread startup did not complete"
+                )
             logger.warning("rank %s: keep thread already running", self.rank)
             return
 
@@ -106,12 +110,36 @@ def keep(self) -> None:
             raise ValueError("vram_to_keep must be positive")
 
         self._stop_evt = threading.Event()
+        startup_evt = threading.Event()
+        startup_errors: list[Exception] = []
         self._thread = threading.Thread(
             target=self._keep_loop,
+            args=(startup_evt, startup_errors),
             name=f"gpu-keeper-{self.rank}",
             daemon=True,  # daemon so program can exit cleanly
         )
         self._thread.start()
+        startup_timeout = 5.0
+        if not startup_evt.wait(startup_timeout):
+            stop_evt = self._stop_evt
+            if stop_evt is not None:
+                stop_evt.set()
+            self._thread.join(timeout=1.0)
+            if self._thread.is_alive():
+                raise RuntimeError(
+                    f"rank {self.rank}: keep thread did not complete startup within "
+                    f"{startup_timeout:.1f}s"
+                )
+            self._thread = None
+            self._stop_evt = None
+            raise RuntimeError(
+                f"rank {self.rank}: keep thread exited before startup completed"
+            )
+        if startup_errors:
+            self._thread.join(timeout=1.0)
+            self._thread = None
+            self._stop_evt = None
+            raise startup_errors[0]
         logger.info("rank %s: keep thread started", self.rank)
 
     def release(self) -> None:
@@ -154,21 +182,45 @@ def __exit__(self, exc_type, exc, tb):
     # ------------------------------------------------------------------
     # Background loop
     # ------------------------------------------------------------------
-    def _keep_loop(self) -> None:
+    def _keep_loop(
+        self,
+        startup_evt: Optional[threading.Event] = None,
+        startup_errors: Optional[list[Exception]] = None,
+    ) -> None:
         """Internal: run workloads until stop event is set."""
         stop_evt = self._stop_evt
         if stop_evt is None:
-            logger.error("rank %s: stop event not initialized", self.rank)
+            exc = RuntimeError(f"rank {self.rank}: stop event not initialized")
+            logger.error("%s", exc)
+            if startup_errors is not None:
+                startup_errors.append(exc)
+            if startup_evt is not None:
+                startup_evt.set()
             return
         assert stop_evt is not None
 
-        torch.cuda.set_device(self.rank)
+        try:
+            torch.cuda.set_device(self.rank)
+        except Exception as exc:  # noqa: BLE001 - surface backend startup failure
+            logger.error("rank %s: CUDA startup failed: %s", self.rank, exc)
+            if startup_errors is not None:
+                startup_errors.append(exc)
+            if startup_evt is not None:
+                startup_evt.set()
+            return
         num_elements = self._num_elements if self._num_elements is not None else 0
         if num_elements <= 0:
-            logger.error(
-                "rank %s: invalid vram_to_keep=%s", self.rank, self.vram_to_keep
+            exc = RuntimeError(
+                f"rank {self.rank}: invalid vram_to_keep={self.vram_to_keep}"
             )
+            logger.error("%s", exc)
+            if startup_errors is not None:
+                startup_errors.append(exc)
+            if startup_evt is not None:
+                startup_evt.set()
             return
+        if startup_evt is not None:
+            startup_evt.set()
         matrix = None
         while not stop_evt.is_set():
             try:

src/keep_gpu/single_gpu_controller/rocm_gpu_controller.py

@@ -54,6 +54,10 @@ def __init__(
 
     def keep(self) -> None:
         if self._thread and self._thread.is_alive():
+            if self._stop_evt is not None and self._stop_evt.is_set():
+                raise RuntimeError(
+                    f"rank {self.rank}: previous keep thread startup did not complete"
+                )
             logger.warning("rank %s: keep thread already running", self.rank)
             return
         self._failure_exc = None
@@ -67,14 +71,48 @@ def keep(self) -> None:
                 logger.debug("rsmi_init failed: %s", exc)
 
         self._stop_evt = threading.Event()
+        startup_evt = threading.Event()
+        startup_errors: list[Exception] = []
         self._thread = threading.Thread(
             target=self._keep_loop,
+            args=(startup_evt, startup_errors),
             name=f"gpu-keeper-rocm-{self.rank}",
             daemon=True,
         )
         self._thread.start()
+        startup_timeout = 5.0
+        if not startup_evt.wait(startup_timeout):
+            stop_evt = self._stop_evt
+            if stop_evt is not None:
+                stop_evt.set()
+            self._thread.join(timeout=1.0)
+            if self._thread.is_alive():
+                self._shutdown_rocm_smi()
+                raise RuntimeError(
+                    f"rank {self.rank}: ROCm keep thread did not complete startup "
+                    f"within {startup_timeout:.1f}s"
+                )
+            self._thread = None
+            self._stop_evt = None
+            self._shutdown_rocm_smi()
+            raise RuntimeError(
+                f"rank {self.rank}: ROCm keep thread exited before startup completed"
+            )
+        if startup_errors:
+            self._thread.join(timeout=1.0)
+            self._thread = None
+            self._stop_evt = None
+            self._shutdown_rocm_smi()
+            raise startup_errors[0]
         logger.info("rank %s: ROCm keep thread started", self.rank)
 
+    def _shutdown_rocm_smi(self) -> None:
+        if self._rocm_smi:
+            try:
+                self._rocm_smi.rsmi_shut_down()
+            except Exception as exc:  # noqa: BLE001  # pragma: no cover - best effort
+                logger.debug("rsmi_shut_down failed: %s", exc)
+
     def release(self) -> None:
         try:
             if self._thread and self._thread.is_alive():
@@ -99,11 +137,7 @@ def release(self) -> None:
                 logger.warning("rank %s: keep thread not running", self.rank)
                 return
         finally:
-            if self._rocm_smi:
-                try:
-                    self._rocm_smi.rsmi_shut_down()
-                except Exception as exc:  # pragma: no cover - best effort
-                    logger.debug("rsmi_shut_down failed: %s", exc)
+            self._shutdown_rocm_smi()
         logger.info("rank %s: keep thread stopped & cache cleared", self.rank)
 
     def __enter__(self):
@@ -126,22 +160,46 @@ def _query_utilization(self) -> Optional[int]:
             logger.debug("ROCm utilization query failed: %s", exc)
             return None
 
-    def _keep_loop(self) -> None:
+    def _keep_loop(
+        self,
+        startup_evt: Optional[threading.Event] = None,
+        startup_errors: Optional[list[Exception]] = None,
+    ) -> None:
         stop_evt = self._stop_evt
         if stop_evt is None:
-            logger.error("rank %s: stop event not initialized", self.rank)
+            exc = RuntimeError(f"rank {self.rank}: stop event not initialized")
+            logger.error("%s", exc)
+            if startup_errors is not None:
+                startup_errors.append(exc)
+            if startup_evt is not None:
+                startup_evt.set()
             return
         assert stop_evt is not None
 
-        torch.cuda.set_device(self.rank)
+        try:
+            torch.cuda.set_device(self.rank)
+        except Exception as exc:  # noqa: BLE001 - surface backend startup failure
+            logger.error("rank %s: ROCm startup failed: %s", self.rank, exc)
+            if startup_errors is not None:
+                startup_errors.append(exc)
+            if startup_evt is not None:
+                startup_evt.set()
+            return
         tensor = None
         attempts = 0
         num_elements = self._num_elements if self._num_elements is not None else 0
         if num_elements <= 0:
-            logger.error(
-                "rank %s: invalid vram_to_keep=%s", self.rank, self.vram_to_keep
+            exc = RuntimeError(
+                f"rank {self.rank}: invalid vram_to_keep={self.vram_to_keep}"
             )
+            logger.error("%s", exc)
+            if startup_errors is not None:
+                startup_errors.append(exc)
+            if startup_evt is not None:
+                startup_evt.set()
             return
+        if startup_evt is not None:
+            startup_evt.set()
         while not stop_evt.is_set():
             try:
                 util = self._query_utilization()

tests/cuda_controller/test_keep_and_release.py

@@ -1,11 +1,52 @@
 import pytest
+import threading
 import time
 import torch
 
 from keep_gpu.single_gpu_controller.cuda_gpu_controller import CudaGPUController
 from tests.polling import wait_until
 
 
+def test_cuda_keep_raises_when_worker_startup_fails(monkeypatch):
+    import keep_gpu.single_gpu_controller.cuda_gpu_controller as cuda_module
+
+    ctrl = CudaGPUController(
+        rank=0,
+        interval=0.01,
+        vram_to_keep=4,
+        busy_threshold=-1,
+    )
+
+    def fail_set_device(_rank):
+        raise RuntimeError("cuda startup failed")
+
+    monkeypatch.setattr(cuda_module.torch.cuda, "set_device", fail_set_device)
+
+    with pytest.raises(RuntimeError, match="cuda startup failed"):
+        ctrl.keep()
+
+    assert not (ctrl._thread and ctrl._thread.is_alive())
+
+
+def test_cuda_keep_rejects_retry_while_startup_thread_is_stopping():
+    class AliveThread:
+        def is_alive(self):
+            return True
+
+    ctrl = CudaGPUController(
+        rank=0,
+        interval=0.01,
+        vram_to_keep=4,
+        busy_threshold=-1,
+    )
+    ctrl._thread = AliveThread()
+    ctrl._stop_evt = threading.Event()
+    ctrl._stop_evt.set()
+
+    with pytest.raises(RuntimeError, match="startup did not complete"):
+        ctrl.keep()
+
+
 @pytest.mark.skipif(
     not torch.cuda.is_available(),
     reason="Only run CUDA tests when CUDA is available",