From 8c175f47ab0006fdc0643d3231a80ca1d9e575ce Mon Sep 17 00:00:00 2001 From: rookiestar28 Date: Fri, 31 Jul 2026 05:40:58 +0800 Subject: [PATCH] feat(startup): model bootstrap lifecycle outcomes --- __init__.py | 12 +- api/route_handlers.py | 18 +- services/route_bootstrap.py | 388 ++++++++--- services/startup_lifecycle.py | 782 +++++++++++++++++---- tests/exception_boundary_policy.json | 14 +- tests/static_analysis_policy.json | 42 -- tests/test_bootstrap_lifecycle_outcome.py | 804 ++++++++++++++++++++++ tests/test_r188_startup_lifecycle.py | 41 +- 8 files changed, 1808 insertions(+), 293 deletions(-) create mode 100644 tests/test_bootstrap_lifecycle_outcome.py diff --git a/__init__.py b/__init__.py index d910827..812bb81 100644 --- a/__init__.py +++ b/__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() diff --git a/api/route_handlers.py b/api/route_handlers.py index f131968..5d8021a 100644 --- a/api/route_handlers.py +++ b/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: diff --git a/services/route_bootstrap.py b/services/route_bootstrap.py index 00266da..8d89d06 100644 --- a/services/route_bootstrap.py +++ b/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( - "R188: failed to start optional startup warmups" + logging.getLogger("ComfyUI-OpenClaw").error( + "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) - except Exception: + mark_startup_fatal(phase, exc, reason_code=reason_code) + except Exception as diagnostics_exc: # IMPORTANT: preserve the original bootstrap exception even if diagnostics fail. - logging.getLogger("ComfyUI-OpenClaw").exception( - "R188: failed to record fatal startup state" + logging.getLogger("ComfyUI-OpenClaw").error( + "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)", - db_path, + "IdempotencyStore durable backend configured (strict_mode=True)" ) 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: - """R25: Retry route registration while PromptServer is warming up.""" +def _resolve_prompt_server(): + 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") - PromptServer = getattr(ps_mod, "PromptServer", None) if ps_mod else None - if PromptServer and getattr(PromptServer, "instance", None) is not None: - _do_full_registration(PromptServer.instance) - _routes_registered = True - logger.info( - "Routes registered successfully on attempt %s", attempts + 1 + server = _resolve_prompt_server() + except Exception as exc: + _mark_startup_fatal("route_registration", exc) + _store_registration_failure(exc, generation=owner_generation) + logger.error( + "PromptServer resolution failed (attempt=%s, error_type=%s)", + 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( - "PromptServer.instance not ready (attempt %s)", attempts + 1 + _store_registration_success(generation=owner_generation) + logger.info( + "Routes registered successfully (attempt=%s)", + attempt, ) - except Exception: - logger.exception("Error registering routes (attempt %s)", attempts + 1) + return - time.sleep(delay) - delay = min(delay * 1.5, 30) - attempts += 1 + _mark_host_waiting(attempt=attempt, max_attempts=max_attempts) + logger.debug("PromptServer not ready (attempt=%s)", attempt) + if attempt < max_attempts: + sleep_fn(delay) + delay = min(delay * 1.5, 30.0) - 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, - ) + failure = RuntimeError("route registration retry exhausted") + _mark_startup_fatal( + "host_wait", + failure, + reason_code="retry_exhausted", + ) + _store_registration_failure(failure, generation=owner_generation) + logger.error( + "Route registration retry exhausted (attempts=%s)", + 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 - if _routes_registered: - return + """Initialize and register routes through one process-wide bootstrap owner.""" + + 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() - _initialize_registries_and_security_gate() - except Exception as exc: - _mark_startup_fatal("required_startup", exc) - raise - - try: - ps_mod = sys.modules.get("server") - PromptServer = getattr(ps_mod, "PromptServer", None) if ps_mod else None - - 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." + _mark_required_initialization_started() + try: + _register_plugins_and_shutdown_hooks() + _initialize_registries_and_security_gate() + except Exception as exc: + _mark_startup_fatal("required_initialization", exc) + _store_registration_failure(exc, generation=generation) + logger.error( + "Required startup initialization failed (error_type=%s)", + type(exc).__name__, ) - else: - logging.getLogger("ComfyUI-OpenClaw").info( - "PromptServer not ready, starting background registration retry loop..." - ) - _start_registration_retry_loop() - except Exception: - _exc_type, exc, _tb = sys.exc_info() - if exc is not None: + raise + + try: + server = _resolve_prompt_server() + except Exception as exc: _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. + _store_registration_failure(exc, generation=generation) + logger.error( + "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() diff --git a/services/startup_lifecycle.py b/services/startup_lifecycle.py index b5a3644..f1c6064 100644 --- a/services/startup_lifecycle.py +++ b/services/startup_lifecycle.py @@ -1,214 +1,698 @@ -""" -Startup lifecycle diagnostics and optional warmup boundaries. +"""Typed startup lifecycle outcomes and redacted public diagnostics. -Required startup work still fails closed in callers. This module only tracks -readiness and runs optional warmups without delaying route availability. +Required startup work still fails closed in callers. This module owns only the +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" -STARTUP_READY = "ready" -STARTUP_DEGRADED_WARMUP = "degraded-warmup" -STARTUP_FATAL = "fatal-startup" +SCHEMA_VERSION = 1 +MAX_DIAGNOSTIC_MS = 86_400_000 +MAX_WARMUPS = 16 +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" -WARMUP_SUCCEEDED = "succeeded" -WARMUP_FAILED = "failed" -WARMUP_TIMED_OUT = "timed_out" + +class StartupPhase(str, Enum): + PACKAGE_IMPORT = "package_import" + REQUIRED_INITIALIZATION = "required_initialization" + 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() -_READY = False -_READY_PHASE: Optional[str] = None -_READY_AT: Optional[float] = None -_FATAL: Optional[Dict[str, Any]] = None -_WARMUPS: Dict[str, Dict[str, Any]] = {} + +class StartupTransitionError(RuntimeError): + """Stable transition failure that never embeds caller/source content.""" + + def __init__(self, code: str): + self.code = code + 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.""" - 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() + """Compatibility facade that reaches the required ready transition.""" + + _ = phase + outcome = _LIFECYCLE.snapshot() + if outcome.state is StartupState.STARTING: + _LIFECYCLE.mark_required_initialization_started() + _LIFECYCLE.mark_route_registration_started() + _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: - """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 mark_startup_fatal( + phase: str, + exc: BaseException | None = None, + *, + reason_code: StartupReason | str | None = None, +) -> None: + """Record a stable fatal classification without retaining ``exc``.""" + + _ = exc + 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 list(specs or []): + for name, fn, timeout_sec in tuple(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() + _LIFECYCLE.reset() 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, - } - + should_start, generation, safe_name = _LIFECYCLE.begin_warmup(name, timeout_sec) + if not should_start: + return monitor = threading.Thread( target=_warmup_monitor, - args=(name, fn, timeout_sec), - name=f"openclaw-warmup-monitor-{name}", + args=(safe_name, generation, fn, max(0.01, min(timeout_sec, 60.0))), + 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: - started_at = time.time() +def _warmup_monitor( + 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: - if name in _WARMUPS: - _WARMUPS[name]["state"] = WARMUP_RUNNING - _WARMUPS[name]["started_at"] = started_at - + _LIFECYCLE.mark_warmup_running(name, generation) worker = threading.Thread( target=_worker, name=f"openclaw-warmup-{name}", daemon=True, ) - worker.start() - - if not done.wait(timeout=timeout_sec): - _finish_warmup( + try: + worker.start() + except Exception as exc: + _LIFECYCLE.finish_warmup( name, - WARMUP_TIMED_OUT, - started_at, - error_type="TimeoutError", - error=f"optional warmup exceeded {timeout_sec:.2f}s", + generation, + state=WarmupState.FAILED, ) 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) - logger.info("R188: optional startup warmup completed: %s", name) + _LIFECYCLE.finish_warmup( + name, + generation, + state=WarmupState.SUCCEEDED, + ) + logger.info("Optional startup warmup completed (component=%s)", name) return - _finish_warmup( + _LIFECYCLE.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", + generation, + state=WarmupState.FAILED, + ) + 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, - } - ) diff --git a/tests/exception_boundary_policy.json b/tests/exception_boundary_policy.json index 16d1464..c4dca9a 100644 --- a/tests/exception_boundary_policy.json +++ b/tests/exception_boundary_policy.json @@ -126,19 +126,19 @@ "review_after": "2027-01-11" }, { - "scope": "_start_registration_retry_loop._retry_worker", - "expected_count": 1, + "scope": "_run_registration_retry_loop", + "expected_count": 2, "classification": "allowed_boundary_guard", - "reason": "Background route warmup retries after transient registration failures.", - "regression_owner": "tests/test_route_registration.py", + "reason": "The bounded retry owner classifies PromptServer resolution and route registration failures without exposing exception content.", + "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.", - "regression_owner": "tests/test_r180_exception_boundary_governance.py", + "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_bootstrap_lifecycle_outcome.py", "review_after": "2027-01-11" } ] diff --git a/tests/static_analysis_policy.json b/tests/static_analysis_policy.json index 78c4c15..3c0eb68 100644 --- a/tests/static_analysis_policy.json +++ b/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", diff --git a/tests/test_bootstrap_lifecycle_outcome.py b/tests/test_bootstrap_lifecycle_outcome.py new file mode 100644 index 0000000..c9d7a54 --- /dev/null +++ b/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() diff --git a/tests/test_r188_startup_lifecycle.py b/tests/test_r188_startup_lifecycle.py index 98c2a0a..1a51095 100644 --- a/tests/test_r188_startup_lifecycle.py +++ b/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.assertIn("RuntimeError", diagnostics["fatal"]["error_type"]) + self.assertTrue(diagnostics["fatal"]) + 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( - diagnostics["warmups"]["slow_provider"]["state"], {"running", "succeeded"} + warmup = next( + 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__":