"""Fail-closed supervisor for the on-demand Hugging Face cloud core. This process is intentionally the container entrypoint. It restores a fresh runtime database, obtains the cross-host continuity fence, publishes the Space endpoint, and only then starts Alpecca's existing FastAPI application. """ from __future__ import annotations from collections.abc import Callable, Mapping, MutableMapping, Sequence from dataclasses import dataclass from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import re import shutil import signal import subprocess import sys import tempfile import threading import time from typing import Any from urllib.parse import urlsplit REPO_ROOT = Path( os.environ.get("ALPECCA_SOURCE_ROOT", str(Path(__file__).resolve().parents[2])) ).expanduser().resolve() if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from alpecca.continuity_lease import ( # noqa: E402 ContinuityLeaseError, ContinuityLeaseGuard, client_from_env, ) MODEL_ID = "Qwen/Qwen3.5-9B" CONTINUITY_ROLE = "cloud-standby" APP_PORT = 7860 LEASE_RENEW_SECONDS = 10.0 LEASE_LOST_EXIT = 75 MAX_APPROVAL_SECONDS = 5 * 60 HARD_KILL_SIGNAL = getattr(signal, "SIGKILL", signal.SIGTERM) _NODE_CHARS = re.compile(r"[^A-Za-z0-9._:-]+") _APPROVAL_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") _REQUIRED_VALUES = ( "ALPECCA_MINDSCAPE_VAULT_URL", "ALPECCA_MINDSCAPE_VAULT_TOKEN", "ALPECCA_MINDSCAPE_VAULT_KEY", "ALPECCA_CONTINUITY_LEASE_URL", "ALPECCA_CONTINUITY_LEASE_TOKEN", "ALPECCA_AUTH_SECRET", "ALPECCA_CREATOR_PASSWORD", ) _TRUE_VALUES = {"1", "true", "yes", "on"} class CloudCoreStartupError(RuntimeError): """A precondition failed before CoreMind was allowed to start.""" def __init__(self, code: str) -> None: super().__init__(code) self.code = code @dataclass(frozen=True, slots=True) class RestoreReceipt: path: Path sequence: int created_at: str sha256: str @dataclass(frozen=True, slots=True) class RestoreApproval: approval_id: str snapshot_digest: str lease_epoch: int issued_at: datetime expires_at: datetime evidence_id: str def _clean_status(value: object) -> str: text = re.sub(r"[^a-z0-9_-]+", "-", str(value or "unknown").lower()) return text.strip("-")[:64] or "unknown" def _approval_time(value: object, field: str) -> datetime: if not isinstance(value, str) or not value: raise CloudCoreStartupError(f"restore_approval_{field}_invalid") candidate = value[:-1] + "+00:00" if value.endswith("Z") else value try: parsed = datetime.fromisoformat(candidate) except ValueError as exc: raise CloudCoreStartupError(f"restore_approval_{field}_invalid") from exc if parsed.tzinfo is None: raise CloudCoreStartupError(f"restore_approval_{field}_invalid") return parsed.astimezone(timezone.utc) def load_restore_approval( environ: Mapping[str, str], receipt: RestoreReceipt, *, now: datetime | None = None, ) -> RestoreApproval: """Validate an external CreatorJD approval against the restored archive.""" raw = str(environ.get("ALPECCA_CLOUD_RESTORE_APPROVAL") or "").strip() if not raw: raise CloudCoreStartupError("explicit_restore_approval_required") if len(raw.encode("utf-8")) > 8 * 1024: raise CloudCoreStartupError("restore_approval_invalid") try: value = json.loads(raw) except json.JSONDecodeError as exc: raise CloudCoreStartupError("restore_approval_invalid") from exc expected_keys = { "approvalId", "purpose", "creatorPrincipal", "snapshotDigest", "leaseEpoch", "issuedAt", "expiresAt", "oneUse", "verification", } if not isinstance(value, dict) or set(value) != expected_keys: raise CloudCoreStartupError("restore_approval_invalid") verification = value.get("verification") if not isinstance(verification, dict) or set(verification) != { "status", "verifier", "evidenceId", }: raise CloudCoreStartupError("restore_approval_invalid") approval_id = value.get("approvalId") evidence_id = verification.get("evidenceId") if not isinstance(approval_id, str) or not _APPROVAL_IDENTIFIER.fullmatch(approval_id): raise CloudCoreStartupError("restore_approval_id_invalid") if not isinstance(evidence_id, str) or not _APPROVAL_IDENTIFIER.fullmatch(evidence_id): raise CloudCoreStartupError("restore_approval_evidence_invalid") if value.get("purpose") != "stage-passive-restore": raise CloudCoreStartupError("restore_approval_purpose_mismatch") if value.get("creatorPrincipal") != "CreatorJD": raise CloudCoreStartupError("restore_approval_creator_mismatch") expected_digest = f"sha256:{receipt.sha256}" if value.get("snapshotDigest") != expected_digest: raise CloudCoreStartupError("restore_approval_digest_mismatch") lease_epoch = value.get("leaseEpoch") if isinstance(lease_epoch, bool) or not isinstance(lease_epoch, int) or lease_epoch < 1: raise CloudCoreStartupError("restore_approval_lease_epoch_invalid") if value.get("oneUse") is not True: raise CloudCoreStartupError("restore_approval_not_one_use") if verification.get("status") != "verified": raise CloudCoreStartupError("restore_approval_unverified") if verification.get("verifier") != "external-creator-verifier": raise CloudCoreStartupError("restore_approval_verifier_invalid") issued_at = _approval_time(value.get("issuedAt"), "issued_at") expires_at = _approval_time(value.get("expiresAt"), "expires_at") current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) if expires_at <= issued_at or (expires_at - issued_at).total_seconds() > MAX_APPROVAL_SECONDS: raise CloudCoreStartupError("restore_approval_window_invalid") if issued_at > current: raise CloudCoreStartupError("restore_approval_not_yet_valid") if current >= expires_at: raise CloudCoreStartupError("restore_approval_expired") return RestoreApproval( approval_id=approval_id, snapshot_digest=expected_digest, lease_epoch=lease_epoch, issued_at=issued_at, expires_at=expires_at, evidence_id=evidence_id, ) def require_current_approval(approval: RestoreApproval) -> None: if datetime.now(timezone.utc) >= approval.expires_at: raise CloudCoreStartupError("restore_approval_expired") def automatic_failover_authorized(environ: Mapping[str, str]) -> bool: """Return whether deployment policy authorizes unattended lease takeover.""" return str(environ.get("ALPECCA_CLOUD_AUTO_FAILOVER") or "").strip().lower() in _TRUE_VALUES def resolve_public_endpoint(environ: Mapping[str, str]) -> str: """Return the exact HTTPS origin that the lease authority may publish.""" configured = str(environ.get("ALPECCA_PUBLIC_ENDPOINT") or "").strip() if configured: candidate = configured else: space_host = str(environ.get("SPACE_HOST") or "").strip() candidate = f"https://{space_host}" if space_host else "" parsed = urlsplit(candidate) if ( parsed.scheme != "https" or not parsed.hostname or parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment or parsed.path not in {"", "/"} ): raise CloudCoreStartupError("public_endpoint_invalid") return f"https://{parsed.netloc}" def configure_runtime( environ: MutableMapping[str, str], runtime_home: Path, endpoint: str, ) -> None: """Pin the cloud runtime before config.py or alpecca.mind is imported.""" forced = { "ALPECCA_HOME": str(runtime_home), "ALPECCA_LLM_BACKEND": "hf", "ALPECCA_HF_MODEL": MODEL_ID, "ALPECCA_MODEL": "qwen3.5:9b", "ALPECCA_FAST_MODEL": "qwen3.5:9b", "ALPECCA_REFLECT_MODEL": "", "ALPECCA_REFLECT_THINK": "0", "ALPECCA_DEEP_BACKEND": "local", "ALPECCA_CHAT_CLOUD_MODEL": "", "ALPECCA_OLLAMA_CLOUD_MODEL": "", "ALPECCA_CHAT_ZEROGPU": "0", "ALPECCA_REMOTE": "1", "ALPECCA_SERVER_HOST": "0.0.0.0", "ALPECCA_SERVER_PORT": str(APP_PORT), "ALPECCA_TUNNEL": "off", "ALPECCA_PUBLIC_URL": endpoint, "ALPECCA_CONTINUITY_ROLE": CONTINUITY_ROLE, # Restore is read-only until Vault writes can validate the exact fence. "ALPECCA_MINDSCAPE": "0", "ALPECCA_MINDSCAPE_VAULT": "0", "ALPECCA_STREAM_CHAT": "0", "ALPECCA_F5_WORKER": "0", "ALPECCA_DISCORD": "0", "ALPECCA_DISCORD_MEDIA": "0", "ALPECCA_DISCORD_VOICE": "0", "ALPECCA_COMPUTER_USE": "0", "ALPECCA_SIGHT": "0", "ALPECCA_FACE": "0", "ALPECCA_VOICE": "0", "ALPECCA_APPS": "", "PYTHONUNBUFFERED": "1", } environ.update(forced) environ.setdefault("ALPECCA_HF_PROVIDER", "auto") hf_token = str( environ.get("HF_TOKEN") or environ.get("HUGGINGFACEHUB_API_TOKEN") or "" ).strip() if hf_token: environ["HF_TOKEN"] = hf_token configured_node = str(environ.get("ALPECCA_CONTINUITY_NODE_ID") or "").strip() if not configured_node: identity = str(environ.get("SPACE_ID") or urlsplit(endpoint).hostname or "space") identity = _NODE_CHARS.sub("-", identity).strip("-._:") or "space" configured_node = f"hf-cloud-standby:{identity}"[:96] environ["ALPECCA_CONTINUITY_NODE_ID"] = configured_node def validate_runtime_configuration(environ: Mapping[str, str]) -> None: missing = [name for name in _REQUIRED_VALUES if not str(environ.get(name) or "").strip()] if not str(environ.get("HF_TOKEN") or "").strip(): missing.append("HF_TOKEN") if missing: raise CloudCoreStartupError("missing_configuration:" + ",".join(sorted(missing))) if len(str(environ["ALPECCA_AUTH_SECRET"]).encode("utf-8")) < 32: raise CloudCoreStartupError("authorization_secret_too_short") if len(str(environ["ALPECCA_CREATOR_PASSWORD"])) < 12: raise CloudCoreStartupError("creator_password_too_short") if str(environ.get("ALPECCA_CONTINUITY_ROLE")) != CONTINUITY_ROLE: raise CloudCoreStartupError("continuity_role_invalid") def create_runtime_home(environ: Mapping[str, str]) -> Path: root = Path( str(environ.get("ALPECCA_CLOUD_RUNTIME_ROOT") or tempfile.gettempdir()) ).expanduser().resolve() root.mkdir(parents=True, exist_ok=True) runtime_home = Path(tempfile.mkdtemp(prefix="alpecca-hf-core-", dir=root)) try: runtime_home.chmod(0o700) except OSError: pass return runtime_home def _file_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def restore_latest_verified_archive( environ: Mapping[str, str], runtime_home: Path, *, vault_module: Any | None = None, ) -> RestoreReceipt: """Restore the latest authenticated Vault archive into a new runtime DB.""" if vault_module is None: from alpecca import mindscape_vault as vault_module destination = (runtime_home / "alpecca.db").resolve() if destination.exists(): raise CloudCoreStartupError("restore_destination_not_fresh") try: recovery_key, _key_source = vault_module.load_or_create_encryption_key(environ) transport_token, _token_source = vault_module.load_or_create_transport_token(environ) except Exception as exc: raise CloudCoreStartupError("vault_credentials_invalid") from exc try: timeout = float(environ.get("ALPECCA_CLOUD_RESTORE_TIMEOUT", "30")) except (TypeError, ValueError) as exc: raise CloudCoreStartupError("restore_timeout_invalid") from exc timeout = max(5.0, min(120.0, timeout)) result = vault_module.fetch_latest_archive( str(environ["ALPECCA_MINDSCAPE_VAULT_URL"]), transport_token, recovery_key, destination, timeout=timeout, ) if not isinstance(result, Mapping) or result.get("ok") is not True: status = result.get("status") if isinstance(result, Mapping) else "invalid_result" raise CloudCoreStartupError(f"vault_restore_failed:{_clean_status(status)}") try: restored_path = Path(str(result["path"])).resolve() sequence = int(result["sequence"]) created_at = str(result["created_at"]) except (KeyError, TypeError, ValueError) as exc: raise CloudCoreStartupError("vault_restore_receipt_invalid") from exc if restored_path != destination or sequence < 0 or not destination.is_file(): raise CloudCoreStartupError("vault_restore_receipt_invalid") return RestoreReceipt( path=destination, sequence=sequence, created_at=created_at, sha256=_file_sha256(destination), ) def server_command() -> list[str]: return [ sys.executable, "-m", "uvicorn", "server:app", "--host", "0.0.0.0", "--port", str(APP_PORT), "--log-level", "warning", "--no-access-log", ] class CloudCoreSupervisor: def __init__( self, *, environ: MutableMapping[str, str] | None = None, vault_module: Any | None = None, lease_client_factory: Callable[..., Any] = client_from_env, lease_guard_factory: Callable[..., Any] = ContinuityLeaseGuard, process_factory: Callable[..., Any] = subprocess.Popen, runtime_home_factory: Callable[[Mapping[str, str]], Path] = create_runtime_home, vrm_installer: Callable[[Path], Path] | None = None, sleep: Callable[[float], None] = time.sleep, ) -> None: self.environ = os.environ if environ is None else environ self.vault_module = vault_module self.lease_client_factory = lease_client_factory self.lease_guard_factory = lease_guard_factory self.process_factory = process_factory self.runtime_home_factory = runtime_home_factory self.vrm_installer = vrm_installer self.sleep = sleep self._shutdown = threading.Event() self._lease_lost = threading.Event() self._child_lock = threading.Lock() self._child: Any | None = None def request_shutdown(self, *_args: object) -> None: self._shutdown.set() @property def shutdown_requested(self) -> bool: """Whether the container asked this promotion attempt to stop.""" return self._shutdown.is_set() def install_signal_handlers(self) -> None: signal.signal(signal.SIGTERM, self.request_shutdown) signal.signal(signal.SIGINT, self.request_shutdown) def _handle_lease_loss(self, _reason: str) -> None: self._lease_lost.set() with self._child_lock: child = self._child if child is not None and child.poll() is None: self._signal_child(child, HARD_KILL_SIGNAL, "kill") def _spawn_child(self, guard: Any) -> Any: with self._child_lock: if self._lease_lost.is_set() or not guard.active: raise CloudCoreStartupError("lease_lost_before_start") child = self.process_factory( server_command(), cwd=str(REPO_ROOT), env=dict(self.environ), start_new_session=True, ) self._child = child return child @staticmethod def _signal_child(child: Any, sig: int, fallback: str) -> None: pid = getattr(child, "pid", None) if os.name == "posix" and isinstance(pid, int) and pid > 0: try: os.killpg(pid, sig) return except (OSError, ProcessLookupError): pass getattr(child, fallback)() @staticmethod def _stop_child(child: Any, *, graceful: bool) -> None: if child.poll() is not None: return if graceful: CloudCoreSupervisor._signal_child(child, signal.SIGTERM, "terminate") else: CloudCoreSupervisor._signal_child(child, HARD_KILL_SIGNAL, "kill") try: child.wait(timeout=12.0 if graceful else 3.0) except subprocess.TimeoutExpired: CloudCoreSupervisor._signal_child(child, HARD_KILL_SIGNAL, "kill") child.wait(timeout=3.0) def run(self) -> int: endpoint = resolve_public_endpoint(self.environ) runtime_home = self.runtime_home_factory(self.environ) guard: Any | None = None child: Any | None = None try: configure_runtime(self.environ, runtime_home, endpoint) validate_runtime_configuration(self.environ) if self.vrm_installer is not None: installed_vrm = self.vrm_installer(runtime_home) if not isinstance(installed_vrm, Path) or not installed_vrm.is_file(): raise CloudCoreStartupError("locked_vrm_install_failed") receipt = restore_latest_verified_archive( self.environ, runtime_home, vault_module=self.vault_module, ) print( "[hf-cloud-core] verified Vault restore " f"sequence={receipt.sequence} sha256={receipt.sha256}" ) approval: RestoreApproval | None if str(self.environ.get("ALPECCA_CLOUD_RESTORE_APPROVAL") or "").strip(): approval = load_restore_approval(self.environ, receipt) elif automatic_failover_authorized(self.environ): approval = None print( "[hf-cloud-core] unattended promotion authorized by the " "deployment failover policy" ) else: approval = load_restore_approval(self.environ, receipt) client = self.lease_client_factory(role=CONTINUITY_ROLE) if client is None: raise CloudCoreStartupError("continuity_lease_not_configured") guard = self.lease_guard_factory( client, renew_seconds=LEASE_RENEW_SECONDS, endpoint="", on_loss=self._handle_lease_loss, ) grant = guard.start() if self._lease_lost.is_set() or not guard.active: raise CloudCoreStartupError("lease_lost_before_endpoint_publication") if approval is not None: if approval.lease_epoch != grant.fencing_epoch: raise CloudCoreStartupError("restore_approval_lease_epoch_mismatch") require_current_approval(approval) published = client.publish_endpoint(grant, endpoint) if not isinstance(published, Mapping) or published.get("ok") is not True: raise CloudCoreStartupError("lease_endpoint_publication_rejected") if self._lease_lost.is_set() or not guard.active: raise CloudCoreStartupError("lease_lost_before_start") if approval is not None: require_current_approval(approval) guard.endpoint = endpoint self.environ["ALPECCA_CONTINUITY_LEASE_ID"] = str(grant.lease_id) self.environ["ALPECCA_CONTINUITY_FENCING_EPOCH"] = str(grant.fencing_epoch) self.environ["ALPECCA_CONTINUITY_LEASE_HOLDER"] = str(grant.holder) self.environ["ALPECCA_RESTORE_APPROVAL_ID"] = ( approval.approval_id if approval is not None else "deployment-auto-failover-v1" ) self.environ.pop("ALPECCA_CLOUD_RESTORE_APPROVAL", None) child = self._spawn_child(guard) print( "[hf-cloud-core] cloud core started under continuity fence " f"epoch={grant.fencing_epoch}" ) while True: if self._lease_lost.is_set() or not guard.active: self._stop_child(child, graceful=False) return LEASE_LOST_EXIT if self._shutdown.is_set(): self._stop_child(child, graceful=True) return int(child.poll() or 0) exit_code = child.poll() if exit_code is not None: return int(exit_code) self.sleep(0.25) finally: try: if child is not None and child.poll() is None: self._stop_child(child, graceful=not self._lease_lost.is_set()) finally: try: if guard is not None: guard.stop() finally: with self._child_lock: self._child = None shutil.rmtree(runtime_home, ignore_errors=True) def run_supervisor_once( _argv: Sequence[str] | None = None, *, vrm_installer: Callable[[Path], Path] | None = None, ) -> tuple[int, bool]: """Run one fenced promotion attempt and report if shutdown caused it.""" supervisor = CloudCoreSupervisor(vrm_installer=vrm_installer) supervisor.install_signal_handlers() try: result = supervisor.run() except CloudCoreStartupError as exc: print(f"[hf-cloud-core] startup blocked: {exc.code}", file=sys.stderr) result = 2 except ContinuityLeaseError: print("[hf-cloud-core] startup blocked: continuity_lease_unavailable", file=sys.stderr) result = 3 except Exception as exc: print( f"[hf-cloud-core] startup blocked: unexpected_{type(exc).__name__}", file=sys.stderr, ) result = 1 return result, supervisor.shutdown_requested def main( _argv: Sequence[str] | None = None, *, vrm_installer: Callable[[Path], Path] | None = None, ) -> int: result, _shutdown_requested = run_supervisor_once( _argv, vrm_installer=vrm_installer, ) return result if __name__ == "__main__": raise SystemExit(main())