diff --git a/docs/architecture/service_domain_packages.md b/docs/architecture/service_domain_packages.md new file mode 100644 index 0000000..b200ffc --- /dev/null +++ b/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. diff --git a/scripts/verify_production_dependencies.py b/scripts/verify_production_dependencies.py index 648ef36..250353c 100644 --- a/scripts/verify_production_dependencies.py +++ b/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: diff --git a/services/bootstrap/__init__.py b/services/bootstrap/__init__.py new file mode 100644 index 0000000..bbd7b16 --- /dev/null +++ b/services/bootstrap/__init__.py @@ -0,0 +1,4 @@ +# ruff: noqa: N999 +"""Bootstrap lifecycle and registration implementation package.""" + +__all__ = ["lifecycle", "registration"] diff --git a/services/bootstrap/lifecycle.py b/services/bootstrap/lifecycle.py new file mode 100644 index 0000000..f1c6064 --- /dev/null +++ b/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, + ) diff --git a/services/bootstrap/registration.py b/services/bootstrap/registration.py new file mode 100644 index 0000000..4c0e18a --- /dev/null +++ b/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() diff --git a/services/effective_security_posture.py b/services/effective_security_posture.py index 1d7f091..17fbb61 100644 --- a/services/effective_security_posture.py +++ b/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 diff --git a/services/posture/__init__.py b/services/posture/__init__.py new file mode 100644 index 0000000..31006e7 --- /dev/null +++ b/services/posture/__init__.py @@ -0,0 +1,4 @@ +# ruff: noqa: N999 +"""Process-static security posture implementation package.""" + +__all__ = ["effective"] diff --git a/services/posture/effective.py b/services/posture/effective.py new file mode 100644 index 0000000..a18ffd9 --- /dev/null +++ b/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), + } diff --git a/services/route_bootstrap.py b/services/route_bootstrap.py index cc08161..a905b6b 100644 --- a/services/route_bootstrap.py +++ b/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 diff --git a/services/startup_lifecycle.py b/services/startup_lifecycle.py index f1c6064..700f5ae 100644 --- a/services/startup_lifecycle.py +++ b/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 diff --git a/tests/architecture_dependency_policy.json b/tests/architecture_dependency_policy.json index bb78c7b..51bf342 100644 --- a/tests/architecture_dependency_policy.json +++ b/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": [ diff --git a/tests/exception_boundary_policy.json b/tests/exception_boundary_policy.json index c4dca9a..735a020 100644 --- a/tests/exception_boundary_policy.json +++ b/tests/exception_boundary_policy.json @@ -90,7 +90,7 @@ } ] }, - "services/route_bootstrap.py": { + "services/bootstrap/registration.py": { "coverage": "all_broad_catches", "broad_catches": [ { diff --git a/tests/static_analysis_policy.json b/tests/static_analysis_policy.json index fde0667..283619c 100644 --- a/tests/static_analysis_policy.json +++ b/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", diff --git a/tests/test_architecture_dependency_policy.py b/tests/test_architecture_dependency_policy.py index 7de83e5..22ca86e 100644 --- a/tests/test_architecture_dependency_policy.py +++ b/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" diff --git a/tests/test_r180_exception_boundary_governance.py b/tests/test_r180_exception_boundary_governance.py index 168e46f..ab7b30a 100644 --- a/tests/test_r180_exception_boundary_governance.py +++ b/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()) diff --git a/tests/test_r219_exception_boundary_phase2.py b/tests/test_r219_exception_boundary_phase2.py index 2a673c0..7fa7ea1 100644 --- a/tests/test_r219_exception_boundary_phase2.py +++ b/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( diff --git a/tests/test_r233_service_domain_packages.py b/tests/test_r233_service_domain_packages.py new file mode 100644 index 0000000..ffec2b3 --- /dev/null +++ b/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()