feat(runtime): report startup warmup status

api/routes.py

@@ -456,6 +456,16 @@ async def health_handler(request: web.Request) -> web.Response:
     except Exception:
         executor_snapshot = {}
 
+    try:
+        if __package__ and "." in __package__:
+            from ..services.startup_lifecycle import get_startup_diagnostics
+        else:
+            from services.startup_lifecycle import get_startup_diagnostics
+
+        startup_diagnostics = get_startup_diagnostics()
+    except Exception:
+        startup_diagnostics = {"state": "unknown", "ready": False, "warmups": {}}
+
     # Job Event Store Stats (Backpressure)
     job_stats = {}
     try:
@@ -507,6 +517,7 @@ async def health_handler(request: web.Request) -> web.Response:
                 "executors": executor_snapshot,  # R129
                 "observability": job_stats,  # R87
             },
+            "startup": startup_diagnostics,
             # S15: Exposure Detection
             "access_policy": {
                 "observability": policy_mode,

docs/openapi.yaml

@@ -22,7 +22,7 @@ paths:
   /health:
     get:
       operationId: "get_health"
-      summary: "System status, uptime, and dependencies."
+      summary: "System status, uptime, dependencies, and startup lifecycle diagnostics."
       responses:
         200:
           description: "OK"

docs/release/api_contract.md

@@ -37,7 +37,7 @@ All new integrations should use the `/openclaw/` prefix. Use of `/moltbot/` is d
 
 | Method | Path | Legacy Path | Auth | Description |
 | :--- | :--- | :--- | :--- | :--- |
-| `GET` | `/health` | `/moltbot/health` | None | System status, uptime, and dependencies. |
+| `GET` | `/health` | `/moltbot/health` | None | System status, uptime, dependencies, and startup lifecycle diagnostics. |
 | `GET` | `/capabilities` | `/moltbot/capabilities` | None | Feature flags and supported extensions (includes optional UX/runtime features such as assist streaming support). |
 | `GET` | `/logs/tail` | `/moltbot/logs/tail` | Observability | Tail recent log lines (rate-limited). |
 | `GET` | `/trace/{prompt_id}` | `/moltbot/trace/{id}` | Observability | Get execution trace by prompt ID. |

docs/release/config_secrets_contract.md

@@ -201,6 +201,7 @@ Contractual limits to prevent resource exhaustion.
 | :--- | :--- |
 | `OPENCLAW_STATE_DIR` | Directory for persistent state (DBs, history, logs). Default: `ComfyUI/user/default/openclaw` |
 | `OPENCLAW_LOG_TRUNCATE_ON_START` | Set `1` to truncate active log file (`openclaw.log`) once at process startup before new handlers write records. |
+| `OPENCLAW_STARTUP_WARMUP_TIMEOUT_SEC` | Optional timeout for non-blocking startup warmups. Warmup timeout degrades health diagnostics but does not block required route startup. |
 | `OPENCLAW_DIAGNOSTICS` | Comma-separated list of subsystems to enable debug logging for (e.g. `webhook.*,templates`). Safe-redacted. |
 | `OPENCLAW_CONNECTOR_DEBUG` | Set `1` to enable verbose debug logging in Connector. |
 

docs/runtime_hardening_and_startup.md

@@ -48,6 +48,22 @@ In `minimal` mode, these checks are warning-first for local/LAN posture, but `pu
 Startup bootstrap no longer swallows fatal security-gate errors.
 If a critical startup gate fails, initialization aborts deterministically instead of continuing with partial route registration.
 
+### Startup lifecycle diagnostics
+
+The health response includes a `startup` diagnostic object with:
+
+- `state`: `starting`, `ready`, `degraded-warmup`, or `fatal-startup`
+- `ready`: whether required route/service startup completed
+- `fatal`: bounded fatal-startup details when required startup fails
+- `warmups`: bounded status for optional background warmups
+
+Required startup work still fails closed. Optional warmups such as model inventory refresh run after route registration and do not block baseline API availability. Their failures or timeouts are reported as `degraded-warmup`.
+
+Optional warmup timeout can be tuned with:
+
+- `OPENCLAW_STARTUP_WARMUP_TIMEOUT_SEC`
+- legacy alias: `MOLTBOT_STARTUP_WARMUP_TIMEOUT_SEC`
+
 ## Public deployment shared-surface acknowledgement
 
 When running deployment profile checks for public posture (`OPENCLAW_DEPLOYMENT_PROFILE=public`),

services/route_bootstrap.py

@@ -15,6 +15,56 @@ import time
 _routes_registered = False
 
 
+def _resolve_optional_warmup_timeout_sec() -> float:
+    raw = (
+        os.environ.get("OPENCLAW_STARTUP_WARMUP_TIMEOUT_SEC")
+        or os.environ.get("MOLTBOT_STARTUP_WARMUP_TIMEOUT_SEC")
+        or "5"
+    )
+    try:
+        return max(0.1, min(float(raw), 60.0))
+    except (TypeError, ValueError):
+        return 5.0
+
+
+def _warm_model_inventory_snapshot() -> None:
+    from .preflight import get_model_inventory_snapshot
+
+    get_model_inventory_snapshot(trigger_refresh=True)
+
+
+def _build_optional_startup_warmups():
+    timeout_sec = _resolve_optional_warmup_timeout_sec()
+    return [
+        ("model_inventory", _warm_model_inventory_snapshot, timeout_sec),
+    ]
+
+
+def _mark_startup_ready_and_start_warmups() -> None:
+    try:
+        from .startup_lifecycle import mark_startup_ready, start_optional_warmups
+
+        mark_startup_ready("routes")
+        start_optional_warmups(_build_optional_startup_warmups())
+    except Exception:
+        # IMPORTANT: optional warmup diagnostics must not undo successful route startup.
+        logging.getLogger("ComfyUI-OpenClaw").exception(
+            "R188: failed to start optional startup warmups"
+        )
+
+
+def _mark_startup_fatal(phase: str, exc: BaseException) -> None:
+    try:
+        from .startup_lifecycle import mark_startup_fatal
+
+        mark_startup_fatal(phase, exc)
+    except Exception:
+        # IMPORTANT: preserve the original bootstrap exception even if diagnostics fail.
+        logging.getLogger("ComfyUI-OpenClaw").exception(
+            "R188: failed to record fatal startup state"
+        )
+
+
 def _register_plugins_and_shutdown_hooks() -> None:
     # R67: Best-effort process shutdown hook for scheduler/failover flush.
     try:
@@ -174,6 +224,7 @@ def _do_full_registration(server) -> None:
         require_admin_token_fn=require_admin_token,
         submit_fn=unified_submit_fn,
     )
+    _mark_startup_ready_and_start_warmups()
 
 
 _BRIDGE_ROUTE_SPECS = (
@@ -240,6 +291,12 @@ def _start_registration_retry_loop() -> None:
             attempts += 1
 
         if not _routes_registered:
+            _mark_startup_fatal(
+                "route_registration_retry",
+                RuntimeError(
+                    f"Failed to register routes after {max_attempts} attempts"
+                ),
+            )
             logger.error(
                 "Failed to register routes after %s attempts. API endpoints unavailable.",
                 max_attempts,
@@ -254,8 +311,12 @@ def register_routes_once() -> None:
     if _routes_registered:
         return
 
-    _register_plugins_and_shutdown_hooks()
-    _initialize_registries_and_security_gate()
+    try:
+        _register_plugins_and_shutdown_hooks()
+        _initialize_registries_and_security_gate()
+    except Exception as exc:
+        _mark_startup_fatal("required_startup", exc)
+        raise
 
     try:
         ps_mod = sys.modules.get("server")
@@ -273,6 +334,9 @@ def register_routes_once() -> None:
             )
             _start_registration_retry_loop()
     except Exception:
+        _exc_type, exc, _tb = sys.exc_info()
+        if exc is not None:
+            _mark_startup_fatal("route_registration", exc)
         logging.getLogger("ComfyUI-OpenClaw").exception("Route registration failed")
         # CRITICAL: initial registration failures must fail closed. The retry loop is
         # only for PromptServer warm-up, not for hiding broken route/bootstrap state.

services/startup_lifecycle.py

@@ -0,0 +1,214 @@
+"""
+Startup lifecycle diagnostics and optional warmup boundaries.
+
+Required startup work still fails closed in callers. This module only tracks
+readiness and runs optional warmups without delaying route availability.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+import time
+from typing import Any, Callable, Dict, Iterable, Optional
+
+logger = logging.getLogger("ComfyUI-OpenClaw.services.startup_lifecycle")
+
+STARTUP_STARTING = "starting"
+STARTUP_READY = "ready"
+STARTUP_DEGRADED_WARMUP = "degraded-warmup"
+STARTUP_FATAL = "fatal-startup"
+
+WARMUP_PENDING = "pending"
+WARMUP_RUNNING = "running"
+WARMUP_SUCCEEDED = "succeeded"
+WARMUP_FAILED = "failed"
+WARMUP_TIMED_OUT = "timed_out"
+
+WarmupSpec = tuple[str, Callable[[], Any], float]
+
+_LOCK = threading.RLock()
+_STARTED_AT = time.time()
+_READY = False
+_READY_PHASE: Optional[str] = None
+_READY_AT: Optional[float] = None
+_FATAL: Optional[Dict[str, Any]] = None
+_WARMUPS: Dict[str, Dict[str, Any]] = {}
+
+
+def mark_startup_ready(phase: str = "routes") -> None:
+    """Mark required startup work as ready."""
+    global _READY, _READY_AT, _READY_PHASE
+    with _LOCK:
+        if _FATAL is not None:
+            return
+        _READY = True
+        _READY_PHASE = str(phase or "routes")
+        _READY_AT = time.time()
+
+
+def mark_startup_fatal(phase: str, exc: BaseException) -> None:
+    """Record a fatal required-startup failure."""
+    global _FATAL, _READY
+    with _LOCK:
+        _READY = False
+        _FATAL = {
+            "phase": str(phase or "startup"),
+            "error_type": type(exc).__name__,
+            "error": str(exc)[:500],
+            "ts": time.time(),
+        }
+
+
+def start_optional_warmups(specs: Iterable[WarmupSpec]) -> None:
+    """Start optional warmups in background monitor threads."""
+    for name, fn, timeout_sec in list(specs or []):
+        _start_optional_warmup(str(name), fn, float(timeout_sec))
+
+
+def get_startup_diagnostics() -> Dict[str, Any]:
+    """Return a bounded diagnostic snapshot for health/operator views."""
+    with _LOCK:
+        warmups = {name: dict(record) for name, record in _WARMUPS.items()}
+        fatal = dict(_FATAL) if _FATAL else None
+        ready = bool(_READY and fatal is None)
+        degraded = any(
+            record.get("state") in {WARMUP_FAILED, WARMUP_TIMED_OUT}
+            for record in warmups.values()
+        )
+        if fatal:
+            state = STARTUP_FATAL
+        elif ready and degraded:
+            state = STARTUP_DEGRADED_WARMUP
+        elif ready:
+            state = STARTUP_READY
+        else:
+            state = STARTUP_STARTING
+        return {
+            "state": state,
+            "ready": ready,
+            "ready_phase": _READY_PHASE,
+            "started_at": _STARTED_AT,
+            "ready_at": _READY_AT,
+            "fatal": fatal,
+            "warmups": warmups,
+        }
+
+
+def reset_startup_lifecycle_for_tests() -> None:
+    """Reset in-memory lifecycle state for tests."""
+    global _READY, _READY_AT, _READY_PHASE, _FATAL, _STARTED_AT
+    with _LOCK:
+        _STARTED_AT = time.time()
+        _READY = False
+        _READY_PHASE = None
+        _READY_AT = None
+        _FATAL = None
+        _WARMUPS.clear()
+
+
+def _start_optional_warmup(
+    name: str, fn: Callable[[], Any], timeout_sec: float
+) -> None:
+    timeout_sec = max(0.01, min(float(timeout_sec or 5.0), 60.0))
+    with _LOCK:
+        existing = _WARMUPS.get(name)
+        if existing and existing.get("state") in {WARMUP_RUNNING, WARMUP_SUCCEEDED}:
+            return
+        _WARMUPS[name] = {
+            "state": WARMUP_PENDING,
+            "timeout_sec": timeout_sec,
+            "started_at": None,
+            "completed_at": None,
+            "duration_sec": None,
+            "error_type": None,
+            "error": None,
+        }
+
+    monitor = threading.Thread(
+        target=_warmup_monitor,
+        args=(name, fn, timeout_sec),
+        name=f"openclaw-warmup-monitor-{name}",
+        daemon=True,
+    )
+    monitor.start()
+
+
+def _warmup_monitor(name: str, fn: Callable[[], Any], timeout_sec: float) -> None:
+    started_at = time.time()
+    done = threading.Event()
+    result: Dict[str, Any] = {}
+
+    def _worker() -> None:
+        try:
+            result["value"] = fn()
+            result["ok"] = True
+        except Exception as exc:  # pragma: no cover - defensive outer guard
+            result["ok"] = False
+            result["exc"] = exc
+        finally:
+            done.set()
+
+    with _LOCK:
+        if name in _WARMUPS:
+            _WARMUPS[name]["state"] = WARMUP_RUNNING
+            _WARMUPS[name]["started_at"] = started_at
+
+    worker = threading.Thread(
+        target=_worker,
+        name=f"openclaw-warmup-{name}",
+        daemon=True,
+    )
+    worker.start()
+
+    if not done.wait(timeout=timeout_sec):
+        _finish_warmup(
+            name,
+            WARMUP_TIMED_OUT,
+            started_at,
+            error_type="TimeoutError",
+            error=f"optional warmup exceeded {timeout_sec:.2f}s",
+        )
+        logger.warning(
+            "R188: optional startup warmup timed out: %s (%.2fs)",
+            name,
+            timeout_sec,
+        )
+        return
+
+    exc = result.get("exc")
+    if result.get("ok"):
+        _finish_warmup(name, WARMUP_SUCCEEDED, started_at)
+        logger.info("R188: optional startup warmup completed: %s", name)
+        return
+
+    _finish_warmup(
+        name,
+        WARMUP_FAILED,
+        started_at,
+        error_type=type(exc).__name__ if exc else "Exception",
+        error=str(exc)[:500] if exc else "unknown warmup failure",
+    )
+    logger.warning("R188: optional startup warmup failed: %s: %s", name, exc)
+
+
+def _finish_warmup(
+    name: str,
+    state: str,
+    started_at: float,
+    *,
+    error_type: Optional[str] = None,
+    error: Optional[str] = None,
+) -> None:
+    completed_at = time.time()
+    with _LOCK:
+        record = _WARMUPS.setdefault(name, {})
+        record.update(
+            {
+                "state": state,
+                "completed_at": completed_at,
+                "duration_sec": max(0.0, completed_at - started_at),
+                "error_type": error_type,
+                "error": error,
+            }
+        )

tests/exception_boundary_policy.json

@@ -5,9 +5,9 @@
       "broad_catches": [
         {
           "scope": "health_handler",
-          "expected_count": 5,
+          "expected_count": 6,
           "classification": "allowed_boundary_guard",
-          "reason": "Health diagnostics must degrade to partial snapshots when optional provider, metrics, event-store, or profile probes fail."
+          "reason": "Health diagnostics must degrade to partial snapshots when optional provider, metrics, startup-lifecycle, event-store, or profile probes fail."
         },
         {
           "scope": "logs_tail_handler",
@@ -65,6 +65,18 @@
     },
     "services/route_bootstrap.py": {
       "broad_catches": [
+        {
+          "scope": "_mark_startup_ready_and_start_warmups",
+          "expected_count": 1,
+          "classification": "allowed_boundary_guard",
+          "reason": "Optional warmup diagnostics must not block route availability after required startup succeeds."
+        },
+        {
+          "scope": "_mark_startup_fatal",
+          "expected_count": 1,
+          "classification": "allowed_boundary_guard",
+          "reason": "Fatal startup-state recording is diagnostic only and must not mask the original bootstrap exception."
+        },
         {
           "scope": "_register_plugins_and_shutdown_hooks",
           "expected_count": 1,
@@ -85,9 +97,9 @@
         },
         {
           "scope": "register_routes_once",
-          "expected_count": 1,
+          "expected_count": 2,
           "classification": "allowed_boundary_guard",
-          "reason": "Initial route registration logs context and re-raises so unexpected bootstrap failures are visible and fail-closed."
+          "reason": "Required startup and initial route registration log context and re-raise so unexpected bootstrap failures are visible and fail-closed."
         }
       ]
     }

tests/test_r188_startup_lifecycle.py

@@ -0,0 +1,209 @@
+import threading
+import time
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+
+class TestStartupLifecycleDiagnostics(unittest.TestCase):
+    def setUp(self):
+        from services.startup_lifecycle import reset_startup_lifecycle_for_tests
+
+        reset_startup_lifecycle_for_tests()
+
+    def tearDown(self):
+        from services.startup_lifecycle import reset_startup_lifecycle_for_tests
+
+        reset_startup_lifecycle_for_tests()
+
+    def test_optional_warmup_timeout_degrades_without_blocking_ready(self):
+        from services.startup_lifecycle import (
+            get_startup_diagnostics,
+            mark_startup_ready,
+            start_optional_warmups,
+        )
+
+        release = threading.Event()
+
+        def slow_warmup():
+            release.wait(timeout=1.0)
+
+        mark_startup_ready("routes")
+        started_at = time.monotonic()
+        start_optional_warmups([("slow_provider", slow_warmup, 0.01)])
+        elapsed = time.monotonic() - started_at
+
+        self.assertLess(elapsed, 0.05)
+
+        deadline = time.monotonic() + 1.0
+        diagnostics = get_startup_diagnostics()
+        while time.monotonic() < deadline:
+            diagnostics = get_startup_diagnostics()
+            if diagnostics["warmups"]["slow_provider"]["state"] == "timed_out":
+                break
+            time.sleep(0.01)
+
+        release.set()
+        self.assertEqual(diagnostics["state"], "degraded-warmup")
+        self.assertEqual(diagnostics["ready"], True)
+        self.assertEqual(diagnostics["warmups"]["slow_provider"]["state"], "timed_out")
+
+    def test_fatal_startup_state_is_distinct_from_warmup_degradation(self):
+        from services.startup_lifecycle import (
+            get_startup_diagnostics,
+            mark_startup_fatal,
+        )
+
+        mark_startup_fatal("security_gate", RuntimeError("blocked"))
+        diagnostics = get_startup_diagnostics()
+
+        self.assertEqual(diagnostics["state"], "fatal-startup")
+        self.assertFalse(diagnostics["ready"])
+        self.assertEqual(diagnostics["fatal"]["phase"], "security_gate")
+        self.assertIn("RuntimeError", diagnostics["fatal"]["error_type"])
+
+
+class _DummyRoutes:
+    def __init__(self):
+        self.calls = []
+
+    def _decorator(self, method, path):
+        def _wrap(handler):
+            self.calls.append((method, path, handler))
+            return handler
+
+        return _wrap
+
+    def get(self, path):
+        return self._decorator("GET", path)
+
+    def post(self, path):
+        return self._decorator("POST", path)
+
+    def put(self, path):
+        return self._decorator("PUT", path)
+
+    def delete(self, path):
+        return self._decorator("DELETE", path)
+
+
+class _DummyRouter:
+    def __init__(self):
+        self.calls = []
+
+    def add_route(self, method, path, handler):
+        self.calls.append((method, path, handler))
+
+    def add_post(self, path, handler):
+        self.calls.append(("POST", path, handler))
+
+    def add_get(self, path, handler):
+        self.calls.append(("GET", path, handler))
+
+
+class _DummyBridgeHandlers:
+    def __init__(self, submit_service=None):
+        self.submit_service = submit_service
+
+    async def submit_handler(self, request=None):
+        return request
+
+    async def deliver_handler(self, request=None):
+        return request
+
+    async def health_handler(self, request=None):
+        return request
+
+
+class TestRouteBootstrapWarmupBoundary(unittest.TestCase):
+    def setUp(self):
+        from services.startup_lifecycle import reset_startup_lifecycle_for_tests
+
+        reset_startup_lifecycle_for_tests()
+
+    def tearDown(self):
+        from services.startup_lifecycle import reset_startup_lifecycle_for_tests
+
+        reset_startup_lifecycle_for_tests()
+
+    def test_full_registration_marks_ready_before_optional_warmup_finishes(self):
+        from services import route_bootstrap
+        from services.startup_lifecycle import get_startup_diagnostics
+
+        release = threading.Event()
+
+        def slow_warmup():
+            release.wait(timeout=1.0)
+
+        app = SimpleNamespace(router=_DummyRouter())
+        server = SimpleNamespace(routes=_DummyRoutes(), app=app)
+
+        contract = {
+            "register_routes": lambda server: setattr(server, "core_routes", True),
+            "register_preset_routes": lambda app: setattr(app, "presets", True),
+            "register_schedule_routes": lambda app, require_admin_token_fn=None: setattr(
+                app, "schedules", True
+            ),
+            "BridgeHandlers": _DummyBridgeHandlers,
+            "register_trigger_routes": lambda app, **kwargs: setattr(
+                app, "triggers", True
+            ),
+            "register_approval_routes": lambda app, **kwargs: setattr(
+                app, "approvals", True
+            ),
+        }
+
+        with (
+            patch(
+                "services.route_bootstrap_contract.load_route_bootstrap_contract",
+                return_value=contract,
+            ),
+            patch("services.scheduler.runner.get_scheduler_runner") as get_runner,
+            patch("services.scheduler.runner.start_scheduler"),
+            patch(
+                "services.route_bootstrap._build_optional_startup_warmups",
+                return_value=[("slow_provider", slow_warmup, 0.5)],
+                create=True,
+            ),
+        ):
+            get_runner.return_value = MagicMock()
+            started_at = time.monotonic()
+            route_bootstrap._do_full_registration(server)
+            elapsed = time.monotonic() - started_at
+
+        diagnostics = get_startup_diagnostics()
+        release.set()
+
+        self.assertLess(elapsed, 0.5)
+        self.assertTrue(server.core_routes)
+        self.assertTrue(app.triggers)
+        self.assertTrue(app.approvals)
+        self.assertTrue(diagnostics["ready"])
+        self.assertIn(
+            diagnostics["warmups"]["slow_provider"]["state"], {"running", "succeeded"}
+        )
+
+    def test_register_routes_once_marks_fatal_when_required_startup_fails(self):
+        from services import route_bootstrap
+        from services.startup_lifecycle import get_startup_diagnostics
+
+        with (
+            patch.object(route_bootstrap, "_routes_registered", False),
+            patch.object(route_bootstrap, "_register_plugins_and_shutdown_hooks"),
+            patch.object(
+                route_bootstrap,
+                "_initialize_registries_and_security_gate",
+                side_effect=RuntimeError("security blocked"),
+            ),
+        ):
+            with self.assertRaises(RuntimeError):
+                route_bootstrap.register_routes_once()
+
+        diagnostics = get_startup_diagnostics()
+        self.assertEqual(diagnostics["state"], "fatal-startup")
+        self.assertFalse(diagnostics["ready"])
+        self.assertEqual(diagnostics["fatal"]["phase"], "required_startup")
+
+
+if __name__ == "__main__":
+    unittest.main()