minecraft_stack.py

from __future__ import annotations

import hashlib
import uuid
import json
import datetime
import os
import re
import shutil
import socket
import subprocess
import tarfile
import threading
import tempfile
import time
import urllib.request
from pathlib import Path
from typing import Any

import yaml

BASE = Path(os.environ.get("MINECRAFT_HOME", "/tmp/minecraft-stack"))
PERSIST_ROOT = Path(os.environ.get("MINECRAFT_PERSIST_HOME", "/persist/minecraft-friends"))
JAVA_DIR = BASE / "java"
VIA_DIR = BASE / "viaproxy"
# Microsoft/Xbox OAuth cache must survive Space rebuilds/restarts. The HF Bucket
# mounted at /persist is private and read-write; nether2rak refreshes token.json
# in-place, so using this directory preserves both initial login and token rotation.
RELAY_DIR = PERSIST_ROOT / "relay"
WORLD_DIR = PERSIST_ROOT / "world-20260929-features-y64-v1"
# Generator settings are only read when the world is created, so replacing the map
# means pointing at a new directory. The previous world is kept so the human
# player's Bedrock session can be carried over instead of being reset.
PREVIOUS_WORLD_DIR = PERSIST_ROOT / "world-20260928-ores-y64-v1"
WORLD_PROFILE = "minecraft:flat"
WORLD_SURFACE_Y = 64
# Resource-rich farming map, bottom to top. Two constraints fix this shape:
# 1.18+ superflat stacks the layers upward from the -64 world floor, and Voyager's
# placement, navigation and prompts are all pinned to a walkable surface at Y=64,
# so the heights below must total exactly WORLD_FLOOR + WORLD_SURFACE_Y.
# Every ore is duplicated: a shallow band 6-24 blocks under the topsoil that a
# mining bot reaches from the surface, and a deepslate band at the bottom of the
# column. Plains with no lakes and no features keeps the surface predictable for
# the pathfinder and for players building on it.
WORLD_FLAT_LAYERS = (
    ("minecraft:bedrock", 1),
    ("minecraft:deepslate", 6),
    ("minecraft:deepslate_coal_ore", 3),
    ("minecraft:deepslate_copper_ore", 3),
    ("minecraft:deepslate_iron_ore", 3),
    ("minecraft:deepslate_gold_ore", 2),
    ("minecraft:deepslate_redstone_ore", 2),
    ("minecraft:deepslate_diamond_ore", 1),
    ("minecraft:deepslate_lapis_ore", 1),
    ("minecraft:deepslate_emerald_ore", 1),
    ("minecraft:stone", 81),
    ("minecraft:coal_ore", 4),
    ("minecraft:copper_ore", 4),
    ("minecraft:iron_ore", 4),
    ("minecraft:gold_ore", 2),
    ("minecraft:redstone_ore", 2),
    ("minecraft:diamond_ore", 1),
    ("minecraft:lapis_ore", 1),
    ("minecraft:emerald_ore", 1),
    ("minecraft:gravel", 1),
    ("minecraft:dirt", 3),
    ("minecraft:grass_block", 1),
)
WORLD_FLOOR_Y = -64
if sum(height for _, height in WORLD_FLAT_LAYERS) != WORLD_SURFACE_Y - WORLD_FLOOR_Y:
    raise RuntimeError(
        "World layers must total the distance from the world floor to WORLD_SURFACE_Y"
    )
FLAT_GENERATOR_SETTINGS = json.dumps(
    {
        "biome": "minecraft:plains",
        "layers": [{"block": block, "height": height} for block, height in WORLD_FLAT_LAYERS],
        # Per minecraft.wiki/w/Superflat/Settings (Java 1.19+):
        #   features            - "whether or not to generate biome-specific placed
        #                         features". Without it a plains superflat has no
        #                         trees, grass or flowers, so the official curriculum
        #                         asks for wood it can never obtain.
        #   lakes               - only controls lava lakes, so it stays off.
        #   structure_overrides - omitted on purpose: the documented default is
        #                         "all the structure sets". An empty list, which is
        #                         what this used to pass, disables every structure.
        "lakes": False,
        "features": True,
    },
    separators=(",", ":"),
)
BOT_USERNAME = os.environ.get("MC_USERNAME", "Naldo_Bot")
ADMIN_USERNAME = "IffyBinkie6134"
MAX_PLAYERS = 100
TOOLS_DIR = BASE / "tools"
BIN_DIR = BASE / "bin"
LOG_PATH = BASE / "stack.log"
READY_FILE = BASE / "minecraft.ready"

FABRIC_INSTALLER_URL = "https://maven.fabricmc.net/net/fabricmc/fabric-installer/0.11.2/fabric-installer-0.11.2.jar"
FABRIC_MC_VERSION = "1.19"
FABRIC_LOADER_VERSION = "0.14.18"

# Official Voyager Fabric 1.19 server-side mod stack.
FABRIC_API_URL = "https://cdn.modrinth.com/data/P7dR8mSH/versions/0.58.0%2B1.19/fabric-api-0.58.0%2B1.19.jar"
FABRIC_API_SHA256 = "0a3d2dba1ed45b002483ae7c08e2c0bcfcb8d2b166801a6c29d8ea64dba64370"
COMPLETE_CONFIG_URL = "https://cdn.modrinth.com/data/GtqG8z1h/versions/2.0.0/completeconfig-2.0.0.jar"
COMPLETE_CONFIG_SHA256 = "d34a3e167bf7942348d3f8c650fcefead6284878b39277ba43282ef69fa2d566"
SERVER_PAUSE_URL = "https://mediafilez.forgecdn.net/files/3822/586/Multiplayer-Server-Pause-1.19-Fabric-1.1.0.jar"

# Not part of the official Voyager mod set. Measured need: the Bedrock player
# received ~100 "moved too quickly" corrections in under a minute, with a single
# delta of -3556 blocks in x, so the server kept snapping their position back.
# Geyser translates Bedrock movement to Java and the two disagree, which upstream
# has not resolved (GeyserMC/Geyser#3407 is still open), and the Spigot/Paper
# thresholds that issue mentions do not exist on a Fabric server. This mixin is
# the server-side fix: it removes the vanilla speed check that produced the
# corrections. Trade-off, stated plainly: it also stops the server rejecting an
# actually-too-fast client, which is acceptable on a private world with friends.
TOO_FAST_URL = "https://cdn.modrinth.com/data/w6JSkKSH/versions/ARqi4mgF/toofast-1.19-1.0.0.jar"
TOO_FAST_SHA256 = "f703fed1f9be34ede7444069ec1a8ebcd50aeaac89ff507bfc98a062f435418c"
SERVER_PAUSE_SHA256 = "4f92d4a4cef4627acbdfa1c601f0c58e392eee4159a69515fd0d622d3eeccd66"
BETTER_RESPAWN_URL = "https://cdn.modrinth.com/data/ZUQINcGy/versions/fabric-1.19-2.0.0/better-respawn-fabric-1.19-2.0.0.jar"
BETTER_RESPAWN_SHA256 = "9405e2ae53f031d5c32505dc3ee774c3f723c2d465230aaed7e6f7b3cdec0948"

VIAPROXY_URL = "https://github.com/ViaVersion/ViaProxy/releases/download/v3.4.13/ViaProxy-3.4.13.jar"
VIAPROXY_SHA256 = "4bbb6a6b9d3dd6a2028773ed44b97a98751e4a48e416aaad4bf6d9c860083be9"

GEYSER_URL = "https://download.geysermc.org/v2/projects/geyser/versions/2.11.3/builds/1247/downloads/viaproxy"
GEYSER_SHA256 = "85eea53e7fe58e4d97c2144a9d061577d37f619ab635227b68a948aa62d22924"

GO_URL = "https://go.dev/dl/go1.25.14.linux-amd64.tar.gz"
GO_SHA256 = "a21ae5633a269bcd7e90cf767e48225633795e99d831742cbf3397064fee7712"

JAVA_URL = "https://github.com/adoptium/temurin21-binaries/releases/download/jdk-21.0.12.1%2B1/OpenJDK21U-jre_x64_linux_hotspot_21.0.12.1_1.tar.gz"
JAVA_SHA256 = "2413149700df0f7d440500a84a8f764c535f21e5a5e87d38328b64eec2c5b500"
MC_JAVA_URL = "https://github.com/adoptium/temurin17-binaries/releases/download/jdk-17.0.20.1%2B1/OpenJDK17U-jre_x64_linux_hotspot_17.0.20.1_1.tar.gz"
MC_JAVA_SHA256 = "0b2b640e3046b64c8ec504de0ab9d91bb5610182bda21fad454681ce54d45a62"

BEDROCK_PROTOCOL = int(os.environ.get("BEDROCK_PROTOCOL", "2193"))
BEDROCK_VERSION = os.environ.get("BEDROCK_VERSION", "1.26.51")
WORLD_NAME = os.environ.get("MINECRAFT_WORLD_NAME", "Voyager AI World")
HOST_NAME = os.environ.get("MINECRAFT_HOST_NAME", "Voyager AI")


def _mkdirs() -> None:
    for path in (BASE, PERSIST_ROOT, JAVA_DIR, VIA_DIR, RELAY_DIR, WORLD_DIR, TOOLS_DIR, BIN_DIR):
        path.mkdir(parents=True, exist_ok=True)


def _log(message: str) -> None:
    _mkdirs()
    line = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {message}"
    print(line, flush=True)
    with LOG_PATH.open("a", encoding="utf-8", errors="replace") as fh:
        fh.write(line + "\n")


def _sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as fh:
        for chunk in iter(lambda: fh.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def _download(url: str, path: Path, sha256: str | None = None) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.exists() and (not sha256 or _sha256(path) == sha256):
        return path
    tmp = path.with_suffix(path.suffix + ".part")
    tmp.unlink(missing_ok=True)
    req = urllib.request.Request(url, headers={"User-Agent": "minecraft-friends-ai-zerogpu/1.0"})
    _log(f"download {url}")
    with urllib.request.urlopen(req, timeout=120) as response, tmp.open("wb") as out:
        shutil.copyfileobj(response, out)
    if sha256 and _sha256(tmp) != sha256:
        tmp.unlink(missing_ok=True)
        raise RuntimeError(f"SHA256 mismatch for {url}")
    tmp.replace(path)
    return path


def _wait_tcp(host: str, port: int, timeout: float) -> bool:
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with socket.create_connection((host, port), timeout=1):
                return True
        except OSError:
            time.sleep(1)
    return False


def _run_capture(args: list[str], cwd: Path, timeout: int = 300, env: dict[str, str] | None = None) -> str:
    _log("exec " + " ".join(args))
    proc = subprocess.run(
        args,
        cwd=str(cwd),
        env=env,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        timeout=timeout,
        check=False,
    )
    output = proc.stdout or ""
    for line in output.splitlines()[-80:]:
        _log("setup> " + line)
    if proc.returncode != 0:
        raise RuntimeError(f"command failed rc={proc.returncode}: {' '.join(args)}")
    return output


def _patch_scalar(text: str, key: str, value: str) -> str:
    pattern = rf"(?m)^(\s*{re.escape(key)}:\s*).*$"
    if re.search(pattern, text):
        return re.sub(pattern, lambda m: m.group(1) + value, text, count=1)
    return text + f"\n{key}: {value}\n"


def _relay_diagnostics(lines: list[str]) -> dict[str, Any]:
    """Expose relay milestones without leaking Xbox IDs, OAuth data or raw logs."""
    events = [line for line in lines if "friends>" in line]
    markers = {
        "session_live": "session is live - should now be visible on the friends tab",
        "session_ended": "session ended, restarting",
        "renewing": "restarting session for scheduled token renewal",
        "session_created": "xbox live session created",
    }
    latest = "not_seen"
    latest_at = None
    last_live_at = None
    for line in events:
        lowered = line.lower()
        for label, phrase in markers.items():
            if phrase in lowered:
                latest = label
                latest_at = line[1:20] if line.startswith("[") else None
                if label == "session_live":
                    last_live_at = latest_at
                break
    errors = [line.lower() for line in events if "level=error" in line.lower()
              or "level=warn" in line.lower()]
    reason = None
    if errors:
        last_error = errors[-1]
        for code, pattern in (
            ("owner_inactive", "owner isn't active"),
            ("unauthorized", "401"), ("forbidden", "403"),
            ("rate_limited", "429"), ("timeout", "timeout"),
            ("connection_refused", "connection refused"),
            ("signaling", "signaling"), ("presence", "presence"),
            ("auth", "auth"),
        ):
            if pattern in last_error:
                reason = code
                break
        reason = reason or "other"
    return {
        "session_live": latest == "session_live",
        "last_session_event": latest,
        "last_session_event_at": latest_at,
        "last_live_at": last_live_at,
        "last_error_category": reason,
        "session_failures": sum("session ended, restarting" in line for line in events),
        "signaling_warnings": sum("signaling unavailable" in line for line in events),
        "join_attempts": sum("player authenticated via xbox live" in line for line in events),
        "backend_joins": sum("relaying player into backend" in line for line in events),
        "backend_failures": sum("failed to dial backend" in line or "failed to log in to backend" in line
                                for line in events),
    }


RELAY_PING_PORT = 7777
RELAY_PING_URL = f"http://127.0.0.1:{RELAY_PING_PORT}/ping"


def _relay_player_latency(events) -> dict[str, Any]:
    """Report the joining player's real end-to-end RTT, as measured by the relay.

    nether2rak registers each connected player's client-leg and backend-leg
    rolling round-trip time and exposes the sum over a loopback-only HTTP API
    (bridge/pingserver.go). Nothing else measures what the player actually
    experiences, so publish it rather than guessing from CPU counters. The full
    XUID is a personal identifier and is deliberately not published.
    """
    player = xuid = None
    for line in events:
        if "player authenticated via xbox live" not in line:
            continue
        name = re.search(r"displayName=\"?([^\"\s]+)\"?", line)
        ident = re.search(r"xuid=(\d+)", line)
        if ident:
            player = name.group(1) if name else player
            xuid = ident.group(1)
    if not xuid:
        return {"available": False, "reason": "no authenticated player in the relay log yet"}
    try:
        with urllib.request.urlopen(f"{RELAY_PING_URL}?xuid={xuid}", timeout=3) as response:
            payload = json.loads(response.read().decode())
    except Exception as exc:
        return {
            "available": False,
            "player": player,
            "reason": f"{type(exc).__name__}: {exc}",
        }
    return {
        "available": True,
        "player": player,
        "xuid_suffix": xuid[-4:],
        "rtt_ms": int(payload.get("ms", 0)),
        "measured_by": "nether2rak client leg + backend leg, rolling average",
    }


def _offline_player_uuid(username: str) -> str:
    return str(uuid.UUID(bytes=hashlib.md5(
        f"OfflinePlayer:{username}".encode("utf-8")
    ).digest(), version=3))


def _carry_player_saves(previous: Path, current: Path) -> None:
    """Copy human player saves into a newly generated world.

    The Xbox OAuth token lives in RELAY_DIR and is never touched here, so a map
    change must not cost a player their session. Player saves are copied instead
    of the world itself: they only store a player's position, inventory and
    gamemode, which are valid in any world of the same dimension.

    The bot is deliberately excluded. A full bot inventory is what pins the
    official CurriculumAgent to its hardcoded full-inventory branch, so a fresh
    world has to start the bot from an empty save.
    """
    source = previous / "playerdata"
    if not source.is_dir():
        return
    destination = current / "playerdata"
    destination.mkdir(parents=True, exist_ok=True)
    bot_save = f"{_offline_player_uuid(BOT_USERNAME)}.dat"
    carried = []
    for save in sorted(source.glob("*.dat")):
        if save.name == bot_save:
            continue
        try:
            shutil.copy2(save, destination / save.name)
        except OSError as exc:
            _log(f"could not carry player save {save.name}: {exc}")
            continue
        carried.append(save.name)
    if carried:
        _log(f"Carried {len(carried)} player save(s) into the new world: {', '.join(carried)}")
    else:
        _log("No human player saves to carry into the new world")


def _server_performance(lines: list[str], now: datetime.datetime | None = None) -> dict[str, Any]:
    """Report vanilla tick-lag signals without treating movement warnings as TPS."""
    now = now or datetime.datetime.now()
    cutoff = (now - datetime.timedelta(minutes=5)).strftime("%Y-%m-%d %H:%M:%S")
    recent = [line for line in lines if line.startswith("[") and line[1:20] >= cutoff]
    tick_lag = [line for line in recent if "Can't keep up!" in line]
    delays = [int(match.group(1)) for line in tick_lag
              if (match := re.search(r"Running (\d+)ms or \d+ ticks behind", line))]
    movement = [line for line in recent
                if "moved too quickly" in line or "moved wrongly" in line]
    # The count alone cannot say whether this is one entity joining or the whole
    # server degrading. Measured: the counter went 0 -> 107 in under a minute and
    # then stopped, which is a burst, so the entity behind it has to be visible.
    # Deduplicate on the message, not the raw line: a burst carries a distinct
    # timestamp on every entry, so comparing raw lines would fill the payload with
    # five copies of one problem instead of showing five distinct ones.
    samples: list[str] = []
    seen: set[str] = set()
    for line in movement:
        trimmed = line.strip()[:200]
        key = re.sub(r"^\[[^\]]*\]\s*", "", trimmed)
        if key in seen:
            continue
        seen.add(key)
        samples.append(trimmed)
        if len(samples) >= 5:
            break
    return {
        "window_seconds": 300,
        "tick_lag_events": len(tick_lag),
        "max_reported_tick_delay_ms": max(delays, default=None),
        "movement_warnings": len(movement),
        "movement_warning_samples": samples,
        "last_tick_lag_at": tick_lag[-1][1:20] if tick_lag else None,
    }


def _process_rss_gib(pid: int) -> float | None:
    try:
        pages = int(Path(f"/proc/{pid}/statm").read_text().split()[1])
        return round(pages * os.sysconf("SC_PAGE_SIZE") / (1024**3), 3)
    except (OSError, ValueError, IndexError):
        return None


def _access_state(state: str, started_at: float | None, relay: dict[str, Any],
                  now: datetime.datetime | None = None) -> str:
    if state != "running":
        return state
    if relay.get("session_live"):
        return "ready"
    now = now or datetime.datetime.now()
    try:
        last_event = datetime.datetime.strptime(relay["last_session_event_at"], "%Y-%m-%d %H:%M:%S")
    except (KeyError, TypeError, ValueError):
        last_event = datetime.datetime.fromtimestamp(started_at) if started_at else now
    return "recovering" if (now - last_event).total_seconds() < 180 else "degraded"


def _ensure_admin_operator(path: Path, username: str) -> None:
    """Seed the exact offline-mode Java profile used by the Bedrock relay."""
    if not re.fullmatch(r"[A-Za-z0-9_]{1,16}", username):
        raise ValueError("Invalid Minecraft operator username")
    operators = json.loads(path.read_text(encoding="utf-8")) if path.exists() else []
    if not isinstance(operators, list):
        raise ValueError("Minecraft ops.json must be a list")
    identity = _offline_player_uuid(username)
    entry = {"uuid": identity, "name": username, "level": 4, "bypassesPlayerLimit": True}
    preserved = [item for item in operators if not (
        isinstance(item, dict) and (item.get("uuid") == identity or item.get("name") == username)
    )]
    preserved.append(entry)
    path.write_text(json.dumps(preserved, indent=2) + "\n", encoding="utf-8")


class MinecraftStack:
    def __init__(self) -> None:
        self.lock = threading.RLock()
        self.worker: threading.Thread | None = None
        self.processes: dict[str, subprocess.Popen[str]] = {}
        self.state = "stopped"
        self.error = ""
        self.started_at: float | None = None
        self.minecraft_ready = threading.Event()

    def _pump(self, name: str, proc: subprocess.Popen[str]) -> None:
        assert proc.stdout is not None
        try:
            for line in proc.stdout:
                clean = line.rstrip()
                _log(f"{name}> {clean}")
                if name == "minecraft" and "Done (" in clean:
                    self.minecraft_ready.set()
        except Exception as exc:
            _log(f"{name} log pump error: {exc!r}")

    def _spawn(self, name: str, args: list[str], cwd: Path, env: dict[str, str] | None = None) -> subprocess.Popen[str]:
        _log(f"start {name}: {' '.join(args)}")
        proc = subprocess.Popen(
            args,
            cwd=str(cwd),
            env=env,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
        )
        self.processes[name] = proc
        threading.Thread(target=self._pump, args=(name, proc), daemon=True, name=f"log-{name}").start()
        return proc

    def _stop_process(self, name: str) -> None:
        proc = self.processes.get(name)
        if not proc or proc.poll() is not None:
            return
        try:
            if name == "minecraft" and proc.stdin:
                proc.stdin.write("stop\n")
                proc.stdin.flush()
                proc.wait(timeout=20)
                return
        except Exception:
            pass
        try:
            proc.terminate()
            proc.wait(timeout=12)
        except Exception:
            try:
                proc.kill()
            except Exception:
                pass

    def _ensure_java(self) -> Path:
        root = TOOLS_DIR / "jre21"
        existing = list(root.glob("*/bin/java")) if root.exists() else []
        if existing:
            return existing[0]
        archive = _download(JAVA_URL, TOOLS_DIR / "temurin21-jre.tar.gz", JAVA_SHA256)
        if root.exists():
            shutil.rmtree(root)
        root.mkdir(parents=True, exist_ok=True)
        with tarfile.open(archive, "r:gz") as tf:
            tf.extractall(root)
        found = list(root.glob("*/bin/java"))
        if not found:
            raise RuntimeError("Temurin JRE 21 extraction failed")
        found[0].chmod(0o755)
        version = _run_capture([str(found[0]), "-version"], BASE, timeout=30)
        _log("portable Java ready: " + version.splitlines()[0])
        return found[0]

    def _ensure_minecraft_java(self) -> Path:
        root = TOOLS_DIR / "jre17"
        existing = list(root.glob("*/bin/java")) if root.exists() else []
        if existing:
            return existing[0]
        archive = _download(MC_JAVA_URL, TOOLS_DIR / "temurin17-jre.tar.gz", MC_JAVA_SHA256)
        if root.exists():
            shutil.rmtree(root)
        root.mkdir(parents=True, exist_ok=True)
        with tarfile.open(archive, "r:gz") as tf:
            tf.extractall(root)
        found = list(root.glob("*/bin/java"))
        if not found:
            raise RuntimeError("Temurin JRE 17 extraction failed")
        found[0].chmod(0o755)
        version = _run_capture([str(found[0]), "-version"], BASE, timeout=30)
        _log("Minecraft Java 17 ready: " + version.splitlines()[0])
        return found[0]

    def _ensure_persistent_world(self) -> None:
        WORLD_DIR.mkdir(parents=True, exist_ok=True)
        # level.dat only exists once the server has generated the world, so its
        # absence means this is the first boot of a new map.
        if not (WORLD_DIR / "level.dat").exists() and PREVIOUS_WORLD_DIR.is_dir():
            _carry_player_saves(PREVIOUS_WORLD_DIR, WORLD_DIR)
        # Probe the exact temporary-file + rename pattern used by player saves.
        # This never touches a player's .dat file or existing world data.
        player_dir = WORLD_DIR / "playerdata"
        probe = None
        target = player_dir / ".voyager-storage-probe"
        try:
            player_dir.mkdir(parents=True, exist_ok=True)
            marker = player_dir / ".keep"
            if not marker.exists():
                marker.write_text("Preserve playerdata directory in object storage.\n", encoding="utf-8")
            fd, filename = tempfile.mkstemp(prefix=".voyager-probe-", suffix=".dat", dir=player_dir)
            probe = Path(filename)
            with os.fdopen(fd, "wb") as handle:
                handle.write(b"voyager storage check")
            probe.replace(target)
            assert target.read_bytes() == b"voyager storage check"
            _log("Player storage probe: temporary create/write/rename/read succeeded")
        except Exception as exc:
            _log(f"PLAYER STORAGE PROBE FAILED: {type(exc).__name__}: {exc}")
        finally:
            for candidate in (probe, target):
                if candidate is not None:
                    try:
                        candidate.unlink(missing_ok=True)
                    except OSError:
                        pass
        WORLD_DIR.mkdir(parents=True, exist_ok=True)
        world_link = JAVA_DIR / "world"
        if world_link.is_symlink():
            try:
                if world_link.resolve() == WORLD_DIR.resolve():
                    return
            except OSError:
                pass
            world_link.unlink(missing_ok=True)
        elif world_link.exists():
            if not any(WORLD_DIR.iterdir()):
                shutil.copytree(world_link, WORLD_DIR, dirs_exist_ok=True)
                _log("Migrated existing world into persistent storage")
            shutil.rmtree(world_link)
        world_link.symlink_to(WORLD_DIR, target_is_directory=True)
        _log(f"Persistent Minecraft world linked: {world_link} -> {WORLD_DIR}")

    def _ensure_voyager_mods(self) -> None:
        mods_dir = JAVA_DIR / "mods"
        mods_dir.mkdir(parents=True, exist_ok=True)
        mods = (
            ("fabric-api-0.58.0+1.19.jar", FABRIC_API_URL, FABRIC_API_SHA256),
            ("completeconfig-2.0.0.jar", COMPLETE_CONFIG_URL, COMPLETE_CONFIG_SHA256),
            ("Multiplayer-Server-Pause-1.19-Fabric-1.1.0.jar", SERVER_PAUSE_URL, SERVER_PAUSE_SHA256),
            # Local addition, not an upstream requirement: see TOO_FAST_URL.
            ("toofast-1.19-1.0.0.jar", TOO_FAST_URL, TOO_FAST_SHA256),
        )
        for filename, url, sha256 in mods:
            _download(url, mods_dir / filename, sha256)

        # The pinned Better Respawn 1.19 mixin dereferences an uninitialized
        # respawnDimension while saving a newly connected Creative player.
        # Remove only this faulty mod from the ephemeral Fabric installation;
        # never edit the persistent world, playerdata, or Voyager checkpoint.
        faulty_mod = mods_dir / "better-respawn-fabric-1.19-2.0.0.jar"
        if faulty_mod.exists():
            faulty_mod.unlink()
            _log("Removed faulty Better Respawn 1.19 mod from Fabric mods directory")
        _log(
            "Official Voyager Fabric mods ready: Fabric API 0.58.0+1.19, "
            "CompleteConfig 2.0.0, Multiplayer Server Pause 1.1.0, "
            "TooFast 1.0.0 (local, removes the Bedrock movement corrections). "
            "Better Respawn 1.19-2.0.0 is disabled due to its player-save NPE. Mod Menu is client-only and is "
            "intentionally omitted on this headless dedicated server."
        )

    def _ensure_fabric(self) -> Path:
        java = self._ensure_minecraft_java()
        installer = _download(FABRIC_INSTALLER_URL, TOOLS_DIR / "fabric-installer-0.11.2.jar")
        JAVA_DIR.mkdir(parents=True, exist_ok=True)
        launcher = JAVA_DIR / "fabric-server-launch.jar"
        if not launcher.exists():
            _run_capture(
                [
                    str(java), "-jar", str(installer), "server",
                    "-mcversion", FABRIC_MC_VERSION,
                    "-loader", FABRIC_LOADER_VERSION,
                    "-downloadMinecraft",
                    "-dir", str(JAVA_DIR),
                ],
                JAVA_DIR,
                timeout=300,
            )
        self._ensure_voyager_mods()
        (JAVA_DIR / "eula.txt").write_text("eula=true\n", encoding="utf-8")
        props = {
            "server-ip": "127.0.0.1",
            "server-port": "25565",
            "online-mode": "false",
            "enforce-secure-profile": "false",
            "enable-command-block": "true",
            "spawn-protection": "0",
            "allow-flight": "true",
            "gamemode": "creative",
            "force-gamemode": "true",
            "difficulty": "peaceful",
            "max-players": str(MAX_PLAYERS),
            "op-permission-level": "4",
            "view-distance": "8",
            "simulation-distance": "6",
            "level-name": "world",
            "level-type": "minecraft:flat",
            "generator-settings": FLAT_GENERATOR_SETTINGS,
            "generate-structures": "true",
            "motd": "Voyager AI + Bedrock Friends",
        }
        (JAVA_DIR / "server.properties").write_text(
            "\n".join(f"{k}={v}" for k, v in props.items()) + "\n",
            encoding="utf-8",
        )
        _ensure_admin_operator(JAVA_DIR / "ops.json", ADMIN_USERNAME)
        self._ensure_persistent_world()
        return launcher

    def _start_fabric(self) -> None:
        launcher = self._ensure_fabric()
        self.minecraft_ready.clear()
        READY_FILE.unlink(missing_ok=True)
        proc = self._spawn(
            "minecraft",
            [str(self._ensure_minecraft_java()), "-Xms1G", "-Xmx8G",
             "--add-exports=java.base/jdk.internal.org.objectweb.asm=ALL-UNNAMED",
             f"-javaagent:{Path(__file__).resolve().parent / 'diagnostics/player-save-diagnostics.jar'}",
             "-jar", str(launcher), "nogui"],
            JAVA_DIR,
        )
        if not _wait_tcp("127.0.0.1", 25565, 120):
            raise RuntimeError("Minecraft Java 1.19 did not open 127.0.0.1:25565")
        if not self.minecraft_ready.wait(timeout=180):
            raise RuntimeError("Minecraft Java opened 25565 but world did not finish loading")
        if proc.stdin:
            for cmd in (
                f"op {BOT_USERNAME}",
                f"gamemode creative {BOT_USERNAME}",
                "gamerule keepInventory true",
                "gamerule doDaylightCycle false",
                # The cycle is frozen, so set the time once: the world used to be
                # frozen at night, which is what the bot reported in every
                # observation and what a player saw on join.
                "time set day",
                "difficulty peaceful",
            ):
                proc.stdin.write(cmd + "\n")
            proc.stdin.flush()
        READY_FILE.write_text("ready\n", encoding="utf-8")
        _log("Minecraft Java 1.19/Fabric fully ready on 127.0.0.1:25565")

    def _ensure_viaproxy(self) -> tuple[Path, Path]:
        jar = _download(VIAPROXY_URL, VIA_DIR / "ViaProxy-3.4.13.jar", VIAPROXY_SHA256)
        plugin_dir = VIA_DIR / "plugins"
        plugin_dir.mkdir(parents=True, exist_ok=True)
        _download(GEYSER_URL, plugin_dir / "Geyser-ViaProxy.jar", GEYSER_SHA256)
        cfg = VIA_DIR / "viaproxy.yml"
        if not cfg.exists():
            proc = subprocess.Popen(
                [str(self._ensure_java()), "-jar", str(jar), "config", str(cfg)],
                cwd=str(VIA_DIR),
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
            )
            deadline = time.time() + 25
            while time.time() < deadline and not cfg.exists() and proc.poll() is None:
                time.sleep(1)
            if proc.poll() is None:
                proc.terminate()
                try:
                    proc.wait(timeout=8)
                except Exception:
                    proc.kill()
            if not cfg.exists():
                raise RuntimeError("ViaProxy did not generate viaproxy.yml")
        text = cfg.read_text(encoding="utf-8", errors="replace")
        text = _patch_scalar(text, "bind-address", "127.0.0.1:25568")
        text = _patch_scalar(text, "target-address", "127.0.0.1:25565")
        text = _patch_scalar(text, "proxy-online-mode", "false")
        text = _patch_scalar(text, "auth-method", "none")
        cfg.write_text(text, encoding="utf-8")
        return jar, cfg

    def _patch_geyser(self, path: Path) -> None:
        data: dict[str, Any] = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
        bedrock = data.setdefault("bedrock", {})
        bedrock["address"] = "127.0.0.1"
        bedrock["port"] = 19132
        bedrock["clone-remote-port"] = False
        java = data.setdefault("java", {})
        java["auth-type"] = "offline"
        advanced = data.setdefault("advanced", {})
        adv_bedrock = advanced.setdefault("bedrock", {})
        adv_bedrock["broadcast-port"] = 19132
        adv_bedrock["validate-bedrock-login"] = False
        data["debug-mode"] = False
        path.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8")
        _log("Geyser configured loopback-only; Xbox/NetherNet relay remains the auth boundary")

    def _start_viaproxy(self) -> None:
        jar, cfg = self._ensure_viaproxy()
        geyser_cfg = VIA_DIR / "plugins" / "Geyser" / "config.yml"
        if not geyser_cfg.exists():
            proc = self._spawn("viaproxy-bootstrap", [str(self._ensure_java()), "-jar", str(jar), "config", str(cfg)], VIA_DIR)
            deadline = time.time() + 35
            while time.time() < deadline and not geyser_cfg.exists() and proc.poll() is None:
                time.sleep(1)
            self._stop_process("viaproxy-bootstrap")
            if not geyser_cfg.exists():
                raise RuntimeError("Geyser-ViaProxy did not generate plugins/Geyser/config.yml")
        self._patch_geyser(geyser_cfg)
        proc = self._spawn("viaproxy", [str(self._ensure_java()), "-jar", str(jar), "config", str(cfg)], VIA_DIR)
        time.sleep(12)
        if proc.poll() is not None:
            raise RuntimeError(f"ViaProxy/Geyser exited rc={proc.returncode}")
        _log("ViaProxy + Geyser running; Bedrock backend is 127.0.0.1:19132")

    def _ensure_go(self) -> Path:
        go = TOOLS_DIR / "go" / "bin" / "go"
        if go.exists():
            return go
        archive = _download(GO_URL, TOOLS_DIR / "go1.25.14.linux-amd64.tar.gz", GO_SHA256)
        target = TOOLS_DIR / "go"
        if target.exists():
            shutil.rmtree(target)
        with tarfile.open(archive, "r:gz") as tf:
            tf.extractall(TOOLS_DIR)
        if not go.exists():
            raise RuntimeError("Go toolchain extraction failed")
        return go

    def _ensure_nether2rak(self) -> Path:
        binary = BIN_DIR / "nether2rak"
        if binary.exists():
            return binary
        go = self._ensure_go()
        src = Path("/app/third_party/nether2rak")
        if not src.exists():
            src = Path(__file__).resolve().parent / "third_party" / "nether2rak"
        if not src.exists():
            raise RuntimeError("Vendored nether2rak source is missing")
        env = os.environ.copy()
        env.update(
            {
                "PATH": f"{go.parent}:{env.get('PATH', '')}",
                "GOTOOLCHAIN": "local",
                "GOCACHE": str(BASE / "go-cache"),
                "GOMODCACHE": str(BASE / "go-mod-cache"),
                "CGO_ENABLED": "0",
            }
        )
        _run_capture([str(go), "build", "-trimpath", "-o", str(binary), "."], src, timeout=900, env=env)
        binary.chmod(0o755)
        return binary

    def _write_relay_config(self) -> Path:
        cfg = {
            "backend_transport": "raknet",
            "geyser_address": "127.0.0.1:19132",
            "nethernet_backend_address": "http://127.0.0.1:19134",
            "host_name": HOST_NAME,
            "world_name": WORLD_NAME,
            "max_players": MAX_PLAYERS,
            "fake_player_count": 0,
            "protocol": BEDROCK_PROTOCOL,
            "version": BEDROCK_VERSION,
            "update_interval_seconds": 30,
            "allowed_xuids": [],
            "compression": "snappy",
            "compression_threshold": 256,
            "fix_native_bds_persistence": False,
            "disable_client_encryption": False,
            "relay_direct_ip": False,
            "relay_direct_ip_listen_address": ":19134",
            "invite_port": 7782,
            "ping_port": 7777,
            "pprof_port": 6060,
            "extra_broadcasts": [],
        }
        path = RELAY_DIR / "config.json"
        path.write_text(json.dumps(cfg, indent=2), encoding="utf-8")
        return path

    def _start_relay(self) -> None:
        binary = self._ensure_nether2rak()
        self._write_relay_config()
        proc = self._spawn("friends", [str(binary)], RELAY_DIR)
        time.sleep(4)
        if proc.poll() is not None:
            raise RuntimeError(f"Xbox Friends relay exited rc={proc.returncode}")
        _log("Xbox Friends relay started; Microsoft device-code login may be required")

    def _boot(self) -> None:
        with self.lock:
            self.state = "starting"
            self.error = ""
            self.started_at = time.time()
        try:
            _mkdirs()
            _log("=== Minecraft stack bootstrap ===")
            self._start_fabric()
            self._start_viaproxy()
            self._start_relay()
            with self.lock:
                self.state = "running"
            _log("=== Minecraft stack running ===")
        except Exception as exc:
            with self.lock:
                self.state = "error"
                self.error = repr(exc)
            _log(f"STACK ERROR: {exc!r}")

    def start_async(self) -> dict[str, Any]:
        with self.lock:
            if self.worker and self.worker.is_alive():
                return self.status()
            if any(p.poll() is None for p in self.processes.values()) and self.state == "running":
                return self.status()
            self.worker = threading.Thread(target=self._boot, daemon=True, name="minecraft-stack")
            self.worker.start()
        return self.status()

    def stop(self) -> dict[str, Any]:
        with self.lock:
            for name in ("friends", "viaproxy", "viaproxy-bootstrap", "minecraft"):
                self._stop_process(name)
            self.state = "stopped"
        _log("Minecraft stack stopped")
        return self.status()

    def tail(self, lines: int = 160) -> str:
        if not LOG_PATH.exists():
            return "Sem logs do Minecraft ainda."
        items = LOG_PATH.read_text(encoding="utf-8", errors="replace").splitlines()
        return "\n".join(items[-max(10, min(lines, 500)):])

    def status(self) -> dict[str, Any]:
        with self.lock:
            processes = {
                name: {"pid": proc.pid, "running": proc.poll() is None, "returncode": proc.poll()}
                for name, proc in self.processes.items()
            }
            for name, process in processes.items():
                if process["running"]:
                    process["rss_gib"] = _process_rss_gib(process["pid"])
            try:
                all_lines = LOG_PATH.read_text(encoding="utf-8", errors="replace").splitlines()
                boot = next((i for i in range(len(all_lines) - 1, -1, -1)
                             if "=== Minecraft stack bootstrap ===" in all_lines[i]), 0)
                server_lines = [line for line in all_lines[boot:] if "minecraft>" in line]
            except OSError:
                all_lines = []
                server_lines = []
            log = "\n".join(all_lines[-120:]) if all_lines else "Sem logs do Minecraft ainda."
            relay_health = _relay_diagnostics(all_lines[boot:] if all_lines else [])
            access_state = _access_state(self.state, self.started_at, relay_health)
            performance = _server_performance(server_lines)
            try:
                operators = json.loads((JAVA_DIR / "ops.json").read_text(encoding="utf-8"))
                expected_uuid = _offline_player_uuid(ADMIN_USERNAME)
                admin_operator = any(
                    isinstance(entry, dict) and entry.get("uuid") == expected_uuid
                    and entry.get("name") == ADMIN_USERNAME and entry.get("level") == 4
                    for entry in operators
                )
            except (OSError, ValueError, TypeError):
                admin_operator = False
            admin_player_file_exists = (WORLD_DIR / "playerdata" / f"{_offline_player_uuid(ADMIN_USERNAME)}.dat").exists()
            try:
                properties = (JAVA_DIR / "server.properties").read_text(encoding="utf-8")
                java_player_limit = int(re.search(r"(?m)^max-players=(\d+)$", properties).group(1))
            except (OSError, AttributeError, ValueError):
                java_player_limit = None
            try:
                xbox_player_limit = int(json.loads((RELAY_DIR / "config.json").read_text(encoding="utf-8"))["max_players"])
            except (OSError, ValueError, KeyError, TypeError):
                xbox_player_limit = None
            offline_uuid = uuid.UUID(bytes=hashlib.md5(
                f"OfflinePlayer:{BOT_USERNAME}".encode("utf-8")
            ).digest(), version=3)
            player_file = WORLD_DIR / "playerdata" / f"{offline_uuid}.dat"
            try:
                player_stat = player_file.stat()
            except OSError:
                player_stat = None
            player_save = {
                "better_respawn_jar_present": (JAVA_DIR / "mods" / "better-respawn-fabric-1.19-2.0.0.jar").exists(),
                "failed_saves_since_boot": sum("Failed to save player data" in line for line in server_lines),
                "respawn_npe_since_boot": sum("respawnDimension" in line and "null" in line
                                             for line in server_lines),
                "player_file_exists": player_stat is not None,
                "player_file_size": player_stat.st_size if player_stat else None,
                "player_file_mtime": player_stat.st_mtime if player_stat else None,
            }
            login_url = None
            device_code = None
            url_match = re.findall(r"https://(?:www\.)?microsoft\.com/link", log, flags=re.I)
            if url_match:
                login_url = url_match[-1]
            for pattern in (
                r"(?:code|c[oó]digo)\s*[:=]\s*([A-Z0-9-]{6,16})",
                r"enter\s+([A-Z0-9-]{6,16})\s+(?:to|at)",
                r"using\s+the\s+code\s+([A-Z0-9-]{6,16})",
            ):
                found = re.findall(pattern, log, flags=re.I)
                if found:
                    device_code = found[-1].upper()
                    break
            return {
                "state": self.state,
                "error": self.error or None,
                "started_at": self.started_at,
                "java_server": "127.0.0.1:25565",
                "bedrock_backend": "127.0.0.1:19132",
                "access": "Xbox/Microsoft Friends via NetherNet; no public UDP tunnel",
                "auth_persistence": str(RELAY_DIR / "token.json"),
                "auth_token_cached": (RELAY_DIR / "token.json").exists(),
                "persistent_root": str(PERSIST_ROOT),
                "world_persistence": str(WORLD_DIR),
                "world_profile": WORLD_PROFILE,
                "world_surface_y": WORLD_SURFACE_Y,
                "player_save": player_save,
                "relay_health": relay_health,
                "relay_latency": _relay_player_latency(all_lines[boot:] if all_lines else []),
                "access_state": access_state,
                "performance": performance,
                "player_limit": {"java": java_player_limit, "xbox_friends": xbox_player_limit},
                "admin_operator": admin_operator,
                "admin_player_file_exists": admin_player_file_exists,
                "bot_username": BOT_USERNAME,
                "bot_display_name": "Naldo Bot",
                "bedrock_protocol": BEDROCK_PROTOCOL,
                "bedrock_version": BEDROCK_VERSION,
                "microsoft_login_url": login_url,
                "microsoft_device_code": device_code,
                "processes": processes,
                "log_tail": log,
            }


STACK = MinecraftStack()