gemini_bot_stack.py

"""Second full Voyager agent backed by Gemini CLI, isolated from Naldo_Bot."""
from __future__ import annotations

import json
import os
import re
import subprocess
import sys
import threading
import time
import urllib.request
from pathlib import Path
from typing import Any

from voyager_stack import ROOT as VOYAGER_ROOT, SRC as VOYAGER_SRC, CKPT_DIR as NALDO_CKPT_DIR, EMBED_BASE_URL as VOYAGER_EMBED_BASE_URL, VOYAGER, _wait_tcp, restart_backoff_after, tail_lines
from voyager_runtime_contract import UPSTREAM_COMMIT, read_attestation
from voyager_learning import read_learning_state

ROOT = Path(os.environ.get("GEMINI_BOT_RUNTIME_HOME", "/tmp/gemini-bot-runtime"))
PERSIST_ROOT = Path(os.environ.get("MINECRAFT_PERSIST_HOME", "/persist/minecraft-friends"))
CKPT_DIR = Path(os.environ.get(
    "GEMINI_VOYAGER_CKPT",
    str(PERSIST_ROOT / "gemini-voyager" / "ckpt-20260927-flat-y64-v1"),
))
LOG_PATH = ROOT / "voyager.log"
PROXY_LOG_PATH = ROOT / "proxy.log"
CLI_VERSION = "0.61.0"
MODEL = "gemini-3.5-flash-lite"
PROXY_PORT = int(os.environ.get("GEMINI_VOYAGER_PROXY_PORT", "7863"))
MINEFLAYER_PORT = int(os.environ.get("GEMINI_VOYAGER_MINEFLAYER_PORT", "3001"))
KEY_NAMES = tuple(f"GEMINI_BOT_KEY_{i}" for i in range(1, 7))


def _node_version() -> int:
    output = subprocess.check_output(["node", "--version"], text=True, timeout=5)
    match = re.fullmatch(r"v(\d+)\.\d+\.\d+\s*", output)
    return int(match.group(1)) if match else 0


def _wait_embedding_ready(base_url: str, timeout: float = 180.0) -> bool:
    deadline = time.monotonic() + timeout
    payload = json.dumps({"model": "text-embedding-ada-002", "input": "health"}).encode("utf-8")
    while time.monotonic() < deadline:
        request = urllib.request.Request(
            base_url.rstrip("/") + "/embeddings", data=payload, method="POST",
            headers={"Content-Type": "application/json"},
        )
        try:
            remaining = max(1.0, deadline - time.monotonic())
            with urllib.request.urlopen(request, timeout=min(remaining, 120.0)) as response:
                body = json.loads(response.read().decode("utf-8"))
            if isinstance(body, dict) and body.get("data"):
                return True
        except Exception:
            pass
        time.sleep(1.0)
    return False


def _terminate(proc: subprocess.Popen[str] | None, timeout: int = 15) -> None:
    if not proc or proc.poll() is not None:
        return
    proc.terminate()
    try:
        proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.kill()
        proc.wait(timeout=5)


def _read_json(path: Path) -> list[str]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
        return value if isinstance(value, list) else []
    except (OSError, ValueError):
        return []


class GeminiBotStack:
    def __init__(self) -> None:
        self.lock = threading.RLock()
        self.worker: threading.Thread | None = None
        self.proc: subprocess.Popen[str] | None = None
        self.proxy_proc: subprocess.Popen[str] | None = None
        self.stop_event = threading.Event()
        self.state = "stopped"
        self.error = ""
        self.started_at: float | None = None
        self.restart_count = 0

    def _prepare(self) -> tuple[Path, Path]:
        ROOT.mkdir(parents=True, exist_ok=True)
        if _node_version() < 20:
            raise RuntimeError("Gemini CLI 0.61.0 needs Node.js 20 or newer")

        node_modules = VOYAGER_SRC / "voyager" / "env" / "mineflayer" / "node_modules"
        marker = VOYAGER_ROOT / ".node-ready"
        deadline = time.monotonic() + 900
        while not self.stop_event.is_set() and (
            not marker.exists() or not (node_modules / "mineflayer").exists()
        ):
            if time.monotonic() > deadline:
                raise RuntimeError("Voyager Mineflayer dependencies did not become ready")
            self.stop_event.wait(2)
        if self.stop_event.is_set():
            raise RuntimeError("stopped")

        cli_dir = ROOT / "cli"
        cli = cli_dir / "node_modules" / ".bin" / "gemini"
        version_marker = cli_dir / ".installed-version"
        if (
            not cli.exists()
            or not version_marker.exists()
            or version_marker.read_text().strip() != CLI_VERSION
        ):
            cli_dir.mkdir(parents=True, exist_ok=True)
            result = subprocess.run(
                [
                    "npm", "install", "--no-audit", "--no-fund", "--prefix", str(cli_dir),
                    f"@google/gemini-cli@{CLI_VERSION}",
                ],
                cwd=ROOT,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                timeout=600,
                check=False,
            )
            if result.returncode or not cli.exists():
                raise RuntimeError(f"Gemini CLI install failed (rc={result.returncode})")
            version_marker.write_text(CLI_VERSION + "\n", encoding="utf-8")

        workspace = ROOT / "workspace"
        settings = workspace / ".gemini" / "settings.json"
        settings.parent.mkdir(parents=True, exist_ok=True)
        settings.write_text(json.dumps({
            "general": {"enableAutoUpdate": False, "plan": {"modelRouting": False}},
            "tools": {"core": [], "exclude": [
                "run_shell_command", "write_file", "replace", "edit_file", "web_fetch",
                "google_web_search", "read_file", "read_many_files", "glob", "grep",
                "list_directory",
            ]},
            "model": {"maxSessionTurns": 1},
            "modelConfigs": {"customOverrides": [{
                "match": {"model": MODEL},
                "modelConfig": {"generateContentConfig": {"temperature": 0}},
            }]},
            "telemetry": {"enabled": False, "logPrompts": False},
            "privacy": {"usageStatisticsEnabled": False},
        }), encoding="utf-8")
        (settings.parent / "system.md").write_text(
            "You are the language model inside the Voyager Minecraft agent. "
            "Follow the supplied Voyager system and user prompts exactly. "
            "Return only the format requested by the current prompt. "
            "Do not use Gemini CLI tools, run commands, read files, or reveal credentials.\n",
            encoding="utf-8",
        )
        home = ROOT / "home"
        home.mkdir(parents=True, exist_ok=True)
        CKPT_DIR.mkdir(parents=True, exist_ok=True)
        return cli, workspace

    def _pump(self, proc: subprocess.Popen[str], path: Path, label: str) -> None:
        assert proc.stdout is not None
        with path.open("a", encoding="utf-8", errors="replace") as log:
            for line in proc.stdout:
                log.write(line)
                log.flush()
                print(f"[{label}] " + line.rstrip(), flush=True)

    def _child_env(self, cli: Path, workspace: Path) -> dict[str, str]:
        env = os.environ.copy()
        env.update({
            "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
            "HOME": str(ROOT / "home"),
            "NODE_PATH": str(
                VOYAGER_SRC / "voyager" / "env" / "mineflayer" / "node_modules"
            ),
            "GEMINI_BOT_CLI": str(cli),
            "GEMINI_BOT_WORKSPACE": str(workspace),
            "GEMINI_BOT_HOME": str(ROOT / "home"),
            "GEMINI_BOT_SYSTEM_MD": str(workspace / ".gemini" / "system.md"),
            "GEMINI_VOYAGER_MODEL": MODEL,
            "GEMINI_VOYAGER_PROXY_PORT": str(PROXY_PORT),
            "GEMINI_VOYAGER_EMBED_BASE": VOYAGER_EMBED_BASE_URL,
        })
        return env

    def _start_proxy(self, env: dict[str, str]) -> None:
        entrypoint = Path(__file__).resolve().parent / "gemini_voyager_proxy.py"
        PROXY_LOG_PATH.write_text("", encoding="utf-8")
        self.proxy_proc = subprocess.Popen(
            [sys.executable, "-u", str(entrypoint)],
            cwd=str(ROOT),
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
        )
        threading.Thread(
            target=self._pump,
            args=(self.proxy_proc, PROXY_LOG_PATH, "gemini-voyager-proxy"),
            daemon=True,
        ).start()
        if not _wait_tcp("127.0.0.1", PROXY_PORT, 60):
            raise RuntimeError("Gemini Voyager proxy did not become ready")
        if self.proxy_proc.poll() is not None:
            raise RuntimeError(
                f"Gemini Voyager proxy exited rc={self.proxy_proc.returncode}"
            )
        if not _wait_embedding_ready(f"http://127.0.0.1:{PROXY_PORT}/v1", timeout=180.0):
            raise RuntimeError("Gemini embedding backend did not become ready within 180s")
        print("[gemini-voyager] embedding probe: ready", flush=True)

    def _start_voyager(self, env: dict[str, str]) -> None:
        if CKPT_DIR.resolve() == NALDO_CKPT_DIR.resolve():
            raise RuntimeError("Gemini and Naldo must have separate Voyager checkpoints")
        runner = Path(__file__).resolve().parent / "voyager_runner.py"
        voyager_env = env.copy()
        voyager_env.update({
            "VOYAGER_HOME": str(VOYAGER_SRC),
            "VOYAGER_ATTESTATION_FILE": str(ROOT / "engine-attestation.json"),
            "VOYAGER_OPENAI_BASE": f"http://127.0.0.1:{PROXY_PORT}/v1",
            "VOYAGER_BRAIN_PROFILE": "gemini-cli",
            "VOYAGER_API_KEY": "local-gemini-voyager",
            "VOYAGER_MODEL": MODEL,
            "VOYAGER_AUX_MODEL": MODEL,
            "VOYAGER_CKPT_DIR": str(CKPT_DIR),
            "MC_USERNAME": "Gemini_Bot",
            "BOT_DISPLAY_NAME": "Gemini Bot",
            "MC_PORT": "25565",
            "MINEFLAYER_API_PORT": str(MINEFLAYER_PORT),
            "MAX_ITERATIONS": os.environ.get(
                "GEMINI_VOYAGER_MAX_ITERATIONS", "1000000"
            ),
            "OPENAI_API_REQUEST_TIMEOUT": "300",
            "ANONYMIZED_TELEMETRY": "False",
            "PYTHONPATH": str(VOYAGER_SRC),
        })
        LOG_PATH.write_text("", encoding="utf-8")
        self.proc = subprocess.Popen(
            [sys.executable, "-u", str(runner)],
            cwd=str(VOYAGER_SRC),
            env=voyager_env,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
        )
        threading.Thread(
            target=self._pump,
            args=(self.proc, LOG_PATH, "gemini-voyager"),
            daemon=True,
        ).start()

    def _boot(self) -> None:
        self.started_at = time.time()
        try:
            with self.lock:
                self.state = "waiting-minecraft"
                self.error = ""
            if not _wait_tcp("127.0.0.1", 25565, 240):
                raise RuntimeError("Minecraft Java is unavailable")

            ready_file = Path(os.environ.get(
                "MINECRAFT_READY_FILE", "/tmp/minecraft-stack/minecraft.ready"
            ))
            deadline = time.monotonic() + 240
            while (
                not self.stop_event.is_set()
                and not ready_file.exists()
                and time.monotonic() < deadline
            ):
                self.stop_event.wait(1)
            if not ready_file.exists():
                raise RuntimeError("Minecraft world did not finish loading")

            if not any(os.environ.get(name) for name in KEY_NAMES):
                with self.lock:
                    self.state = "waiting-credentials"
                    self.error = "Set GEMINI_BOT_KEY_1 as a Space secret"
                return

            with self.lock:
                self.state = "preparing"
            cli, workspace = self._prepare()
            env = self._child_env(cli, workspace)
            backoff = 5

            while not self.stop_event.is_set():
                run_started = time.monotonic()
                self._start_proxy(env)
                self._start_voyager(env)
                time.sleep(3)
                if self.proc.poll() is None and self.proxy_proc.poll() is None:
                    with self.lock:
                        self.state = "running"
                        self.error = ""

                while (
                    not self.stop_event.is_set()
                    and self.proc.poll() is None
                    and self.proxy_proc.poll() is None
                ):
                    self.stop_event.wait(1)

                if self.stop_event.is_set():
                    break

                runner_rc = self.proc.returncode if self.proc else None
                proxy_rc = self.proxy_proc.returncode if self.proxy_proc else None
                _terminate(self.proc)
                _terminate(self.proxy_proc)
                self.restart_count += 1
                with self.lock:
                    self.state = "restarting"
                    self.error = (
                        f"Gemini Voyager child exit runner={runner_rc} "
                        f"proxy={proxy_rc}; retry {self.restart_count}"
                    )
                if self.stop_event.wait(backoff):
                    break
                backoff = restart_backoff_after(backoff, time.monotonic() - run_started)

            _terminate(self.proc)
            _terminate(self.proxy_proc)
            with self.lock:
                self.state = "stopped"
        except Exception as exc:
            _terminate(self.proc)
            _terminate(self.proxy_proc)
            with self.lock:
                self.state = "error"
                self.error = str(exc)[:200]
            print(
                f"[gemini-voyager] startup error: {type(exc).__name__}: {self.error}",
                flush=True,
            )

    def start_async(self) -> dict[str, Any]:
        with self.lock:
            if self.worker and self.worker.is_alive():
                return self.status()
            self.stop_event.clear()
            self.worker = threading.Thread(
                target=self._boot, daemon=True, name="gemini-voyager-bootstrap"
            )
            self.worker.start()
        return self.status()

    def stop(self) -> dict[str, Any]:
        self.stop_event.set()
        with self.lock:
            runner = self.proc
            proxy = self.proxy_proc
            self.state = "stopped"
        _terminate(runner)
        _terminate(proxy)
        return self.status()

    def _mineflayer_status(self) -> dict[str, Any]:
        try:
            with urllib.request.urlopen(
                f"http://127.0.0.1:{MINEFLAYER_PORT}/health", timeout=1.5
            ) as response:
                data = json.load(response)
            return data if isinstance(data, dict) else {"ready": False}
        except Exception:
            return {"ready": False}

    def _proxy_status(self) -> dict[str, Any]:
        try:
            with urllib.request.urlopen(
                f"http://127.0.0.1:{PROXY_PORT}/health", timeout=1.5
            ) as response:
                data = json.load(response)
            return data if isinstance(data, dict) else {"ready": False}
        except Exception:
            return {"ready": False}

    def status(self) -> dict[str, Any]:
        completed = _read_json(CKPT_DIR / "curriculum" / "completed_tasks.json")
        failed = _read_json(CKPT_DIR / "curriculum" / "failed_tasks.json")
        lines = tail_lines(LOG_PATH)
        mineflayer = self._mineflayer_status()
        learning = read_learning_state(CKPT_DIR)
        proxy = self._proxy_status()
        with self.lock:
            proc = self.proc
            proxy_proc = self.proxy_proc
            runner_running = bool(proc and proc.poll() is None)
            proxy_running = bool(proxy_proc and proxy_proc.poll() is None)
            engine = read_attestation(ROOT / "engine-attestation.json", proc.pid if runner_running else None)
            naldo_proc = VOYAGER.proc
            naldo_running = bool(naldo_proc and naldo_proc.poll() is None)
            naldo_engine = read_attestation(VOYAGER_ROOT / "engine-attestation.json", naldo_proc.pid if naldo_running else None)
            return {
                "state": self.state,
                "error": self.error or None,
                "running": runner_running and proxy_running,
                "pid": proc.pid if runner_running else None,
                "proxy_pid": proxy_proc.pid if proxy_running else None,
                "restart_count": self.restart_count,
                "started_at": self.started_at,
                "architecture": "Voyager",
                "same_voyager_runtime_as_naldo": bool(engine and naldo_engine and engine.get("source_root") == naldo_engine.get("source_root") == str(VOYAGER_SRC.resolve()) and engine.get("upstream_commit") == naldo_engine.get("upstream_commit") == UPSTREAM_COMMIT),
                "engine_runtime": engine,
                "bot_username": "Gemini_Bot",
                "model": MODEL,
                "decision_brain": "gemini-cli-only",
                "brain_profile": "gemini-cli",
                "embedding_backend": VOYAGER_EMBED_BASE_URL,
                "cli_version": CLI_VERSION,
                "configured_key_count": sum(
                    bool(os.environ.get(name)) for name in KEY_NAMES
                ),
                "checkpoint_dir": str(CKPT_DIR),
                "checkpoint_persistent": str(CKPT_DIR).startswith("/persist/"),
                "learning": learning,
                "activity": {
                    "completed_count": len(completed),
                    "failed_count": len(failed),
                    "recent_completed": completed[-5:],
                    "recent_failed": failed[-8:],
                    "curriculum_retry_exhausted": sum(
                        "Max retries reached, failed to propose ai task" in line for line in lines
                    ),
                    "action_model_messages": sum(
                        "****Action Agent ai message****" in line for line in lines
                    ),
                    "curriculum_model_messages": sum(
                        "****Curriculum Agent ai message****" in line for line in lines
                    ),
                    "partial_read_errors": sum("PartialReadError" in line for line in lines),
                    "file_not_found_errors": sum("FileNotFoundError" in line for line in lines),
                    "index_errors": sum("IndexError" in line for line in lines),
                    "syntax_errors": sum("SyntaxError" in line for line in lines),
                },
                "mineflayer": mineflayer,
                "proxy": proxy,
                "game": mineflayer,
            }


GEMINI_BOT = GeminiBotStack()