shaibu01/Titan-Engine
0
1#!/usr/bin/env python32"""Run Titan's autonomous cloud core independently of the Streamlit session.3 4Hugging Face can keep a Space container alive without an attached browser5session. The execution/provider core therefore must be owned by the container6process, not by Streamlit's per-session script lifecycle.7"""8 9from __future__ import annotations10 11import os12import json13import signal14import shutil15import subprocess16import sys17import time18from pathlib import Path19from typing import IO, Dict, List, Optional20 21 22ROOT = Path(__file__).resolve().parent23RESTART_FLOOR_SEC = 3.024RESTART_CEILING_SEC = 30.025 26 27def _writable_dir(*candidates: Path) -> Path:28 for candidate in candidates:29 try:30 candidate.mkdir(parents=True, exist_ok=True)31 probe = candidate / ".write_probe"32 probe.write_text("ok", encoding="utf-8")33 probe.unlink(missing_ok=True)34 return candidate35 except Exception:36 continue37 fallback = ROOT / ".titan_runtime"38 fallback.mkdir(parents=True, exist_ok=True)39 return fallback40 41 42def durable_root() -> Path:43 configured = os.getenv("TITAN_SPACE_DURABLE_ROOT", "").strip()44 candidates: List[Path] = []45 if configured:46 candidates.append(Path(configured))47 candidates.extend([Path("/data/titan_engine"), ROOT / ".titan_engine_storage", Path("/tmp/titan_engine")])48 return _writable_dir(*candidates)49 50 51def _truthy_env(name: str, default: bool = False) -> bool:52 text = str(os.getenv(name, "1" if default else "0")).strip().lower()53 if text in {"1", "true", "yes", "on", "enabled"}:54 return True55 if text in {"0", "false", "no", "off", "disabled"}:56 return False57 return bool(default)58 59 60def _fresh_start_scrub_default(root: Path) -> bool:61 root_text = str(root)62 hf_runtime = bool(os.getenv("SPACE_ID") or os.getenv("SPACE_HOST") or os.getenv("HF_SPACE_ID"))63 return hf_runtime or root_text == "/data" or root_text.startswith("/data/")64 65 66def fresh_start_scrub_once(root: Path) -> Dict[str, object]:67 scrub_id = os.getenv("TITAN_CLOUD_FRESH_START_SCRUB_ID", "ops_stability_20260826").strip() or "ops_stability_20260826"68 enabled = _truthy_env("TITAN_CLOUD_FRESH_START_SCRUB_ONCE", _fresh_start_scrub_default(root))69 marker = root / f".fresh_start_scrub_{scrub_id}.done"70 report_path = root / "fresh_start_scrub_report.json"71 report: Dict[str, object] = {72 "enabled": bool(enabled),73 "scrub_id": scrub_id,74 "root": str(root),75 "marker": str(marker),76 "deleted": [],77 "errors": [],78 "ts": time.time(),79 }80 if not enabled:81 report["skipped"] = "disabled"82 return report83 if marker.exists():84 report["skipped"] = "already_scrubbed"85 return report86 87 targets = [88 root / "runtime",89 root / "brain_models",90 root / "brain_models_staging",91 root / "storage",92 root / "mlflow",93 root / "logs",94 root / "cache",95 root / "tmp",96 root / "paper_trade_unlock.json",97 ROOT / "runtime",98 ROOT / "brain_models",99 ROOT / "brain_models_staging",100 ROOT / ".titan_engine_storage",101 ROOT / ".pytest_cache",102 ROOT / "__pycache__",103 ROOT / "neural_forge_status.json",104 ROOT / "neural_trainer_boot.log",105 ROOT / "titan_core.log",106 ]107 for pattern in ("*.log", "*.jsonl", "*.tmp", "*.bak", "*.bak_*", "*.sqlite", "*.db"):108 targets.extend(ROOT.glob(pattern))109 110 seen = set()111 root_resolved = str(root.resolve())112 for target in targets:113 try:114 resolved = str(target.resolve()) if target.exists() else str(target)115 if resolved in seen or resolved == str(marker) or resolved == root_resolved:116 continue117 seen.add(resolved)118 if target.is_dir():119 shutil.rmtree(target)120 report["deleted"].append(resolved)121 elif target.exists():122 target.unlink()123 report["deleted"].append(resolved)124 except Exception as exc:125 report["errors"].append({"path": str(target), "error": str(exc)})126 127 for required in (root / "runtime", root / "brain_models", root / "storage", root / "mlflow", root / "cache", root / "logs"):128 try:129 required.mkdir(parents=True, exist_ok=True)130 except Exception as exc:131 report["errors"].append({"path": str(required), "error": str(exc)})132 try:133 report_path.write_text(json.dumps(report, indent=2, sort_keys=True), encoding="utf-8")134 marker.write_text(json.dumps({"scrub_id": scrub_id, "ts": time.time()}, indent=2), encoding="utf-8")135 except Exception as exc:136 report["errors"].append({"path": str(report_path), "error": str(exc)})137 return report138 139 140DURABLE_ROOT = durable_root()141FRESH_START_SCRUB_REPORT = fresh_start_scrub_once(DURABLE_ROOT)142LOG_DIR = _writable_dir(DURABLE_ROOT / "logs", ROOT)143 144 145def child_environment() -> Dict[str, str]:146 env = os.environ.copy()147 runtime_dir = DURABLE_ROOT / "runtime"148 models_dir = DURABLE_ROOT / "brain_models"149 storage_root = DURABLE_ROOT / "storage"150 cache_dir = DURABLE_ROOT / "cache"151 mlflow_dir = DURABLE_ROOT / "mlflow"152 env.update(153 {154 "PYTHONUNBUFFERED": "1",155 "TITAN_CORE_SUPERVISED": "1",156 # The dashboard must never create a second execution core.157 "TITAN_ENABLE_BACKGROUND_CORE": "0",158 "TITAN_SPACE_DURABLE_ROOT": str(DURABLE_ROOT),159 "TITAN_RUNTIME_DIR": str(runtime_dir),160 "TITAN_MODELS_DIR": str(models_dir),161 "TITAN_STORAGE_LOCAL_ROOT": str(storage_root),162 "TITAN_MLFLOW_DIR": str(mlflow_dir),163 "TITAN_TRAINER_STATUS_FILE": str(mlflow_dir / "neural_forge_status.json"),164 "TITAN_OPTUNA_EVENTS_FILE": str(mlflow_dir / "optuna_trials.jsonl"),165 "TITAN_TRAINER_LOG_FILE": str(LOG_DIR / "neural_trainer_boot.log"),166 "TITAN_CORE_LOG_FILE": str(LOG_DIR / "titan_core.log"),167 "TITAN_ALT_DATA_LOG_FILE": str(LOG_DIR / "titan_alt_data.log"),168 "TITAN_ALPACA_LOG_FILE": str(LOG_DIR / "alpaca_direct_feed.log"),169 "TITAN_STREAMLIT_LOG_FILE": str(LOG_DIR / "streamlit.log"),170 "TITAN_PAPER_PROBATION_FILE": env.get("TITAN_PAPER_PROBATION_FILE", str(runtime_dir / "paper_trade_unlock.json")),171 "HF_HOME": str(cache_dir / "hf_home"),172 "TRANSFORMERS_CACHE": str(cache_dir / "transformers"),173 "MPLCONFIGDIR": str(cache_dir / "matplotlib"),174 "TITAN_PROVIDER_ISOLATED_ACCOUNTING": env.get("TITAN_PROVIDER_ISOLATED_ACCOUNTING", "1"),175 "TITAN_ALPACA_INDEPENDENT_MODE": env.get("TITAN_ALPACA_INDEPENDENT_MODE", "1"),176 # PU Prime remains connected for telemetry and close-only risk177 # management, but cannot receive new exposure until its sizing and178 # duplicate-entry behavior has completed postmortem validation.179 # PU Prime is quarantined for this release. Keep telemetry, exits,180 # and the isolated $100 shadow audit active without live entries.181 "TITAN_CLOUD_PUPRIME_ENTRY_ENABLED": "0",182 "TITAN_CLOUD_ALPACA_ENTRY_ENABLED": env.get("TITAN_CLOUD_ALPACA_ENTRY_ENABLED", "1"),183 "TITAN_NEURAL_RETRAIN_REQUIRE_FLAT": env.get("TITAN_NEURAL_RETRAIN_REQUIRE_FLAT", "1"),184 "TITAN_REQUIRE_NEURAL_FOR_ENTRIES": env.get("TITAN_REQUIRE_NEURAL_FOR_ENTRIES", "1"),185 "TITAN_REQUIRE_PAPER_PROBATION": env.get("TITAN_REQUIRE_PAPER_PROBATION", "1"),186 "TITAN_PAPER_PROBATION_REQUIRED_WINS": env.get("TITAN_PAPER_PROBATION_REQUIRED_WINS", "3"),187 "TITAN_PAPER_PROBATION_MIN_NET_PNL_PCT": env.get("TITAN_PAPER_PROBATION_MIN_NET_PNL_PCT", "0.01"),188 "TITAN_PAPER_PROBATION_SLIPPAGE_BPS": env.get("TITAN_PAPER_PROBATION_SLIPPAGE_BPS", "8"),189 "TITAN_PAPER_PROBATION_FEE_BPS": env.get("TITAN_PAPER_PROBATION_FEE_BPS", "2"),190 "TITAN_CLOUD_FRESH_START_SCRUB_REPORT": str(DURABLE_ROOT / "fresh_start_scrub_report.json"),191 }192 )193 for path in (194 runtime_dir,195 models_dir,196 storage_root,197 mlflow_dir,198 cache_dir,199 cache_dir / "hf_home",200 cache_dir / "transformers",201 cache_dir / "matplotlib",202 ):203 path.mkdir(parents=True, exist_ok=True)204 return env205 206 207class Child:208 def __init__(self, name: str, command: List[str], log_name: str) -> None:209 self.name = name210 self.command = command211 self.log_path = LOG_DIR / log_name212 self.process: Optional[subprocess.Popen] = None213 self.log_handle: Optional[IO[bytes]] = None214 self.restart_delay = RESTART_FLOOR_SEC215 self.last_start = 0.0216 217 def start(self, env: Dict[str, str]) -> None:218 self.close_log()219 self.log_handle = self.log_path.open("ab", buffering=0)220 self.process = subprocess.Popen(221 self.command,222 cwd=str(ROOT),223 env=env,224 stdout=self.log_handle,225 stderr=subprocess.STDOUT,226 start_new_session=True,227 )228 self.last_start = time.monotonic()229 print(f"[SUPERVISOR] started {self.name} pid={self.process.pid}", flush=True)230 231 def restart_if_needed(self, env: Dict[str, str]) -> None:232 if self.process is not None and self.process.poll() is None:233 return234 return_code = None if self.process is None else self.process.returncode235 alive_for = max(0.0, time.monotonic() - self.last_start)236 if alive_for >= 60.0:237 self.restart_delay = RESTART_FLOOR_SEC238 print(239 f"[SUPERVISOR] {self.name} exited rc={return_code}; "240 f"restart in {self.restart_delay:.1f}s",241 flush=True,242 )243 time.sleep(self.restart_delay)244 self.restart_delay = min(RESTART_CEILING_SEC, self.restart_delay * 2.0)245 self.start(env)246 247 def stop(self) -> None:248 process = self.process249 if process is not None and process.poll() is None:250 try:251 os.killpg(process.pid, signal.SIGTERM)252 process.wait(timeout=10.0)253 except Exception:254 try:255 os.killpg(process.pid, signal.SIGKILL)256 except Exception:257 pass258 self.close_log()259 260 def close_log(self) -> None:261 if self.log_handle is not None:262 try:263 self.log_handle.close()264 except Exception:265 pass266 self.log_handle = None267 268 269def brain_artifacts_status(models_dir: Path) -> Dict[str, object]:270 try:271 models_dir.mkdir(parents=True, exist_ok=True)272 except Exception:273 pass274 experts = 0275 for idx in range(1, 51):276 if (models_dir / f"expert_{idx}_gru.pth").exists():277 experts += 1278 params_ready = (models_dir / "optimal_params.json").exists()279 return {280 "experts": int(experts),281 "params_ready": bool(params_ready),282 "ready": bool(params_ready and experts >= 50),283 "models_dir": str(models_dir),284 }285 286 287def _tail_log(path: Path, lines: int = 24) -> str:288 try:289 if not path.exists():290 return ""291 return "".join(path.read_text(encoding="utf-8", errors="ignore").splitlines(True)[-int(lines):])[-4000:]292 except Exception as exc:293 return f"unable to read trainer log: {exc}"294 295 296def _write_training_supervisor_status(env: Dict[str, str], stage: str, **payload: object) -> None:297 path = Path(env.get("TITAN_TRAINER_STATUS_FILE", str(DURABLE_ROOT / "mlflow" / "neural_forge_status.json")))298 try:299 path.parent.mkdir(parents=True, exist_ok=True)300 status = {301 "stage": str(stage),302 "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),303 "supervisor_pid": os.getpid(),304 **payload,305 }306 tmp = path.with_suffix(path.suffix + ".tmp")307 tmp.write_text(json.dumps(status, indent=2, sort_keys=True), encoding="utf-8")308 tmp.replace(path)309 except Exception as exc:310 print(f"[SUPERVISOR] trainer status write failed: {exc}", flush=True)311 312 313class TrainerChild(Child):314 def __init__(self, name: str, command: List[str], log_name: str) -> None:315 super().__init__(name, command, log_name)316 self.last_idle_status = 0.0317 318 def _write_idle_status(self, env: Dict[str, str], models: Dict[str, object], *, reason: str) -> None:319 now = time.time()320 interval = max(30.0, float(env.get("TITAN_TRAINER_IDLE_STATUS_SEC", os.getenv("TITAN_TRAINER_IDLE_STATUS_SEC", "300")) or 300))321 if now - self.last_idle_status < interval:322 return323 self.last_idle_status = now324 _write_training_supervisor_status(325 env,326 "COMPLETE_IDLE",327 forged_experts=models["experts"],328 params_ready=models["params_ready"],329 models_dir=models["models_dir"],330 trainer_running=False,331 idle=True,332 next_training_check_sec=interval,333 reason=reason,334 )335 336 def start(self, env: Dict[str, str]) -> None:337 models = brain_artifacts_status(Path(env["TITAN_MODELS_DIR"]))338 force = _truthy_env("TITAN_FORCE_RETRAIN", False)339 if models["ready"] and not force:340 self._write_idle_status(env, models, reason="complete 50-expert brain already exists; trainer is intentionally stopped")341 print("[SUPERVISOR] neural-trainer skipped; 50-expert brain already exists", flush=True)342 return343 _write_training_supervisor_status(344 env,345 "SUPERVISOR_STARTING_TRAINER",346 forged_experts=models["experts"],347 params_ready=models["params_ready"],348 models_dir=models["models_dir"],349 reason="brain incomplete; trainer child is required",350 )351 super().start(env)352 353 def restart_if_needed(self, env: Dict[str, str]) -> None:354 if self.process is not None and self.process.poll() is None:355 return356 models = brain_artifacts_status(Path(env["TITAN_MODELS_DIR"]))357 if models["ready"]:358 self._write_idle_status(env, models, reason="trainer exited after complete brain artifacts were created; idle until retrain is requested")359 return360 return_code = None if self.process is None else self.process.returncode361 _write_training_supervisor_status(362 env,363 "TRAINER_EXITED_WITHOUT_COMPLETE_SWARM",364 forged_experts=models["experts"],365 params_ready=models["params_ready"],366 models_dir=models["models_dir"],367 return_code=return_code,368 reason=f"neural trainer exited before 50 experts were ready; restart in {self.restart_delay:.1f}s",369 recent_log=_tail_log(self.log_path),370 )371 super().restart_if_needed(env)372 373 374def main() -> int:375 env = child_environment()376 children = [377 TrainerChild("neural-trainer", [sys.executable, "-u", "neural_trainer.py"], "neural_trainer_boot.log"),378 Child("titan-core", [sys.executable, "-u", "titan_core.py"], "titan_core.log"),379 Child("alt-data", [sys.executable, "-u", "titan_alt_data.py"], "titan_alt_data.log"),380 ]381 streamlit = Child(382 "streamlit",383 [384 sys.executable,385 "-m",386 "streamlit",387 "run",388 "app.py",389 "--server.address=0.0.0.0",390 "--server.port=7860",391 "--server.headless=true",392 ],393 "streamlit.log",394 )395 stopping = False396 397 def request_stop(signum, _frame) -> None:398 nonlocal stopping399 print(f"[SUPERVISOR] received signal {signum}; shutting down", flush=True)400 stopping = True401 402 signal.signal(signal.SIGTERM, request_stop)403 signal.signal(signal.SIGINT, request_stop)404 405 try:406 for child in children:407 child.start(env)408 streamlit.start(env)409 while not stopping:410 streamlit.restart_if_needed(env)411 for child in children:412 child.restart_if_needed(env)413 time.sleep(2.0)414 finally:415 streamlit.stop()416 for child in reversed(children):417 child.stop()418 return 0419 420 421if __name__ == "__main__":422 raise SystemExit(main())423