altoclef_stack.py

"""Third player: upstream AltoClef/Baritone, supervised by keyless OpenCode."""
from __future__ import annotations

import hashlib
import json
import os
import shutil
import subprocess
import threading
import time
import uuid
from pathlib import Path
from typing import Any

import requests

from minecraft_stack import STACK, PERSIST_ROOT, _download, _wait_tcp
from opencode_config import (OPENCODE_VERSION, MUSE_MODEL, configuration, coach_user_message,
                             child_environment, parse_decision, goal_completed)
from voyager_processes import stop_runner

SOURCE = Path(__file__).resolve().parent
ROOT = Path(os.environ.get("ALTOCLEF_HOME", str(PERSIST_ROOT / "altoclef")))
# Runtime-only client dir: Fabric remap, assets and libraries are slow on the
# persistent bucket mount, so they live in /tmp while status/log/checkpoint
# stay in ROOT. Jars are cached in ROOT and copied in (see _ensure_client).
CLIENT = Path(os.environ.get("ALTOCLEF_CLIENT_HOME", "/tmp/altoclef-client")) / "minecraft-client"
MOD_CACHE = ROOT / "mod-cache"
COACH = ROOT / "opencode"
MC_VERSION = "1.19.4"
LOADER_VERSION = "0.16.2"
# Single autonomous account. minecraft_stack, voyager_stack and this module all
# resolve the same default so no second duplicate AltoClef account can appear.
USERNAME = os.environ.get("MC_USERNAME", "Naldo_Bot")
BRIDGE_PORT = 3002
BRIDGE_READY_TIMEOUT = float(os.environ.get("ALTOCLEF_BRIDGE_TIMEOUT", "1800"))
CONNECT_TIMEOUT = float(os.environ.get("ALTOCLEF_CONNECT_TIMEOUT", "600"))
GAMER_COOLDOWN = float(os.environ.get("ALTOCLEF_GAMER_COOLDOWN", "60"))
# Loading grace: the bridge heartbeat (bridge alive) is accepted while the
# client remaps/loads, but connected telemetry must always be fresh (see _observe).
LOADING_GRACE = float(os.environ.get("ALTOCLEF_LOADING_GRACE", "900"))
# /state is reachable before the first 2s scheduler tick publishes a
# heartbeat (exact initial payload has heartbeat_at 0). Accept that payload
# only inside this short window measured from first bridge contact.
FIRST_HEARTBEAT_WINDOW = float(os.environ.get("ALTOCLEF_FIRST_HEARTBEAT_WINDOW", "30"))
# After play started, a respawn/dimension change shows loading briefly; allow
# disconnected-with-heartbeat for up to this long before failing the run.
# Kept short on purpose: a truly frozen Naldo must fail fast, not wait out a
# 15-minute budget. Initial warmup budgets above stay generous instead.
RECONNECT_GRACE = float(os.environ.get("ALTOCLEF_RECONNECT_GRACE", "120"))
# High-level supervision only, via the free OpenCode Muse model. Deterministic
# AltoClef gameplay continues between (and without) model calls.
COACH_MODEL = MUSE_MODEL
ALTOCLEF_URL = "https://github.com/MiranCZ/altoclef/releases/download/0.19/altoclef-1.19.4-0.19.jar"
ALTOCLEF_SHA256 = "fe6f6f8b52ed8a8d8a382f1b0e8c85145f038d2c5ec831710eb4f74a6db489b8"
FABRIC_API_URL = "https://maven.fabricmc.net/net/fabricmc/fabric-api/fabric-api/0.87.2+1.19.4/fabric-api-0.87.2+1.19.4.jar"

# Assisted survival run: no creative equipment is ever given to this player, no
# teleport, no portal construction and no dragon-kill command. AltoClef plays
# the real upstream survival speedrun (BeatMinecraftTask) on the natural map.
# The existing player inventory is retained untouched (never cleared, given or
# archived away); completion is therefore quicker than a fresh speedrun and is
# reported honestly as assisted.


class AltoClefStack:
    def __init__(self) -> None:
        self.lock = threading.RLock()
        self.worker: threading.Thread | None = None
        self.client: subprocess.Popen | None = None
        self.display: subprocess.Popen | None = None
        self.coach: subprocess.Popen | None = None
        self.cancel = threading.Event()
        self.state = "stopped"
        self.error = ""
        self.restart_count = 0
        self.observation: dict[str, Any] = {}
        self.model_checks = 0
        self.model_errors = 0
        self.last_decision: dict[str, Any] = {}
        self.opencode_version = ""
        self.opencode_model = os.environ.get("ALTOCLEF_OPENCODE_MODEL", COACH_MODEL)
        self.last_started_at: float | None = None
        self.next_retry_at = 0.0
        self._last_gamer_at = 0.0
        self._last_mode_set = 0.0
        self._bridge_seen_at: float | None = None
        self._last_connected_at: float | None = None
        saved = ROOT / "status.json"
        if saved.exists():
            try:
                previous = json.loads(saved.read_text())
                self.model_checks = int(previous.get("model_checks") or 0)
                self.model_errors = int(previous.get("model_errors") or 0)
                self.last_decision = previous.get("last_decision") or {}
                if goal_completed(previous.get("observation") or {}):
                    self.observation = previous["observation"]
                    self.state = "completed"
            except (ValueError, OSError, TypeError):
                pass

    def _log(self, text: str) -> None:
        ROOT.mkdir(parents=True, exist_ok=True)
        with (ROOT / "bot.log").open("a", encoding="utf-8") as out:
            out.write(time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + " " + text + "\n")
        print("[altoclef] " + text, flush=True)

    def _pump(self, proc: subprocess.Popen, label: str) -> None:
        if proc.stdout:
            for line in proc.stdout:
                self._log(label + "> " + line.rstrip())

    def _ensure_opencode(self) -> Path:
        COACH.mkdir(parents=True, exist_ok=True)
        runtime = Path(os.environ.get("ALTOCLEF_RUNTIME_HOME", "/tmp/altoclef-runtime"))
        install = runtime / "opencode-cli"
        package = install / "node_modules/opencode-ai/package.json"
        shim = install / "node_modules/.bin" / ("opencode.cmd" if os.name == "nt" else "opencode")
        current = json.loads(package.read_text()).get("version") if package.exists() else None
        if current != OPENCODE_VERSION or not shim.exists():
            shutil.rmtree(install, ignore_errors=True)
            install.mkdir(parents=True, exist_ok=True)
            self._log("installing runtime-only OpenCode " + OPENCODE_VERSION)
            subprocess.run(["npm", "install", "--prefix", str(install), "--no-audit", "--no-fund",
                            "opencode-ai@" + OPENCODE_VERSION], check=True, timeout=600)
        if not shim.exists():
            raise RuntimeError("OpenCode npm shim missing after install")
        report = subprocess.run([str(shim), "--version"], capture_output=True, text=True, timeout=45, check=True)
        self.opencode_version = report.stdout.strip()
        if self.opencode_version != OPENCODE_VERSION:
            raise RuntimeError("Installed OpenCode version differs from reviewed version")
        (COACH / "opencode.json").write_text(json.dumps(configuration(), indent=2))
        return shim

    def _ensure_client(self, java: Path) -> list[str]:
        import minecraft_launcher_lib as launcher
        CLIENT.mkdir(parents=True, exist_ok=True)
        version = f"fabric-loader-{LOADER_VERSION}-{MC_VERSION}"
        if not (CLIENT / "versions" / version / (version + ".json")).exists():
            self._log("installing official Minecraft client " + MC_VERSION + " and Fabric " + LOADER_VERSION
                      + " into runtime-only " + str(CLIENT))
            launcher.fabric.install_fabric(MC_VERSION, str(CLIENT), loader_version=LOADER_VERSION, java=str(java),
                                           callback={"setStatus": lambda text: self._log("installer: " + text)})
        mods = CLIENT / "mods"
        mods.mkdir(exist_ok=True)
        # Jars are verified once into the persistent cache, then copied into
        # the runtime-only client so a /tmp wipe costs a copy, not a download.
        MOD_CACHE.mkdir(parents=True, exist_ok=True)
        _download(ALTOCLEF_URL, MOD_CACHE / "altoclef-1.19.4-0.19.jar", ALTOCLEF_SHA256)
        _download(FABRIC_API_URL, MOD_CACHE / "fabric-api-0.87.2+1.19.4.jar")
        shutil.copyfile(MOD_CACHE / "altoclef-1.19.4-0.19.jar", mods / "altoclef-1.19.4-0.19.jar")
        shutil.copyfile(MOD_CACHE / "fabric-api-0.87.2+1.19.4.jar", mods / "fabric-api-0.87.2+1.19.4.jar")
        shutil.copyfile(SOURCE / "altoclef/space-altoclef-bridge.jar", mods / "space-altoclef-bridge.jar")
        settings = CLIENT / "options.txt"
        defaults = ("maxFps:20\nrenderDistance:4\nsimulationDistance:4\nentityDistanceScaling:0.5\n"
                    "graphicsMode:0\nparticles:2\nsoundCategory_master:0.0\nuseVbo:true\n"
                    "pauseOnLostFocus:false\nfullscreen:false\nshowSubtitles:false\n"
                    # 1.19.4 first-run accessibility onboarding would park the
                    # headless client on a setup screen; skip it when present.
                    "onboardAccessibility:false\n")
        if not settings.exists():
            settings.write_text(defaults)
        elif "onboardAccessibility" not in settings.read_text(errors="replace"):
            with settings.open("a", encoding="utf-8") as out:
                out.write("onboardAccessibility:false\n")
        config_dir = CLIENT / "altoclef/configs"
        config_dir.mkdir(parents=True, exist_ok=True)
        config = config_dir / "beat_minecraft.json"
        if not config.exists():
            config.write_text(json.dumps({"targetEyes":14,"minimumEyes":12,"sleepThroughNight":False,
                                         "placeSpawnNearEndPortal":False,"requiredBeds":10}))
        # Match Java's offline UUID algorithm without changing server auth.
        player_uuid = str(uuid.UUID(bytes=hashlib.md5(("OfflinePlayer:" + USERNAME).encode()).digest(), version=3))
        # No vanilla --server/--port autojoin: it bypasses the TitleScreen, and
        # upstream AltoClef initializes ONLY there (EntryMixin on
        # TitleScreen.init -> TitleScreenEntryEvent -> onInitializeLoad, which
        # creates the command executor and task chain). The normal title flow
        # renders first; the bridge's readiness-gated fallback then connects
        # after resources are done.
        return launcher.command.get_minecraft_command(version, str(CLIENT), {
            "username": USERNAME, "uuid": player_uuid, "token": "0",
            "executablePath": str(java), "jvmArguments": ["-Xms256M", "-Xmx2G", "-XX:ActiveProcessorCount=2"],
            "customResolution":True, "resolutionWidth":"640", "resolutionHeight":"360",
        })

    def _loading_grace(self) -> bool:
        return self.last_started_at is not None and time.time() - self.last_started_at < LOADING_GRACE

    @staticmethod
    def _survival_verified(data: dict[str, Any]) -> bool:
        # Connected, clean, and creative EXPLICITLY false. A loading payload
        # without a creative flag never counts as Survival.
        return bool(data.get("connected")) and not data.get("error") and data.get("creative") is False

    def _observe(self) -> dict[str, Any]:
        response = requests.get(f"http://127.0.0.1:{BRIDGE_PORT}/state", timeout=5)
        response.raise_for_status()
        data = response.json()
        if not isinstance(data, dict):
            raise RuntimeError("Invalid AltoClef telemetry")
        now = time.time()
        if self._bridge_seen_at is None:
            self._bridge_seen_at = now
        if data.get("connected"):
            # Connected world observations must always be fresh; a stale
            # connected payload is never accepted as alive.
            if now - float(data.get("observed_at") or 0) > 30:
                raise RuntimeError("AltoClef client telemetry stopped updating: "
                                   + str(data.get("error") or "no fresh snapshot from client thread"))
            self._last_connected_at = now
        else:
            heartbeat = float(data.get("heartbeat_at") or 0)
            if heartbeat <= 0:
                # Exact initial payload before the first scheduler tick:
                # accept only inside the short first-heartbeat window.
                if now - self._bridge_seen_at > FIRST_HEARTBEAT_WINDOW:
                    raise RuntimeError("AltoClef bridge never published a heartbeat")
            elif now - heartbeat > 60:
                raise RuntimeError("AltoClef bridge heartbeat stopped updating")
            if self._last_connected_at is None:
                # Startup: loading grace is anchored at bridge readiness (see
                # _boot), so a slow remap never consumes the connect budget.
                if not self._loading_grace():
                    raise RuntimeError("AltoClef client still not connected after loading grace: "
                                       + str(data.get("dispatch_error") or data.get("error") or "no diagnosis"))
            elif now - self._last_connected_at > RECONNECT_GRACE:
                raise RuntimeError("AltoClef client did not reconnect after "
                                   + str(int(RECONNECT_GRACE)) + "s: "
                                   + str(data.get("dispatch_error") or data.get("error") or "no diagnosis"))
        self.observation = data
        self._checkpoint()
        return data

    def _checkpoint(self) -> None:
        data = {"observation": self.observation, "model_checks":self.model_checks,
                "model_errors":self.model_errors, "last_decision":self.last_decision,
                "goal_completed":goal_completed(self.observation), "saved_at":time.time()}
        temp = ROOT / "status.json.tmp"
        temp.write_text(json.dumps(data, ensure_ascii=False, indent=2))
        temp.replace(ROOT / "status.json")

    def _command(self, command: str) -> None:
        response = requests.post(f"http://127.0.0.1:{BRIDGE_PORT}/command", data=command, timeout=8)
        response.raise_for_status()
        self._log("AltoClef command: @" + command)

    def _ensure_survival_mode(self) -> None:
        """Bootstrap/respawn mode assignment: Naldo plays Survival, nothing given.

        Only a console gamemode assignment; the existing inventory is retained
        untouched and no item, teleport or effect is ever granted.
        """
        server = STACK.processes.get("minecraft")
        if server is None or server.poll() is not None or server.stdin is None:
            raise RuntimeError("Minecraft console unavailable for gamemode assignment")
        with STACK.lock:
            server.stdin.write(f"gamemode survival {USERNAME}\n")
            server.stdin.flush()
        self._last_mode_set = time.time()
        self._log(f"mode assignment: gamemode survival {USERNAME} (existing inventory retained)")

    def _consult(self, exe: Path, data: dict[str, Any]) -> None:
        prompt = coach_user_message(data)
        env = child_environment(COACH)
        self.coach = subprocess.Popen([str(exe), "run", "--format", "json", "--agent", "plan",
                                       "--model", self.opencode_model, prompt], cwd=str(COACH), env=env,
                                      stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                                      start_new_session=True)
        try:
            # Empty stdin provides EOF: a non-TTY child that waits on stdin
            # would otherwise park here until the timeout.
            output, error = self.coach.communicate(input="", timeout=300)
            if self.coach.returncode:
                raise RuntimeError("OpenCode failed: " + (error or output)[-300:])
            decision = parse_decision(output)
            self.model_checks += 1
            self.last_decision = {**decision, "model":self.opencode_model, "observed_at":time.time()}
            self._log("OpenCode " + self.opencode_model + ": " + json.dumps(decision))
            fresh = self._observe()
            if decision["action"] in {"start","resume"} and fresh.get("connected") and not fresh.get("task_active") and not goal_completed(fresh):
                self._command("gamer")
        except Exception as exc:
            self.model_errors += 1
            self.last_decision = {"error":str(exc)[:400],"observed_at":time.time()}
            self._log("OpenCode supervision unavailable: " + str(exc)[:400])
            # A model outage never stalls deterministic gameplay.
        finally:
            stop_runner(self.coach, timeout=3)
            self.coach = None
            self._checkpoint()

    def _client_tree_size(self) -> int:
        total = 0
        try:
            for path in CLIENT.rglob("*"):
                try:
                    if path.is_file():
                        total += path.stat().st_size
                except OSError:
                    pass
        except OSError:
            pass
        return total

    def _wait_bridge_ready(self) -> None:
        """Bounded warmup that never kills a healthy growing process.

        The first Fabric remap plus asset download can take tens of minutes on
        a slow mount. While the client process is alive we wait up to
        BRIDGE_READY_TIMEOUT, logging live progress (runtime dir growth). A
        client exit fails fast with the log tail instead of a bare timeout.
        """
        deadline = time.time() + BRIDGE_READY_TIMEOUT
        last_size = -1
        last_note = 0.0
        while time.time() < deadline:
            if self.cancel.is_set():
                return
            code = self.client.poll() if self.client else None
            if code is not None:
                raise RuntimeError("AltoClef client exited during startup rc=" + str(code)
                                   + "; log tail:\n" + self.tail(30))
            if _wait_tcp("127.0.0.1", BRIDGE_PORT, 5):
                return
            size = self._client_tree_size()
            if size != last_size and time.time() - last_note > 120:
                last_size = size
                last_note = time.time()
                self._log(f"client warmup in progress: runtime dir {size // 1024 // 1024} MiB, "
                          f"bridge not up yet (budget {int(deadline - time.time())}s left)")
            self.cancel.wait(5)
        raise RuntimeError(f"AltoClef telemetry bridge did not start within {int(BRIDGE_READY_TIMEOUT)}s; "
                           "client still alive, log tail:\n" + self.tail(30))

    def _reset_launch_clocks(self) -> None:
        # Per-launch reset: a second boot after a crash must not reuse the
        # previous launch's clocks, or the new initial loading payload is
        # rejected against expired grace/heartbeat timestamps.
        self._bridge_seen_at = None
        self._last_connected_at = None
        self.last_started_at = None
        self._last_gamer_at = 0.0
        self._last_mode_set = 0.0

    def _boot(self) -> None:
        try:
            ROOT.mkdir(parents=True, exist_ok=True)
            self._reset_launch_clocks()
            if not STACK.minecraft_ready.wait(timeout=600) or self.cancel.is_set():
                raise RuntimeError("Minecraft server did not become ready")
            exe = self._ensure_opencode()
            java = STACK._ensure_minecraft_java()
            command = self._ensure_client(java)
            if not shutil.which("Xvfb"):
                raise RuntimeError("Xvfb is missing; rebuild the Space with packages.txt")
            if not _wait_tcp("127.0.0.1", 25568, 180):
                raise RuntimeError("Existing ViaProxy did not become ready")
            self.display = subprocess.Popen(["Xvfb", ":97", "-screen", "0", "640x360x24", "-nolisten", "tcp"],
                                             stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, start_new_session=True)
            threading.Thread(target=self._pump,args=(self.display,"display"),daemon=True).start()
            if self.cancel.wait(2):
                return
            env = child_environment(COACH)
            env.update({"DISPLAY":":97","LIBGL_ALWAYS_SOFTWARE":"1","GALLIUM_DRIVER":"llvmpipe",
                        "LP_NUM_THREADS":"2","ALTOCLEF_BRIDGE_PORT":str(BRIDGE_PORT),"ALTOCLEF_SERVER":"127.0.0.1:25568",
                        "CUDA_VISIBLE_DEVICES":"","HIP_VISIBLE_DEVICES":"","ROCR_VISIBLE_DEVICES":""})
            self.client = subprocess.Popen(command,cwd=str(CLIENT),env=env,stdout=subprocess.PIPE,
                                           stderr=subprocess.STDOUT,text=True,start_new_session=True)
            threading.Thread(target=self._pump,args=(self.client,"client"),daemon=True).start()
            self.last_started_at = time.time()
            self._log("starting upstream AltoClef 0.19 player " + USERNAME + " using CPU rendering")
            self._wait_bridge_ready()
            if self.cancel.is_set():
                return
            # Re-anchor startup clocks at bridge readiness: the warmup budget
            # above must not consume the loading/connect budgets below.
            self.last_started_at = time.time()
            self._log("bridge ready; loading grace and connect budget start now")
            deadline = time.time() + CONNECT_TIMEOUT
            data: dict[str, Any] = {}
            while not self.cancel.is_set():
                data = self._observe()
                if data.get("connected") and not data.get("error"):
                    break
                if self.client.poll() is not None or time.time() > deadline:
                    raise RuntimeError("AltoClef could not connect: " + str(data.get("error")))
                self.cancel.wait(3)
            if self.cancel.is_set():
                return
            # Coherent mode assignment BEFORE gamer: Naldo must be Survival
            # (upstream speedrun needs hostile mobs/loot); humans keep theirs.
            self._ensure_survival_mode()
            self.cancel.wait(4)
            data = self._observe()
            # Strict verification: connected, no error, and creative EXPLICITLY
            # false. A loading payload without a creative flag is not Survival.
            if not self._survival_verified(data):
                raise RuntimeError("mode verification failed: connected=" + str(data.get("connected"))
                                   + " error=" + str(data.get("error"))
                                   + " creative=" + str(data.get("creative"))
                                   + "; gamer not started")
            self._log("mode verified: " + USERNAME + " is Survival; existing inventory retained")
            self.state = "running"
            self._command("gamer")
            self._last_gamer_at = time.time()
            next_coach = 0.0
            while not self.cancel.wait(5):
                if self.client.poll() is not None:
                    raise RuntimeError("AltoClef client exited with " + str(self.client.returncode)
                                       + "; log tail:\n" + self.tail(30))
                data = self._observe()
                if goal_completed(data):
                    self.state = "completed"
                    self._log("Verified completion: entered End and observed Minecraft credits")
                    break
                now = time.time()
                if data.get("creative"):
                    # Respawn/mode drift: re-assign Survival instead of gaming Creative.
                    if now - self._last_mode_set > 300:
                        self._ensure_survival_mode()
                elif data.get("connected") and not data.get("task_active") and not goal_completed(data):
                    # Cooldown: re-issuing gamer while the task boots would
                    # restart it every poll instead of letting it run.
                    if now - self._last_gamer_at >= GAMER_COOLDOWN:
                        self._command("gamer")
                        self._last_gamer_at = now
                if time.time() >= next_coach:
                    self._consult(exe, data)
                    next_coach = time.time() + 600
        except Exception as exc:
            self.error = f"{type(exc).__name__}: {exc}"
            self.state = "error"
            self.next_retry_at = time.time() + min(30 * 2 ** min(self.restart_count, 4), 480)
            self._log(self.error)
        finally:
            if self.state != "completed":
                stop_runner(self.coach,timeout=3)
                stop_runner(self.client,timeout=8)
                stop_runner(self.display,timeout=3)
            if self.cancel.is_set():
                self.state = "stopped"

    def start_async(self) -> dict[str, Any]:
        with self.lock:
            if goal_completed(self.observation):
                self.state = "completed"
                return self.status()
            if self.worker and self.worker.is_alive() or self.state == "completed":
                return self.status()
            if self.state == "error" and time.time() < self.next_retry_at:
                return self.status()
            self.restart_count += int(self.state == "error")
            self.cancel.clear()
            self.error = ""
            self.state = "starting"
            self.worker = threading.Thread(target=self._boot,daemon=True,name="altoclef-bot")
            self.worker.start()
            return self.status()

    def stop(self) -> dict[str, Any]:
        self.cancel.set()
        stop_runner(self.coach,timeout=3)
        stop_runner(self.client,timeout=8)
        stop_runner(self.display,timeout=3)
        if self.worker and self.worker is not threading.current_thread():
            self.worker.join(timeout=10)
        self.state = "stopped"
        return self.status()

    def status(self) -> dict[str, Any]:
        running = bool(self.client and self.client.poll() is None)
        return {"state":self.state,"running":running,"username":USERNAME,"error":self.error,
                "engine":"AltoClef 0.19 / Baritone (upstream)","source":"https://github.com/MiranCZ/altoclef",
                "client_version":MC_VERSION,"server_version":"1.19","connection":"127.0.0.1:25568 (ViaProxy)",
                "opencode_version":self.opencode_version,"model":self.opencode_model,"api_key_required":False,
                "model_checks":self.model_checks,"model_errors":self.model_errors,"last_decision":self.last_decision,
                "goal_completed":goal_completed(self.observation),"world_mode":"Survival / Normal (assisted: existing inventory retained)",
                "restart_count":self.restart_count,"started_at":self.last_started_at,"checkpoint":str(ROOT),
                "next_retry_at":self.next_retry_at,
                "cpu_only":True,"rendering":"Mesa llvmpipe (software)","observation":self.observation}

    def tail(self, lines: int = 100) -> str:
        path = ROOT / "bot.log"
        return "\n".join(path.read_text(errors="replace").splitlines()[-max(1,min(lines,500)):]) if path.exists() else ""


ALTOCLEF = AltoClefStack()