refactor(services): package bootstrap and posture domains

docs/architecture/service_domain_packages.md

@@ -0,0 +1,27 @@
+# Service Domain Packages
+
+Bootstrap lifecycle, route registration, and effective security posture have explicit
+implementation owners:
+
+- `services/bootstrap/lifecycle.py` owns startup phase, outcome, and optional-warmup state.
+- `services/bootstrap/registration.py` owns host route registration and retry coordination.
+- `services/posture/effective.py` owns the immutable process security-posture snapshot.
+
+The historical modules remain compatibility aliases:
+
+- `services/startup_lifecycle.py`
+- `services/route_bootstrap.py`
+- `services/effective_security_posture.py`
+
+Each alias maps its module name to the implementation module object. This preserves one
+process singleton and keeps existing imports and patch points compatible. Do not replace
+these aliases with copied re-exports: copied module globals can diverge from the state used
+by implementation functions. Type-checker-only exports may describe the legacy interface,
+but they must stay behind `TYPE_CHECKING` and must not become a second runtime owner.
+
+New implementation code should import the domain-owned modules. Existing consumers may
+continue to use the compatibility paths. An implementation module must never import its
+compatibility alias; the repository dependency policy enforces that direction.
+
+Package initializers are navigation-only. They must not register routes, resolve posture,
+start threads, or re-export mutable process state during import.

scripts/verify_production_dependencies.py

@@ -29,6 +29,7 @@ _TOP_LEVEL_KEYS = {
     "domains",
     "allowed_dependencies",
     "compatibility_exceptions",
+    "facade_contracts",
     "accepted_cycles",
     "dynamic_imports",
 }
@@ -45,6 +46,13 @@ _EXCEPTION_KEYS = {
     "rationale",
     "review_condition",
 }
+_FACADE_KEYS = {
+    "facade",
+    "implementation",
+    "owner",
+    "rationale",
+    "review_condition",
+}
 _CYCLE_KEYS = {"modules", "owner", "rationale", "review_condition"}
 _DYNAMIC_KEYS = {
     "path",
@@ -123,6 +131,7 @@ class _PolicyContext:
     module_paths: dict[str, str]
     allowed_dependencies: dict[str, set[str]]
     compatibility_exceptions: set[tuple[str, str]]
+    facade_contracts: set[tuple[str, str]]
     accepted_cycles: set[frozenset[str]]
     dynamic_imports: dict[tuple[str, str, str, str, str], Mapping[str, Any]]
 
@@ -379,6 +388,32 @@ def _validate_policy(
         if edge[0] not in module_paths or edge[1] not in module_paths:
             findings.append(_finding("DEP_EXCEPTION_MODULE_UNKNOWN", subject=subject))
 
+    facade_contracts: set[tuple[str, str]] = set()
+    facade_entries = policy.get("facade_contracts", [])
+    if not isinstance(facade_entries, list):
+        findings.append(_finding("FACADES_INVALID"))
+        facade_entries = []
+    for index, entry in enumerate(facade_entries):
+        subject = f"facade_contracts[{index}]"
+        if not isinstance(entry, Mapping):
+            findings.append(_finding("FACADES_INVALID", subject=subject))
+            continue
+        for key in sorted(set(entry) - _FACADE_KEYS):
+            findings.append(_finding("POLICY_UNKNOWN_KEY", subject=f"{subject}.{key}"))
+        _validate_review_metadata(entry, path=subject, findings=findings)
+        edge = (
+            str(entry.get("facade", "")),
+            str(entry.get("implementation", "")),
+        )
+        if not edge[0] or not edge[1] or edge[0] == edge[1]:
+            findings.append(_finding("FACADES_INVALID", subject=subject))
+            continue
+        if edge in facade_contracts:
+            findings.append(_finding("FACADE_DUPLICATE", subject=subject))
+        facade_contracts.add(edge)
+        if edge[0] not in module_paths or edge[1] not in module_paths:
+            findings.append(_finding("FACADE_MODULE_UNKNOWN", subject=subject))
+
     accepted_cycles: set[frozenset[str]] = set()
     cycle_entries = policy.get("accepted_cycles")
     if not isinstance(cycle_entries, list):
@@ -457,6 +492,7 @@ def _validate_policy(
         module_paths=module_paths,
         allowed_dependencies=allowed_dependencies,
         compatibility_exceptions=compatibility_exceptions,
+        facade_contracts=facade_contracts,
         accepted_cycles=accepted_cycles,
         dynamic_imports=dynamic_imports,
     )
@@ -724,6 +760,26 @@ def analyze_repository(
                 )
             )
 
+    for facade, implementation in sorted(context.facade_contracts):
+        if (facade, implementation) not in edges:
+            path = context.module_paths.get(facade, ".")
+            findings.append(
+                _finding(
+                    "FACADE_STALE",
+                    path=path,
+                    subject=f"{facade}->{implementation}",
+                )
+            )
+        if (implementation, facade) in edges:
+            path = context.module_paths.get(implementation, ".")
+            findings.append(
+                _finding(
+                    "FACADE_REVERSE_DEPENDENCY",
+                    path=path,
+                    subject=f"{implementation}->{facade}",
+                )
+            )
+
     cycles = _strongly_connected_components(context.module_paths, edges)
     current_cycle_sets = {frozenset(cycle) for cycle in cycles}
     for cycle in cycles:

services/bootstrap/__init__.py

@@ -0,0 +1,4 @@
+# ruff: noqa: N999
+"""Bootstrap lifecycle and registration implementation package."""
+
+__all__ = ["lifecycle", "registration"]

services/bootstrap/lifecycle.py

@@ -0,0 +1,698 @@
+"""Typed startup lifecycle outcomes and redacted public diagnostics.
+
+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 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")
+
+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",
+)
+
+
+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]
+
+
+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:
+    """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 = 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:
+    for name, fn, timeout_sec in tuple(specs or ()):
+        _start_optional_warmup(str(name), fn, float(timeout_sec))
+
+
+def reset_startup_lifecycle_for_tests() -> None:
+    _LIFECYCLE.reset()
+
+
+def _start_optional_warmup(
+    name: str, fn: Callable[[], Any], timeout_sec: float
+) -> None:
+    should_start, generation, safe_name = _LIFECYCLE.begin_warmup(name, timeout_sec)
+    if not should_start:
+        return
+    monitor = threading.Thread(
+        target=_warmup_monitor,
+        args=(safe_name, generation, fn, max(0.01, min(timeout_sec, 60.0))),
+        name=f"openclaw-warmup-monitor-{safe_name}",
+        daemon=True,
+    )
+    try:
+        monitor.start()
+    except Exception:
+        _LIFECYCLE.finish_warmup(
+            safe_name,
+            generation,
+            state=WarmupState.FAILED,
+        )
+        raise
+
+
+def _warmup_monitor(
+    name: str,
+    generation: int,
+    fn: Callable[[], Any],
+    timeout_sec: float,
+) -> None:
+    done = threading.Event()
+    result: dict[str, bool] = {}
+
+    def _worker() -> None:
+        try:
+            fn()
+            result["ok"] = True
+        except Exception:
+            # SECURITY: never retain or log arbitrary exception content.
+            result["ok"] = False
+        finally:
+            done.set()
+
+    _LIFECYCLE.mark_warmup_running(name, generation)
+    worker = threading.Thread(
+        target=_worker,
+        name=f"openclaw-warmup-{name}",
+        daemon=True,
+    )
+    try:
+        worker.start()
+    except Exception as exc:
+        _LIFECYCLE.finish_warmup(
+            name,
+            generation,
+            state=WarmupState.FAILED,
+        )
+        logger.warning(
+            "Optional startup warmup worker could not start "
+            "(component=%s, error_type=%s)",
+            name,
+            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
+
+    if result.get("ok"):
+        _LIFECYCLE.finish_warmup(
+            name,
+            generation,
+            state=WarmupState.SUCCEEDED,
+        )
+        logger.info("Optional startup warmup completed (component=%s)", name)
+        return
+
+    _LIFECYCLE.finish_warmup(
+        name,
+        generation,
+        state=WarmupState.FAILED,
+    )
+    logger.warning(
+        "Optional startup warmup failed (component=%s, reason_code=%s)",
+        name,
+        StartupReason.WARMUP_FAILED.value,
+    )

services/bootstrap/registration.py

@@ -0,0 +1,621 @@
+"""
+Route/bootstrap orchestration implementation owner.
+
+Keeps __init__.py thin while preserving startup behavior and fallback handling.
+"""
+
+from __future__ import annotations
+
+import logging
+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:
+    raw = (
+        os.environ.get("OPENCLAW_STARTUP_WARMUP_TIMEOUT_SEC")
+        or os.environ.get("MOLTBOT_STARTUP_WARMUP_TIMEOUT_SEC")
+        or "5"
+    )
+    try:
+        return max(0.1, min(float(raw), 60.0))
+    except (TypeError, ValueError):
+        return 5.0
+
+
+def _warm_model_inventory_snapshot() -> None:
+    from ..preflight import get_model_inventory_snapshot
+
+    get_model_inventory_snapshot(trigger_refresh=True)
+
+
+def _build_optional_startup_warmups():
+    timeout_sec = _resolve_optional_warmup_timeout_sec()
+    return [
+        ("model_inventory", _warm_model_inventory_snapshot, timeout_sec),
+    ]
+
+
+def _mark_startup_ready_and_start_warmups() -> None:
+    from .lifecycle import mark_startup_ready, start_optional_warmups
+
+    # 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 as exc:
+        # IMPORTANT: optional warmup diagnostics must not undo successful route startup.
+        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,
+    *,
+    reason_code=None,
+) -> None:
+    try:
+        from .lifecycle import mark_startup_fatal
+
+        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").error(
+            "Startup diagnostics update failed (error_type=%s)",
+            type(diagnostics_exc).__name__,
+        )
+
+
+def _mark_required_initialization_started() -> None:
+    from .lifecycle import mark_required_initialization_started
+
+    mark_required_initialization_started()
+
+
+def _mark_host_waiting(*, attempt: int, max_attempts: int) -> None:
+    from .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 .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."""
+
+    from ..plugins.builtin import register_all
+    from ..runtime_lifecycle import register_shutdown_hooks
+
+    return register_shutdown_hooks, register_all
+
+
+def _register_plugins_and_shutdown_hooks() -> None:
+    # R67: Best-effort process shutdown hook and built-in plugin registration.
+    try:
+        register_shutdown_hooks, register_all = _load_plugin_shutdown_registrars()
+    except ImportError as exc:
+        logging.getLogger("ComfyUI-OpenClaw").error(
+            "Optional startup registrars unavailable (error_type=%s)",
+            type(exc).__name__,
+        )
+        return
+
+    logger = logging.getLogger("ComfyUI-OpenClaw")
+    for component, registrar in (
+        ("shutdown_hooks", register_shutdown_hooks),
+        ("builtin_plugins", register_all),
+    ):
+        try:
+            registrar()
+        except Exception as exc:
+            # IMPORTANT: these optional steps are independent. Keep startup available,
+            # do not echo exception content, and do not catch BaseException cancellation.
+            logger.error(
+                "Optional startup registrar failed (component=%s, error_type=%s)",
+                component,
+                type(exc).__name__,
+            )
+
+
+def _initialize_registries_and_security_gate() -> None:
+    # R63/R84: Initialize Service & Module Registries.
+    try:
+        from ..modules import ModuleCapability, ModuleRegistry, enable_module
+        from ..registry import SVC_RUNTIME_CONFIG, ServiceRegistry
+        from ..runtime_config import get_config
+
+        config = get_config()
+        ServiceRegistry.register(SVC_RUNTIME_CONFIG, config)
+
+        from ..posture.effective import (
+            get_effective_security_posture,
+            resolve_effective_security_posture,
+        )
+
+        posture = get_effective_security_posture(required=False)
+        if posture is None:
+            # Direct compatibility/test invocation does not own process installation.
+            posture = resolve_effective_security_posture()
+
+        # Always-on modules
+        enable_module(ModuleCapability.CORE)
+        enable_module(ModuleCapability.SECURITY)
+        enable_module(ModuleCapability.OBSERVABILITY)
+
+        # S50: initialize durable idempotency storage early.
+        from ..idempotency_store import IdempotencyStore
+        from ..state_dir import get_state_dir
+
+        db_path = os.path.join(get_state_dir(), "idempotency.db")
+        # 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 (strict_mode=True)"
+        )
+
+        if config.bridge_enabled:
+            enable_module(ModuleCapability.BRIDGE)
+
+        # Core runtime modules stay enabled; runners decide active behavior.
+        enable_module(ModuleCapability.SCHEDULER)
+        enable_module(ModuleCapability.WEBHOOK)
+        enable_module(ModuleCapability.CONNECTOR)
+
+        ModuleRegistry.lock()
+        logging.getLogger("ComfyUI-OpenClaw").info(
+            "Initialized modules: %s", ModuleRegistry.get_enabled_list()
+        )
+
+        from ..security_gate import enforce_startup_gate
+
+        enforce_startup_gate(posture=posture)
+    except Exception as exc:
+        logging.getLogger("ComfyUI-OpenClaw").error(
+            "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.
+        raise
+
+
+def _do_full_registration(server) -> None:
+    """Register all OpenClaw routes including bridge/scheduler bindings."""
+    from ..access_control import require_admin_token
+    from ..parameter_lab_queue_receipt import (
+        register_parameter_lab_queue_receipt_handler,
+    )
+    from ..plugins.async_bridge import run_async_in_sync_context
+    from ..queue_submit import submit_prompt
+    from ..route_bootstrap_contract import load_route_bootstrap_contract
+    from ..scheduler.runner import get_scheduler_runner, start_scheduler
+    from ..templates import get_template_service
+
+    contract = load_route_bootstrap_contract(__package__)
+    register_approval_routes = contract["register_approval_routes"]
+    BridgeHandlers = contract["BridgeHandlers"]
+    register_preset_routes = contract["register_preset_routes"]
+    register_routes = contract["register_routes"]
+    register_schedule_routes = contract["register_schedule_routes"]
+    register_trigger_routes = contract["register_trigger_routes"]
+
+    register_routes(server)
+    # CRITICAL: receipt promotion is required for exact Parameter Lab run ownership.
+    register_parameter_lab_queue_receipt_handler(server)
+    register_preset_routes(server.app)
+    register_schedule_routes(server.app, require_admin_token_fn=require_admin_token)
+
+    class QueueSubmitService:
+        def submit(self, job_req):
+            tmpl_svc = get_template_service()
+            workflow = tmpl_svc.render_template(job_req.template_id, job_req.inputs)
+
+            async def _do_submit():
+                return await submit_prompt(
+                    workflow,
+                    client_id=job_req.session_id or "bridge",
+                    extra_data={
+                        "openclaw": {"trace_id": job_req.trace_id},
+                        # Legacy key kept for existing tooling that expects this blob.
+                        "moltbot": {"trace_id": job_req.trace_id},
+                    },
+                    source="bridge",
+                    trace_id=job_req.trace_id,
+                )
+
+            return run_async_in_sync_context(_do_submit())
+
+    bridge_handlers = BridgeHandlers(submit_service=QueueSubmitService())
+    _register_bridge_routes(server.app.router, bridge_handlers)
+
+    async def unified_submit_fn(
+        template_id,
+        inputs,
+        trace_id,
+        idempotency_key,
+        delivery=None,
+        source="unknown",
+    ):
+        """Submit function for scheduler and trigger-triggered runs."""
+        # NOTE: Use IdempotencyStore API (check_and_record/update_prompt_id).
+        # Avoid legacy get_store/get/set usage; wrong API here breaks route registration at runtime.
+        from ..idempotency_store import IdempotencyStore
+        from ..queue_submit import submit_prompt as _submit_prompt
+        from ..templates import get_template_service as _get_template_service
+
+        store = IdempotencyStore()
+        is_dup, existing_prompt_id = store.check_and_record(idempotency_key)
+        if is_dup:
+            return {"prompt_id": existing_prompt_id, "deduped": True}
+
+        tmpl_svc = _get_template_service()
+        workflow = tmpl_svc.render_template(template_id, inputs)
+
+        result = await _submit_prompt(
+            workflow,
+            extra_data={
+                "openclaw": {"trace_id": trace_id, "source": "automation"},
+                "moltbot": {"trace_id": trace_id, "source": "automation"},
+            },
+            source=source,
+            trace_id=trace_id,
+        )
+
+        if result.get("prompt_id"):
+            store.update_prompt_id(idempotency_key, result["prompt_id"])
+        return result
+
+    runner = get_scheduler_runner()
+    runner._submit_fn = unified_submit_fn
+    start_scheduler()
+
+    register_trigger_routes(
+        server.app,
+        require_admin_token_fn=require_admin_token,
+        submit_fn=unified_submit_fn,
+    )
+    register_approval_routes(
+        server.app,
+        require_admin_token_fn=require_admin_token,
+        submit_fn=unified_submit_fn,
+    )
+
+
+_BRIDGE_ROUTE_SPECS = (
+    ("add_post", "/moltbot/bridge/submit", "submit_handler"),
+    ("add_post", "/moltbot/bridge/deliver", "deliver_handler"),
+    ("add_get", "/moltbot/bridge/health", "health_handler"),
+    ("add_post", "/openclaw/bridge/submit", "submit_handler"),
+    ("add_post", "/openclaw/bridge/deliver", "deliver_handler"),
+    ("add_get", "/openclaw/bridge/health", "health_handler"),
+    ("add_post", "/api/moltbot/bridge/submit", "submit_handler"),
+    ("add_post", "/api/moltbot/bridge/deliver", "deliver_handler"),
+    ("add_get", "/api/moltbot/bridge/health", "health_handler"),
+    ("add_post", "/api/openclaw/bridge/submit", "submit_handler"),
+    ("add_post", "/api/openclaw/bridge/deliver", "deliver_handler"),
+    ("add_get", "/api/openclaw/bridge/health", "health_handler"),
+)
+
+
+def _register_bridge_routes(router, bridge_handlers) -> None:
+    # IMPORTANT: keep bridge route registration table-driven.
+    # Missing one alias path here silently breaks one control-plane surface while
+    # leaving the rest apparently healthy, which is hard to diagnose during startup.
+    for method_name, path, handler_name in _BRIDGE_ROUTE_SPECS:
+        registrar = getattr(router, method_name, None)
+        if registrar is None:
+            continue
+        try:
+            registrar(path, getattr(bridge_handlers, handler_name))
+        except RuntimeError:
+            if path.startswith("/api/"):
+                continue
+            raise
+
+
+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 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()
+    try:
+        from ..posture.effective import reset_effective_security_posture_for_tests
+
+        reset_effective_security_posture_for_tests()
+    except ImportError:
+        # Dependency-light test/import mode may omit the posture module.
+        pass
+
+
+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:
+                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
+                _store_registration_success(generation=owner_generation)
+                logger.info(
+                    "Routes registered successfully (attempt=%s)",
+                    attempt,
+                )
+                return
+
+            _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)
+
+        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()
+
+
+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:
+    """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:
+        from ..posture.effective import get_or_create_effective_security_posture
+
+        # CRITICAL: this required startup owner installs process-static posture once.
+        # Direct helper/API invocations resolve ephemeral snapshots instead.
+        get_or_create_effective_security_posture()
+        _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__,
+            )
+            raise
+
+        try:
+            server = _resolve_prompt_server()
+        except Exception as exc:
+            _mark_startup_fatal("route_registration", exc)
+            _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()

services/effective_security_posture.py

@@ -1,511 +1,34 @@
-"""Immutable process-static security posture contract (R232)."""
+"""Compatibility alias for the effective security posture implementation module."""
 
 from __future__ import annotations
 
-import os
 import sys
-import threading
-from collections.abc import Mapping
-from dataclasses import dataclass
-from typing import Any
+from typing import TYPE_CHECKING
 
-SCHEMA_VERSION = 1
-_TRUTHY = frozenset({"1", "true", "yes", "on"})
-_FALSY = frozenset({"0", "false", "no", "off"})
-_CONTROL_PLANE_TRUTHY = frozenset({"1", "true", "yes"})
-_VALID_DEPLOYMENT_PROFILES = frozenset({"local", "lan", "public"})
-_VALID_WEBHOOK_MODES = frozenset({"bearer", "hmac", "bearer_or_hmac"})
-_installed_posture: EffectiveSecurityPosture | None = None
-_posture_lock = threading.RLock()
+from .posture import effective as _implementation
 
-
-@dataclass(frozen=True, slots=True, kw_only=True)
-class PostureFinding:
-    severity: str
-    code: str
-    message: str
-    remediation: str = ""
-
-
-@dataclass(frozen=True, slots=True, kw_only=True)
-class EffectiveSecurityPosture:
-    schema_version: int
-    runtime_profile: str
-    deployment_profile: str
-    mae_profile: str
-    network_exposed: bool
-    admin_token_configured: bool
-    observability_token_configured: bool
-    dangerous_profile_override: bool
-    dangerous_bind_override: bool
-    localhost_no_origin_override: bool
-    allow_any_public_llm_host: bool
-    allow_insecure_base_url: bool
-    webhook_auth_mode: str
-    webhook_bearer_configured: bool
-    webhook_hmac_configured: bool
-    webhook_replay_protection_required: bool
-    remote_admin_enabled: bool
-    trust_x_forwarded_for: bool
-    trusted_proxies_configured: bool
-    callback_allow_hosts_configured: bool
-    external_tools_enabled: bool
-    registry_sync_enabled: bool
-    transforms_enabled: bool
-    bridge_enabled: bool
-    bridge_device_token_configured: bool
-    bridge_mtls_enabled: bool
-    bridge_device_cert_map_configured: bool
-    bridge_allowed_device_ids_configured: bool
-    public_shared_surface_acknowledged: bool
-    control_plane_mode: str
-    control_plane_url_configured: bool
-    control_plane_token_configured: bool
-    control_plane_prerequisites_satisfied: bool
-    control_plane_compat_override: bool
-    connector_active_platforms: tuple[str, ...]
-    connector_unguarded_platforms: tuple[str, ...]
-    connector_recommended_allowlist_vars: tuple[str, ...]
-    deployment_checks: tuple[PostureFinding, ...]
-    deployment_pass_codes: tuple[str, ...]
-    deployment_warn_codes: tuple[str, ...]
-    deployment_fail_codes: tuple[str, ...]
-    startup_profile_passed: bool
-    startup_profile_overridden: bool
-    startup_profile_violation_codes: tuple[str, ...]
-    blocked_surface_ids: tuple[str, ...]
-    decision_codes: tuple[str, ...]
-    reason_codes: tuple[str, ...]
-
-
-def _read(
-    environ: Mapping[str, str],
-    primary: str,
-    legacy: str | None = None,
-    default: str = "",
-) -> str:
-    try:
-        if primary in environ:
-            value = environ.get(primary, default)
-        elif legacy and legacy in environ:
-            value = environ.get(legacy, default)
-        else:
-            value = default
-    except Exception:
-        # CRITICAL: malformed environment providers must fail closed without echoing
-        # exception content or the attempted value into diagnostics.
-        raise ValueError("security posture input unavailable") from None
-    if value is None:
-        return ""
-    try:
-        return str(value)
-    except Exception:
-        raise ValueError("security posture input is not scalar") from None
-
-
-def _normalized(
-    environ: Mapping[str, str],
-    primary: str,
-    legacy: str | None = None,
-    default: str = "",
-) -> str:
-    return _read(environ, primary, legacy, default).strip().lower()
-
-
-def _enabled(
-    environ: Mapping[str, str],
-    primary: str,
-    legacy: str | None = None,
-) -> bool:
-    return _normalized(environ, primary, legacy) in _TRUTHY
-
-
-def _configured(
-    environ: Mapping[str, str],
-    primary: str,
-    legacy: str | None = None,
-) -> bool:
-    return bool(_read(environ, primary, legacy).strip())
-
-
-def _network_exposed_from_argv() -> bool:
-    # Preserve the accepted S41 heuristic exactly: only the explicit --listen flag
-    # changes this process-static decision.
-    return "--listen" in sys.argv
-
-
-def _deployment_report(profile: str, environ: Mapping[str, str]):
-    try:
-        from .deployment_profile import evaluate_deployment_profile
-    except ImportError:  # pragma: no cover - top-level compatibility mode
-        from services.deployment_profile import evaluate_deployment_profile
-
-    return evaluate_deployment_profile(profile, environ)
-
-
-def _connector_posture(environ: Mapping[str, str]) -> Mapping[str, Any]:
-    try:
-        from .connector_allowlist_posture import evaluate_connector_allowlist_posture
-    except ImportError:  # pragma: no cover - top-level compatibility mode
-        from services.connector_allowlist_posture import (
-            evaluate_connector_allowlist_posture,
-        )
-
-    return evaluate_connector_allowlist_posture(environ)
-
-
-def _blocked_surface_ids(profile: str, mode: str) -> tuple[str, ...]:
-    if profile != "public" or mode != "split":
-        return ()
-    # IMPORTANT: these are the stable scalar IDs from the S62 registry. Importing
-    # control_plane here would create a dependency cycle before R233 packages the domain.
-    return (
-        "callback_egress",
-        "registry_sync",
-        "secrets_write",
-        "tool_execution",
-        "transforms_exec",
-        "webhook_execute",
+if TYPE_CHECKING:
+    # Static-only exports keep legacy imports typed without duplicating installed state.
+    from .posture.effective import EffectiveSecurityPosture as EffectiveSecurityPosture
+    from .posture.effective import PostureFinding as PostureFinding
+    from .posture.effective import (
+        effective_security_posture_diagnostics as effective_security_posture_diagnostics,
+    )
+    from .posture.effective import (
+        get_effective_security_posture as get_effective_security_posture,
+    )
+    from .posture.effective import (
+        get_or_create_effective_security_posture as get_or_create_effective_security_posture,
+    )
+    from .posture.effective import (
+        install_effective_security_posture as install_effective_security_posture,
+    )
+    from .posture.effective import (
+        reset_effective_security_posture_for_tests as reset_effective_security_posture_for_tests,
+    )
+    from .posture.effective import (
+        resolve_effective_security_posture as resolve_effective_security_posture,
     )
 
-
-def _safe_finding(check: Any) -> PostureFinding:
-    message = str(check.message)
-    if str(check.code) == "DP-WEBHOOK-005":
-        # IMPORTANT: the legacy evaluator includes the raw invalid environment value.
-        # The immutable boundary retains the stable code but never the untrusted value.
-        message = "Unsupported webhook auth mode."
-    return PostureFinding(
-        severity=str(check.severity),
-        code=str(check.code),
-        message=message,
-        remediation=str(check.remediation),
-    )
-
-
-def resolve_effective_security_posture(
-    environ: Mapping[str, str] | None = None,
-    *,
-    network_exposed: bool | None = None,
-) -> EffectiveSecurityPosture:
-    # IMPORTANT: an explicitly supplied empty mapping means empty input. Do not use
-    # `environ or os.environ`; doing so makes tests and lifecycle injection ambient.
-    env = os.environ if environ is None else environ
-
-    resolved_network_exposed = (
-        _network_exposed_from_argv()
-        if network_exposed is None
-        else bool(network_exposed)
-    )
-    deployment_profile = _normalized(
-        env, "OPENCLAW_DEPLOYMENT_PROFILE", default="local"
-    )
-    if deployment_profile not in _VALID_DEPLOYMENT_PROFILES:
-        raise ValueError("unsupported deployment profile")
-
-    raw_runtime_profile = _normalized(
-        env, "OPENCLAW_RUNTIME_PROFILE", default="minimal"
-    )
-    runtime_profile = "hardened" if raw_runtime_profile == "hardened" else "minimal"
-    mae_profile = (
-        "hardened"
-        if runtime_profile == "hardened" and deployment_profile != "public"
-        else deployment_profile
-    )
-
-    try:
-        report = _deployment_report(deployment_profile, env)
-    except Exception:
-        # CRITICAL: delegated evaluators must not expose hostile mapping values or
-        # exception text across the immutable posture boundary.
-        raise ValueError("security posture evaluation failed") from None
-    findings = tuple(_safe_finding(check) for check in report.checks)
-    pass_codes = tuple(item.code for item in findings if item.severity == "pass")
-    warn_codes = tuple(item.code for item in findings if item.severity == "warn")
-    fail_codes = tuple(item.code for item in findings if item.severity == "fail")
-
-    dangerous_profile_override = _enabled(
-        env, "OPENCLAW_SECURITY_DANGEROUS_PROFILE_OVERRIDE"
-    )
-    startup_violations = () if deployment_profile == "local" else fail_codes
-    startup_overridden = bool(startup_violations and dangerous_profile_override)
-    startup_passed = (
-        deployment_profile == "local" or not startup_violations or startup_overridden
-    )
-
-    explicit_control_mode = _normalized(env, "OPENCLAW_CONTROL_PLANE_MODE")
-    if explicit_control_mode in {"embedded", "split"}:
-        control_plane_mode = explicit_control_mode
-    elif deployment_profile == "public":
-        control_plane_mode = "split"
-    else:
-        control_plane_mode = "embedded"
-
-    control_plane_url_configured = _configured(env, "OPENCLAW_CONTROL_PLANE_URL")
-    control_plane_token_configured = _configured(env, "OPENCLAW_CONTROL_PLANE_TOKEN")
-    control_plane_prerequisites_satisfied = (
-        control_plane_url_configured and control_plane_token_configured
-    )
-    control_plane_compat_override = (
-        _normalized(env, "OPENCLAW_SPLIT_COMPAT_OVERRIDE") in _CONTROL_PLANE_TRUTHY
-    )
-
-    try:
-        connector = _connector_posture(env)
-    except Exception:
-        raise ValueError("security posture evaluation failed") from None
-    active_platforms = tuple(
-        sorted({str(item) for item in connector["active_platforms"]})
-    )
-    unguarded_platforms = tuple(
-        sorted({str(item) for item in connector["unguarded_platforms"]})
-    )
-    recommended_allowlist_vars = tuple(
-        sorted({str(item) for item in connector["recommended_allowlist_vars"]})
-    )
-
-    reason_codes = list(startup_violations)
-    if deployment_profile == "public" and control_plane_mode == "split":
-        if not control_plane_url_configured:
-            reason_codes.append("CP-URL-MISSING")
-        if not control_plane_token_configured:
-            reason_codes.append("CP-TOKEN-MISSING")
-    elif deployment_profile == "public" and control_plane_mode == "embedded":
-        if not control_plane_compat_override:
-            reason_codes.append("CP-PUBLIC-EMBEDDED")
-    reason_codes.extend(
-        f"CONNECTOR-ALLOWLIST-{platform.upper()}" for platform in unguarded_platforms
-    )
-    if raw_runtime_profile not in {"", "minimal", "hardened"}:
-        reason_codes.append("RUNTIME-PROFILE-DEFAULTED")
-
-    decision_codes = [
-        (
-            "STARTUP-OVERRIDDEN"
-            if startup_overridden
-            else "STARTUP-PASS" if startup_passed else "STARTUP-DENY"
-        ),
-        (
-            "CONTROL-PLANE-PASS"
-            if (
-                deployment_profile != "public"
-                or (
-                    control_plane_mode == "split"
-                    and control_plane_prerequisites_satisfied
-                )
-                or (control_plane_mode == "embedded" and control_plane_compat_override)
-            )
-            else "CONTROL-PLANE-DENY"
-        ),
-        (
-            "CONNECTORS-NONE"
-            if not active_platforms
-            else "CONNECTORS-UNGUARDED" if unguarded_platforms else "CONNECTORS-GUARDED"
-        ),
-        "NETWORK-EXPOSED" if resolved_network_exposed else "NETWORK-LOOPBACK",
-    ]
-
-    raw_webhook_mode = _normalized(
-        env,
-        "OPENCLAW_WEBHOOK_AUTH_MODE",
-        "MOLTBOT_WEBHOOK_AUTH_MODE",
-    )
-    webhook_auth_mode = (
-        raw_webhook_mode
-        if raw_webhook_mode in _VALID_WEBHOOK_MODES
-        else "unset" if not raw_webhook_mode else "invalid"
-    )
-    replay_value = _normalized(
-        env,
-        "OPENCLAW_WEBHOOK_REQUIRE_REPLAY_PROTECTION",
-        "MOLTBOT_WEBHOOK_REQUIRE_REPLAY_PROTECTION",
-    )
-
-    return EffectiveSecurityPosture(
-        schema_version=SCHEMA_VERSION,
-        runtime_profile=runtime_profile,
-        deployment_profile=deployment_profile,
-        mae_profile=mae_profile,
-        network_exposed=resolved_network_exposed,
-        admin_token_configured=_configured(
-            env, "OPENCLAW_ADMIN_TOKEN", "MOLTBOT_ADMIN_TOKEN"
-        ),
-        observability_token_configured=_configured(
-            env, "OPENCLAW_OBSERVABILITY_TOKEN", "MOLTBOT_OBSERVABILITY_TOKEN"
-        ),
-        dangerous_profile_override=dangerous_profile_override,
-        dangerous_bind_override=_enabled(
-            env,
-            "OPENCLAW_SECURITY_DANGEROUS_BIND_OVERRIDE",
-            "MOLTBOT_SECURITY_DANGEROUS_BIND_OVERRIDE",
-        ),
-        localhost_no_origin_override=(
-            _normalized(env, "OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN") == "true"
-        ),
-        allow_any_public_llm_host=_enabled(
-            env,
-            "OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST",
-            "MOLTBOT_ALLOW_ANY_PUBLIC_LLM_HOST",
-        ),
-        allow_insecure_base_url=_enabled(
-            env,
-            "OPENCLAW_ALLOW_INSECURE_BASE_URL",
-            "MOLTBOT_ALLOW_INSECURE_BASE_URL",
-        ),
-        webhook_auth_mode=webhook_auth_mode,
-        webhook_bearer_configured=_configured(
-            env,
-            "OPENCLAW_WEBHOOK_BEARER_TOKEN",
-            "MOLTBOT_WEBHOOK_BEARER_TOKEN",
-        ),
-        webhook_hmac_configured=_configured(
-            env,
-            "OPENCLAW_WEBHOOK_HMAC_SECRET",
-            "MOLTBOT_WEBHOOK_HMAC_SECRET",
-        ),
-        webhook_replay_protection_required=replay_value not in _FALSY,
-        remote_admin_enabled=_enabled(
-            env, "OPENCLAW_ALLOW_REMOTE_ADMIN", "MOLTBOT_ALLOW_REMOTE_ADMIN"
-        ),
-        trust_x_forwarded_for=_enabled(
-            env,
-            "OPENCLAW_TRUST_X_FORWARDED_FOR",
-            "MOLTBOT_TRUST_X_FORWARDED_FOR",
-        ),
-        trusted_proxies_configured=_configured(
-            env, "OPENCLAW_TRUSTED_PROXIES", "MOLTBOT_TRUSTED_PROXIES"
-        ),
-        callback_allow_hosts_configured=_configured(
-            env,
-            "OPENCLAW_CALLBACK_ALLOW_HOSTS",
-            "MOLTBOT_CALLBACK_ALLOW_HOSTS",
-        ),
-        external_tools_enabled=_enabled(env, "OPENCLAW_ENABLE_EXTERNAL_TOOLS"),
-        registry_sync_enabled=_enabled(env, "OPENCLAW_ENABLE_REGISTRY_SYNC"),
-        transforms_enabled=_enabled(env, "OPENCLAW_ENABLE_TRANSFORMS"),
-        bridge_enabled=_enabled(
-            env, "OPENCLAW_BRIDGE_ENABLED", "MOLTBOT_BRIDGE_ENABLED"
-        ),
-        bridge_device_token_configured=_configured(
-            env,
-            "OPENCLAW_BRIDGE_DEVICE_TOKEN",
-            "MOLTBOT_BRIDGE_DEVICE_TOKEN",
-        ),
-        bridge_mtls_enabled=_enabled(env, "OPENCLAW_BRIDGE_MTLS_ENABLED"),
-        bridge_device_cert_map_configured=_configured(
-            env, "OPENCLAW_BRIDGE_DEVICE_CERT_MAP"
-        ),
-        bridge_allowed_device_ids_configured=_configured(
-            env,
-            "OPENCLAW_BRIDGE_ALLOWED_DEVICE_IDS",
-            "MOLTBOT_BRIDGE_ALLOWED_DEVICE_IDS",
-        ),
-        public_shared_surface_acknowledged=_enabled(
-            env,
-            "OPENCLAW_PUBLIC_SHARED_SURFACE_BOUNDARY_ACK",
-            "MOLTBOT_PUBLIC_SHARED_SURFACE_BOUNDARY_ACK",
-        ),
-        control_plane_mode=control_plane_mode,
-        control_plane_url_configured=control_plane_url_configured,
-        control_plane_token_configured=control_plane_token_configured,
-        control_plane_prerequisites_satisfied=control_plane_prerequisites_satisfied,
-        control_plane_compat_override=control_plane_compat_override,
-        connector_active_platforms=active_platforms,
-        connector_unguarded_platforms=unguarded_platforms,
-        connector_recommended_allowlist_vars=recommended_allowlist_vars,
-        deployment_checks=findings,
-        deployment_pass_codes=pass_codes,
-        deployment_warn_codes=warn_codes,
-        deployment_fail_codes=fail_codes,
-        startup_profile_passed=startup_passed,
-        startup_profile_overridden=startup_overridden,
-        startup_profile_violation_codes=startup_violations,
-        blocked_surface_ids=_blocked_surface_ids(
-            deployment_profile, control_plane_mode
-        ),
-        decision_codes=tuple(decision_codes),
-        reason_codes=tuple(dict.fromkeys(reason_codes)),
-    )
-
-
-def install_effective_security_posture(
-    posture: EffectiveSecurityPosture,
-) -> EffectiveSecurityPosture:
-    if not isinstance(posture, EffectiveSecurityPosture):
-        raise TypeError("posture must be EffectiveSecurityPosture")
-    global _installed_posture
-    with _posture_lock:
-        if _installed_posture is None:
-            _installed_posture = posture
-        elif _installed_posture is not posture:
-            # CRITICAL: silently replacing process posture creates contradictory
-            # authorization decisions. Reset is an explicit lifecycle/test operation.
-            raise RuntimeError("effective security posture is already installed")
-        return _installed_posture
-
-
-def get_effective_security_posture(
-    *, required: bool = True
-) -> EffectiveSecurityPosture | None:
-    with _posture_lock:
-        posture = _installed_posture
-    if posture is None and required:
-        raise RuntimeError("effective security posture is not installed")
-    return posture
-
-
-def get_or_create_effective_security_posture(
-    environ: Mapping[str, str] | None = None,
-    *,
-    network_exposed: bool | None = None,
-) -> EffectiveSecurityPosture:
-    with _posture_lock:
-        if _installed_posture is not None:
-            return _installed_posture
-        posture = resolve_effective_security_posture(
-            environ,
-            network_exposed=network_exposed,
-        )
-        # The RLock makes this identity-stable even under concurrent startup.
-        return install_effective_security_posture(posture)
-
-
-def reset_effective_security_posture_for_tests() -> None:
-    global _installed_posture
-    with _posture_lock:
-        _installed_posture = None
-
-
-def effective_security_posture_diagnostics(
-    posture: EffectiveSecurityPosture | None = None,
-) -> dict[str, Any]:
-    resolved = posture or get_effective_security_posture()
-    assert resolved is not None
-    return {
-        "schema_version": resolved.schema_version,
-        "runtime_profile": resolved.runtime_profile,
-        "deployment_profile": resolved.deployment_profile,
-        "mae_profile": resolved.mae_profile,
-        "network_exposed": resolved.network_exposed,
-        "authentication": {
-            "admin_configured": resolved.admin_token_configured,
-            "observability_configured": resolved.observability_token_configured,
-        },
-        "startup_gate": {
-            "passed": resolved.startup_profile_passed,
-            "overridden": resolved.startup_profile_overridden,
-            "violation_codes": list(resolved.startup_profile_violation_codes),
-        },
-        "control_plane": {
-            "mode": resolved.control_plane_mode,
-            "prerequisites_satisfied": (resolved.control_plane_prerequisites_satisfied),
-            "compat_override": resolved.control_plane_compat_override,
-            "blocked_surface_count": len(resolved.blocked_surface_ids),
-        },
-        "connectors": {
-            "active_count": len(resolved.connector_active_platforms),
-            "unguarded_count": len(resolved.connector_unguarded_platforms),
-        },
-        "decision_codes": list(resolved.decision_codes),
-        "reason_codes": list(resolved.reason_codes),
-    }
+# IMPORTANT: alias the module object; copied re-exports split installed posture state.
+sys.modules[__name__] = _implementation

services/posture/__init__.py

@@ -0,0 +1,4 @@
+# ruff: noqa: N999
+"""Process-static security posture implementation package."""
+
+__all__ = ["effective"]

services/posture/effective.py

@@ -0,0 +1,511 @@
+"""Immutable process-static security posture implementation."""
+
+from __future__ import annotations
+
+import os
+import sys
+import threading
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+SCHEMA_VERSION = 1
+_TRUTHY = frozenset({"1", "true", "yes", "on"})
+_FALSY = frozenset({"0", "false", "no", "off"})
+_CONTROL_PLANE_TRUTHY = frozenset({"1", "true", "yes"})
+_VALID_DEPLOYMENT_PROFILES = frozenset({"local", "lan", "public"})
+_VALID_WEBHOOK_MODES = frozenset({"bearer", "hmac", "bearer_or_hmac"})
+_installed_posture: EffectiveSecurityPosture | None = None
+_posture_lock = threading.RLock()
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class PostureFinding:
+    severity: str
+    code: str
+    message: str
+    remediation: str = ""
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class EffectiveSecurityPosture:
+    schema_version: int
+    runtime_profile: str
+    deployment_profile: str
+    mae_profile: str
+    network_exposed: bool
+    admin_token_configured: bool
+    observability_token_configured: bool
+    dangerous_profile_override: bool
+    dangerous_bind_override: bool
+    localhost_no_origin_override: bool
+    allow_any_public_llm_host: bool
+    allow_insecure_base_url: bool
+    webhook_auth_mode: str
+    webhook_bearer_configured: bool
+    webhook_hmac_configured: bool
+    webhook_replay_protection_required: bool
+    remote_admin_enabled: bool
+    trust_x_forwarded_for: bool
+    trusted_proxies_configured: bool
+    callback_allow_hosts_configured: bool
+    external_tools_enabled: bool
+    registry_sync_enabled: bool
+    transforms_enabled: bool
+    bridge_enabled: bool
+    bridge_device_token_configured: bool
+    bridge_mtls_enabled: bool
+    bridge_device_cert_map_configured: bool
+    bridge_allowed_device_ids_configured: bool
+    public_shared_surface_acknowledged: bool
+    control_plane_mode: str
+    control_plane_url_configured: bool
+    control_plane_token_configured: bool
+    control_plane_prerequisites_satisfied: bool
+    control_plane_compat_override: bool
+    connector_active_platforms: tuple[str, ...]
+    connector_unguarded_platforms: tuple[str, ...]
+    connector_recommended_allowlist_vars: tuple[str, ...]
+    deployment_checks: tuple[PostureFinding, ...]
+    deployment_pass_codes: tuple[str, ...]
+    deployment_warn_codes: tuple[str, ...]
+    deployment_fail_codes: tuple[str, ...]
+    startup_profile_passed: bool
+    startup_profile_overridden: bool
+    startup_profile_violation_codes: tuple[str, ...]
+    blocked_surface_ids: tuple[str, ...]
+    decision_codes: tuple[str, ...]
+    reason_codes: tuple[str, ...]
+
+
+def _read(
+    environ: Mapping[str, str],
+    primary: str,
+    legacy: str | None = None,
+    default: str = "",
+) -> str:
+    try:
+        if primary in environ:
+            value = environ.get(primary, default)
+        elif legacy and legacy in environ:
+            value = environ.get(legacy, default)
+        else:
+            value = default
+    except Exception:
+        # CRITICAL: malformed environment providers must fail closed without echoing
+        # exception content or the attempted value into diagnostics.
+        raise ValueError("security posture input unavailable") from None
+    if value is None:
+        return ""
+    try:
+        return str(value)
+    except Exception:
+        raise ValueError("security posture input is not scalar") from None
+
+
+def _normalized(
+    environ: Mapping[str, str],
+    primary: str,
+    legacy: str | None = None,
+    default: str = "",
+) -> str:
+    return _read(environ, primary, legacy, default).strip().lower()
+
+
+def _enabled(
+    environ: Mapping[str, str],
+    primary: str,
+    legacy: str | None = None,
+) -> bool:
+    return _normalized(environ, primary, legacy) in _TRUTHY
+
+
+def _configured(
+    environ: Mapping[str, str],
+    primary: str,
+    legacy: str | None = None,
+) -> bool:
+    return bool(_read(environ, primary, legacy).strip())
+
+
+def _network_exposed_from_argv() -> bool:
+    # Preserve the accepted S41 heuristic exactly: only the explicit --listen flag
+    # changes this process-static decision.
+    return "--listen" in sys.argv
+
+
+def _deployment_report(profile: str, environ: Mapping[str, str]):
+    try:
+        from ..deployment_profile import evaluate_deployment_profile
+    except ImportError:  # pragma: no cover - top-level compatibility mode
+        from services.deployment_profile import evaluate_deployment_profile
+
+    return evaluate_deployment_profile(profile, environ)
+
+
+def _connector_posture(environ: Mapping[str, str]) -> Mapping[str, Any]:
+    try:
+        from ..connector_allowlist_posture import evaluate_connector_allowlist_posture
+    except ImportError:  # pragma: no cover - top-level compatibility mode
+        from services.connector_allowlist_posture import (
+            evaluate_connector_allowlist_posture,
+        )
+
+    return evaluate_connector_allowlist_posture(environ)
+
+
+def _blocked_surface_ids(profile: str, mode: str) -> tuple[str, ...]:
+    if profile != "public" or mode != "split":
+        return ()
+    # IMPORTANT: these are the stable scalar IDs from the S62 registry. Importing
+    # control_plane here would create a dependency cycle before R233 packages the domain.
+    return (
+        "callback_egress",
+        "registry_sync",
+        "secrets_write",
+        "tool_execution",
+        "transforms_exec",
+        "webhook_execute",
+    )
+
+
+def _safe_finding(check: Any) -> PostureFinding:
+    message = str(check.message)
+    if str(check.code) == "DP-WEBHOOK-005":
+        # IMPORTANT: the legacy evaluator includes the raw invalid environment value.
+        # The immutable boundary retains the stable code but never the untrusted value.
+        message = "Unsupported webhook auth mode."
+    return PostureFinding(
+        severity=str(check.severity),
+        code=str(check.code),
+        message=message,
+        remediation=str(check.remediation),
+    )
+
+
+def resolve_effective_security_posture(
+    environ: Mapping[str, str] | None = None,
+    *,
+    network_exposed: bool | None = None,
+) -> EffectiveSecurityPosture:
+    # IMPORTANT: an explicitly supplied empty mapping means empty input. Do not use
+    # `environ or os.environ`; doing so makes tests and lifecycle injection ambient.
+    env = os.environ if environ is None else environ
+
+    resolved_network_exposed = (
+        _network_exposed_from_argv()
+        if network_exposed is None
+        else bool(network_exposed)
+    )
+    deployment_profile = _normalized(
+        env, "OPENCLAW_DEPLOYMENT_PROFILE", default="local"
+    )
+    if deployment_profile not in _VALID_DEPLOYMENT_PROFILES:
+        raise ValueError("unsupported deployment profile")
+
+    raw_runtime_profile = _normalized(
+        env, "OPENCLAW_RUNTIME_PROFILE", default="minimal"
+    )
+    runtime_profile = "hardened" if raw_runtime_profile == "hardened" else "minimal"
+    mae_profile = (
+        "hardened"
+        if runtime_profile == "hardened" and deployment_profile != "public"
+        else deployment_profile
+    )
+
+    try:
+        report = _deployment_report(deployment_profile, env)
+    except Exception:
+        # CRITICAL: delegated evaluators must not expose hostile mapping values or
+        # exception text across the immutable posture boundary.
+        raise ValueError("security posture evaluation failed") from None
+    findings = tuple(_safe_finding(check) for check in report.checks)
+    pass_codes = tuple(item.code for item in findings if item.severity == "pass")
+    warn_codes = tuple(item.code for item in findings if item.severity == "warn")
+    fail_codes = tuple(item.code for item in findings if item.severity == "fail")
+
+    dangerous_profile_override = _enabled(
+        env, "OPENCLAW_SECURITY_DANGEROUS_PROFILE_OVERRIDE"
+    )
+    startup_violations = () if deployment_profile == "local" else fail_codes
+    startup_overridden = bool(startup_violations and dangerous_profile_override)
+    startup_passed = (
+        deployment_profile == "local" or not startup_violations or startup_overridden
+    )
+
+    explicit_control_mode = _normalized(env, "OPENCLAW_CONTROL_PLANE_MODE")
+    if explicit_control_mode in {"embedded", "split"}:
+        control_plane_mode = explicit_control_mode
+    elif deployment_profile == "public":
+        control_plane_mode = "split"
+    else:
+        control_plane_mode = "embedded"
+
+    control_plane_url_configured = _configured(env, "OPENCLAW_CONTROL_PLANE_URL")
+    control_plane_token_configured = _configured(env, "OPENCLAW_CONTROL_PLANE_TOKEN")
+    control_plane_prerequisites_satisfied = (
+        control_plane_url_configured and control_plane_token_configured
+    )
+    control_plane_compat_override = (
+        _normalized(env, "OPENCLAW_SPLIT_COMPAT_OVERRIDE") in _CONTROL_PLANE_TRUTHY
+    )
+
+    try:
+        connector = _connector_posture(env)
+    except Exception:
+        raise ValueError("security posture evaluation failed") from None
+    active_platforms = tuple(
+        sorted({str(item) for item in connector["active_platforms"]})
+    )
+    unguarded_platforms = tuple(
+        sorted({str(item) for item in connector["unguarded_platforms"]})
+    )
+    recommended_allowlist_vars = tuple(
+        sorted({str(item) for item in connector["recommended_allowlist_vars"]})
+    )
+
+    reason_codes = list(startup_violations)
+    if deployment_profile == "public" and control_plane_mode == "split":
+        if not control_plane_url_configured:
+            reason_codes.append("CP-URL-MISSING")
+        if not control_plane_token_configured:
+            reason_codes.append("CP-TOKEN-MISSING")
+    elif deployment_profile == "public" and control_plane_mode == "embedded":
+        if not control_plane_compat_override:
+            reason_codes.append("CP-PUBLIC-EMBEDDED")
+    reason_codes.extend(
+        f"CONNECTOR-ALLOWLIST-{platform.upper()}" for platform in unguarded_platforms
+    )
+    if raw_runtime_profile not in {"", "minimal", "hardened"}:
+        reason_codes.append("RUNTIME-PROFILE-DEFAULTED")
+
+    decision_codes = [
+        (
+            "STARTUP-OVERRIDDEN"
+            if startup_overridden
+            else "STARTUP-PASS" if startup_passed else "STARTUP-DENY"
+        ),
+        (
+            "CONTROL-PLANE-PASS"
+            if (
+                deployment_profile != "public"
+                or (
+                    control_plane_mode == "split"
+                    and control_plane_prerequisites_satisfied
+                )
+                or (control_plane_mode == "embedded" and control_plane_compat_override)
+            )
+            else "CONTROL-PLANE-DENY"
+        ),
+        (
+            "CONNECTORS-NONE"
+            if not active_platforms
+            else "CONNECTORS-UNGUARDED" if unguarded_platforms else "CONNECTORS-GUARDED"
+        ),
+        "NETWORK-EXPOSED" if resolved_network_exposed else "NETWORK-LOOPBACK",
+    ]
+
+    raw_webhook_mode = _normalized(
+        env,
+        "OPENCLAW_WEBHOOK_AUTH_MODE",
+        "MOLTBOT_WEBHOOK_AUTH_MODE",
+    )
+    webhook_auth_mode = (
+        raw_webhook_mode
+        if raw_webhook_mode in _VALID_WEBHOOK_MODES
+        else "unset" if not raw_webhook_mode else "invalid"
+    )
+    replay_value = _normalized(
+        env,
+        "OPENCLAW_WEBHOOK_REQUIRE_REPLAY_PROTECTION",
+        "MOLTBOT_WEBHOOK_REQUIRE_REPLAY_PROTECTION",
+    )
+
+    return EffectiveSecurityPosture(
+        schema_version=SCHEMA_VERSION,
+        runtime_profile=runtime_profile,
+        deployment_profile=deployment_profile,
+        mae_profile=mae_profile,
+        network_exposed=resolved_network_exposed,
+        admin_token_configured=_configured(
+            env, "OPENCLAW_ADMIN_TOKEN", "MOLTBOT_ADMIN_TOKEN"
+        ),
+        observability_token_configured=_configured(
+            env, "OPENCLAW_OBSERVABILITY_TOKEN", "MOLTBOT_OBSERVABILITY_TOKEN"
+        ),
+        dangerous_profile_override=dangerous_profile_override,
+        dangerous_bind_override=_enabled(
+            env,
+            "OPENCLAW_SECURITY_DANGEROUS_BIND_OVERRIDE",
+            "MOLTBOT_SECURITY_DANGEROUS_BIND_OVERRIDE",
+        ),
+        localhost_no_origin_override=(
+            _normalized(env, "OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN") == "true"
+        ),
+        allow_any_public_llm_host=_enabled(
+            env,
+            "OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST",
+            "MOLTBOT_ALLOW_ANY_PUBLIC_LLM_HOST",
+        ),
+        allow_insecure_base_url=_enabled(
+            env,
+            "OPENCLAW_ALLOW_INSECURE_BASE_URL",
+            "MOLTBOT_ALLOW_INSECURE_BASE_URL",
+        ),
+        webhook_auth_mode=webhook_auth_mode,
+        webhook_bearer_configured=_configured(
+            env,
+            "OPENCLAW_WEBHOOK_BEARER_TOKEN",
+            "MOLTBOT_WEBHOOK_BEARER_TOKEN",
+        ),
+        webhook_hmac_configured=_configured(
+            env,
+            "OPENCLAW_WEBHOOK_HMAC_SECRET",
+            "MOLTBOT_WEBHOOK_HMAC_SECRET",
+        ),
+        webhook_replay_protection_required=replay_value not in _FALSY,
+        remote_admin_enabled=_enabled(
+            env, "OPENCLAW_ALLOW_REMOTE_ADMIN", "MOLTBOT_ALLOW_REMOTE_ADMIN"
+        ),
+        trust_x_forwarded_for=_enabled(
+            env,
+            "OPENCLAW_TRUST_X_FORWARDED_FOR",
+            "MOLTBOT_TRUST_X_FORWARDED_FOR",
+        ),
+        trusted_proxies_configured=_configured(
+            env, "OPENCLAW_TRUSTED_PROXIES", "MOLTBOT_TRUSTED_PROXIES"
+        ),
+        callback_allow_hosts_configured=_configured(
+            env,
+            "OPENCLAW_CALLBACK_ALLOW_HOSTS",
+            "MOLTBOT_CALLBACK_ALLOW_HOSTS",
+        ),
+        external_tools_enabled=_enabled(env, "OPENCLAW_ENABLE_EXTERNAL_TOOLS"),
+        registry_sync_enabled=_enabled(env, "OPENCLAW_ENABLE_REGISTRY_SYNC"),
+        transforms_enabled=_enabled(env, "OPENCLAW_ENABLE_TRANSFORMS"),
+        bridge_enabled=_enabled(
+            env, "OPENCLAW_BRIDGE_ENABLED", "MOLTBOT_BRIDGE_ENABLED"
+        ),
+        bridge_device_token_configured=_configured(
+            env,
+            "OPENCLAW_BRIDGE_DEVICE_TOKEN",
+            "MOLTBOT_BRIDGE_DEVICE_TOKEN",
+        ),
+        bridge_mtls_enabled=_enabled(env, "OPENCLAW_BRIDGE_MTLS_ENABLED"),
+        bridge_device_cert_map_configured=_configured(
+            env, "OPENCLAW_BRIDGE_DEVICE_CERT_MAP"
+        ),
+        bridge_allowed_device_ids_configured=_configured(
+            env,
+            "OPENCLAW_BRIDGE_ALLOWED_DEVICE_IDS",
+            "MOLTBOT_BRIDGE_ALLOWED_DEVICE_IDS",
+        ),
+        public_shared_surface_acknowledged=_enabled(
+            env,
+            "OPENCLAW_PUBLIC_SHARED_SURFACE_BOUNDARY_ACK",
+            "MOLTBOT_PUBLIC_SHARED_SURFACE_BOUNDARY_ACK",
+        ),
+        control_plane_mode=control_plane_mode,
+        control_plane_url_configured=control_plane_url_configured,
+        control_plane_token_configured=control_plane_token_configured,
+        control_plane_prerequisites_satisfied=control_plane_prerequisites_satisfied,
+        control_plane_compat_override=control_plane_compat_override,
+        connector_active_platforms=active_platforms,
+        connector_unguarded_platforms=unguarded_platforms,
+        connector_recommended_allowlist_vars=recommended_allowlist_vars,
+        deployment_checks=findings,
+        deployment_pass_codes=pass_codes,
+        deployment_warn_codes=warn_codes,
+        deployment_fail_codes=fail_codes,
+        startup_profile_passed=startup_passed,
+        startup_profile_overridden=startup_overridden,
+        startup_profile_violation_codes=startup_violations,
+        blocked_surface_ids=_blocked_surface_ids(
+            deployment_profile, control_plane_mode
+        ),
+        decision_codes=tuple(decision_codes),
+        reason_codes=tuple(dict.fromkeys(reason_codes)),
+    )
+
+
+def install_effective_security_posture(
+    posture: EffectiveSecurityPosture,
+) -> EffectiveSecurityPosture:
+    if not isinstance(posture, EffectiveSecurityPosture):
+        raise TypeError("posture must be EffectiveSecurityPosture")
+    global _installed_posture
+    with _posture_lock:
+        if _installed_posture is None:
+            _installed_posture = posture
+        elif _installed_posture is not posture:
+            # CRITICAL: silently replacing process posture creates contradictory
+            # authorization decisions. Reset is an explicit lifecycle/test operation.
+            raise RuntimeError("effective security posture is already installed")
+        return _installed_posture
+
+
+def get_effective_security_posture(
+    *, required: bool = True
+) -> EffectiveSecurityPosture | None:
+    with _posture_lock:
+        posture = _installed_posture
+    if posture is None and required:
+        raise RuntimeError("effective security posture is not installed")
+    return posture
+
+
+def get_or_create_effective_security_posture(
+    environ: Mapping[str, str] | None = None,
+    *,
+    network_exposed: bool | None = None,
+) -> EffectiveSecurityPosture:
+    with _posture_lock:
+        if _installed_posture is not None:
+            return _installed_posture
+        posture = resolve_effective_security_posture(
+            environ,
+            network_exposed=network_exposed,
+        )
+        # The RLock makes this identity-stable even under concurrent startup.
+        return install_effective_security_posture(posture)
+
+
+def reset_effective_security_posture_for_tests() -> None:
+    global _installed_posture
+    with _posture_lock:
+        _installed_posture = None
+
+
+def effective_security_posture_diagnostics(
+    posture: EffectiveSecurityPosture | None = None,
+) -> dict[str, Any]:
+    resolved = posture or get_effective_security_posture()
+    assert resolved is not None
+    return {
+        "schema_version": resolved.schema_version,
+        "runtime_profile": resolved.runtime_profile,
+        "deployment_profile": resolved.deployment_profile,
+        "mae_profile": resolved.mae_profile,
+        "network_exposed": resolved.network_exposed,
+        "authentication": {
+            "admin_configured": resolved.admin_token_configured,
+            "observability_configured": resolved.observability_token_configured,
+        },
+        "startup_gate": {
+            "passed": resolved.startup_profile_passed,
+            "overridden": resolved.startup_profile_overridden,
+            "violation_codes": list(resolved.startup_profile_violation_codes),
+        },
+        "control_plane": {
+            "mode": resolved.control_plane_mode,
+            "prerequisites_satisfied": (resolved.control_plane_prerequisites_satisfied),
+            "compat_override": resolved.control_plane_compat_override,
+            "blocked_surface_count": len(resolved.blocked_surface_ids),
+        },
+        "connectors": {
+            "active_count": len(resolved.connector_active_platforms),
+            "unguarded_count": len(resolved.connector_unguarded_platforms),
+        },
+        "decision_codes": list(resolved.decision_codes),
+        "reason_codes": list(resolved.reason_codes),
+    }

services/route_bootstrap.py

@@ -1,623 +1,18 @@
-"""
-R130 route/bootstrap orchestration extracted from package entrypoint.
-
-Keeps __init__.py thin while preserving startup behavior and fallback handling.
-"""
+"""Compatibility alias for the bootstrap registration implementation module."""
 
 from __future__ import annotations
 
-import logging
-import os
 import sys
-import threading
-import time
-from collections.abc import Callable
+from typing import TYPE_CHECKING
 
-_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
+from .bootstrap import registration as _implementation
 
-
-def _resolve_optional_warmup_timeout_sec() -> float:
-    raw = (
-        os.environ.get("OPENCLAW_STARTUP_WARMUP_TIMEOUT_SEC")
-        or os.environ.get("MOLTBOT_STARTUP_WARMUP_TIMEOUT_SEC")
-        or "5"
-    )
-    try:
-        return max(0.1, min(float(raw), 60.0))
-    except (TypeError, ValueError):
-        return 5.0
-
-
-def _warm_model_inventory_snapshot() -> None:
-    from .preflight import get_model_inventory_snapshot
-
-    get_model_inventory_snapshot(trigger_refresh=True)
-
-
-def _build_optional_startup_warmups():
-    timeout_sec = _resolve_optional_warmup_timeout_sec()
-    return [
-        ("model_inventory", _warm_model_inventory_snapshot, timeout_sec),
-    ]
-
-
-def _mark_startup_ready_and_start_warmups() -> None:
-    from .startup_lifecycle import mark_startup_ready, start_optional_warmups
-
-    # 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 as exc:
-        # IMPORTANT: optional warmup diagnostics must not undo successful route startup.
-        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,
-    *,
-    reason_code=None,
-) -> None:
-    try:
-        from .startup_lifecycle import mark_startup_fatal
-
-        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").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,
+if TYPE_CHECKING:
+    # Static-only exports keep legacy imports typed without duplicating module globals.
+    from .bootstrap.registration import register_routes_once as register_routes_once
+    from .bootstrap.registration import (
+        reset_route_bootstrap_for_tests as reset_route_bootstrap_for_tests,
     )
 
-
-def _load_plugin_shutdown_registrars():
-    """Load optional startup registrars behind one patchable compatibility seam."""
-
-    from .plugins.builtin import register_all
-    from .runtime_lifecycle import register_shutdown_hooks
-
-    return register_shutdown_hooks, register_all
-
-
-def _register_plugins_and_shutdown_hooks() -> None:
-    # R67: Best-effort process shutdown hook and built-in plugin registration.
-    try:
-        register_shutdown_hooks, register_all = _load_plugin_shutdown_registrars()
-    except ImportError as exc:
-        logging.getLogger("ComfyUI-OpenClaw").error(
-            "Optional startup registrars unavailable (error_type=%s)",
-            type(exc).__name__,
-        )
-        return
-
-    logger = logging.getLogger("ComfyUI-OpenClaw")
-    for component, registrar in (
-        ("shutdown_hooks", register_shutdown_hooks),
-        ("builtin_plugins", register_all),
-    ):
-        try:
-            registrar()
-        except Exception as exc:
-            # IMPORTANT: these optional steps are independent. Keep startup available,
-            # do not echo exception content, and do not catch BaseException cancellation.
-            logger.error(
-                "Optional startup registrar failed (component=%s, error_type=%s)",
-                component,
-                type(exc).__name__,
-            )
-
-
-def _initialize_registries_and_security_gate() -> None:
-    # R63/R84: Initialize Service & Module Registries.
-    try:
-        from .modules import ModuleCapability, ModuleRegistry, enable_module
-        from .registry import SVC_RUNTIME_CONFIG, ServiceRegistry
-        from .runtime_config import get_config
-
-        config = get_config()
-        ServiceRegistry.register(SVC_RUNTIME_CONFIG, config)
-
-        from .effective_security_posture import (
-            get_effective_security_posture,
-            resolve_effective_security_posture,
-        )
-
-        posture = get_effective_security_posture(required=False)
-        if posture is None:
-            # Direct compatibility/test invocation does not own process installation.
-            posture = resolve_effective_security_posture()
-
-        # Always-on modules
-        enable_module(ModuleCapability.CORE)
-        enable_module(ModuleCapability.SECURITY)
-        enable_module(ModuleCapability.OBSERVABILITY)
-
-        # S50: initialize durable idempotency storage early.
-        from .idempotency_store import IdempotencyStore
-        from .state_dir import get_state_dir
-
-        db_path = os.path.join(get_state_dir(), "idempotency.db")
-        # 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 (strict_mode=True)"
-        )
-
-        if config.bridge_enabled:
-            enable_module(ModuleCapability.BRIDGE)
-
-        # Core runtime modules stay enabled; runners decide active behavior.
-        enable_module(ModuleCapability.SCHEDULER)
-        enable_module(ModuleCapability.WEBHOOK)
-        enable_module(ModuleCapability.CONNECTOR)
-
-        ModuleRegistry.lock()
-        logging.getLogger("ComfyUI-OpenClaw").info(
-            "Initialized modules: %s", ModuleRegistry.get_enabled_list()
-        )
-
-        from .security_gate import enforce_startup_gate
-
-        enforce_startup_gate(posture=posture)
-    except Exception as exc:
-        logging.getLogger("ComfyUI-OpenClaw").error(
-            "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.
-        raise
-
-
-def _do_full_registration(server) -> None:
-    """Register all OpenClaw routes including bridge/scheduler bindings."""
-    from .access_control import require_admin_token
-    from .parameter_lab_queue_receipt import (
-        register_parameter_lab_queue_receipt_handler,
-    )
-    from .plugins.async_bridge import run_async_in_sync_context
-    from .queue_submit import submit_prompt
-    from .route_bootstrap_contract import load_route_bootstrap_contract
-    from .scheduler.runner import get_scheduler_runner, start_scheduler
-    from .templates import get_template_service
-
-    contract = load_route_bootstrap_contract(__package__)
-    register_approval_routes = contract["register_approval_routes"]
-    BridgeHandlers = contract["BridgeHandlers"]
-    register_preset_routes = contract["register_preset_routes"]
-    register_routes = contract["register_routes"]
-    register_schedule_routes = contract["register_schedule_routes"]
-    register_trigger_routes = contract["register_trigger_routes"]
-
-    register_routes(server)
-    # CRITICAL: receipt promotion is required for exact Parameter Lab run ownership.
-    register_parameter_lab_queue_receipt_handler(server)
-    register_preset_routes(server.app)
-    register_schedule_routes(server.app, require_admin_token_fn=require_admin_token)
-
-    class QueueSubmitService:
-        def submit(self, job_req):
-            tmpl_svc = get_template_service()
-            workflow = tmpl_svc.render_template(job_req.template_id, job_req.inputs)
-
-            async def _do_submit():
-                return await submit_prompt(
-                    workflow,
-                    client_id=job_req.session_id or "bridge",
-                    extra_data={
-                        "openclaw": {"trace_id": job_req.trace_id},
-                        # Legacy key kept for existing tooling that expects this blob.
-                        "moltbot": {"trace_id": job_req.trace_id},
-                    },
-                    source="bridge",
-                    trace_id=job_req.trace_id,
-                )
-
-            return run_async_in_sync_context(_do_submit())
-
-    bridge_handlers = BridgeHandlers(submit_service=QueueSubmitService())
-    _register_bridge_routes(server.app.router, bridge_handlers)
-
-    async def unified_submit_fn(
-        template_id,
-        inputs,
-        trace_id,
-        idempotency_key,
-        delivery=None,
-        source="unknown",
-    ):
-        """Submit function for scheduler and trigger-triggered runs."""
-        # NOTE: Use IdempotencyStore API (check_and_record/update_prompt_id).
-        # Avoid legacy get_store/get/set usage; wrong API here breaks route registration at runtime.
-        from .idempotency_store import IdempotencyStore
-        from .queue_submit import submit_prompt as _submit_prompt
-        from .templates import get_template_service as _get_template_service
-
-        store = IdempotencyStore()
-        is_dup, existing_prompt_id = store.check_and_record(idempotency_key)
-        if is_dup:
-            return {"prompt_id": existing_prompt_id, "deduped": True}
-
-        tmpl_svc = _get_template_service()
-        workflow = tmpl_svc.render_template(template_id, inputs)
-
-        result = await _submit_prompt(
-            workflow,
-            extra_data={
-                "openclaw": {"trace_id": trace_id, "source": "automation"},
-                "moltbot": {"trace_id": trace_id, "source": "automation"},
-            },
-            source=source,
-            trace_id=trace_id,
-        )
-
-        if result.get("prompt_id"):
-            store.update_prompt_id(idempotency_key, result["prompt_id"])
-        return result
-
-    runner = get_scheduler_runner()
-    runner._submit_fn = unified_submit_fn
-    start_scheduler()
-
-    register_trigger_routes(
-        server.app,
-        require_admin_token_fn=require_admin_token,
-        submit_fn=unified_submit_fn,
-    )
-    register_approval_routes(
-        server.app,
-        require_admin_token_fn=require_admin_token,
-        submit_fn=unified_submit_fn,
-    )
-
-
-_BRIDGE_ROUTE_SPECS = (
-    ("add_post", "/moltbot/bridge/submit", "submit_handler"),
-    ("add_post", "/moltbot/bridge/deliver", "deliver_handler"),
-    ("add_get", "/moltbot/bridge/health", "health_handler"),
-    ("add_post", "/openclaw/bridge/submit", "submit_handler"),
-    ("add_post", "/openclaw/bridge/deliver", "deliver_handler"),
-    ("add_get", "/openclaw/bridge/health", "health_handler"),
-    ("add_post", "/api/moltbot/bridge/submit", "submit_handler"),
-    ("add_post", "/api/moltbot/bridge/deliver", "deliver_handler"),
-    ("add_get", "/api/moltbot/bridge/health", "health_handler"),
-    ("add_post", "/api/openclaw/bridge/submit", "submit_handler"),
-    ("add_post", "/api/openclaw/bridge/deliver", "deliver_handler"),
-    ("add_get", "/api/openclaw/bridge/health", "health_handler"),
-)
-
-
-def _register_bridge_routes(router, bridge_handlers) -> None:
-    # IMPORTANT: keep bridge route registration table-driven.
-    # Missing one alias path here silently breaks one control-plane surface while
-    # leaving the rest apparently healthy, which is hard to diagnose during startup.
-    for method_name, path, handler_name in _BRIDGE_ROUTE_SPECS:
-        registrar = getattr(router, method_name, None)
-        if registrar is None:
-            continue
-        try:
-            registrar(path, getattr(bridge_handlers, handler_name))
-        except RuntimeError:
-            if path.startswith("/api/"):
-                continue
-            raise
-
-
-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 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()
-    try:
-        from .effective_security_posture import (
-            reset_effective_security_posture_for_tests,
-        )
-
-        reset_effective_security_posture_for_tests()
-    except ImportError:
-        # Dependency-light test/import mode may omit the posture module.
-        pass
-
-
-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:
-                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
-                _store_registration_success(generation=owner_generation)
-                logger.info(
-                    "Routes registered successfully (attempt=%s)",
-                    attempt,
-                )
-                return
-
-            _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)
-
-        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()
-
-
-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:
-    """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:
-        from .effective_security_posture import get_or_create_effective_security_posture
-
-        # CRITICAL: this required startup owner installs process-static posture once.
-        # Direct helper/API invocations resolve ephemeral snapshots instead.
-        get_or_create_effective_security_posture()
-        _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__,
-            )
-            raise
-
-        try:
-            server = _resolve_prompt_server()
-        except Exception as exc:
-            _mark_startup_fatal("route_registration", exc)
-            _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()
+# IMPORTANT: alias the module object; copied re-exports break accepted patch seams.
+sys.modules[__name__] = _implementation

services/startup_lifecycle.py

@@ -1,698 +1,55 @@
-"""Typed startup lifecycle outcomes and redacted public diagnostics.
-
-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.
-"""
+"""Compatibility alias for the bootstrap lifecycle implementation module."""
 
 from __future__ import annotations
 
-import logging
-import math
-import re
-import threading
-import time
-from collections.abc import Callable, Iterable
-from dataclasses import dataclass
-from enum import Enum
-from typing import Any
+import sys
+from typing import TYPE_CHECKING
 
-logger = logging.getLogger("ComfyUI-OpenClaw.services.startup_lifecycle")
+from .bootstrap import lifecycle as _implementation
 
-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",
-)
-
-
-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]
-
-
-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,
+if TYPE_CHECKING:
+    # Static-only exports keep legacy imports typed without a second runtime owner.
+    from .bootstrap.lifecycle import MAX_DIAGNOSTIC_MS as MAX_DIAGNOSTIC_MS
+    from .bootstrap.lifecycle import MAX_WARMUPS as MAX_WARMUPS
+    from .bootstrap.lifecycle import SCHEMA_VERSION as SCHEMA_VERSION
+    from .bootstrap.lifecycle import STARTUP_DEGRADED_WARMUP as STARTUP_DEGRADED_WARMUP
+    from .bootstrap.lifecycle import STARTUP_DIAGNOSTIC_KEYS as STARTUP_DIAGNOSTIC_KEYS
+    from .bootstrap.lifecycle import STARTUP_FATAL as STARTUP_FATAL
+    from .bootstrap.lifecycle import STARTUP_READY as STARTUP_READY
+    from .bootstrap.lifecycle import STARTUP_STARTING as STARTUP_STARTING
+    from .bootstrap.lifecycle import WARMUP_FAILED as WARMUP_FAILED
+    from .bootstrap.lifecycle import WARMUP_PENDING as WARMUP_PENDING
+    from .bootstrap.lifecycle import WARMUP_RUNNING as WARMUP_RUNNING
+    from .bootstrap.lifecycle import WARMUP_SUCCEEDED as WARMUP_SUCCEEDED
+    from .bootstrap.lifecycle import WARMUP_TIMED_OUT as WARMUP_TIMED_OUT
+    from .bootstrap.lifecycle import StartupLifecycle as StartupLifecycle
+    from .bootstrap.lifecycle import StartupOutcome as StartupOutcome
+    from .bootstrap.lifecycle import StartupPhase as StartupPhase
+    from .bootstrap.lifecycle import StartupReason as StartupReason
+    from .bootstrap.lifecycle import StartupState as StartupState
+    from .bootstrap.lifecycle import StartupTransitionError as StartupTransitionError
+    from .bootstrap.lifecycle import WarmupOutcome as WarmupOutcome
+    from .bootstrap.lifecycle import WarmupSpec as WarmupSpec
+    from .bootstrap.lifecycle import WarmupState as WarmupState
+    from .bootstrap.lifecycle import get_startup_diagnostics as get_startup_diagnostics
+    from .bootstrap.lifecycle import get_startup_outcome as get_startup_outcome
+    from .bootstrap.lifecycle import (
+        mark_bootstrap_import_failed as mark_bootstrap_import_failed,
     )
-
-
-def mark_startup_ready(phase: str = "routes") -> None:
-    """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 = 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,
+    from .bootstrap.lifecycle import mark_host_waiting as mark_host_waiting
+    from .bootstrap.lifecycle import (
+        mark_required_initialization_started as mark_required_initialization_started,
     )
-
-
-def mark_bootstrap_import_failed(exc: BaseException) -> None:
-    mark_startup_fatal(
-        "package_import",
-        exc,
-        reason_code=StartupReason.BOOTSTRAP_IMPORT_FAILED,
+    from .bootstrap.lifecycle import mark_retry_exhausted as mark_retry_exhausted
+    from .bootstrap.lifecycle import (
+        mark_route_registration_started as mark_route_registration_started,
     )
-
-
-def mark_retry_exhausted() -> None:
-    _LIFECYCLE.mark_retry_exhausted()
-
-
-def start_optional_warmups(specs: Iterable[WarmupSpec]) -> None:
-    for name, fn, timeout_sec in tuple(specs or ()):
-        _start_optional_warmup(str(name), fn, float(timeout_sec))
-
-
-def reset_startup_lifecycle_for_tests() -> None:
-    _LIFECYCLE.reset()
-
-
-def _start_optional_warmup(
-    name: str, fn: Callable[[], Any], timeout_sec: float
-) -> None:
-    should_start, generation, safe_name = _LIFECYCLE.begin_warmup(name, timeout_sec)
-    if not should_start:
-        return
-    monitor = threading.Thread(
-        target=_warmup_monitor,
-        args=(safe_name, generation, fn, max(0.01, min(timeout_sec, 60.0))),
-        name=f"openclaw-warmup-monitor-{safe_name}",
-        daemon=True,
+    from .bootstrap.lifecycle import mark_startup_fatal as mark_startup_fatal
+    from .bootstrap.lifecycle import mark_startup_ready as mark_startup_ready
+    from .bootstrap.lifecycle import (
+        reset_startup_lifecycle_for_tests as reset_startup_lifecycle_for_tests,
     )
-    try:
-        monitor.start()
-    except Exception:
-        _LIFECYCLE.finish_warmup(
-            safe_name,
-            generation,
-            state=WarmupState.FAILED,
-        )
-        raise
+    from .bootstrap.lifecycle import start_optional_warmups as start_optional_warmups
 
-
-def _warmup_monitor(
-    name: str,
-    generation: int,
-    fn: Callable[[], Any],
-    timeout_sec: float,
-) -> None:
-    done = threading.Event()
-    result: dict[str, bool] = {}
-
-    def _worker() -> None:
-        try:
-            fn()
-            result["ok"] = True
-        except Exception:
-            # SECURITY: never retain or log arbitrary exception content.
-            result["ok"] = False
-        finally:
-            done.set()
-
-    _LIFECYCLE.mark_warmup_running(name, generation)
-    worker = threading.Thread(
-        target=_worker,
-        name=f"openclaw-warmup-{name}",
-        daemon=True,
-    )
-    try:
-        worker.start()
-    except Exception as exc:
-        _LIFECYCLE.finish_warmup(
-            name,
-            generation,
-            state=WarmupState.FAILED,
-        )
-        logger.warning(
-            "Optional startup warmup worker could not start "
-            "(component=%s, error_type=%s)",
-            name,
-            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
-
-    if result.get("ok"):
-        _LIFECYCLE.finish_warmup(
-            name,
-            generation,
-            state=WarmupState.SUCCEEDED,
-        )
-        logger.info("Optional startup warmup completed (component=%s)", name)
-        return
-
-    _LIFECYCLE.finish_warmup(
-        name,
-        generation,
-        state=WarmupState.FAILED,
-    )
-    logger.warning(
-        "Optional startup warmup failed (component=%s, reason_code=%s)",
-        name,
-        StartupReason.WARMUP_FAILED.value,
-    )
+# IMPORTANT: alias the module object; copied re-exports split singleton and patch state.
+sys.modules[__name__] = _implementation

tests/architecture_dependency_policy.json

@@ -167,6 +167,9 @@
       "services/automation_composer.py",
       "services/bridge_handshake.py",
       "services/bridge_token_lifecycle.py",
+      "services/bootstrap/__init__.py",
+      "services/bootstrap/lifecycle.py",
+      "services/bootstrap/registration.py",
       "services/cache/__init__.py",
       "services/cache/ttl_cache.py",
       "services/callback_delivery.py",
@@ -248,6 +251,8 @@
       "services/plugins/manager.py",
       "services/pnginfo.py",
       "services/policy_posture.py",
+      "services/posture/__init__.py",
+      "services/posture/effective.py",
       "services/preflight.py",
       "services/presets/__init__.py",
       "services/presets/models.py",
@@ -436,6 +441,29 @@
       "review_condition": "Remove when the importer no longer requires the higher-level domain."
     }
   ],
+  "facade_contracts": [
+    {
+      "facade": "services.startup_lifecycle",
+      "implementation": "services.bootstrap.lifecycle",
+      "owner": "architecture-maintainers",
+      "rationale": "Preserve accepted lifecycle imports, singleton identity, and patch seams while implementation ownership lives in the bootstrap package.",
+      "review_condition": "Review when the legacy startup_lifecycle import path can be removed without compatibility impact."
+    },
+    {
+      "facade": "services.route_bootstrap",
+      "implementation": "services.bootstrap.registration",
+      "owner": "architecture-maintainers",
+      "rationale": "Preserve accepted route-bootstrap imports and patch seams while implementation ownership lives in the bootstrap package.",
+      "review_condition": "Review when the legacy route_bootstrap import path can be removed without compatibility impact."
+    },
+    {
+      "facade": "services.effective_security_posture",
+      "implementation": "services.posture.effective",
+      "owner": "architecture-maintainers",
+      "rationale": "Preserve accepted posture imports and installed-snapshot identity while implementation ownership lives in the posture package.",
+      "review_condition": "Review when the legacy effective_security_posture import path can be removed without compatibility impact."
+    }
+  ],
   "accepted_cycles": [
     {
       "modules": [

tests/exception_boundary_policy.json

@@ -90,7 +90,7 @@
         }
       ]
     },
-    "services/route_bootstrap.py": {
+    "services/bootstrap/registration.py": {
       "coverage": "all_broad_catches",
       "broad_catches": [
         {

tests/static_analysis_policy.json

@@ -4837,6 +4837,13 @@
       "message": "Use `X | None` for type annotations",
       "count": 11
     },
+    {
+      "tool": "ruff",
+      "path": "services/bootstrap/registration.py",
+      "code": "N806",
+      "message": "Variable `BridgeHandlers` in function should be lowercase",
+      "count": 1
+    },
     {
       "tool": "ruff",
       "path": "services/bridge_handshake.py",
@@ -8351,13 +8358,6 @@
       "message": "Use `X | None` for type annotations",
       "count": 15
     },
-    {
-      "tool": "ruff",
-      "path": "services/route_bootstrap.py",
-      "code": "N806",
-      "message": "Variable `BridgeHandlers` in function should be lowercase",
-      "count": 1
-    },
     {
       "tool": "ruff",
       "path": "services/runtime_config.py",

tests/test_architecture_dependency_policy.py

@@ -60,6 +60,7 @@ class ArchitecturePolicyFixture(unittest.TestCase):
                 "core": ["core"],
             },
             "compatibility_exceptions": [],
+            "facade_contracts": [],
             "accepted_cycles": [],
             "dynamic_imports": [],
         }
@@ -98,6 +99,91 @@ class ArchitecturePolicyFixture(unittest.TestCase):
         self._write("core/util.py", "VALUE = 1\n")
         self.assertIn("DEP_STALE_EXCEPTION", self._codes(policy))
 
+    def test_same_domain_facade_reverse_dependency_fails(self):
+        self._write("app/facade.py", "from . import implementation\n")
+        self._write("app/implementation.py", "VALUE = 1\n")
+        policy = self._policy()
+        policy["domains"]["app"].extend(
+            [
+                "app/facade.py",
+                "app/implementation.py",
+            ]
+        )
+        policy["facade_contracts"] = [
+            {
+                "facade": "app.facade",
+                "implementation": "app.implementation",
+                **self._review_metadata(),
+            }
+        ]
+        self.assertEqual(self._verify(policy), ())
+
+        self._write("app/implementation.py", "from . import facade\n")
+        self.assertIn("FACADE_REVERSE_DEPENDENCY", self._codes(policy))
+
+    def test_facade_contract_rejects_unknown_stale_and_duplicate_entries(self):
+        self._write("app/facade.py", "from . import implementation\n")
+        self._write("app/implementation.py", "VALUE = 1\n")
+        policy = self._policy()
+        policy["domains"]["app"].extend(
+            [
+                "app/facade.py",
+                "app/implementation.py",
+            ]
+        )
+        entry = {
+            "facade": "app.facade",
+            "implementation": "app.implementation",
+            **self._review_metadata(),
+        }
+        policy["facade_contracts"] = [entry, copy.deepcopy(entry)]
+        self.assertIn("FACADE_DUPLICATE", self._codes(policy))
+
+        policy["facade_contracts"] = [
+            {
+                **entry,
+                "implementation": "app.missing",
+            }
+        ]
+        self.assertIn("FACADE_MODULE_UNKNOWN", self._codes(policy))
+
+        policy["facade_contracts"] = [entry]
+        self._write("app/facade.py", "VALUE = 1\n")
+        self.assertIn("FACADE_STALE", self._codes(policy))
+
+        policy["facade_contracts"] = "invalid"
+        self.assertIn("FACADES_INVALID", self._codes(policy))
+
+        policy["facade_contracts"] = [
+            {
+                "facade": "app.facade",
+                "implementation": "app.implementation",
+                "owner": "",
+                "rationale": "",
+                "review_condition": "",
+                "unexpected": True,
+            }
+        ]
+        codes = self._codes(policy)
+        self.assertIn("POLICY_REVIEW_METADATA", codes)
+        self.assertIn("POLICY_UNKNOWN_KEY", codes)
+
+        policy["facade_contracts"] = "invalid"
+        self.assertIn("FACADES_INVALID", self._codes(policy))
+
+        policy["facade_contracts"] = [
+            {
+                "facade": "app.facade",
+                "implementation": "app.facade",
+                "owner": "",
+                "rationale": "",
+                "review_condition": "",
+            }
+        ]
+        codes = self._codes(policy)
+        self.assertIn("FACADES_INVALID", codes)
+        self.assertIn("POLICY_REVIEW_METADATA", codes)
+
     def test_new_cycle_and_stale_accepted_cycle_fail(self):
         self._write("core/util.py", "from app.main import VALUE\n")
         policy = self._policy()
@@ -289,10 +375,11 @@ class RepositoryArchitecturePolicyTests(unittest.TestCase):
         analysis = dependency_policy.analyze_repository(self.repo_root, policy)
 
         self.assertEqual(analysis.findings, ())
-        self.assertEqual(len(analysis.owned_paths), 300)
+        self.assertEqual(len(analysis.owned_paths), 305)
         self.assertEqual(len(policy["accepted_cycles"]), 2)
         self.assertEqual(len(policy["dynamic_imports"]), 8)
         self.assertEqual(len(policy["compatibility_exceptions"]), 9)
+        self.assertEqual(len(policy["facade_contracts"]), 3)
 
     def test_policy_change_does_not_weaken_static_analysis_governance(self):
         static_policy_path = self.repo_root / "tests" / "static_analysis_policy.json"

tests/test_r180_exception_boundary_governance.py

@@ -8,6 +8,13 @@ from services import route_bootstrap
 
 
 class TestExceptionBoundaryGovernance(unittest.TestCase):
+    def setUp(self):
+        route_bootstrap.reset_route_bootstrap_for_tests()
+
+    def tearDown(self):
+        # IMPORTANT: registration installs process-static posture; keep test order inert.
+        route_bootstrap.reset_route_bootstrap_for_tests()
+
     def test_register_routes_once_reraises_initial_registration_failure(self):
         route_bootstrap._routes_registered = False
         server = SimpleNamespace(app=object())

tests/test_r219_exception_boundary_phase2.py

@@ -39,7 +39,7 @@ class TestPolicyV2(unittest.TestCase):
                 "connector/router_admin_handlers.py",
                 "connector/router_dispatch.py",
                 "connector/router_execution_handlers.py",
-                "services/route_bootstrap.py",
+                "services/bootstrap/registration.py",
                 "api/config_projection_handlers.py",
                 "api/config_llm_handlers.py",
                 "connector/platforms/slack_installation_handlers.py",
@@ -155,6 +155,20 @@ class TestPolicyV2(unittest.TestCase):
 
 
 class TestMaeProfileBoundary(unittest.TestCase):
+    def setUp(self):
+        from services.effective_security_posture import (
+            reset_effective_security_posture_for_tests,
+        )
+
+        reset_effective_security_posture_for_tests()
+
+    def tearDown(self):
+        from services.effective_security_posture import (
+            reset_effective_security_posture_for_tests,
+        )
+
+        reset_effective_security_posture_for_tests()
+
     def test_explicit_deployment_profile_precedes_runtime_probe(self):
         with (
             patch.dict(

tests/test_r233_service_domain_packages.py

@@ -0,0 +1,166 @@
+"""Contract tests for bootstrap/posture implementation package ownership."""
+
+from __future__ import annotations
+
+import ast
+import importlib
+import importlib.machinery
+import sys
+import types
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+ROOT = Path(__file__).resolve().parents[1]
+
+FACADE_PAIRS = (
+    ("services.startup_lifecycle", "services.bootstrap.lifecycle"),
+    ("services.route_bootstrap", "services.bootstrap.registration"),
+    ("services.effective_security_posture", "services.posture.effective"),
+)
+
+
+class ServiceDomainPackageContractTests(unittest.TestCase):
+    def test_old_and_new_paths_resolve_to_the_same_module_objects(self):
+        for facade_name, implementation_name in FACADE_PAIRS:
+            with self.subTest(facade=facade_name):
+                facade = importlib.import_module(facade_name)
+                implementation = importlib.import_module(implementation_name)
+
+                self.assertIs(facade, implementation)
+                self.assertIs(sys.modules[facade_name], implementation)
+
+    def test_file_loader_package_namespace_preserves_module_identity(self):
+        package_name = "r233_comfyui_loader_probe"
+        package = types.ModuleType(package_name)
+        package.__path__ = [str(ROOT)]
+        package.__package__ = package_name
+        package.__spec__ = importlib.machinery.ModuleSpec(
+            package_name,
+            loader=None,
+            is_package=True,
+        )
+        sys.modules[package_name] = package
+        try:
+            for facade_name, implementation_name in FACADE_PAIRS:
+                qualified_facade = f"{package_name}.{facade_name}"
+                qualified_implementation = f"{package_name}.{implementation_name}"
+                with self.subTest(facade=qualified_facade):
+                    facade = importlib.import_module(qualified_facade)
+                    implementation = importlib.import_module(qualified_implementation)
+                    self.assertIs(facade, implementation)
+                    self.assertIs(sys.modules[qualified_facade], implementation)
+
+            posture = importlib.import_module(
+                f"{package_name}.services.posture.effective"
+            )
+            snapshot = posture.resolve_effective_security_posture({})
+            self.assertEqual(snapshot.deployment_profile, "local")
+        finally:
+            for module_name in tuple(sys.modules):
+                if module_name == package_name or module_name.startswith(
+                    f"{package_name}."
+                ):
+                    sys.modules.pop(module_name, None)
+
+    def test_lifecycle_and_posture_singletons_are_not_duplicated(self):
+        legacy_lifecycle = importlib.import_module("services.startup_lifecycle")
+        owned_lifecycle = importlib.import_module("services.bootstrap.lifecycle")
+        legacy_posture = importlib.import_module("services.effective_security_posture")
+        owned_posture = importlib.import_module("services.posture.effective")
+
+        self.assertIs(legacy_lifecycle._LIFECYCLE, owned_lifecycle._LIFECYCLE)
+        self.assertIs(legacy_posture._posture_lock, owned_posture._posture_lock)
+
+        owned_posture.reset_effective_security_posture_for_tests()
+        try:
+            installed = legacy_posture.get_or_create_effective_security_posture({})
+            self.assertIs(owned_posture.get_effective_security_posture(), installed)
+        finally:
+            legacy_posture.reset_effective_security_posture_for_tests()
+
+    def test_old_path_patches_mutate_the_owned_modules(self):
+        owned_lifecycle = importlib.import_module("services.bootstrap.lifecycle")
+        owned_registration = importlib.import_module("services.bootstrap.registration")
+        owned_posture = importlib.import_module("services.posture.effective")
+
+        lifecycle_thread = MagicMock()
+        with patch(
+            "services.startup_lifecycle.threading.Thread",
+            lifecycle_thread,
+        ):
+            self.assertIs(owned_lifecycle.threading.Thread, lifecycle_thread)
+
+        registration_step = MagicMock()
+        with patch(
+            "services.route_bootstrap._do_full_registration",
+            registration_step,
+        ):
+            self.assertIs(
+                owned_registration._do_full_registration,
+                registration_step,
+            )
+
+        deployment_report = MagicMock()
+        with patch(
+            "services.effective_security_posture._deployment_report",
+            deployment_report,
+        ):
+            self.assertIs(owned_posture._deployment_report, deployment_report)
+
+    def test_compatibility_facades_are_bounded_module_aliases(self):
+        expected_imports = {
+            "startup_lifecycle.py": "from .bootstrap import lifecycle",
+            "route_bootstrap.py": "from .bootstrap import registration",
+            "effective_security_posture.py": "from .posture import effective",
+        }
+        allowed_statement_types = {
+            ast.Expr,
+            ast.Import,
+            ast.ImportFrom,
+            ast.If,
+            ast.Assign,
+        }
+
+        for filename, implementation_suffix in expected_imports.items():
+            with self.subTest(filename=filename):
+                path = ROOT / "services" / filename
+                tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+                self.assertTrue(
+                    all(
+                        type(statement) in allowed_statement_types
+                        for statement in tree.body
+                    )
+                )
+                imports = [
+                    node
+                    for node in ast.walk(tree)
+                    if isinstance(node, (ast.Import, ast.ImportFrom))
+                ]
+                rendered = "\n".join(ast.unparse(node) for node in imports)
+                self.assertIn(implementation_suffix, rendered)
+                self.assertIn("sys", rendered)
+                self.assertLessEqual(len(tree.body), 7)
+                type_only_blocks = [
+                    statement
+                    for statement in tree.body
+                    if isinstance(statement, ast.If)
+                ]
+                self.assertEqual(len(type_only_blocks), 1)
+                self.assertIsInstance(type_only_blocks[0].test, ast.Name)
+                self.assertEqual(type_only_blocks[0].test.id, "TYPE_CHECKING")
+
+    def test_package_initializers_are_navigation_only(self):
+        for relative_path in (
+            "services/bootstrap/__init__.py",
+            "services/posture/__init__.py",
+        ):
+            with self.subTest(path=relative_path):
+                path = ROOT / relative_path
+                tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+                calls = [node for node in ast.walk(tree) if isinstance(node, ast.Call)]
+                self.assertEqual(calls, [])
+
+
+if __name__ == "__main__":
+    unittest.main()