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-20261003-normal-v1"
# Generator settings are only read when the world is created, so replacing the map
# means pointing at a new directory. The previous worlds are kept so the human
# player's Bedrock session can be carried over instead of being reset, and so the
# old maps can be revisited. Nothing here ever deletes a world directory.
PREVIOUS_WORLD_DIR = PERSIST_ROOT / "world-20260928-ores-y64-v1"
# The new map is normal terrain (no flat resource layers). WORLD_PROFILE selects
# the server.properties generator; the flat constants below stay for the old maps
# and for the map-contract tests.
WORLD_PROFILE = "minecraft:normal"
# Reference surface for status output. On normal terrain the real spawn Y comes
# from the generated world, never from this constant.
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],
        "lakes": False,
        "features": False,
        "structure_overrides": [],
    },
    separators=(",", ":"),
)


def _server_properties() -> dict[str, str]:
    """Build server.properties from WORLD_PROFILE.



    Flat worlds need the layer stack; normal terrain must NOT receive flat

    generator settings (they would flatten the new map). Extracted so the map

    contract can be tested without booting a server.

    """
    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",
        # Default gamemode is Creative so human Bedrock players keep a
        # building-friendly mode, but force-gamemode is false so Naldo_Bot
        # keeps its assigned Survival mode (set at bootstrap/respawn by the
        # AltoClef supervisor). Difficulty Normal: blazes/endermen must exist
        # for the survival speedrun (eyes, rods, pearls).
        "gamemode": "creative",
        "force-gamemode": "false",
        "difficulty": "normal",
        "max-players": str(MAX_PLAYERS),
        "op-permission-level": "4",
        "view-distance": "8",
        "simulation-distance": "6",
        "level-name": "world",
        "motd": "Voyager AI + Bedrock Friends",
    }
    if WORLD_PROFILE == "minecraft:flat":
        props.update({
            "level-type": "minecraft:flat",
            "generator-settings": FLAT_GENERATOR_SETTINGS,
            "generate-structures": "false",
        })
    else:
        # Normal terrain: default overworld with structures; no flat layers.
        props.update({
            "level-type": "minecraft:normal",
            "generate-structures": "true",
        })
    return props
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"
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 _tail_lines(path: Path, max_lines: int = 5000) -> list[str]:
    """Read at most max_lines trailing lines without loading a days-long log.

    status() runs on every supervisor tick and UI click, so a full read of an
    ever-growing stack.log would stall the watchdog it is meant to serve.
    """
    try:
        with path.open("rb") as fh:
            fh.seek(0, os.SEEK_END)
            size = fh.tell()
            chunk_size = 65536
            data = b""
            position = size
            while position > 0 and data.count(b"\n") <= max_lines:
                step = min(chunk_size, position)
                position -= step
                fh.seek(position)
                data = fh.read(step) + data
                if len(data) > (max_lines + 1) * 512:
                    break
        return data.decode("utf-8", errors="replace").splitlines()[-max_lines:]
    except OSError:
        return []


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),
    }


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


def _should_carry_player_saves(current: Path, previous: Path) -> bool:
    """Whether an old world's player saves may be reused in a new directory.



    Only flat-map replacements reuse saves. A flat->normal move must spawn

    everyone at the new map's natural spawn: importing an old position drops

    the player inside a mountain, far from Naldo. Old playerdata and

    inventories always stay intact in the old world (plus operator backup).

    """
    return (
        WORLD_PROFILE == "minecraft:flat"
        and not (current / "level.dat").exists()
        and previous.is_dir()
    )


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))]
    return {
        "window_seconds": 300,
        "tick_lag_events": len(tick_lag),
        "max_reported_tick_delay_ms": max(delays, default=None),
        "movement_warnings": sum("moved too quickly" in line or "moved wrongly" in line
                                 for line in recent),
        "last_tick_lag_at": tick_lag[-1][1:20] if tick_lag else None,
    }


def _recent_players(server_lines: list[str]) -> dict[str, Any]:
    """Read-only presence diagnostic from this boot's server log.



    Counts joins/leaves without sending any console command. Only counts and

    timestamps are published, never player names, so no identity leaks through

    status. Exact live positions still need the server console/RCON.

    """
    joins = sum("joined the game" in line for line in server_lines)
    leaves = sum("left the game" in line for line in server_lines)

    def _last(marker: str) -> str | None:
        for line in reversed(server_lines):
            if marker in line and line.startswith("["):
                return line[1:20]
        return None

    return {
        "joins_this_boot": joins,
        "leaves_this_boot": leaves,
        "estimated_online": max(0, joins - leaves),
        "last_join_at": _last("joined the game"),
        "last_leave_at": _last("left the game"),
    }


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()
        self.failed_saves_since_boot = 0
        self.respawn_npe_since_boot = 0

    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":
                    with self.lock:
                        if "Failed to save player data" in clean:
                            self.failed_saves_since_boot += 1
                        if "respawnDimension" in clean and "null" in clean:
                            self.respawn_npe_since_boot += 1
                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 _should_carry_player_saves(WORLD_DIR, PREVIOUS_WORLD_DIR):
            _carry_player_saves(PREVIOUS_WORLD_DIR, WORLD_DIR)
        elif WORLD_PROFILE != "minecraft:flat":
            _log("Normal terrain: no playerdata import; everyone spawns fresh, old saves stay in the old world")
        # 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),
        )
        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. "
            "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_properties()
        (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 _reset_boot_counters(self) -> None:
        with self.lock:
            self.failed_saves_since_boot = 0
            self.respawn_npe_since_boot = 0

    def _start_fabric(self) -> None:
        launcher = self._ensure_fabric()
        self.minecraft_ready.clear()
        self._reset_boot_counters()
        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}",
                # Naldo plays Survival on Normal; humans keep their own mode
                # because force-gamemode is false. Daylight cycle stays on:
                # endermen (pearls) only spawn at night. keepInventory stays as
                # configured (existing setting, retained on request).
                f"gamemode survival {BOT_USERNAME}",
                "gamerule keepInventory true",
                "gamerule doDaylightCycle true",
                "difficulty normal",
            ):
                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 save_all(self, timeout: float = 30.0) -> dict[str, Any]:
        """Flush the world to disk without stopping anything.



        Best-effort console write (save-all, then the 1.19 save-all flush).

        Never raises; reports whether the server accepted input. Call this

        before any redeploy, then stop() (vanilla saves again on "stop").

        """
        with self.lock:
            proc = self.processes.get("minecraft")
            running = bool(proc and proc.poll() is None)
            if not running or not proc.stdin:
                return {"saved": False, "reason": "minecraft not running"}
            try:
                proc.stdin.write("save-all\n")
                proc.stdin.flush()
                time.sleep(2)
                proc.stdin.write("save-all flush\n")
                proc.stdin.flush()
            except Exception as exc:
                return {"saved": False, "reason": f"{type(exc).__name__}: {exc}"[:160]}
        deadline = time.time() + timeout
        while time.time() < deadline:
            time.sleep(2)
            tail = _tail_lines(LOG_PATH, 60)
            if any("Saved the game" in line or "Saved the world" in line for line in tail[-10:]):
                return {"saved": True, "reason": None}
        return {"saved": False, "reason": "no save confirmation in log yet (flush was requested)"}

    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 = _tail_lines(LOG_PATH, max(10, min(lines, 500)))
        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"])
            all_lines = _tail_lines(LOG_PATH)
            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]
            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)
            presence = _recent_players(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": self.failed_saves_since_boot,
                "respawn_npe_since_boot": self.respawn_npe_since_boot,
                "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,
                "access_state": access_state,
                "performance": performance,
                "presence": presence,
                "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()