thermal: front-driven central supervisor with cooperative pause + SIGSTOP escalation

The thermal guard was checkpoint-based and ran inside each engine, so it went
blind during a single long native call (llama.cpp prefill / image encode): the
engine can't reach a between-token checkpoint, and the [thermal][debug] heartbeat
stops exactly when the GPU is hottest.

Move the AUTHORITATIVE monitor to the front, which stays responsive regardless of
what an engine is doing:

* Front (engine_supervisor): a thermal thread reads per-card temps via gpu_stats()
  + CPU temp, maps each card to the owning engine by its CODERAI_ENGINE_GPUS
  selectors, and drives pause/resume with hysteresis (pause at *_high, resume only
  back at *_resume). A hot GPU pauses just its engine; a hot CPU pauses all.
* Engines stop cooperatively, as before, but triggered remotely: the front POSTs
  /internal/thermal-pause; thermal.set_external_pause() makes wait_until_safe()/
  checkpoint() block at the next safe point (publishing cooldown state so the Tasks
  page shows it), until /internal/thermal-resume.
* Escalation: if a paused engine keeps generating (inflight > 0) — stuck in a
  native call it can't interrupt — for stop_escalate_checks (default 3) consecutive
  checks, the front SIGSTOPs the engine's process group; SIGCONT on cooldown. Both
  signals target the session group so children freeze too.
* stop_all()/restart_engine() SIGCONT a frozen engine first (a stopped process
  ignores SIGTERM until continued); _spawn() resets the thermal flags.
* Config: thermal.supervisor_enabled (default on), thermal.stop_escalate_checks.
* UI: per-engine temp + pause/frozen state in engines_list and the Tasks cooldown
  banner (covers a SIGSTOPped engine that can't report its own cooldown).

Also: exclude coderai-runtime/*-runtime from the Docker build context (.dockerignore)
— a root-owned runtime temp file was breaking the image build.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RdMufYvtTbtGDWsiZVoXce

.dockerignore

@@ -15,6 +15,8 @@ __pycache__
 !.packaging-cache/build-manifest.json
 
 # Large runtime/generated data
+coderai-runtime
+*-runtime
 models
 offload
 township_output

codai/api/app.py

@@ -331,8 +331,45 @@ async def internal_engine_state():
                        "message": cs.get("message")}
     except Exception:
         cooling = None
+    # Whether the front's thermal supervisor currently holds this engine in a
+    # cooperative pause (so the front can confirm the engine acknowledged it).
+    paused = False
+    try:
+        from codai.models import thermal as _therm
+        paused = _therm.external_pause_active()
+    except Exception:
+        paused = False
     return {"ok": True, "pid": _os.getpid(), "loaded_models": loaded,
-            "vram": vram, "tasks": tasks, "cooling": cooling}
+            "vram": vram, "tasks": tasks, "cooling": cooling, "paused": paused}
+
+
+@app.post("/internal/thermal-pause", include_in_schema=False)
+async def internal_thermal_pause(request: Request):
+    """Front-driven thermal pause: the supervisor (which monitors temps centrally)
+    asks this engine to stop generating. Honoured cooperatively at the next safe
+    point — request entry / between tokens — via codai.models.thermal."""
+    import json as _json
+    try:
+        data = _json.loads((await request.body()) or b"{}") or {}
+    except Exception:
+        data = {}
+    try:
+        from codai.models import thermal as _therm
+        _therm.set_external_pause(True, reason=str(data.get("reason") or "thermal"))
+    except Exception as e:
+        return {"ok": False, "error": str(e)}
+    return {"ok": True, "paused": True}
+
+
+@app.post("/internal/thermal-resume", include_in_schema=False)
+async def internal_thermal_resume(request: Request):
+    """Lift the front-driven thermal pause once the hardware has cooled."""
+    try:
+        from codai.models import thermal as _therm
+        _therm.set_external_pause(False)
+    except Exception as e:
+        return {"ok": False, "error": str(e)}
+    return {"ok": True, "paused": False}
 
 
 @app.post("/internal/reload-config", include_in_schema=False)

codai/config.py

@@ -248,6 +248,14 @@ class ThermalConfig:
     soft_throttle_enabled: bool = False
     soft_throttle_temp: float = 80.0       # engage at/above this temperature (°C)
     soft_throttle_max_sleep: float = 3.0   # max seconds to sleep/checkpoint at the limit
+    # Front-driven thermal supervision. The front proxy monitors temperatures
+    # centrally (it stays responsive even while an engine is GIL-blocked in a long
+    # native call) and tells engines to pause/resume cooperatively. When an engine
+    # ignores a pause — stuck in a native call it can't interrupt — for
+    # `stop_escalate_checks` consecutive monitor checks, the front escalates to an
+    # OS-level SIGSTOP (and SIGCONT to resume after cooldown).
+    supervisor_enabled: bool = True
+    stop_escalate_checks: int = 3
 
 
 @dataclass
@@ -702,6 +710,8 @@ class ConfigManager:
                 "soft_throttle_enabled": self.config.thermal.soft_throttle_enabled,
                 "soft_throttle_temp": self.config.thermal.soft_throttle_temp,
                 "soft_throttle_max_sleep": self.config.thermal.soft_throttle_max_sleep,
+                "supervisor_enabled": self.config.thermal.supervisor_enabled,
+                "stop_escalate_checks": self.config.thermal.stop_escalate_checks,
             },
             "jobs": {
                 "resume_on_restart": self.config.jobs.resume_on_restart,

codai/frontproxy/app.py

@@ -957,6 +957,9 @@ class FrontProxy:
             out.append({"id": e.id, "name": e.name, "backend": e.backend,
                         "gpu": e.gpu, "healthy": e.healthy, "primary": e.primary,
                         "vram": e.vram, "cooling": bool(e.cooling),
+                        "temp": getattr(e, "therm_temp", None),
+                        "thermal_paused": bool(getattr(e, "therm_paused", False)),
+                        "thermal_frozen": bool(getattr(e, "therm_sigstopped", False)),
                         "loaded_models": self._canonical_loaded(e.loaded_models),
                         "pid": pid})
         return out
@@ -1107,13 +1110,24 @@ class FrontProxy:
         return resp
 
     def _cooling_engines(self) -> list:
-        """Which engines are in thermal cooldown right now (for the Tasks banner)."""
+        """Which engines are in thermal cooldown right now (for the Tasks banner).
+
+        Covers both a locally-reported cooldown (engine.cooling, from the engine's
+        own checkpoint loop) and a front-driven supervisor pause (engine.therm_paused)
+        — the latter matters when the engine is SIGSTOPped and can't report at all."""
         out = []
         for e in self.registry.all():
             if e.cooling:
                 out.append({"engine": e.name, "gpu": e.cooling.get("gpu"),
                             "cpu": e.cooling.get("cpu"),
                             "message": e.cooling.get("message")})
+            elif getattr(e, "therm_paused", False):
+                out.append({
+                    "engine": e.name, "gpu": getattr(e, "therm_temp", None),
+                    "cpu": None,
+                    "message": ("thermal pause (frozen)"
+                                if getattr(e, "therm_sigstopped", False)
+                                else "thermal pause")})
         return out
 
     @staticmethod

codai/frontproxy/engine_supervisor.py

@@ -348,6 +348,11 @@ class EngineSupervisor:
             preexec_fn=_engine_preexec if os.name == "posix" else None,
         )
         engine.proc = proc
+        # A freshly (re)spawned engine is never paused/frozen — clear any thermal
+        # supervision state left over from the previous process.
+        engine.therm_paused = False
+        engine.therm_sigstopped = False
+        engine.therm_stop_checks = 0
         # Enlarge the kernel pipe buffer for this engine's stdout. The engine
         # prints (incl. large --debug dumps) synchronously from its event-loop
         # thread; if this pipe fills before the pump thread drains it, that
@@ -477,6 +482,14 @@ class EngineSupervisor:
         self._spawn(sysw)
         self._poll_thread = threading.Thread(target=self._poll_loop, daemon=True)
         self._poll_thread.start()
+        # Central thermal supervisor: monitor temps from the front (responsive even
+        # when an engine is GIL-blocked) and drive cooperative pause/resume, with a
+        # SIGSTOP escalation for an engine that can't honour the pause in time.
+        self._thermal_thread = None
+        if getattr(self.config.thermal, "supervisor_enabled", True):
+            self._thermal_thread = threading.Thread(target=self._thermal_loop,
+                                                    daemon=True)
+            self._thermal_thread.start()
         atexit.register(self.stop_all)
 
     def _push_assignment_if_changed(self, client) -> None:
@@ -540,6 +553,189 @@ class EngineSupervisor:
                 print(f"[front] reload-config push to '{e.name}' failed: {exc}; requeued",
                       flush=True)
 
+    # ------------------------------------------------------------- thermal monitor
+    def _thermal_settings(self):
+        """Build ThermalSettings from the live config.thermal (same thresholds the
+        engines would use), so the front and engines agree on what 'hot' means."""
+        from codai.models.thermal import ThermalSettings
+        tc = self.config.thermal
+        return ThermalSettings(
+            cpu_enabled=tc.cpu_enabled, gpu_enabled=tc.gpu_enabled,
+            cpu_high=tc.cpu_high, cpu_resume=tc.cpu_resume,
+            gpu_high=tc.gpu_high, gpu_resume=tc.gpu_resume,
+            gpu_overrides=getattr(tc, "gpu_overrides", None),
+            poll_seconds=tc.poll_seconds)
+
+    @staticmethod
+    def _engine_cards(engine, all_cards) -> list:
+        """The physical cards this engine owns, by its CODERAI_ENGINE_GPUS selectors
+        — the same matching rule the engine's own thermal check uses. No scope
+        (single-engine / legacy) → it owns every card."""
+        raw = (engine.env or {}).get("CODERAI_ENGINE_GPUS", "") or ""
+        sels = {s.strip() for s in raw.split(",") if s.strip()}
+        if not sels:
+            return list(all_cards)
+        return [c for c in all_cards
+                if c.get("vendor") in sels or (c.get("uuid") and c["uuid"] in sels)]
+
+    def _thermal_signal(self, engine, sig) -> None:
+        """Send a signal to the engine's whole process group (it runs in its own
+        session via setsid), so a SIGSTOP/SIGCONT freezes/resumes the engine and any
+        children (whisper-server, ds4) in one shot."""
+        proc = engine.proc
+        if proc is None:
+            return
+        try:
+            os.killpg(os.getpgid(proc.pid), sig)
+        except Exception:
+            try:
+                proc.send_signal(sig)
+            except Exception:
+                pass
+
+    def _thermal_post(self, client, engine, action: str, reason: str) -> bool:
+        try:
+            client.post(engine.url + f"/internal/thermal-{action}",
+                        json={"reason": reason})
+            return True
+        except Exception:
+            return False
+
+    @staticmethod
+    def _thermal_reason(gtemp, cpu_t, settings) -> str:
+        bits = []
+        if gtemp is not None and gtemp >= settings.gpu_high:
+            bits.append(f"GPU {gtemp:.0f}°C")
+        if cpu_t is not None and cpu_t >= settings.cpu_high:
+            bits.append(f"CPU {cpu_t:.0f}°C")
+        return ", ".join(bits) or "hot"
+
+    def _thermal_resume_if_frozen(self, engine) -> None:
+        """Wake an engine we SIGSTOPped — used before a shutdown/restart so the
+        normal SIGTERM path works (a stopped process ignores SIGTERM until SIGCONT)."""
+        if getattr(engine, "therm_sigstopped", False):
+            self._thermal_signal(engine, signal.SIGCONT)
+            engine.therm_sigstopped = False
+
+    def _thermal_apply(self, client, engine, settings, want_pause: bool,
+                       want_resume_ok: bool, gtemp, cpu_t, escalate_n: int) -> None:
+        """Drive one engine's pause/resume state with hysteresis (pause at *_high,
+        resume only once back at/below *_resume), escalating to SIGSTOP if a paused
+        engine keeps generating (stuck in a native call it can't interrupt)."""
+        if not engine.therm_paused:
+            if not want_pause:
+                return
+            # idle -> paused: ask the engine to stop cooperatively.
+            engine.therm_paused = True
+            engine.therm_stop_checks = 0
+            engine.therm_sigstopped = False
+            reason = self._thermal_reason(gtemp, cpu_t, settings)
+            engine.cooling = {"gpu": gtemp, "cpu": cpu_t,
+                              "message": f"thermal pause: {reason}"}
+            ok = self._thermal_post(client, engine, "pause", reason)
+            print(f"[front] thermal: pausing engine '{engine.name}' ({reason})"
+                  f"{'' if ok else ' [pause msg failed — will escalate if busy]'}",
+                  flush=True)
+            return
+
+        # Already paused.
+        if want_resume_ok:
+            if engine.therm_sigstopped:
+                self._thermal_signal(engine, signal.SIGCONT)
+                engine.therm_sigstopped = False
+                print(f"[front] thermal: SIGCONT engine '{engine.name}' (cooled)",
+                      flush=True)
+            ok = self._thermal_post(client, engine, "resume", "")
+            if ok:
+                engine.therm_paused = False
+                engine.therm_stop_checks = 0
+                engine.cooling = None
+                print(f"[front] thermal: resuming engine '{engine.name}' "
+                      f"(GPU {self._fmt_t(gtemp)} CPU {self._fmt_t(cpu_t)})", flush=True)
+            else:
+                print(f"[front] thermal: resume msg to '{engine.name}' failed — "
+                      f"will retry next check", flush=True)
+            return
+
+        # Still hot and still paused.
+        if engine.therm_sigstopped:
+            return   # already frozen; just wait for cooldown
+        if int(getattr(engine, "inflight", 0) or 0) <= 0:
+            engine.therm_stop_checks = 0   # cooperatively idle — the pause worked
+            return
+        # Engine is still running work after we asked it to pause: it's stuck in a
+        # native call. Count consecutive busy checks, then freeze it at the OS level.
+        engine.therm_stop_checks += 1
+        if engine.therm_stop_checks >= max(1, escalate_n):
+            self._thermal_signal(engine, signal.SIGSTOP)
+            engine.therm_sigstopped = True
+            engine.cooling = {"gpu": gtemp, "cpu": cpu_t,
+                              "message": "thermal pause (SIGSTOP — engine was busy)"}
+            print(f"[front] thermal: engine '{engine.name}' ignored pause for "
+                  f"{engine.therm_stop_checks} checks (inflight={engine.inflight}) "
+                  f"— SIGSTOP", flush=True)
+
+    @staticmethod
+    def _fmt_t(t) -> str:
+        return f"{t:.0f}°C" if isinstance(t, (int, float)) else "n/a"
+
+    def _thermal_loop(self) -> None:
+        from codai.frontproxy.gpu_detect import gpu_stats
+        from codai.models import thermal as _thermal
+        _auth = ({"x-coderai-internal": self.internal_token}
+                 if self.internal_token else {})
+        client = httpx.Client(timeout=self.config.server.proxy_status_timeout,
+                              headers=_auth)
+        try:
+            while not self._stopped.is_set():
+                settings = self._thermal_settings()
+                period = max(2.0, settings.poll_seconds)
+                if not (settings.cpu_enabled or settings.gpu_enabled):
+                    self._stopped.wait(period)
+                    continue
+                try:
+                    try:
+                        cards = gpu_stats()
+                    except Exception:
+                        cards = []
+                    cpu_t = _thermal.read_cpu_temp() if settings.cpu_enabled else None
+                    cpu_hot = (cpu_t is not None and cpu_t >= settings.cpu_high)
+                    cpu_warm = (cpu_t is not None and cpu_t > settings.cpu_resume)
+                    escalate_n = int(getattr(self.config.thermal,
+                                             "stop_escalate_checks", 3) or 3)
+                    for engine in self.registry.all():
+                        if getattr(engine, "role", "engine") == "system":
+                            continue   # the cache/downloads worker isn't on the GPU
+                        ecards = (self._engine_cards(engine, cards)
+                                  if settings.gpu_enabled else [])
+                        gpu_hot = gpu_warm = False
+                        temps = []
+                        for c in ecards:
+                            t = c.get("temp")
+                            if t is None:
+                                continue
+                            temps.append(t)
+                            high, resume = settings.gpu_thresholds(c.get("vendor"))
+                            if t >= high:
+                                gpu_hot = True
+                            if t > resume:
+                                gpu_warm = True
+                        engine.therm_temp = max(temps) if temps else None
+                        # Pause when this engine's GPU is over high OR the (global)
+                        # CPU is over high. Resume only when BOTH are back to resume.
+                        want_pause = (settings.gpu_enabled and gpu_hot) or cpu_hot
+                        want_resume_ok = (not (settings.gpu_enabled and gpu_warm)
+                                          and not cpu_warm)
+                        self._thermal_apply(client, engine, settings, want_pause,
+                                            want_resume_ok, engine.therm_temp, cpu_t,
+                                            escalate_n)
+                except Exception as exc:
+                    if self.debug:
+                        print(f"[front] thermal monitor error: {exc}", flush=True)
+                self._stopped.wait(period)
+        finally:
+            client.close()
+
     def _poll_loop(self) -> None:
         _auth = ({"x-coderai-internal": self.internal_token}
                  if self.internal_token else {})
@@ -627,6 +823,10 @@ class EngineSupervisor:
             drain_grace = float(getattr(self.config.server,
                                         "engine_restart_drain_grace", 30.0) or 0.0)
         with self._restart_lock:
+            # If we'd thermally frozen it, wake it so it can drain in-flight work and
+            # honour SIGTERM (a stopped process ignores both until continued).
+            self._thermal_resume_if_frozen(engine)
+            engine.therm_paused = False
             proc = engine.proc
             if proc is not None and proc.poll() is None and drain_grace > 0:
                 engine.draining = True
@@ -697,6 +897,11 @@ class EngineSupervisor:
                 except Exception:
                     pass
 
+        # Wake any thermally-frozen engine first: a SIGSTOPped process ignores
+        # SIGTERM until it's continued, so without this the polite phase below would
+        # do nothing and we'd wait out the full grace before SIGKILL.
+        for e in self.registry.all():
+            self._thermal_resume_if_frozen(e)
         procs = [(e, e.proc) for e in self.registry.all()
                  if e.proc is not None and e.proc.poll() is None]
         # Phase 1: polite SIGTERM to each group.

codai/frontproxy/registry.py

@@ -75,6 +75,13 @@ class Engine:
     draining: bool = False         # restart pending: stop routing NEW requests here
                                     # and let in-flight ones finish (drain grace period)
     inflight: int = 0              # proxied requests currently streaming through
+    # Front-side thermal supervision bookkeeping (written by the EngineSupervisor's
+    # thermal monitor; NOT reported by the engine). therm_temp is this engine's
+    # hottest owned card.
+    therm_temp: Optional[float] = None
+    therm_paused: bool = False     # front asked this engine to pause (cooperatively)
+    therm_sigstopped: bool = False  # front escalated to an OS-level SIGSTOP
+    therm_stop_checks: int = 0     # consecutive checks the engine stayed busy after a pause
     _inflight_lock: object = field(default_factory=threading.Lock, repr=False, compare=False)
     # In-flight request metadata {rid: {"model","kind","path","started_at"}} so the
     # front can synthesize Tasks-page entries for work it dispatched — visible even

codai/models/thermal.py

@@ -118,6 +118,67 @@ def _cooldown_exit() -> None:
             })
 
 
+# ---------------------------------------------------------------------------
+# External (front-driven) pause
+# ---------------------------------------------------------------------------
+# The front proxy runs the AUTHORITATIVE thermal monitor: it reads temperatures
+# centrally and stays responsive even while an engine is GIL-blocked inside a
+# long native call (a single llama.cpp prefill/image-encode that the engine's own
+# between-token checkpoint can't interrupt). When the front decides an engine must
+# stop, it POSTs /internal/thermal-pause, which sets this flag; the engine honours
+# it COOPERATIVELY at the next safe point (request entry / between tokens) — the
+# same mechanism as a local temperature cooldown, just triggered remotely. The
+# front clears it (/internal/thermal-resume) once the hardware has cooled. If an
+# engine ignores the pause (stuck in a native call) the front escalates to an
+# OS-level SIGSTOP, which doesn't need the engine to cooperate.
+_external_lock = threading.Lock()
+_external_pause = {"active": False, "reason": "", "since": 0.0}
+
+
+def set_external_pause(active: bool, reason: str = "") -> None:
+    """Set/clear the front-driven pause. Called from the /internal/thermal-* routes."""
+    with _external_lock:
+        active = bool(active)
+        if active and not _external_pause["active"]:
+            _external_pause["since"] = time.time()
+        if not active:
+            _external_pause["since"] = 0.0
+        _external_pause["active"] = active
+        _external_pause["reason"] = reason if active else ""
+
+
+def external_pause_active() -> bool:
+    with _external_lock:
+        return _external_pause["active"]
+
+
+def external_pause_state() -> dict:
+    with _external_lock:
+        return dict(_external_pause)
+
+
+def _wait_external_pause(settings: "ThermalSettings", context: str = "") -> None:
+    """Block while the front holds an external thermal pause, publishing cooldown
+    state so the Tasks page shows the engine is paused for cooldown."""
+    desc = f" ({context})" if context else ""
+    st = external_pause_state()
+    print(f"[thermal] Paused by supervisor{desc}: {st.get('reason') or 'cooldown'} "
+          f"— holding until resume")
+    waited = 0.0
+    poll = min(2.0, max(0.5, settings.poll_seconds))
+    _cooldown_enter()
+    try:
+        while external_pause_active():
+            _cooldown_update(read_gpu_temp(), read_cpu_temp(), waited,
+                             external_pause_state().get("reason")
+                             or "paused by thermal supervisor")
+            time.sleep(poll)
+            waited += poll
+    finally:
+        _cooldown_exit()
+    print(f"[thermal] Supervisor lifted pause{desc} after {int(waited)}s — resuming")
+
+
 # ---------------------------------------------------------------------------
 # Temperature readers
 # ---------------------------------------------------------------------------
@@ -601,6 +662,14 @@ def wait_until_safe(settings: Optional[ThermalSettings] = None,
     """
     if settings is None:
         settings = _settings_from_global_args()
+
+    # Front-driven pause takes precedence: the central supervisor decided this
+    # engine must stop. Block here — a safe point — until it lifts the pause after
+    # cooldown. Honoured even when local CPU/GPU protection is disabled, since the
+    # front owns this decision (it sees temps the GIL-blocked engine can't).
+    if external_pause_active():
+        _wait_external_pause(settings, context)
+
     if not settings.cpu_enabled and not settings.gpu_enabled:
         _dbg(f"protection disabled (cpu/gpu both off) — proceeding [{context}]")
         return