#!/usr/bin/env python3 import json import os import sys import subprocess from dataclasses import dataclass, field from pathlib import Path from typing import Optional try: import yaml except ImportError: subprocess.check_call([sys.executable, "-m", "pip", "install", "pyyaml", "-q"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) import yaml # ── Paths ──────────────────────────────────────────────────────── HOME = Path("/home/missminute") APP_DIR = HOME / "app" HERMES_HOME = HOME / ".hermes" OPENCLAW_HOME = HOME / ".openclaw" WORKSPACE = APP_DIR / "workspace" LOG_DIR = Path("/tmp/missminute/logs") LLM_CFG_PATH = APP_DIR / "llm_config.json" REQUIRED_SECRETS = ["TELEGRAM_BOT_TOKEN", "TELEGRAM_OWNER_ID"] @dataclass class MissminuteConfig: telegram_token: str telegram_owner_id: str llm_api_key: str llm_base_url: str llm_model: str github_token: Optional[str] = None git_repo_url: Optional[str] = None allowed_users: list = field(default_factory=list) openclaw_port: int = 18789 def _log(msg: str): print(f"[config_fusion] {msg}", file=sys.stderr) def _load_llm_config() -> dict: if not LLM_CFG_PATH.exists(): _log(f"ERROR: llm_config.json not found at {LLM_CFG_PATH}") sys.exit(1) with open(LLM_CFG_PATH) as f: raw = f.read() import re raw = re.sub(r'"_comment"\s*:\s*"[^"]*",?\s*', '', raw) raw = re.sub(r',\s*}', '}', raw) raw = re.sub(r',\s*]', ']', raw) return json.loads(raw) def _load_dotenv(): env_file = HOME / ".missminute" / ".env" if env_file.exists(): with open(env_file) as f: for line in f: line = line.strip() if line and not line.startswith("#") and "=" in line: k, _, v = line.partition("=") os.environ.setdefault(k.strip(), v.strip()) def resolve_config() -> MissminuteConfig: _load_dotenv() cfg_file = _load_llm_config() primary_key = cfg_file["routing"]["primary"] provider = cfg_file["providers"][primary_key] api_key_env = provider["env_secret_name"] api_key = os.environ.get(api_key_env, "").strip().replace("\n", "").replace("\r", "").replace("'", "").replace('"', '') if not api_key: fallback_key = None for name, prov in cfg_file["providers"].items(): if name != primary_key: fb_key = os.environ.get(prov["env_secret_name"], "").strip().replace("\n", "").replace("\r", "") if fb_key: _log(f"WARNING: Primary key '{api_key_env}' not set — falling back to '{name}'") api_key = fb_key provider = prov fallback_key = name break if not api_key: _log(f"ERROR: No LLM API key found. Set at least one of: {[p['env_secret_name'] for p in cfg_file['providers'].values()]}") sys.exit(1) missing = [k for k in REQUIRED_SECRETS if not os.environ.get(k)] if missing: _log(f"ERROR: Missing required HF Space Secrets: {missing}") sys.exit(1) owner_id_raw = os.environ.get("TELEGRAM_OWNER_ID", "") allowed_users = [] for x in owner_id_raw.split(","): x = x.strip() if x: allowed_users.append(x) try: allowed_users.append(int(x)) except ValueError: pass github_token = os.environ.get("GITHUB_PAT") or os.environ.get("GITHUB_TOKEN") git_repo_url = os.environ.get("GIT_REPO_URL", "").strip() return MissminuteConfig( telegram_token=os.environ["TELEGRAM_BOT_TOKEN"], telegram_owner_id=owner_id_raw, llm_api_key=api_key, llm_base_url=provider["base_url"], llm_model=provider["model"], github_token=github_token, git_repo_url=git_repo_url if git_repo_url else None, allowed_users=allowed_users, ) def fetch_dynamic_memory(cfg: MissminuteConfig) -> str: """Clones external memory, configures Git for auto-push, and sponges all text files.""" base_prompt = ( "You are Missminute, an advanced AI operating system built on the PicoClaw framework. " "Your ephemeral container resets frequently. Your absolute source of truth, identity, and long-term memory " "is stored externally in your GitHub repository, which is cloned locally at: /home/missminute/app/workspace/core_memory\n\n" "CRITICAL IMMORTALITY DIRECTIVE:\n" "Whenever you learn something new, complete a project milestone, or need to remember context for the future, " "you MUST update the markdown files in your `core_memory` directory. " "Before finishing your response to the user, you MUST use your shell tool to navigate to that directory and run:\n" "`git add . && git commit -m 'Auto-memory update' && git push`\n" "If you do not push your changes, your memories will die when the system sleeps.\n\n" "--- YOUR RESTORED MEMORY BANK ---\n" ) if not cfg.git_repo_url: _log("WARNING: GIT_REPO_URL not set. Missminute will boot with a blank slate.") return base_prompt + "No external memory repository configured." repo_dir = WORKSPACE / "core_memory" clean_url = cfg.git_repo_url if not clean_url.startswith("http://") and not clean_url.startswith("https://"): clean_url = f"https://{clean_url}" auth_url = clean_url if cfg.github_token and "://" in clean_url: proto, rest = clean_url.split("://", 1) auth_url = f"{proto}://{cfg.github_token}@{rest}" try: if not repo_dir.exists(): _log(f"Pulling external memory from {clean_url}...") res = subprocess.run(["git", "clone", auth_url, str(repo_dir)], capture_output=True, text=True) if res.returncode != 0: _log(f"ERROR: Failed to clone repo. Error: {res.stderr}") return base_prompt + "Failed to access external memory repository." else: subprocess.run(["git", "-C", str(repo_dir), "pull"], capture_output=True) subprocess.run(["git", "-C", str(repo_dir), "config", "user.name", "Missminute OS"], capture_output=True) subprocess.run(["git", "-C", str(repo_dir), "config", "user.email", "missminute@localhost"], capture_output=True) memory_content = "" for root, _, files in os.walk(repo_dir): if ".git" in root: continue for file in files: if file.endswith(('.md', '.txt')): file_path = Path(root) / file try: with open(file_path, "r", encoding="utf-8") as f: rel_path = file_path.relative_to(repo_dir) memory_content += f"\n=== FILE: {rel_path} ===\n" memory_content += f.read().strip() + "\n" except Exception as e: _log(f"WARNING: Could not read {file}: {e}") if not memory_content: memory_content = "Memory repository cloned successfully, but no .md or .txt files were found." return base_prompt + memory_content + "\n--- END OF RESTORED MEMORY ---" except Exception as e: _log(f"WARNING: Error accessing dynamic memory: {e}") return base_prompt + f"System error while loading memory: {e}" def write_openclaw_config(cfg: MissminuteConfig): OPENCLAW_HOME.mkdir(parents=True, exist_ok=True) openclaw_cfg: dict = { "gateway": { "port": cfg.openclaw_port, "mode": "local", "auth": { "token": "missminute-internal-secret-123" }, "remote": { "token": "missminute-internal-secret-123" } } } out = OPENCLAW_HOME / "openclaw.json" with open(out, "w") as f: json.dump(openclaw_cfg, f, indent=2) _log(f"OpenClaw config → {out}") def write_hermes_config(cfg: MissminuteConfig, dynamic_prompt: str): HERMES_HOME.mkdir(parents=True, exist_ok=True) (HERMES_HOME / "skills").mkdir(exist_ok=True) with open(HERMES_HOME / "SOUL.md", "w", encoding="utf-8") as f: f.write(dynamic_prompt) _log(f"Identity fused into → {HERMES_HOME / 'SOUL.md'}") mcp_servers = { "openclaw_bridge": { "command": "openclaw", "args": [ "mcp", "serve", "--url", f"ws://127.0.0.1:{cfg.openclaw_port}" ], "env": { "GATEWAY_TOKEN": "missminute-internal-secret-123", "OPENCLAW_AUTH_TOKEN": "missminute-internal-secret-123" } }, "workspace_fs": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", str(WORKSPACE)], }, } if cfg.github_token: mcp_servers["github"] = { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-github"], "env": {"GITHUB_PERSONAL_ACCESS_TOKEN": cfg.github_token}, } hermes_cfg = { "model": { "provider": "custom", "api_key": cfg.llm_api_key, "base_url": cfg.llm_base_url, "default": cfg.llm_model, # Required: hermes rejects local/custom models with <64K context. # For DeepSeek/cloud providers this is ignored (they report their own limit). # For Ollama (local or remote Space B), this tells hermes what to expect. "context_length": 65536, }, "gateway": { "telegram": { "enabled": True, "token": cfg.telegram_token, "allowed_users": cfg.allowed_users, } }, "memory": { "backend": "sqlite", "db_path": str(HERMES_HOME / "memory.db"), }, "skills_dir": str(HERMES_HOME / "skills"), "workspace": str(WORKSPACE), "mcp_servers": mcp_servers, "tools": { "shell": True, "browser": False, "git": True, "max_tool_calls": 20, }, "cron": { "keepalive": { "schedule": "*/25 * * * *", "task": ( "Send a one-line status ping to confirm Missminute OS is " "operational. Format: '🟢 Missminute OS nominal — [timestamp]'" ), } }, } out = HERMES_HOME / "config.yaml" with open(out, "w") as f: yaml.dump(hermes_cfg, f, default_flow_style=False, allow_unicode=True) _log(f"Hermes config → {out}") def fuse(): _log("═" * 52) _log("MISSMINUTE OS — PRE-IGNITION CONFIG FUSION (HEADLESS)") _log("═" * 52) cfg = resolve_config() for d in [WORKSPACE, HERMES_HOME, OPENCLAW_HOME, LOG_DIR]: d.mkdir(parents=True, exist_ok=True) dynamic_system_prompt = fetch_dynamic_memory(cfg) write_openclaw_config(cfg) write_hermes_config(cfg, dynamic_system_prompt) _log(f"Provider: {cfg.llm_model} @ {cfg.llm_base_url}") _log(f"Telegram: owner_id={cfg.telegram_owner_id}") _log(f"GitHub: {'connected' if cfg.github_token else 'not configured'}") _log("═" * 52) print(f"export TELEGRAM_BOT_TOKEN='{cfg.telegram_token}'") print(f"export TELEGRAM_OWNER_ID='{cfg.telegram_owner_id}'") print(f"export LLM_API_KEY='{cfg.llm_api_key}'") print(f"export LLM_BASE_URL='{cfg.llm_base_url}'") print(f"export LLM_MODEL='{cfg.llm_model}'") if cfg.github_token: print(f"export GITHUB_PAT='{cfg.github_token}'") if __name__ == "__main__": fuse()