Supervise the engine's host: deep health, loop watchdog, one worker

The API image ran four uvicorn workers, and each one built a full flow
controller — four sets of MQTT subscriptions, cron ticks and webhooks.
Runs one worker now; scaling out is the worker split, not more processes.

Adds a loop-lag watchdog and a deep /utils/health/ that fails when the
event loop is wedged or Redis is unreachable, the two failure modes a
process-alive check never sees. Autoheal restarts on that signal, behind
a compose profile because it mounts the Docker socket.

The private user-seeding routes now need an explicit opt-in rather than
just ENVIRONMENT=local, so a deployment that kept the default never
exposes them.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011LF61rxW1FG5YCD2J9YqjY

backend/Dockerfile

@@ -42,4 +42,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
 
 WORKDIR /app/backend/
 
-CMD ["fastapi", "run", "--workers", "4", "app/main.py"]
+# Single worker on purpose: the process hosts the flow engine, and a second
+# worker would be a second engine — duplicated subscriptions, cron ticks and
+# webhooks. Scaling out is the M5 worker split, not more uvicorn processes.
+CMD ["fastapi", "run", "app/main.py"]

backend/app/api/main.py

@@ -1,7 +1,6 @@
 from fastapi import APIRouter
 
 from app.api.routes import flows, login, oauth, private, secrets, users, utils
-from app.core.config import settings
 
 api_router = APIRouter()
 api_router.include_router(login.router)
@@ -15,5 +14,6 @@ api_router.include_router(secrets.router)
 api_router.include_router(oauth.router)
 
 
-if settings.ENVIRONMENT == "local":
-    api_router.include_router(private.router)
+# Always mounted like oauth, so the generated SDK keeps its shape; the
+# endpoints refuse to work unless the private API is explicitly enabled.
+api_router.include_router(private.router)

backend/app/api/routes/private.py

@@ -1,9 +1,10 @@
 from typing import Any
 
-from fastapi import APIRouter
+from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel
 
 from app.api.deps import SessionDep
+from app.core.config import settings
 from app.core.security import get_password_hash
 from app.models import (
     User,
@@ -13,6 +14,11 @@ from app.models import (
 router = APIRouter(tags=["private"], prefix="/private")
 
 
+def _require_private_api() -> None:
+    if not (settings.ENVIRONMENT == "local" and settings.PRIVATE_API_ENABLED):
+        raise HTTPException(status_code=403, detail="Private API is disabled")
+
+
 class PrivateUserCreate(BaseModel):
     email: str
     password: str
@@ -25,6 +31,7 @@ def create_user(user_in: PrivateUserCreate, session: SessionDep) -> Any:
     """
     Create a new user.
     """
+    _require_private_api()
 
     user = User(
         email=user_in.email,

backend/app/api/routes/utils.py

@@ -1,7 +1,12 @@
-from fastapi import APIRouter, Depends
+import time
+from typing import Any
+
+from fastapi import APIRouter, Depends, Request, Response
+from fastapi.concurrency import run_in_threadpool
 from pydantic.networks import EmailStr
 
 from app.api.deps import get_current_active_superuser
+from app.flow.state import RedisState
 from app.models import Message
 from app.utils import generate_test_email, send_email
 
@@ -29,3 +34,61 @@ def test_email(email_to: EmailStr) -> Message:
 @router.get("/health-check/")
 async def health_check() -> bool:
     return True
+
+
+@router.get("/health/")
+async def health(request: Request, response: Response) -> dict[str, Any]:
+    """Deep health: 200 while the engine can serve its purpose, 503 otherwise.
+
+    "Serve its purpose" means the event loop is responsive and the configured
+    state backend answers — the two failure modes a process-alive check never
+    sees. The queue and pool sections are filled by the execution service.
+    """
+    controller = getattr(request.app.state, "flow_controller", None)
+    watchdog = getattr(request.app.state, "watchdog", None)
+    problems: list[str] = []
+
+    loop_lag = watchdog.snapshot() if watchdog else {"ewma": 0.0, "max_60s": 0.0}
+    if watchdog is not None and watchdog.degraded:
+        problems.append("event loop lagging")
+
+    redis_info: dict[str, Any] = {
+        "configured": False,
+        "connected": None,
+        "rtt_ms": None,
+    }
+    engine: dict[str, Any] = {"flows": 0, "nodes": 0, "quarantined": []}
+    queue: dict[str, Any] = {}
+    if controller is not None:
+        state = controller.state
+        if isinstance(state, RedisState):
+            redis_info["configured"] = True
+            start = time.perf_counter()
+            connected = await run_in_threadpool(state.ping)
+            redis_info["connected"] = connected
+            redis_info["rtt_ms"] = round((time.perf_counter() - start) * 1000, 1)
+            if not connected:
+                problems.append("redis unreachable")
+        engine["flows"] = len(getattr(controller, "loaded", {}) or {})
+        engine["nodes"] = len(
+            getattr(getattr(controller, "pipeline", None), "nodes", []) or []
+        )
+        engine["quarantined"] = sorted(getattr(controller, "quarantined", ()) or ())
+        stats = getattr(controller, "queue_stats", None)
+        if callable(stats):
+            queue = await run_in_threadpool(stats)
+            if queue.get("oldest_pending_s", 0) > 120:
+                problems.append("queue stalled")
+
+    status = "degraded" if problems else "ok"
+    if problems:
+        response.status_code = 503
+    return {
+        "status": status,
+        "problems": problems,
+        "loop_lag_ms": loop_lag,
+        "redis": redis_info,
+        "engine": engine,
+        "queue": queue,
+        "ts": time.time(),
+    }

backend/app/core/config.py

@@ -44,6 +44,10 @@ class Settings(BaseSettings):
     # The MCP endpoint, and the OAuth server agents authenticate against. Off
     # until someone asks for it: it opens client registration to the network.
     MCP_ENABLED: bool = False
+    # Unauthenticated test-only endpoints (user seeding). Requires an explicit
+    # opt-in on top of ENVIRONMENT=local, so a deployment that merely kept the
+    # default environment never exposes them.
+    PRIVATE_API_ENABLED: bool = False
     DOMAIN: str = "localhost"
     OAUTH_PRIVATE_KEY_FILE: Path = Path("flow-data/oauth-key.pem")
     OAUTH_CODE_EXPIRE_SECONDS: int = 60

backend/app/flow/watchdog.py

@@ -0,0 +1,83 @@
+"""Engine self-observation: event-loop lag watchdog and the deliberate exit.
+
+Docker's restart policy only fires when the process exits, so a wedged event
+loop would otherwise stay "up" forever. The watchdog measures how late a
+sleeping task wakes up — the standard loop-lag trick — and feeds the deep
+health endpoint; `engine_fatal` is the deliberate handoff to the outer
+supervisor when a clean restart beats limping on.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import time
+from collections import deque
+
+from app.flow.events import EventBus
+
+logger = logging.getLogger(__name__)
+
+INTERVAL = 1.0
+EWMA_ALPHA = 0.2
+# One late wake-up is a busy moment; several in a row is a blocked loop.
+DEGRADED_LAG_S = 5.0
+DEGRADED_STRIKES = 3
+# Sustained lag above this marks the engine degraded in /health.
+DEGRADED_EWMA_MS = 200.0
+
+
+class LoopWatchdog:
+    """Measures event-loop lag and reports it as engine health."""
+
+    def __init__(self, events: EventBus | None = None) -> None:
+        self._events = events
+        self.ewma_ms = 0.0
+        self._window: deque[tuple[float, float]] = deque()  # (monotonic ts, lag ms)
+        self._strikes = 0
+
+    async def run(self) -> None:
+        while True:
+            before = time.monotonic()
+            await asyncio.sleep(INTERVAL)
+            self._record(max(0.0, time.monotonic() - before - INTERVAL))
+
+    def _record(self, lag_s: float) -> None:
+        lag_ms = lag_s * 1000.0
+        self.ewma_ms += EWMA_ALPHA * (lag_ms - self.ewma_ms)
+        now = time.monotonic()
+        self._window.append((now, lag_ms))
+        while self._window and self._window[0][0] < now - 60.0:
+            self._window.popleft()
+        if lag_s >= DEGRADED_LAG_S:
+            self._strikes += 1
+            if self._strikes == DEGRADED_STRIKES and self._events is not None:
+                logger.warning("event loop lagging: %.1fs late", lag_s)
+                self._events.publish(
+                    {
+                        "type": "engine_degraded",
+                        "reason": f"event loop lag {lag_s:.1f}s",
+                        "ts": time.time(),
+                    }
+                )
+        else:
+            self._strikes = 0
+
+    @property
+    def degraded(self) -> bool:
+        return self.ewma_ms > DEGRADED_EWMA_MS
+
+    def snapshot(self) -> dict[str, float]:
+        return {
+            "ewma": round(self.ewma_ms, 1),
+            "max_60s": round(max((lag for _, lag in self._window), default=0.0), 1),
+        }
+
+
+def engine_fatal(reason: str, events: EventBus | None = None) -> None:
+    """Log, tell whoever still listens, and exit so Docker restarts us clean."""
+    logger.critical("engine fatal: %s", reason)
+    if events is not None:
+        events.publish({"type": "engine_fatal", "reason": reason, "ts": time.time()})
+    os._exit(1)

backend/app/main.py

@@ -19,6 +19,7 @@ from app.flow.plugins import load_plugins
 from app.flow.secrets import init_secrets
 from app.flow.state import MemoryState, RedisState, StateBackend
 from app.flow.store import FlowStore
+from app.flow.watchdog import LoopWatchdog
 
 
 def custom_generate_unique_id(route: APIRoute) -> str:
@@ -62,6 +63,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         fastapi_app=app,
     )
     app.state.flow_controller = controller
+    watchdog = LoopWatchdog(event_bus)
+    app.state.watchdog = watchdog
+    watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog")
     await controller.start()
     try:
         # A mounted sub-app gets no lifespan of its own, so the MCP session
@@ -69,6 +73,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         async with _mcp_sessions():
             yield
     finally:
+        watchdog_task.cancel()
         await controller.stop()
         if settings.MCP_ENABLED:
             from app.mcp.http import aclose

backend/tests/__init__.py

@@ -13,3 +13,6 @@ os.environ["POSTGRES_DB"] = "app_test"
 # builds a TestClient — and so a lifespan — per test module. Tests that want the
 # endpoint mount it themselves.
 os.environ["MCP_ENABLED"] = "false"
+# The private seeding endpoints are opt-in; the suite is one of the two places
+# (with the dev stack) where they are meant to work.
+os.environ["PRIVATE_API_ENABLED"] = "true"

backend/tests/api/routes/test_private.py

@@ -1,3 +1,4 @@
+import pytest
 from fastapi.testclient import TestClient
 from sqlmodel import Session, select
 
@@ -24,3 +25,22 @@ def test_create_user(client: TestClient, db: Session) -> None:
     assert user
     assert user.email == "pollo@listo.com"
     assert user.full_name == "Pollo Listo"
+
+
+def test_creating_a_user_is_refused_unless_the_private_api_is_enabled(
+    client: TestClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The route is always mounted so the SDK keeps its shape; the opt-in is
+    what decides whether unauthenticated user seeding actually works."""
+    monkeypatch.setattr(settings, "PRIVATE_API_ENABLED", False)
+
+    r = client.post(
+        f"{settings.API_V1_STR}/private/users/",
+        json={
+            "email": "nobody@listo.com",
+            "password": "password123",
+            "full_name": "Nobody",
+        },
+    )
+
+    assert r.status_code == 403

backend/tests/flow/test_watchdog.py

@@ -0,0 +1,68 @@
+"""Loop-lag watchdog: what the deep health check reads."""
+
+from fastapi.testclient import TestClient
+
+from app.flow.events import EventBus
+from app.flow.watchdog import DEGRADED_STRIKES, LoopWatchdog
+from app.main import app
+
+
+def test_a_responsive_loop_stays_healthy():
+    watchdog = LoopWatchdog()
+
+    for _ in range(10):
+        watchdog._record(0.002)
+
+    assert not watchdog.degraded
+    assert watchdog.snapshot()["max_60s"] == 2.0
+
+
+def test_sustained_lag_marks_the_engine_degraded():
+    watchdog = LoopWatchdog()
+
+    for _ in range(20):
+        watchdog._record(1.0)
+
+    assert watchdog.degraded
+
+
+def test_a_blocked_loop_is_announced_once_it_keeps_happening():
+    events = []
+    bus = EventBus()
+    bus.publish = events.append  # type: ignore[method-assign]
+    watchdog = LoopWatchdog(bus)
+
+    for _ in range(DEGRADED_STRIKES - 1):
+        watchdog._record(6.0)
+    assert events == []
+
+    watchdog._record(6.0)
+    assert [e["type"] for e in events] == ["engine_degraded"]
+
+    # A recovered loop resets the count, so the next stall is announced again.
+    watchdog._record(0.001)
+    for _ in range(DEGRADED_STRIKES):
+        watchdog._record(6.0)
+    assert len(events) == 2
+
+
+def test_health_reports_503_once_the_loop_is_wedged():
+    """The point of the deep check: unhealthy without the process being dead."""
+    watchdog = LoopWatchdog()
+    app.state.watchdog = watchdog
+    # No lifespan here, so there is no controller either — the endpoint has to
+    # cope with a half-built app rather than assume the engine is up.
+    client = TestClient(app)
+    try:
+        assert client.get("/api/v1/utils/health/").status_code == 200
+
+        for _ in range(20):
+            watchdog._record(1.0)
+
+        response = client.get("/api/v1/utils/health/")
+        assert response.status_code == 503
+        body = response.json()
+        assert body["status"] == "degraded"
+        assert body["problems"] == ["event loop lagging"]
+    finally:
+        del app.state.watchdog