feat(startup): model bootstrap lifecycle outcomes

__init__.py

@@ -53,7 +53,17 @@ def _bootstrap_openclaw_routes() -> None:
             from .services.route_bootstrap import register_routes_once
         else:
             from services.route_bootstrap import register_routes_once
-    except Exception:
+    except Exception as exc:
+        try:
+            if __package__:
+                from .services.startup_lifecycle import mark_bootstrap_import_failed
+            else:
+                from services.startup_lifecycle import mark_bootstrap_import_failed
+
+            mark_bootstrap_import_failed(exc)
+        except Exception:
+            # IMPORTANT: diagnostics must not mask the original compatibility fallback.
+            pass
         return
 
     register_routes_once()

api/route_handlers.py

@@ -137,7 +137,23 @@ async def health_response(request: Any, deps: RouteHandlerDependencies) -> Any:
             from services.startup_lifecycle import get_startup_diagnostics
         startup_diagnostics = get_startup_diagnostics()
     except Exception:
-        startup_diagnostics = {"state": "unknown", "ready": False, "warmups": {}}
+        # SECURITY: keep the public fallback deterministic and content-free even when
+        # startup diagnostics cannot be imported.
+        startup_diagnostics = {
+            "schema_version": 1,
+            "phase": "package_import",
+            "state": "fatal",
+            "reason_code": "bootstrap_import_failed",
+            "ready": False,
+            "degraded": False,
+            "fatal": True,
+            "attempt": 0,
+            "max_attempts": 0,
+            "elapsed_ms": 0,
+            "phase_elapsed_ms": 0,
+            "ready_elapsed_ms": None,
+            "warmups": [],
+        }
 
     job_stats = {}
     try:

services/route_bootstrap.py

@@ -11,8 +11,17 @@ import os
 import sys
 import threading
 import time
+from collections.abc import Callable
 
 _routes_registered = False
+_registration_condition = threading.Condition(threading.RLock())
+_registration_inflight = False
+_registration_started = False
+_registration_error: Exception | None = None
+_registration_retry_thread: threading.Thread | None = None
+_registration_generation = 0
+_REGISTRATION_MAX_ATTEMPTS = 10
+_REGISTRATION_INITIAL_DELAY_SEC = 2.0
 
 
 def _resolve_optional_warmup_timeout_sec() -> float:
@@ -41,30 +50,62 @@ def _build_optional_startup_warmups():
 
 
 def _mark_startup_ready_and_start_warmups() -> None:
-    try:
+    from .startup_lifecycle import mark_startup_ready, start_optional_warmups
-        from .startup_lifecycle import mark_startup_ready, start_optional_warmups
 
-        mark_startup_ready("routes")
+    # Required readiness is part of successful route registration and must not be
+    # hidden behind the optional warmup boundary.
+    mark_startup_ready("routes")
+    try:
         start_optional_warmups(_build_optional_startup_warmups())
-    except Exception:
+    except Exception as exc:
         # IMPORTANT: optional warmup diagnostics must not undo successful route startup.
-        logging.getLogger("ComfyUI-OpenClaw").exception(
+        logging.getLogger("ComfyUI-OpenClaw").error(
-            "R188: failed to start optional startup warmups"
+            "Optional startup warmups could not be started (error_type=%s)",
+            type(exc).__name__,
         )
 
 
-def _mark_startup_fatal(phase: str, exc: BaseException) -> None:
+def _mark_startup_fatal(
+    phase: str,
+    exc: BaseException,
+    *,
+    reason_code=None,
+) -> None:
     try:
         from .startup_lifecycle import mark_startup_fatal
 
-        mark_startup_fatal(phase, exc)
+        mark_startup_fatal(phase, exc, reason_code=reason_code)
-    except Exception:
+    except Exception as diagnostics_exc:
         # IMPORTANT: preserve the original bootstrap exception even if diagnostics fail.
-        logging.getLogger("ComfyUI-OpenClaw").exception(
+        logging.getLogger("ComfyUI-OpenClaw").error(
-            "R188: failed to record fatal startup state"
+            "Startup diagnostics update failed (error_type=%s)",
+            type(diagnostics_exc).__name__,
         )
 
 
+def _mark_required_initialization_started() -> None:
+    from .startup_lifecycle import mark_required_initialization_started
+
+    mark_required_initialization_started()
+
+
+def _mark_host_waiting(*, attempt: int, max_attempts: int) -> None:
+    from .startup_lifecycle import mark_host_waiting
+
+    mark_host_waiting(attempt=attempt, max_attempts=max_attempts)
+
+
+def _mark_route_registration_started(
+    *, attempt: int = 0, max_attempts: int = 0
+) -> None:
+    from .startup_lifecycle import mark_route_registration_started
+
+    mark_route_registration_started(
+        attempt=attempt,
+        max_attempts=max_attempts,
+    )
+
+
 def _load_plugin_shutdown_registrars():
     """Load optional startup registrars behind one patchable compatibility seam."""
 
@@ -125,8 +166,7 @@ def _initialize_registries_and_security_gate() -> None:
         # CRITICAL: pass db_path as keyword (first positional arg is backend object).
         IdempotencyStore().configure_durable(db_path=db_path, strict_mode=True)
         logging.getLogger("ComfyUI-OpenClaw").info(
-            "IdempotencyStore durable backend configured at: %s (strict_mode=True)",
+            "IdempotencyStore durable backend configured (strict_mode=True)"
-            db_path,
         )
 
         if config.bridge_enabled:
@@ -145,9 +185,10 @@ def _initialize_registries_and_security_gate() -> None:
         from .security_gate import enforce_startup_gate
 
         enforce_startup_gate()
-    except Exception as e:
+    except Exception as exc:
         logging.getLogger("ComfyUI-OpenClaw").error(
-            f"Failed to initialize registries: {e}"
+            "Required registry initialization failed (error_type=%s)",
+            type(exc).__name__,
         )
         # CRITICAL: keep bootstrap fail-closed; swallowing startup gate errors
         # silently degrades security posture and can expose partial registration.
@@ -249,7 +290,6 @@ 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 = (
@@ -284,85 +324,271 @@ def _register_bridge_routes(router, bridge_handlers) -> None:
             raise
 
 
-def _start_registration_retry_loop() -> None:
+def _resolve_prompt_server():
-    """R25: Retry route registration while PromptServer is warming up."""
+    ps_mod = sys.modules.get("server")
+    prompt_server = getattr(ps_mod, "PromptServer", None) if ps_mod else None
+    return getattr(prompt_server, "instance", None) if prompt_server else None
 
-    def _retry_worker():
-        global _routes_registered
-        attempts = 0
-        max_attempts = 10
-        delay = 2.0
-        logger = logging.getLogger("ComfyUI-OpenClaw")
 
-        while not _routes_registered and attempts < max_attempts:
+def reset_route_bootstrap_for_tests() -> None:
+    """Invalidate background ownership and reset the route bootstrap seam."""
+
+    global _routes_registered
+    global _registration_error
+    global _registration_generation
+    global _registration_inflight
+    global _registration_retry_thread
+    global _registration_started
+    with _registration_condition:
+        _registration_generation += 1
+        _routes_registered = False
+        _registration_inflight = False
+        _registration_started = False
+        _registration_error = None
+        _registration_retry_thread = None
+        _registration_condition.notify_all()
+
+
+def _store_registration_success(*, generation: int | None = None) -> bool:
+    global _routes_registered
+    global _registration_error
+    global _registration_inflight
+    global _registration_started
+    with _registration_condition:
+        if generation is not None and generation != _registration_generation:
+            return False
+        _routes_registered = True
+        _registration_inflight = False
+        _registration_started = True
+        _registration_error = None
+        _registration_condition.notify_all()
+        return True
+
+
+def _store_registration_failure(
+    exc: Exception,
+    *,
+    generation: int | None = None,
+) -> bool:
+    global _registration_error
+    global _registration_inflight
+    global _registration_started
+    with _registration_condition:
+        if generation is not None and generation != _registration_generation:
+            return False
+        _registration_inflight = False
+        _registration_started = True
+        _registration_error = exc
+        _registration_condition.notify_all()
+        return True
+
+
+def _run_registration_retry_loop(
+    *,
+    max_attempts: int = _REGISTRATION_MAX_ATTEMPTS,
+    initial_delay: float = _REGISTRATION_INITIAL_DELAY_SEC,
+    sleep_fn: Callable[[float], None] = time.sleep,
+    generation: int | None = None,
+) -> None:
+    """Run the sole bounded host-wait owner with explicit terminal outcomes."""
+
+    global _registration_retry_thread
+    logger = logging.getLogger("ComfyUI-OpenClaw")
+    with _registration_condition:
+        owner_generation = (
+            _registration_generation if generation is None else generation
+        )
+
+    delay = max(0.0, float(initial_delay))
+    try:
+        for attempt in range(1, max_attempts + 1):
+            with _registration_condition:
+                if owner_generation != _registration_generation:
+                    return
+
             try:
-                ps_mod = sys.modules.get("server")
+                server = _resolve_prompt_server()
-                PromptServer = getattr(ps_mod, "PromptServer", None) if ps_mod else None
+            except Exception as exc:
-                if PromptServer and getattr(PromptServer, "instance", None) is not None:
+                _mark_startup_fatal("route_registration", exc)
-                    _do_full_registration(PromptServer.instance)
+                _store_registration_failure(exc, generation=owner_generation)
-                    _routes_registered = True
+                logger.error(
-                    logger.info(
+                    "PromptServer resolution failed (attempt=%s, error_type=%s)",
-                        "Routes registered successfully on attempt %s", attempts + 1
+                    attempt,
+                    type(exc).__name__,
+                )
+                return
+
+            if server is not None:
+                try:
+                    _mark_route_registration_started(
+                        attempt=attempt,
+                        max_attempts=max_attempts,
+                    )
+                    _do_full_registration(server)
+                    _mark_startup_ready_and_start_warmups()
+                except Exception as exc:
+                    _mark_startup_fatal("route_registration", exc)
+                    _store_registration_failure(exc, generation=owner_generation)
+                    logger.error(
+                        "Route registration failed " "(attempt=%s, error_type=%s)",
+                        attempt,
+                        type(exc).__name__,
                     )
                     return
-                logger.debug(
+                _store_registration_success(generation=owner_generation)
-                    "PromptServer.instance not ready (attempt %s)", attempts + 1
+                logger.info(
+                    "Routes registered successfully (attempt=%s)",
+                    attempt,
                 )
-            except Exception:
+                return
-                logger.exception("Error registering routes (attempt %s)", attempts + 1)
 
-            time.sleep(delay)
+            _mark_host_waiting(attempt=attempt, max_attempts=max_attempts)
-            delay = min(delay * 1.5, 30)
+            logger.debug("PromptServer not ready (attempt=%s)", attempt)
-            attempts += 1
+            if attempt < max_attempts:
+                sleep_fn(delay)
+                delay = min(delay * 1.5, 30.0)
 
-        if not _routes_registered:
+        failure = RuntimeError("route registration retry exhausted")
-            _mark_startup_fatal(
+        _mark_startup_fatal(
-                "route_registration_retry",
+            "host_wait",
-                RuntimeError(
+            failure,
-                    f"Failed to register routes after {max_attempts} attempts"
+            reason_code="retry_exhausted",
-                ),
+        )
-            )
+        _store_registration_failure(failure, generation=owner_generation)
-            logger.error(
+        logger.error(
-                "Failed to register routes after %s attempts. API endpoints unavailable.",
+            "Route registration retry exhausted (attempts=%s)",
-                max_attempts,
+            max_attempts,
-            )
+        )
+    finally:
+        with _registration_condition:
+            if owner_generation == _registration_generation:
+                current = threading.current_thread()
+                if _registration_retry_thread is current:
+                    _registration_retry_thread = None
+                _registration_condition.notify_all()
 
-    t = threading.Thread(target=_retry_worker, name="openclaw-route-retry", daemon=True)
+
-    t.start()
+def _start_registration_retry_loop() -> None:
+    """Start at most one background host-wait owner."""
+
+    global _registration_retry_thread
+    with _registration_condition:
+        existing = _registration_retry_thread
+        if existing is not None and existing.is_alive():
+            return
+        generation = _registration_generation
+        thread = threading.Thread(
+            target=_run_registration_retry_loop,
+            kwargs={"generation": generation},
+            name="openclaw-route-retry",
+            daemon=True,
+        )
+        _registration_retry_thread = thread
+        thread.start()
 
 
 def register_routes_once() -> None:
-    global _routes_registered
+    """Initialize and register routes through one process-wide bootstrap owner."""
-    if _routes_registered:
+
-        return
+    global _registration_inflight
+    global _registration_started
+    logger = logging.getLogger("ComfyUI-OpenClaw")
+
+    # CRITICAL: one condition owns initialization, registration, retry creation, and
+    # terminal error replay. Independent flags reintroduce duplicate side effects.
+    with _registration_condition:
+        if _routes_registered:
+            return
+        if _registration_error is not None:
+            raise _registration_error
+        if _registration_started:
+            while _registration_inflight:
+                _registration_condition.wait()
+            if _registration_error is not None:
+                raise _registration_error
+            return
+        _registration_started = True
+        _registration_inflight = True
+        generation = _registration_generation
 
     try:
-        _register_plugins_and_shutdown_hooks()
+        _mark_required_initialization_started()
-        _initialize_registries_and_security_gate()
+        try:
-    except Exception as exc:
+            _register_plugins_and_shutdown_hooks()
-        _mark_startup_fatal("required_startup", exc)
+            _initialize_registries_and_security_gate()
-        raise
+        except Exception as exc:
-
+            _mark_startup_fatal("required_initialization", exc)
-    try:
+            _store_registration_failure(exc, generation=generation)
-        ps_mod = sys.modules.get("server")
+            logger.error(
-        PromptServer = getattr(ps_mod, "PromptServer", None) if ps_mod else None
+                "Required startup initialization failed (error_type=%s)",
-
+                type(exc).__name__,
-        if PromptServer and getattr(PromptServer, "instance", None) is not None:
-            _do_full_registration(PromptServer.instance)
-            _routes_registered = True
-            logging.getLogger("ComfyUI-OpenClaw").info(
-                "Routes registered successfully on initial attempt."
             )
-        else:
+            raise
-            logging.getLogger("ComfyUI-OpenClaw").info(
+
-                "PromptServer not ready, starting background registration retry loop..."
+        try:
-            )
+            server = _resolve_prompt_server()
-            _start_registration_retry_loop()
+        except Exception as exc:
-    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")
+            _store_registration_failure(exc, generation=generation)
-        # CRITICAL: initial registration failures must fail closed. The retry loop is
+            logger.error(
-        # only for PromptServer warm-up, not for hiding broken route/bootstrap state.
+                "Initial PromptServer resolution failed (error_type=%s)",
+                type(exc).__name__,
+            )
+            raise
+        if server is None:
+            _mark_host_waiting(
+                attempt=0,
+                max_attempts=_REGISTRATION_MAX_ATTEMPTS,
+            )
+            try:
+                _start_registration_retry_loop()
+            except Exception as exc:
+                _mark_startup_fatal(
+                    "host_wait",
+                    exc,
+                    reason_code="retry_exhausted",
+                )
+                _store_registration_failure(exc, generation=generation)
+                logger.error(
+                    "Route registration retry owner failed to start " "(error_type=%s)",
+                    type(exc).__name__,
+                )
+                raise
+            logger.info(
+                "PromptServer not ready; route registration retry owner started"
+            )
+            return
+
+        _mark_route_registration_started()
+        try:
+            _do_full_registration(server)
+            _mark_startup_ready_and_start_warmups()
+        except Exception as exc:
+            _mark_startup_fatal("route_registration", exc)
+            _store_registration_failure(exc, generation=generation)
+            logger.error(
+                "Initial route registration failed (error_type=%s)",
+                type(exc).__name__,
+            )
+            # CRITICAL: only host availability is retryable. Broken route
+            # registration remains fail-closed and replays the same error.
+            raise
+
+        _store_registration_success(generation=generation)
+        logger.info("Routes registered successfully on initial attempt")
+    except BaseException:
+        with _registration_condition:
+            if (
+                generation == _registration_generation
+                and _registration_error is None
+                and not _routes_registered
+            ):
+                _registration_started = False
+            _registration_inflight = False
+            _registration_condition.notify_all()
         raise
+    finally:
+        with _registration_condition:
+            if generation == _registration_generation:
+                _registration_inflight = False
+                _registration_condition.notify_all()

services/startup_lifecycle.py

@@ -1,214 +1,698 @@
-"""
+"""Typed startup lifecycle outcomes and redacted public diagnostics.
-Startup lifecycle diagnostics and optional warmup boundaries.
 
-Required startup work still fails closed in callers. This module only tracks
+Required startup work still fails closed in callers. This module owns only the
-readiness and runs optional warmups without delaying route availability.
+phase/result state machine and optional post-ready warmup observations; it does
+not own ComfyUI's application lifecycle.
 """
 
 from __future__ import annotations
 
 import logging
+import math
+import re
 import threading
 import time
-from typing import Any, Callable, Dict, Iterable, Optional
+from collections.abc import Callable, Iterable
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
 
 logger = logging.getLogger("ComfyUI-OpenClaw.services.startup_lifecycle")
 
-STARTUP_STARTING = "starting"
+SCHEMA_VERSION = 1
-STARTUP_READY = "ready"
+MAX_DIAGNOSTIC_MS = 86_400_000
-STARTUP_DEGRADED_WARMUP = "degraded-warmup"
+MAX_WARMUPS = 16
-STARTUP_FATAL = "fatal-startup"
+STARTUP_DIAGNOSTIC_KEYS = (
+    "schema_version",
+    "phase",
+    "state",
+    "reason_code",
+    "ready",
+    "degraded",
+    "fatal",
+    "attempt",
+    "max_attempts",
+    "elapsed_ms",
+    "phase_elapsed_ms",
+    "ready_elapsed_ms",
+    "warmups",
+)
 
-WARMUP_PENDING = "pending"
+
-WARMUP_RUNNING = "running"
+class StartupPhase(str, Enum):
-WARMUP_SUCCEEDED = "succeeded"
+    PACKAGE_IMPORT = "package_import"
-WARMUP_FAILED = "failed"
+    REQUIRED_INITIALIZATION = "required_initialization"
-WARMUP_TIMED_OUT = "timed_out"
+    HOST_WAIT = "host_wait"
+    ROUTE_REGISTRATION = "route_registration"
+    COMPLETE = "complete"
+    OPTIONAL_WARMUP = "optional_warmup"
+
+
+class StartupState(str, Enum):
+    STARTING = "starting"
+    INITIALIZING = "initializing"
+    WAITING_FOR_HOST = "waiting_for_host"
+    REGISTERING_ROUTES = "registering_routes"
+    READY = "ready"
+    DEGRADED = "degraded"
+    FATAL = "fatal"
+
+
+class StartupReason(str, Enum):
+    BOOTSTRAP_STARTED = "bootstrap_started"
+    BOOTSTRAP_IMPORT_FAILED = "bootstrap_import_failed"
+    REQUIRED_INITIALIZATION_STARTED = "required_initialization_started"
+    REQUIRED_INITIALIZATION_FAILED = "required_initialization_failed"
+    HOST_NOT_READY = "host_not_ready"
+    ROUTE_REGISTRATION_STARTED = "route_registration_started"
+    ROUTE_REGISTRATION_SUCCEEDED = "route_registration_succeeded"
+    ROUTE_REGISTRATION_FAILED = "route_registration_failed"
+    RETRY_EXHAUSTED = "retry_exhausted"
+    WARMUP_STARTED = "warmup_started"
+    WARMUP_SUCCEEDED = "warmup_succeeded"
+    WARMUP_FAILED = "warmup_failed"
+    WARMUP_TIMED_OUT = "warmup_timed_out"
+
+
+class WarmupState(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    TIMED_OUT = "timed_out"
+
+
+# Compatibility constants retained for existing internal imports.
+STARTUP_STARTING = StartupState.STARTING.value
+STARTUP_READY = StartupState.READY.value
+STARTUP_DEGRADED_WARMUP = StartupState.DEGRADED.value
+STARTUP_FATAL = StartupState.FATAL.value
+WARMUP_PENDING = WarmupState.PENDING.value
+WARMUP_RUNNING = WarmupState.RUNNING.value
+WARMUP_SUCCEEDED = WarmupState.SUCCEEDED.value
+WARMUP_FAILED = WarmupState.FAILED.value
+WARMUP_TIMED_OUT = WarmupState.TIMED_OUT.value
 
 WarmupSpec = tuple[str, Callable[[], Any], float]
 
-_LOCK = threading.RLock()
+
-_STARTED_AT = time.time()
+class StartupTransitionError(RuntimeError):
-_READY = False
+    """Stable transition failure that never embeds caller/source content."""
-_READY_PHASE: Optional[str] = None
+
-_READY_AT: Optional[float] = None
+    def __init__(self, code: str):
-_FATAL: Optional[Dict[str, Any]] = None
+        self.code = code
-_WARMUPS: Dict[str, Dict[str, Any]] = {}
+        super().__init__(code)
+
+
+@dataclass(frozen=True)
+class WarmupOutcome:
+    name: str
+    state: WarmupState
+    reason_code: StartupReason
+    timeout_ms: int
+    duration_ms: int
+
+    def to_diagnostics(self) -> dict[str, Any]:
+        return {
+            "name": self.name,
+            "state": self.state.value,
+            "reason_code": self.reason_code.value,
+            "timeout_ms": self.timeout_ms,
+            "duration_ms": self.duration_ms,
+        }
+
+
+@dataclass(frozen=True)
+class StartupOutcome:
+    schema_version: int
+    phase: StartupPhase
+    state: StartupState
+    reason_code: StartupReason
+    ready: bool
+    degraded: bool
+    fatal: bool
+    attempt: int
+    max_attempts: int
+    elapsed_ms: int
+    phase_elapsed_ms: int
+    ready_elapsed_ms: int | None
+    warmups: tuple[WarmupOutcome, ...]
+
+    def to_diagnostics(self) -> dict[str, Any]:
+        """Return a fresh, ordered, JSON-safe public projection."""
+
+        return {
+            "schema_version": self.schema_version,
+            "phase": self.phase.value,
+            "state": self.state.value,
+            "reason_code": self.reason_code.value,
+            "ready": self.ready,
+            "degraded": self.degraded,
+            "fatal": self.fatal,
+            "attempt": self.attempt,
+            "max_attempts": self.max_attempts,
+            "elapsed_ms": self.elapsed_ms,
+            "phase_elapsed_ms": self.phase_elapsed_ms,
+            "ready_elapsed_ms": self.ready_elapsed_ms,
+            "warmups": [warmup.to_diagnostics() for warmup in self.warmups],
+        }
+
+
+@dataclass
+class _WarmupRecord:
+    name: str
+    state: WarmupState
+    reason_code: StartupReason
+    timeout_sec: float
+    started_at: float | None = None
+    completed_at: float | None = None
+
+
+def _bounded_ms(seconds: float) -> int:
+    if not math.isfinite(seconds) or seconds <= 0:
+        return 0
+    return min(round(seconds * 1000.0), MAX_DIAGNOSTIC_MS)
+
+
+_WARMUP_NAME_RE = re.compile(r"[^a-zA-Z0-9_.-]+")
+
+
+def _safe_warmup_name(value: Any) -> str:
+    normalized = _WARMUP_NAME_RE.sub("_", str(value or "warmup")).strip("._-")
+    return (normalized or "warmup")[:64]
+
+
+class StartupLifecycle:
+    """Single lock-protected owner of startup phase and warmup outcomes."""
+
+    def __init__(self, *, monotonic_fn: Callable[[], float] = time.monotonic):
+        self._clock = monotonic_fn
+        self._lock = threading.RLock()
+        self._generation = 0
+        self._reset_locked()
+
+    def _reset_locked(self) -> None:
+        now = self._clock()
+        self._started_at = now
+        self._phase_started_at = now
+        self._ready_at: float | None = None
+        self._phase = StartupPhase.PACKAGE_IMPORT
+        self._state = StartupState.STARTING
+        self._reason_code = StartupReason.BOOTSTRAP_STARTED
+        self._attempt = 0
+        self._max_attempts = 0
+        self._warmups: dict[str, _WarmupRecord] = {}
+
+    def reset(self) -> None:
+        with self._lock:
+            self._generation += 1
+            self._reset_locked()
+
+    def _require_nonterminal(self) -> None:
+        if self._state is StartupState.FATAL:
+            raise StartupTransitionError("TERMINAL_STATE")
+
+    def _set(
+        self,
+        *,
+        phase: StartupPhase,
+        state: StartupState,
+        reason_code: StartupReason,
+        attempt: int | None = None,
+        max_attempts: int | None = None,
+    ) -> None:
+        now = self._clock()
+        if phase is not self._phase:
+            self._phase_started_at = now
+        self._phase = phase
+        self._state = state
+        self._reason_code = reason_code
+        if attempt is not None:
+            self._attempt = attempt
+        if max_attempts is not None:
+            self._max_attempts = max_attempts
+        if state is StartupState.READY and self._ready_at is None:
+            self._ready_at = now
+
+    def mark_required_initialization_started(self) -> None:
+        with self._lock:
+            self._require_nonterminal()
+            if self._state is not StartupState.STARTING:
+                raise StartupTransitionError("INVALID_TRANSITION")
+            self._set(
+                phase=StartupPhase.REQUIRED_INITIALIZATION,
+                state=StartupState.INITIALIZING,
+                reason_code=StartupReason.REQUIRED_INITIALIZATION_STARTED,
+            )
+
+    def mark_host_waiting(self, *, attempt: int, max_attempts: int) -> None:
+        with self._lock:
+            self._require_nonterminal()
+            if max_attempts <= 0 or attempt < 0 or attempt > max_attempts:
+                raise StartupTransitionError("ATTEMPT_OUT_OF_RANGE")
+            if self._state is StartupState.INITIALIZING:
+                if attempt != 0:
+                    raise StartupTransitionError("ATTEMPT_OUT_OF_RANGE")
+            elif self._state is StartupState.WAITING_FOR_HOST:
+                if attempt <= self._attempt:
+                    raise StartupTransitionError("ATTEMPT_NOT_INCREASING")
+                if max_attempts != self._max_attempts:
+                    raise StartupTransitionError("ATTEMPT_BOUND_CHANGED")
+            else:
+                raise StartupTransitionError("INVALID_TRANSITION")
+            self._set(
+                phase=StartupPhase.HOST_WAIT,
+                state=StartupState.WAITING_FOR_HOST,
+                reason_code=StartupReason.HOST_NOT_READY,
+                attempt=attempt,
+                max_attempts=max_attempts,
+            )
+
+    def mark_route_registration_started(
+        self, *, attempt: int = 0, max_attempts: int = 0
+    ) -> None:
+        with self._lock:
+            self._require_nonterminal()
+            if self._state not in {
+                StartupState.INITIALIZING,
+                StartupState.WAITING_FOR_HOST,
+            }:
+                raise StartupTransitionError("INVALID_TRANSITION")
+            if attempt < 0 or max_attempts < 0 or attempt > max_attempts:
+                raise StartupTransitionError("ATTEMPT_OUT_OF_RANGE")
+            if self._state is StartupState.INITIALIZING:
+                if attempt != 0 or max_attempts != 0:
+                    raise StartupTransitionError("ATTEMPT_OUT_OF_RANGE")
+            else:
+                if max_attempts != self._max_attempts:
+                    raise StartupTransitionError("ATTEMPT_BOUND_CHANGED")
+                if attempt <= self._attempt:
+                    raise StartupTransitionError("ATTEMPT_NOT_INCREASING")
+            self._set(
+                phase=StartupPhase.ROUTE_REGISTRATION,
+                state=StartupState.REGISTERING_ROUTES,
+                reason_code=StartupReason.ROUTE_REGISTRATION_STARTED,
+                attempt=attempt,
+                max_attempts=max_attempts,
+            )
+
+    def mark_ready(self) -> None:
+        with self._lock:
+            self._require_nonterminal()
+            if self._state is not StartupState.REGISTERING_ROUTES:
+                raise StartupTransitionError("INVALID_TRANSITION")
+            self._set(
+                phase=StartupPhase.COMPLETE,
+                state=StartupState.READY,
+                reason_code=StartupReason.ROUTE_REGISTRATION_SUCCEEDED,
+            )
+
+    def mark_fatal(
+        self,
+        *,
+        phase: StartupPhase,
+        reason_code: StartupReason,
+    ) -> None:
+        with self._lock:
+            self._require_nonterminal()
+            allowed = {
+                StartupReason.BOOTSTRAP_IMPORT_FAILED,
+                StartupReason.REQUIRED_INITIALIZATION_FAILED,
+                StartupReason.ROUTE_REGISTRATION_FAILED,
+                StartupReason.RETRY_EXHAUSTED,
+            }
+            if reason_code not in allowed:
+                raise StartupTransitionError("INVALID_FATAL_REASON")
+            expected_phase = {
+                StartupReason.BOOTSTRAP_IMPORT_FAILED: StartupPhase.PACKAGE_IMPORT,
+                StartupReason.REQUIRED_INITIALIZATION_FAILED: (
+                    StartupPhase.REQUIRED_INITIALIZATION
+                ),
+                StartupReason.ROUTE_REGISTRATION_FAILED: (
+                    StartupPhase.ROUTE_REGISTRATION
+                ),
+                StartupReason.RETRY_EXHAUSTED: StartupPhase.HOST_WAIT,
+            }[reason_code]
+            if phase is not expected_phase:
+                raise StartupTransitionError("FATAL_PHASE_MISMATCH")
+            allowed_states = {
+                StartupReason.BOOTSTRAP_IMPORT_FAILED: {
+                    StartupState.STARTING,
+                },
+                StartupReason.REQUIRED_INITIALIZATION_FAILED: {
+                    StartupState.INITIALIZING,
+                },
+                StartupReason.ROUTE_REGISTRATION_FAILED: {
+                    StartupState.INITIALIZING,
+                    StartupState.WAITING_FOR_HOST,
+                    StartupState.REGISTERING_ROUTES,
+                },
+                StartupReason.RETRY_EXHAUSTED: {
+                    StartupState.WAITING_FOR_HOST,
+                },
+            }[reason_code]
+            if self._state not in allowed_states:
+                raise StartupTransitionError("INVALID_FATAL_TRANSITION")
+            self._set(
+                phase=phase,
+                state=StartupState.FATAL,
+                reason_code=reason_code,
+            )
+
+    def mark_retry_exhausted(self) -> None:
+        with self._lock:
+            self._require_nonterminal()
+            if self._state is not StartupState.WAITING_FOR_HOST:
+                raise StartupTransitionError("INVALID_TRANSITION")
+            self._set(
+                phase=StartupPhase.HOST_WAIT,
+                state=StartupState.FATAL,
+                reason_code=StartupReason.RETRY_EXHAUSTED,
+            )
+
+    def begin_warmup(self, name: str, timeout_sec: float) -> tuple[bool, int, str]:
+        safe_name = _safe_warmup_name(name)
+        timeout_sec = max(0.01, min(float(timeout_sec or 5.0), 60.0))
+        with self._lock:
+            self._require_nonterminal()
+            if self._state not in {StartupState.READY, StartupState.DEGRADED}:
+                raise StartupTransitionError("WARMUP_BEFORE_READY")
+            existing = self._warmups.get(safe_name)
+            if existing is not None:
+                return False, self._generation, safe_name
+            if len(self._warmups) >= MAX_WARMUPS:
+                raise StartupTransitionError("WARMUP_LIMIT_EXCEEDED")
+            self._warmups[safe_name] = _WarmupRecord(
+                name=safe_name,
+                state=WarmupState.PENDING,
+                reason_code=StartupReason.WARMUP_STARTED,
+                timeout_sec=timeout_sec,
+            )
+            return True, self._generation, safe_name
+
+    def mark_warmup_running(self, name: str, generation: int) -> None:
+        with self._lock:
+            if generation != self._generation:
+                return
+            record = self._warmups.get(name)
+            if record is None:
+                return
+            record.state = WarmupState.RUNNING
+            record.reason_code = StartupReason.WARMUP_STARTED
+            record.started_at = self._clock()
+            if self._state is StartupState.READY:
+                self._set(
+                    phase=StartupPhase.OPTIONAL_WARMUP,
+                    state=StartupState.READY,
+                    reason_code=StartupReason.WARMUP_STARTED,
+                )
+
+    def finish_warmup(
+        self,
+        name: str,
+        generation: int,
+        *,
+        state: WarmupState,
+    ) -> None:
+        reason_by_state = {
+            WarmupState.SUCCEEDED: StartupReason.WARMUP_SUCCEEDED,
+            WarmupState.FAILED: StartupReason.WARMUP_FAILED,
+            WarmupState.TIMED_OUT: StartupReason.WARMUP_TIMED_OUT,
+        }
+        reason = reason_by_state.get(state)
+        if reason is None:
+            raise StartupTransitionError("INVALID_WARMUP_RESULT")
+        with self._lock:
+            if generation != self._generation:
+                return
+            record = self._warmups.get(name)
+            if record is None:
+                return
+            if record.state in {
+                WarmupState.SUCCEEDED,
+                WarmupState.FAILED,
+                WarmupState.TIMED_OUT,
+            }:
+                return
+            record.state = state
+            record.reason_code = reason
+            record.completed_at = self._clock()
+            if state in {WarmupState.FAILED, WarmupState.TIMED_OUT}:
+                self._set(
+                    phase=StartupPhase.OPTIONAL_WARMUP,
+                    state=StartupState.DEGRADED,
+                    reason_code=reason,
+                )
+            elif self._state is StartupState.READY:
+                self._set(
+                    phase=StartupPhase.OPTIONAL_WARMUP,
+                    state=StartupState.READY,
+                    reason_code=reason,
+                )
+
+    def snapshot(self) -> StartupOutcome:
+        with self._lock:
+            now = self._clock()
+            warmups = []
+            for name in sorted(self._warmups):
+                record = self._warmups[name]
+                started_at = record.started_at
+                completed_at = record.completed_at
+                if started_at is None:
+                    duration = 0.0
+                else:
+                    duration = (
+                        completed_at if completed_at is not None else now
+                    ) - started_at
+                warmups.append(
+                    WarmupOutcome(
+                        name=record.name,
+                        state=record.state,
+                        reason_code=record.reason_code,
+                        timeout_ms=_bounded_ms(record.timeout_sec),
+                        duration_ms=_bounded_ms(duration),
+                    )
+                )
+            ready = self._state in {StartupState.READY, StartupState.DEGRADED}
+            return StartupOutcome(
+                schema_version=SCHEMA_VERSION,
+                phase=self._phase,
+                state=self._state,
+                reason_code=self._reason_code,
+                ready=ready,
+                degraded=self._state is StartupState.DEGRADED,
+                fatal=self._state is StartupState.FATAL,
+                attempt=self._attempt,
+                max_attempts=self._max_attempts,
+                elapsed_ms=_bounded_ms(now - self._started_at),
+                phase_elapsed_ms=_bounded_ms(now - self._phase_started_at),
+                ready_elapsed_ms=(
+                    _bounded_ms(now - self._ready_at)
+                    if self._ready_at is not None
+                    else None
+                ),
+                warmups=tuple(warmups),
+            )
+
+
+_LIFECYCLE = StartupLifecycle()
+
+
+def get_startup_outcome() -> StartupOutcome:
+    return _LIFECYCLE.snapshot()
+
+
+def get_startup_diagnostics() -> dict[str, Any]:
+    return get_startup_outcome().to_diagnostics()
+
+
+def mark_required_initialization_started() -> None:
+    _LIFECYCLE.mark_required_initialization_started()
+
+
+def mark_host_waiting(*, attempt: int, max_attempts: int) -> None:
+    _LIFECYCLE.mark_host_waiting(attempt=attempt, max_attempts=max_attempts)
+
+
+def mark_route_registration_started(*, attempt: int = 0, max_attempts: int = 0) -> None:
+    _LIFECYCLE.mark_route_registration_started(
+        attempt=attempt,
+        max_attempts=max_attempts,
+    )
 
 
 def mark_startup_ready(phase: str = "routes") -> None:
-    """Mark required startup work as ready."""
+    """Compatibility facade that reaches the required ready transition."""
-    global _READY, _READY_AT, _READY_PHASE
+
-    with _LOCK:
+    _ = phase
-        if _FATAL is not None:
+    outcome = _LIFECYCLE.snapshot()
-            return
+    if outcome.state is StartupState.STARTING:
-        _READY = True
+        _LIFECYCLE.mark_required_initialization_started()
-        _READY_PHASE = str(phase or "routes")
+        _LIFECYCLE.mark_route_registration_started()
-        _READY_AT = time.time()
+        _LIFECYCLE.mark_ready()
+        return
+    if outcome.state is StartupState.REGISTERING_ROUTES:
+        _LIFECYCLE.mark_ready()
+        return
+    if outcome.state in {StartupState.READY, StartupState.DEGRADED}:
+        return
+    raise StartupTransitionError("INVALID_TRANSITION")
 
 
-def mark_startup_fatal(phase: str, exc: BaseException) -> None:
+def mark_startup_fatal(
-    """Record a fatal required-startup failure."""
+    phase: str,
-    global _FATAL, _READY
+    exc: BaseException | None = None,
-    with _LOCK:
+    *,
-        _READY = False
+    reason_code: StartupReason | str | None = None,
-        _FATAL = {
+) -> None:
-            "phase": str(phase or "startup"),
+    """Record a stable fatal classification without retaining ``exc``."""
-            "error_type": type(exc).__name__,
+
-            "error": str(exc)[:500],
+    _ = exc
-            "ts": time.time(),
+    phase_map = {
+        "package_import": StartupPhase.PACKAGE_IMPORT,
+        "required_startup": StartupPhase.REQUIRED_INITIALIZATION,
+        "required_initialization": StartupPhase.REQUIRED_INITIALIZATION,
+        "route_registration": StartupPhase.ROUTE_REGISTRATION,
+        "route_registration_retry": StartupPhase.HOST_WAIT,
+        "host_wait": StartupPhase.HOST_WAIT,
+    }
+    resolved_phase = phase_map.get(str(phase), StartupPhase.REQUIRED_INITIALIZATION)
+    if reason_code is None:
+        default_reasons = {
+            StartupPhase.PACKAGE_IMPORT: StartupReason.BOOTSTRAP_IMPORT_FAILED,
+            StartupPhase.REQUIRED_INITIALIZATION: (
+                StartupReason.REQUIRED_INITIALIZATION_FAILED
+            ),
+            StartupPhase.ROUTE_REGISTRATION: (StartupReason.ROUTE_REGISTRATION_FAILED),
+            StartupPhase.HOST_WAIT: StartupReason.RETRY_EXHAUSTED,
         }
+        resolved_reason = default_reasons[resolved_phase]
+    else:
+        resolved_reason = (
+            reason_code
+            if isinstance(reason_code, StartupReason)
+            else StartupReason(str(reason_code))
+        )
+    _LIFECYCLE.mark_fatal(
+        phase=resolved_phase,
+        reason_code=resolved_reason,
+    )
+
+
+def mark_bootstrap_import_failed(exc: BaseException) -> None:
+    mark_startup_fatal(
+        "package_import",
+        exc,
+        reason_code=StartupReason.BOOTSTRAP_IMPORT_FAILED,
+    )
+
+
+def mark_retry_exhausted() -> None:
+    _LIFECYCLE.mark_retry_exhausted()
 
 
 def start_optional_warmups(specs: Iterable[WarmupSpec]) -> None:
-    """Start optional warmups in background monitor threads."""
+    for name, fn, timeout_sec in tuple(specs or ()):
-    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."""
+    _LIFECYCLE.reset()
-    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))
+    should_start, generation, safe_name = _LIFECYCLE.begin_warmup(name, timeout_sec)
-    with _LOCK:
+    if not should_start:
-        existing = _WARMUPS.get(name)
+        return
-        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),
+        args=(safe_name, generation, fn, max(0.01, min(timeout_sec, 60.0))),
-        name=f"openclaw-warmup-monitor-{name}",
+        name=f"openclaw-warmup-monitor-{safe_name}",
         daemon=True,
     )
-    monitor.start()
+    try:
+        monitor.start()
+    except Exception:
+        _LIFECYCLE.finish_warmup(
+            safe_name,
+            generation,
+            state=WarmupState.FAILED,
+        )
+        raise
 
 
-def _warmup_monitor(name: str, fn: Callable[[], Any], timeout_sec: float) -> None:
+def _warmup_monitor(
-    started_at = time.time()
+    name: str,
+    generation: int,
+    fn: Callable[[], Any],
+    timeout_sec: float,
+) -> None:
     done = threading.Event()
-    result: Dict[str, Any] = {}
+    result: dict[str, bool] = {}
 
     def _worker() -> None:
         try:
-            result["value"] = fn()
+            fn()
             result["ok"] = True
-        except Exception as exc:  # pragma: no cover - defensive outer guard
+        except Exception:
+            # SECURITY: never retain or log arbitrary exception content.
             result["ok"] = False
-            result["exc"] = exc
         finally:
             done.set()
 
-    with _LOCK:
+    _LIFECYCLE.mark_warmup_running(name, generation)
-        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()
+    try:
-
+        worker.start()
-    if not done.wait(timeout=timeout_sec):
+    except Exception as exc:
-        _finish_warmup(
+        _LIFECYCLE.finish_warmup(
             name,
-            WARMUP_TIMED_OUT,
+            generation,
-            started_at,
+            state=WarmupState.FAILED,
-            error_type="TimeoutError",
-            error=f"optional warmup exceeded {timeout_sec:.2f}s",
         )
         logger.warning(
-            "R188: optional startup warmup timed out: %s (%.2fs)",
+            "Optional startup warmup worker could not start "
+            "(component=%s, error_type=%s)",
             name,
-            timeout_sec,
+            type(exc).__name__,
+        )
+        return
+
+    if not done.wait(timeout=timeout_sec):
+        _LIFECYCLE.finish_warmup(
+            name,
+            generation,
+            state=WarmupState.TIMED_OUT,
+        )
+        logger.warning(
+            "Optional startup warmup timed out (component=%s, reason_code=%s)",
+            name,
+            StartupReason.WARMUP_TIMED_OUT.value,
         )
         return
 
-    exc = result.get("exc")
     if result.get("ok"):
-        _finish_warmup(name, WARMUP_SUCCEEDED, started_at)
+        _LIFECYCLE.finish_warmup(
-        logger.info("R188: optional startup warmup completed: %s", name)
+            name,
+            generation,
+            state=WarmupState.SUCCEEDED,
+        )
+        logger.info("Optional startup warmup completed (component=%s)", name)
         return
 
-    _finish_warmup(
+    _LIFECYCLE.finish_warmup(
         name,
-        WARMUP_FAILED,
+        generation,
-        started_at,
+        state=WarmupState.FAILED,
-        error_type=type(exc).__name__ if exc else "Exception",
+    )
-        error=str(exc)[:500] if exc else "unknown warmup failure",
+    logger.warning(
+        "Optional startup warmup failed (component=%s, reason_code=%s)",
+        name,
+        StartupReason.WARMUP_FAILED.value,
     )
-    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

@@ -126,19 +126,19 @@
           "review_after": "2027-01-11"
         },
         {
-          "scope": "_start_registration_retry_loop._retry_worker",
+          "scope": "_run_registration_retry_loop",
-          "expected_count": 1,
+          "expected_count": 2,
           "classification": "allowed_boundary_guard",
-          "reason": "Background route warmup retries after transient registration failures.",
+          "reason": "The bounded retry owner classifies PromptServer resolution and route registration failures without exposing exception content.",
-          "regression_owner": "tests/test_route_registration.py",
+          "regression_owner": "tests/test_bootstrap_lifecycle_outcome.py",
           "review_after": "2027-01-11"
         },
         {
           "scope": "register_routes_once",
-          "expected_count": 2,
+          "expected_count": 5,
           "classification": "allowed_boundary_guard",
-          "reason": "Required startup failures are recorded and re-raised instead of hidden.",
+          "reason": "Required, host-resolution, retry-owner-start, and route failures are classified and replayed fail-closed, while BaseException cleanup re-raises cancellation unchanged.",
-          "regression_owner": "tests/test_r180_exception_boundary_governance.py",
+          "regression_owner": "tests/test_bootstrap_lifecycle_outcome.py",
           "review_after": "2027-01-11"
         }
       ]

tests/static_analysis_policy.json

@@ -1750,13 +1750,6 @@
       "message": "Unused \"type: ignore\" comment",
       "count": 1
     },
-    {
-      "tool": "mypy",
-      "path": "services/route_bootstrap.py",
-      "code": "misc",
-      "message": "Assignment to variable \"exc\" outside except: block",
-      "count": 1
-    },
     {
       "tool": "mypy",
       "path": "services/runtime_config.py",
@@ -8386,13 +8379,6 @@
       "message": "Variable `BridgeHandlers` in function should be lowercase",
       "count": 1
     },
-    {
-      "tool": "ruff",
-      "path": "services/route_bootstrap.py",
-      "code": "N806",
-      "message": "Variable `PromptServer` in function should be lowercase",
-      "count": 2
-    },
     {
       "tool": "ruff",
       "path": "services/runtime_config.py",
@@ -9667,34 +9653,6 @@
       "message": "Import from `collections.abc` instead: `Mapping`",
       "count": 1
     },
-    {
-      "tool": "ruff",
-      "path": "services/startup_lifecycle.py",
-      "code": "UP006",
-      "message": "Use `dict` instead of `Dict` for type annotation",
-      "count": 5
-    },
-    {
-      "tool": "ruff",
-      "path": "services/startup_lifecycle.py",
-      "code": "UP035",
-      "message": "Import from `collections.abc` instead: `Callable`, `Iterable`",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "services/startup_lifecycle.py",
-      "code": "UP035",
-      "message": "`typing.Dict` is deprecated, use `dict` instead",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "services/startup_lifecycle.py",
-      "code": "UP045",
-      "message": "Use `X | None` for type annotations",
-      "count": 5
-    },
     {
       "tool": "ruff",
       "path": "services/startup_profile_gate.py",

tests/test_bootstrap_lifecycle_outcome.py

@@ -0,0 +1,804 @@
+import asyncio
+import json
+import runpy
+import sys
+import threading
+import time
+import unittest
+from dataclasses import FrozenInstanceError
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from services import route_bootstrap
+from services.startup_lifecycle import (
+    STARTUP_DIAGNOSTIC_KEYS,
+    MAX_DIAGNOSTIC_MS,
+    MAX_WARMUPS,
+    StartupLifecycle,
+    StartupPhase,
+    StartupReason,
+    StartupState,
+    StartupTransitionError,
+    WarmupState,
+    get_startup_diagnostics,
+    get_startup_outcome,
+    mark_bootstrap_import_failed,
+    reset_startup_lifecycle_for_tests,
+)
+
+
+class _Clock:
+    def __init__(self, value: float = 10.0):
+        self.value = value
+
+    def __call__(self) -> float:
+        return self.value
+
+    def advance(self, seconds: float) -> None:
+        self.value += seconds
+
+
+class TestStartupOutcomeContract(unittest.TestCase):
+    def test_initial_snapshot_is_frozen_versioned_and_deterministic(self):
+        clock = _Clock()
+        lifecycle = StartupLifecycle(monotonic_fn=clock)
+
+        outcome = lifecycle.snapshot()
+        first = outcome.to_diagnostics()
+        second = outcome.to_diagnostics()
+
+        self.assertEqual(tuple(first), STARTUP_DIAGNOSTIC_KEYS)
+        self.assertEqual(first, second)
+        self.assertEqual(first["schema_version"], 1)
+        self.assertEqual(first["phase"], "package_import")
+        self.assertEqual(first["state"], "starting")
+        self.assertEqual(first["reason_code"], "bootstrap_started")
+        self.assertFalse(first["ready"])
+        self.assertFalse(first["degraded"])
+        self.assertFalse(first["fatal"])
+        self.assertEqual(first["warmups"], [])
+        self.assertIsInstance(outcome.warmups, tuple)
+        with self.assertRaises(FrozenInstanceError):
+            outcome.ready = True
+
+    def test_legal_transition_path_has_bounded_monotonic_timing(self):
+        clock = _Clock()
+        lifecycle = StartupLifecycle(monotonic_fn=clock)
+
+        lifecycle.mark_required_initialization_started()
+        clock.advance(0.125)
+        lifecycle.mark_host_waiting(attempt=0, max_attempts=3)
+        clock.advance(0.125)
+        lifecycle.mark_host_waiting(attempt=1, max_attempts=3)
+        lifecycle.mark_route_registration_started(attempt=2, max_attempts=3)
+        clock.advance(200_000)
+        lifecycle.mark_ready()
+
+        diagnostics = lifecycle.snapshot().to_diagnostics()
+        self.assertEqual(diagnostics["phase"], "complete")
+        self.assertEqual(diagnostics["state"], "ready")
+        self.assertEqual(
+            diagnostics["reason_code"], "route_registration_succeeded"
+        )
+        self.assertTrue(diagnostics["ready"])
+        self.assertFalse(diagnostics["fatal"])
+        self.assertEqual(diagnostics["attempt"], 2)
+        self.assertEqual(diagnostics["max_attempts"], 3)
+        for field in ("elapsed_ms", "phase_elapsed_ms", "ready_elapsed_ms"):
+            self.assertIsInstance(diagnostics[field], int)
+            self.assertGreaterEqual(diagnostics[field], 0)
+            self.assertLessEqual(diagnostics[field], MAX_DIAGNOSTIC_MS)
+
+    def test_invalid_and_post_fatal_transitions_do_not_mutate_state(self):
+        lifecycle = StartupLifecycle()
+        before = lifecycle.snapshot()
+
+        with self.assertRaises(StartupTransitionError) as invalid:
+            lifecycle.mark_ready()
+        self.assertEqual(invalid.exception.code, "INVALID_TRANSITION")
+        self.assertEqual(lifecycle.snapshot(), before)
+
+        lifecycle.mark_fatal(
+            phase=StartupPhase.PACKAGE_IMPORT,
+            reason_code=StartupReason.BOOTSTRAP_IMPORT_FAILED,
+        )
+        fatal = lifecycle.snapshot()
+        with self.assertRaises(StartupTransitionError) as terminal:
+            lifecycle.mark_required_initialization_started()
+        self.assertEqual(terminal.exception.code, "TERMINAL_STATE")
+        self.assertEqual(lifecycle.snapshot(), fatal)
+
+    def test_retry_attempt_must_increase_and_stay_within_bound(self):
+        lifecycle = StartupLifecycle()
+        lifecycle.mark_required_initialization_started()
+        lifecycle.mark_host_waiting(attempt=0, max_attempts=2)
+
+        for attempt, code in ((0, "ATTEMPT_NOT_INCREASING"), (3, "ATTEMPT_OUT_OF_RANGE")):
+            with self.subTest(attempt=attempt):
+                before = lifecycle.snapshot()
+                with self.assertRaises(StartupTransitionError) as ctx:
+                    lifecycle.mark_host_waiting(attempt=attempt, max_attempts=2)
+                self.assertEqual(ctx.exception.code, code)
+                self.assertEqual(lifecycle.snapshot(), before)
+
+        with self.assertRaises(StartupTransitionError) as route_regression:
+            lifecycle.mark_route_registration_started(
+                attempt=0,
+                max_attempts=2,
+            )
+        self.assertEqual(
+            route_regression.exception.code,
+            "ATTEMPT_NOT_INCREASING",
+        )
+
+    def test_fatal_reason_must_match_its_phase(self):
+        lifecycle = StartupLifecycle()
+        before = lifecycle.snapshot()
+
+        with self.assertRaises(StartupTransitionError) as mismatch:
+            lifecycle.mark_fatal(
+                phase=StartupPhase.PACKAGE_IMPORT,
+                reason_code=StartupReason.ROUTE_REGISTRATION_FAILED,
+            )
+
+        self.assertEqual(mismatch.exception.code, "FATAL_PHASE_MISMATCH")
+        self.assertEqual(lifecycle.snapshot(), before)
+
+        with self.assertRaises(StartupTransitionError) as source:
+            lifecycle.mark_fatal(
+                phase=StartupPhase.ROUTE_REGISTRATION,
+                reason_code=StartupReason.ROUTE_REGISTRATION_FAILED,
+            )
+        self.assertEqual(source.exception.code, "INVALID_FATAL_TRANSITION")
+        self.assertEqual(lifecycle.snapshot(), before)
+
+    def test_exception_payload_is_never_retained_or_serialized(self):
+        marker = "PRIVATE_R231_SECRET C:/private/token.txt"
+
+        reset_startup_lifecycle_for_tests()
+        mark_bootstrap_import_failed(RuntimeError(marker))
+        outcome = get_startup_outcome()
+        rendered = json.dumps(get_startup_diagnostics(), sort_keys=True)
+
+        self.assertEqual(outcome.state, StartupState.FATAL)
+        self.assertEqual(outcome.reason_code, StartupReason.BOOTSTRAP_IMPORT_FAILED)
+        self.assertNotIn(marker, repr(outcome))
+        self.assertNotIn(marker, rendered)
+        self.assertNotIn("RuntimeError", rendered)
+        self.assertNotIn("started_at", rendered)
+        self.assertNotIn("traceback", rendered.lower())
+
+
+class TestStartupWarmupProjection(unittest.TestCase):
+    def setUp(self):
+        reset_startup_lifecycle_for_tests()
+        from services.startup_lifecycle import mark_startup_ready
+
+        mark_startup_ready("routes")
+
+    def tearDown(self):
+        reset_startup_lifecycle_for_tests()
+
+    def _wait_for_warmup(self, name: str, state: str) -> dict:
+        deadline = time.monotonic() + 1.0
+        while time.monotonic() < deadline:
+            diagnostics = get_startup_diagnostics()
+            warmup = next(
+                (
+                    item
+                    for item in diagnostics["warmups"]
+                    if item["name"] == name
+                ),
+                None,
+            )
+            if warmup and warmup["state"] == state:
+                return diagnostics
+            time.sleep(0.005)
+        self.fail(f"warmup {name} did not reach {state}")
+
+    def test_failure_and_timeout_degrade_without_leaking_exception_content(self):
+        from services.startup_lifecycle import start_optional_warmups
+
+        release = threading.Event()
+        marker = "PRIVATE_WARMUP_FAILURE C:/private/model"
+
+        def fail():
+            raise RuntimeError(marker)
+
+        def block():
+            release.wait(timeout=1)
+
+        with self.assertLogs("ComfyUI-OpenClaw", level="WARNING") as captured:
+            start_optional_warmups(
+                [
+                    ("z_failure", fail, 0.5),
+                    ("a_timeout", block, 0.01),
+                ]
+            )
+            self._wait_for_warmup("z_failure", "failed")
+            diagnostics = self._wait_for_warmup("a_timeout", "timed_out")
+            release.set()
+            time.sleep(0.02)
+
+        self.assertTrue(diagnostics["ready"])
+        self.assertTrue(diagnostics["degraded"])
+        self.assertFalse(diagnostics["fatal"])
+        self.assertEqual(diagnostics["state"], "degraded")
+        self.assertEqual(
+            [item["name"] for item in diagnostics["warmups"]],
+            ["a_timeout", "z_failure"],
+        )
+        rendered = json.dumps(diagnostics, sort_keys=True)
+        self.assertNotIn(marker, rendered)
+        self.assertNotIn("RuntimeError", rendered)
+        self.assertNotIn("error", rendered)
+        self.assertNotIn(marker, "\n".join(captured.output))
+        final = get_startup_diagnostics()
+        timed_out = next(
+            item for item in final["warmups"] if item["name"] == "a_timeout"
+        )
+        self.assertEqual(timed_out["state"], "timed_out")
+
+    def test_terminal_warmup_cannot_restart_or_be_overwritten(self):
+        lifecycle = StartupLifecycle()
+        lifecycle.mark_required_initialization_started()
+        lifecycle.mark_route_registration_started()
+        lifecycle.mark_ready()
+
+        should_start, generation, name = lifecycle.begin_warmup("provider", 0.01)
+        self.assertTrue(should_start)
+        lifecycle.mark_warmup_running(name, generation)
+        lifecycle.finish_warmup(
+            name,
+            generation,
+            state=WarmupState.TIMED_OUT,
+        )
+        should_restart, _, _ = lifecycle.begin_warmup("provider", 0.01)
+        lifecycle.finish_warmup(
+            name,
+            generation,
+            state=WarmupState.SUCCEEDED,
+        )
+
+        self.assertFalse(should_restart)
+        outcome = lifecycle.snapshot()
+        self.assertEqual(outcome.state, StartupState.DEGRADED)
+        self.assertEqual(outcome.warmups[0].state, WarmupState.TIMED_OUT)
+
+    def test_warmup_projection_has_a_non_mutating_cardinality_bound(self):
+        lifecycle = StartupLifecycle()
+        lifecycle.mark_required_initialization_started()
+        lifecycle.mark_route_registration_started()
+        lifecycle.mark_ready()
+
+        for index in range(MAX_WARMUPS):
+            should_start, _, _ = lifecycle.begin_warmup(
+                f"provider_{index}",
+                0.01,
+            )
+            self.assertTrue(should_start)
+        before = lifecycle.snapshot()
+
+        with self.assertRaises(StartupTransitionError) as limit:
+            lifecycle.begin_warmup("one_too_many", 0.01)
+
+        self.assertEqual(limit.exception.code, "WARMUP_LIMIT_EXCEEDED")
+        self.assertEqual(lifecycle.snapshot(), before)
+        self.assertEqual(len(before.warmups), MAX_WARMUPS)
+
+    def test_monitor_thread_start_failure_degrades_and_reraises_content_free(self):
+        from services.startup_lifecycle import start_optional_warmups
+
+        marker = "PRIVATE_MONITOR_START C:/private/monitor"
+        failure = RuntimeError(marker)
+        with patch(
+            "services.startup_lifecycle.threading.Thread"
+        ) as thread_factory:
+            thread_factory.return_value.start.side_effect = failure
+            with self.assertRaises(RuntimeError) as ctx:
+                start_optional_warmups(
+                    [("monitor_provider", lambda: None, 0.01)]
+                )
+
+        self.assertIs(ctx.exception, failure)
+        diagnostics = get_startup_diagnostics()
+        self.assertTrue(diagnostics["degraded"])
+        warmup = next(
+            item
+            for item in diagnostics["warmups"]
+            if item["name"] == "monitor_provider"
+        )
+        self.assertEqual(warmup["state"], "failed")
+        self.assertNotIn(marker, json.dumps(diagnostics))
+
+    def test_worker_thread_start_failure_degrades_without_escaping_monitor(self):
+        from services.startup_lifecycle import _LIFECYCLE, _warmup_monitor
+
+        marker = "PRIVATE_WORKER_START C:/private/worker"
+        failure = RuntimeError(marker)
+        _, generation, name = _LIFECYCLE.begin_warmup(
+            "worker_provider",
+            0.01,
+        )
+
+        with (
+            patch(
+                "services.startup_lifecycle.threading.Thread"
+            ) as thread_factory,
+            self.assertLogs("ComfyUI-OpenClaw", level="WARNING") as captured,
+        ):
+            thread_factory.return_value.start.side_effect = failure
+            _warmup_monitor(name, generation, lambda: None, 0.01)
+
+        diagnostics = get_startup_diagnostics()
+        self.assertTrue(diagnostics["degraded"])
+        warmup = next(
+            item
+            for item in diagnostics["warmups"]
+            if item["name"] == "worker_provider"
+        )
+        self.assertEqual(warmup["state"], "failed")
+        self.assertNotIn(marker, json.dumps(diagnostics))
+        self.assertNotIn(marker, "\n".join(captured.output))
+
+
+class TestRouteBootstrapOutcomeIntegration(unittest.TestCase):
+    def setUp(self):
+        route_bootstrap.reset_route_bootstrap_for_tests()
+        reset_startup_lifecycle_for_tests()
+
+    def tearDown(self):
+        route_bootstrap.reset_route_bootstrap_for_tests()
+        reset_startup_lifecycle_for_tests()
+
+    @staticmethod
+    def _host_module(instance):
+        return SimpleNamespace(PromptServer=SimpleNamespace(instance=instance))
+
+    def test_duplicate_concurrent_calls_share_one_sync_owner_and_retry_owner(self):
+        entered = threading.Event()
+        release = threading.Event()
+        errors = []
+
+        def required_init():
+            entered.set()
+            release.wait(timeout=1)
+
+        with (
+            patch.object(route_bootstrap, "_register_plugins_and_shutdown_hooks") as optional,
+            patch.object(
+                route_bootstrap,
+                "_initialize_registries_and_security_gate",
+                side_effect=required_init,
+            ) as required,
+            patch.object(route_bootstrap, "_start_registration_retry_loop") as retry,
+            patch.dict(sys.modules, {"server": self._host_module(None)}),
+        ):
+            first = threading.Thread(
+                target=self._call_registration,
+                args=(errors,),
+            )
+            second = threading.Thread(
+                target=self._call_registration,
+                args=(errors,),
+            )
+            first.start()
+            self.assertTrue(entered.wait(timeout=1))
+            second.start()
+            release.set()
+            first.join(timeout=1)
+            second.join(timeout=1)
+
+        self.assertFalse(first.is_alive())
+        self.assertFalse(second.is_alive())
+        self.assertEqual(errors, [])
+        self.assertEqual(optional.call_count, 1)
+        self.assertEqual(required.call_count, 1)
+        self.assertEqual(retry.call_count, 1)
+        diagnostics = get_startup_diagnostics()
+        self.assertEqual(diagnostics["state"], "waiting_for_host")
+        self.assertFalse(diagnostics["ready"])
+
+    @staticmethod
+    def _call_registration(errors):
+        try:
+            route_bootstrap.register_routes_once()
+        except BaseException as exc:  # test capture; production must not swallow
+            errors.append(exc)
+
+    def test_required_failure_is_shared_fail_closed_and_content_free(self):
+        marker = "PRIVATE_REQUIRED_FAILURE C:/private/config"
+        failure = RuntimeError(marker)
+
+        with (
+            patch.object(route_bootstrap, "_register_plugins_and_shutdown_hooks"),
+            patch.object(
+                route_bootstrap,
+                "_initialize_registries_and_security_gate",
+                side_effect=failure,
+            ) as required,
+            patch.dict(sys.modules, {"server": self._host_module(None)}),
+        ):
+            with self.assertLogs("ComfyUI-OpenClaw", level="ERROR") as captured:
+                observed = []
+                for _ in range(2):
+                    with self.assertRaises(RuntimeError) as ctx:
+                        route_bootstrap.register_routes_once()
+                    observed.append(ctx.exception)
+
+        self.assertEqual(required.call_count, 1)
+        self.assertIs(observed[0], failure)
+        self.assertIs(observed[1], failure)
+        diagnostics = get_startup_diagnostics()
+        self.assertEqual(diagnostics["state"], "fatal")
+        self.assertEqual(
+            diagnostics["reason_code"], "required_initialization_failed"
+        )
+        self.assertNotIn(marker, json.dumps(diagnostics))
+        self.assertNotIn(marker, "\n".join(captured.output))
+
+    def test_initial_registration_failure_is_fatal_and_reraised(self):
+        marker = "PRIVATE_ROUTE_FAILURE C:/private/route"
+        failure = RuntimeError(marker)
+        server = SimpleNamespace(app=object())
+
+        with (
+            patch.object(route_bootstrap, "_register_plugins_and_shutdown_hooks"),
+            patch.object(route_bootstrap, "_initialize_registries_and_security_gate"),
+            patch.object(
+                route_bootstrap,
+                "_do_full_registration",
+                side_effect=failure,
+            ),
+            patch.dict(sys.modules, {"server": self._host_module(server)}),
+        ):
+            with self.assertLogs("ComfyUI-OpenClaw", level="ERROR") as captured:
+                with self.assertRaises(RuntimeError) as ctx:
+                    route_bootstrap.register_routes_once()
+
+        self.assertIs(ctx.exception, failure)
+        diagnostics = get_startup_diagnostics()
+        self.assertEqual(diagnostics["state"], "fatal")
+        self.assertEqual(diagnostics["reason_code"], "route_registration_failed")
+        self.assertNotIn(marker, json.dumps(diagnostics))
+        self.assertNotIn(marker, "\n".join(captured.output))
+
+    def test_initial_host_resolution_failure_is_terminal_and_replayed(self):
+        marker = "PRIVATE_HOST_RESOLUTION C:/private/host"
+        failure = RuntimeError(marker)
+
+        with (
+            patch.object(route_bootstrap, "_register_plugins_and_shutdown_hooks"),
+            patch.object(route_bootstrap, "_initialize_registries_and_security_gate"),
+            patch.object(
+                route_bootstrap,
+                "_resolve_prompt_server",
+                side_effect=failure,
+            ) as resolve,
+        ):
+            observed = []
+            for _ in range(2):
+                with self.assertRaises(RuntimeError) as ctx:
+                    route_bootstrap.register_routes_once()
+                observed.append(ctx.exception)
+
+        self.assertEqual(resolve.call_count, 1)
+        self.assertIs(observed[0], failure)
+        self.assertIs(observed[1], failure)
+        diagnostics = get_startup_diagnostics()
+        self.assertTrue(diagnostics["fatal"])
+        self.assertEqual(diagnostics["reason_code"], "route_registration_failed")
+        self.assertNotIn(marker, json.dumps(diagnostics))
+
+    def test_retry_owner_start_failure_is_terminal_and_replayed(self):
+        marker = "PRIVATE_THREAD_START C:/private/thread"
+        failure = RuntimeError(marker)
+
+        with (
+            patch.object(route_bootstrap, "_register_plugins_and_shutdown_hooks"),
+            patch.object(route_bootstrap, "_initialize_registries_and_security_gate"),
+            patch.object(route_bootstrap, "_resolve_prompt_server", return_value=None),
+            patch.object(
+                route_bootstrap,
+                "_start_registration_retry_loop",
+                side_effect=failure,
+            ) as start_retry,
+        ):
+            observed = []
+            for _ in range(2):
+                with self.assertRaises(RuntimeError) as ctx:
+                    route_bootstrap.register_routes_once()
+                observed.append(ctx.exception)
+
+        self.assertEqual(start_retry.call_count, 1)
+        self.assertIs(observed[0], failure)
+        self.assertIs(observed[1], failure)
+        diagnostics = get_startup_diagnostics()
+        self.assertTrue(diagnostics["fatal"])
+        self.assertEqual(diagnostics["reason_code"], "retry_exhausted")
+        self.assertNotIn(marker, json.dumps(diagnostics))
+
+    def test_retry_success_and_exhaustion_have_distinct_outcomes(self):
+        from services.startup_lifecycle import (
+            mark_host_waiting,
+            mark_required_initialization_started,
+        )
+
+        server = SimpleNamespace(app=object())
+        sequence = iter([None, server])
+        mark_required_initialization_started()
+        mark_host_waiting(attempt=0, max_attempts=3)
+        with (
+            patch.object(
+                route_bootstrap,
+                "_resolve_prompt_server",
+                side_effect=lambda: next(sequence),
+            ),
+            patch.object(route_bootstrap, "_do_full_registration") as register,
+            patch.object(
+                route_bootstrap,
+                "_build_optional_startup_warmups",
+                return_value=[],
+            ),
+        ):
+            route_bootstrap._run_registration_retry_loop(
+                max_attempts=3,
+                initial_delay=0,
+                sleep_fn=lambda _delay: None,
+            )
+
+        register.assert_called_once_with(server)
+        diagnostics = get_startup_diagnostics()
+        self.assertTrue(diagnostics["ready"])
+        self.assertEqual(
+            diagnostics["reason_code"], "route_registration_succeeded"
+        )
+
+        reset_startup_lifecycle_for_tests()
+        route_bootstrap.reset_route_bootstrap_for_tests()
+        mark_required_initialization_started()
+        mark_host_waiting(attempt=0, max_attempts=2)
+        with patch.object(
+            route_bootstrap,
+            "_resolve_prompt_server",
+            return_value=None,
+        ):
+            route_bootstrap._run_registration_retry_loop(
+                max_attempts=2,
+                initial_delay=0,
+                sleep_fn=lambda _delay: None,
+            )
+
+        diagnostics = get_startup_diagnostics()
+        self.assertFalse(diagnostics["ready"])
+        self.assertTrue(diagnostics["fatal"])
+        self.assertEqual(diagnostics["reason_code"], "retry_exhausted")
+        self.assertEqual(diagnostics["attempt"], 2)
+
+    def test_base_exception_is_not_swallowed_or_converted_to_fatal(self):
+        signal = KeyboardInterrupt()
+
+        with (
+            patch.object(route_bootstrap, "_register_plugins_and_shutdown_hooks"),
+            patch.object(
+                route_bootstrap,
+                "_initialize_registries_and_security_gate",
+                side_effect=signal,
+            ),
+        ):
+            with self.assertRaises(KeyboardInterrupt) as ctx:
+                route_bootstrap.register_routes_once()
+
+        self.assertIs(ctx.exception, signal)
+        diagnostics = get_startup_diagnostics()
+        self.assertFalse(diagnostics["fatal"])
+        self.assertEqual(diagnostics["state"], "initializing")
+
+
+class TestPublicHealthLifecycleProjection(unittest.TestCase):
+    def setUp(self):
+        reset_startup_lifecycle_for_tests()
+
+    def tearDown(self):
+        reset_startup_lifecycle_for_tests()
+
+    @staticmethod
+    def _health_payload(*, diagnostics_side_effect=None):
+        from api.route_handlers import health_response
+
+        web = SimpleNamespace(json_response=lambda data, **_kwargs: data)
+        deps = SimpleNamespace(
+            web=web,
+            pack_start_time=time.time(),
+            pack_name="openclaw",
+            pack_version="test",
+            metrics=SimpleNamespace(
+                get_snapshot=lambda: {
+                    "errors_captured": 0,
+                    "logs_processed": 0,
+                }
+            ),
+            get_executor_diagnostics=lambda: {},
+            check_dependency=lambda _name: True,
+        )
+        client = MagicMock()
+        client.get_provider_summary.return_value = {
+            "provider": "openai",
+            "model": "test",
+            "key_configured": False,
+        }
+        diagnostics_patch = patch(
+            "services.startup_lifecycle.get_startup_diagnostics",
+            side_effect=diagnostics_side_effect,
+        )
+        if diagnostics_side_effect is None:
+            diagnostics_patch = patch(
+                "services.startup_lifecycle.get_startup_diagnostics",
+                wraps=get_startup_diagnostics,
+            )
+
+        with (
+            patch("services.llm_client.LLMClient", return_value=client),
+            patch("services.providers.keys.requires_api_key", return_value=True),
+            patch(
+                "services.job_events.get_job_event_store",
+                return_value=SimpleNamespace(stats=lambda: {}),
+            ),
+            patch(
+                "services.capabilities._get_control_plane_info",
+                return_value={},
+            ),
+            patch(
+                "services.runtime_profile.get_runtime_profile",
+                return_value="minimal",
+            ),
+            diagnostics_patch,
+        ):
+            return asyncio.run(health_response(SimpleNamespace(), deps))
+
+    def test_health_uses_exact_schema_and_content_free_fallback(self):
+        normal = self._health_payload()
+        self.assertEqual(
+            tuple(normal["startup"]),
+            STARTUP_DIAGNOSTIC_KEYS,
+        )
+
+        marker = "PRIVATE_HEALTH_FAILURE C:/private/health"
+        fallback = self._health_payload(
+            diagnostics_side_effect=RuntimeError(marker)
+        )
+        self.assertEqual(
+            tuple(fallback["startup"]),
+            STARTUP_DIAGNOSTIC_KEYS,
+        )
+        self.assertEqual(fallback["startup"]["schema_version"], 1)
+        self.assertEqual(fallback["startup"]["state"], "fatal")
+        self.assertEqual(
+            fallback["startup"]["reason_code"],
+            "bootstrap_import_failed",
+        )
+        self.assertNotIn(marker, json.dumps(fallback["startup"]))
+
+    def test_health_metadata_remains_public_and_all_aliases_share_one_handler(self):
+        from api.route_registrars import build_core_route_specs
+        from api.routes import health_handler
+        from services.endpoint_manifest import AuthTier, get_metadata
+
+        metadata = get_metadata(health_handler)
+        self.assertIsNotNone(metadata)
+        self.assertEqual(metadata.auth_tier, AuthTier.PUBLIC)
+
+        handlers = {"health_handler": health_handler}
+        sentinel = MagicMock()
+        handlers.update(
+            {
+                key: sentinel
+                for key in (
+                    "remote_admin_page_handler",
+                    "logs_tail_handler",
+                    "jobs_handler",
+                    "trace_handler",
+                    "webhook_handler",
+                    "webhook_submit_handler",
+                    "webhook_validate_handler",
+                    "capabilities_handler",
+                    "config_get_handler",
+                    "config_put_handler",
+                    "llm_test_handler",
+                    "llm_chat_handler",
+                    "llm_models_handler",
+                    "templates_list_handler",
+                    "preflight_handler",
+                    "inventory_handler",
+                    "pnginfo_handler",
+                    "list_checkpoints_handler",
+                    "create_checkpoint_handler",
+                    "get_checkpoint_handler",
+                    "delete_checkpoint_handler",
+                    "rewrite_recipes_list_handler",
+                    "rewrite_recipe_create_handler",
+                    "rewrite_recipe_get_handler",
+                    "rewrite_recipe_update_handler",
+                    "rewrite_recipe_delete_handler",
+                    "rewrite_recipe_dry_run_handler",
+                    "rewrite_recipe_apply_handler",
+                    "model_search_handler",
+                    "model_download_create_handler",
+                    "model_download_list_handler",
+                    "model_download_get_handler",
+                    "model_download_cancel_handler",
+                    "model_import_handler",
+                    "model_installations_list_handler",
+                    "secrets_status_handler",
+                    "secrets_put_handler",
+                    "events_stream_handler",
+                    "events_poll_handler",
+                    "secrets_delete_handler",
+                    "security_doctor_handler",
+                    "tools_list_handler",
+                    "tools_run_handler",
+                    "create_sweep_handler",
+                    "create_compare_handler",
+                    "list_experiments_handler",
+                    "get_experiment_handler",
+                    "update_experiment_handler",
+                    "select_apply_winner_handler",
+                )
+            }
+        )
+
+        for prefix in ("/openclaw", "/api/openclaw", "/moltbot", "/api/moltbot"):
+            specs = build_core_route_specs(prefix, handlers)
+            health = next(spec for spec in specs if spec.path == f"{prefix}/health")
+            self.assertIs(health.handler, health_handler)
+
+
+class TestPackageBootstrapLifecycleProjection(unittest.TestCase):
+    def setUp(self):
+        reset_startup_lifecycle_for_tests()
+
+    def tearDown(self):
+        reset_startup_lifecycle_for_tests()
+
+    @staticmethod
+    def _run_entrypoint_with_import_signal(signal):
+        original_import = __import__
+
+        def guarded_import(name, *args, **kwargs):
+            if name == "services.route_bootstrap":
+                raise signal
+            return original_import(name, *args, **kwargs)
+
+        root = Path(__file__).resolve().parents[1]
+        with patch("builtins.__import__", side_effect=guarded_import):
+            return runpy.run_path(
+                str(root / "__init__.py"),
+                run_name="openclaw_bootstrap_lifecycle_probe",
+            )
+
+    def test_package_import_failure_is_stably_classified_without_payload(self):
+        marker = "PRIVATE_IMPORT_FAILURE C:/private/import"
+
+        self._run_entrypoint_with_import_signal(ImportError(marker))
+
+        diagnostics = get_startup_diagnostics()
+        self.assertEqual(diagnostics["state"], "fatal")
+        self.assertEqual(
+            diagnostics["reason_code"],
+            "bootstrap_import_failed",
+        )
+        self.assertNotIn(marker, json.dumps(diagnostics))
+
+    def test_package_import_base_exception_is_reraised_unchanged(self):
+        signal = KeyboardInterrupt()
+
+        with self.assertRaises(KeyboardInterrupt) as ctx:
+            self._run_entrypoint_with_import_signal(signal)
+
+        self.assertIs(ctx.exception, signal)
+        self.assertFalse(get_startup_diagnostics()["fatal"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_r188_startup_lifecycle.py

@@ -39,28 +39,39 @@ class TestStartupLifecycleDiagnostics(unittest.TestCase):
         diagnostics = get_startup_diagnostics()
         while time.monotonic() < deadline:
             diagnostics = get_startup_diagnostics()
-            if diagnostics["warmups"]["slow_provider"]["state"] == "timed_out":
+            warmup = next(
+                (
+                    item
+                    for item in diagnostics["warmups"]
+                    if item["name"] == "slow_provider"
+                ),
+                None,
+            )
+            if warmup and warmup["state"] == "timed_out":
                 break
             time.sleep(0.01)
 
         release.set()
-        self.assertEqual(diagnostics["state"], "degraded-warmup")
+        self.assertEqual(diagnostics["state"], "degraded")
         self.assertEqual(diagnostics["ready"], True)
-        self.assertEqual(diagnostics["warmups"]["slow_provider"]["state"], "timed_out")
+        self.assertEqual(warmup["state"], "timed_out")
 
     def test_fatal_startup_state_is_distinct_from_warmup_degradation(self):
         from services.startup_lifecycle import (
             get_startup_diagnostics,
+            mark_required_initialization_started,
             mark_startup_fatal,
         )
 
-        mark_startup_fatal("security_gate", RuntimeError("blocked"))
+        mark_required_initialization_started()
+        mark_startup_fatal("required_initialization", RuntimeError("blocked"))
         diagnostics = get_startup_diagnostics()
 
-        self.assertEqual(diagnostics["state"], "fatal-startup")
+        self.assertEqual(diagnostics["state"], "fatal")
         self.assertFalse(diagnostics["ready"])
-        self.assertEqual(diagnostics["fatal"]["phase"], "security_gate")
+        self.assertTrue(diagnostics["fatal"])
-        self.assertIn("RuntimeError", diagnostics["fatal"]["error_type"])
+        self.assertEqual(diagnostics["phase"], "required_initialization")
+        self.assertEqual(diagnostics["reason_code"], "required_initialization_failed")
 
 
 class _DummyRoutes:
@@ -117,13 +128,17 @@ class _DummyBridgeHandlers:
 
 class TestRouteBootstrapWarmupBoundary(unittest.TestCase):
     def setUp(self):
+        from services import route_bootstrap
         from services.startup_lifecycle import reset_startup_lifecycle_for_tests
 
+        route_bootstrap.reset_route_bootstrap_for_tests()
         reset_startup_lifecycle_for_tests()
 
     def tearDown(self):
+        from services import route_bootstrap
         from services.startup_lifecycle import reset_startup_lifecycle_for_tests
 
+        route_bootstrap.reset_route_bootstrap_for_tests()
         reset_startup_lifecycle_for_tests()
 
     def test_full_registration_marks_ready_before_optional_warmup_finishes(self):
@@ -169,6 +184,7 @@ class TestRouteBootstrapWarmupBoundary(unittest.TestCase):
             get_runner.return_value = MagicMock()
             started_at = time.monotonic()
             route_bootstrap._do_full_registration(server)
+            route_bootstrap._mark_startup_ready_and_start_warmups()
             elapsed = time.monotonic() - started_at
 
         diagnostics = get_startup_diagnostics()
@@ -179,16 +195,16 @@ class TestRouteBootstrapWarmupBoundary(unittest.TestCase):
         self.assertTrue(app.triggers)
         self.assertTrue(app.approvals)
         self.assertTrue(diagnostics["ready"])
-        self.assertIn(
+        warmup = next(
-            diagnostics["warmups"]["slow_provider"]["state"], {"running", "succeeded"}
+            item for item in diagnostics["warmups"] if item["name"] == "slow_provider"
         )
+        self.assertIn(warmup["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,
@@ -200,9 +216,10 @@ class TestRouteBootstrapWarmupBoundary(unittest.TestCase):
                 route_bootstrap.register_routes_once()
 
         diagnostics = get_startup_diagnostics()
-        self.assertEqual(diagnostics["state"], "fatal-startup")
+        self.assertEqual(diagnostics["state"], "fatal")
         self.assertFalse(diagnostics["ready"])
-        self.assertEqual(diagnostics["fatal"]["phase"], "required_startup")
+        self.assertTrue(diagnostics["fatal"])
+        self.assertEqual(diagnostics["reason_code"], "required_initialization_failed")
 
 
 if __name__ == "__main__":