agentmc/mcfold/server.py

"""空白测试服的起停与探活（mcfold M0）。

## 为什么是 vanilla 1.21.1 / 0 mods

用户 2026-09-26 定向：起一个空白新服测功能，不用现役的。
理由不只是"干净"，而是**可预期**：

- 现役服 `/home/minecraft`：world 15G、mods 64 个、neoforge。
  任何协议层改动都可能污染它，而且**测不出边界**（模组太多、状态太杂）。
- 我们要验的是**协议与行为**：已知方块集、已知出生点、可重置的世界。

## 版本隔离（三层，与我们给 agent 建的实例同一套口径）

    store/vanilla-1.21.1/server/     内容根  服务端 jar + 配置 + world（可只读共享）
    runtime/test-server/             运行期  日志 / 截图 / pid

★ 端口用 **25566**（避开现役 25565）。
★ 日志在运行期层，不在内容根 —— 那是运行产物。

## 用法

    python3 -m mcfold.server start     # 起（首次会生成世界）
    python3 -m mcfold.server status    # 探活（协议级状态查询）
    python3 -m mcfold.server stop
    python3 -m mcfold.server reset     # 删世界，下次 start 重生成
"""
from __future__ import annotations

import json
import os
import socket
import struct
import subprocess
import sys
import time
from pathlib import Path

# ── 三层路径（与 mcview/lib/versions.cjs 同一套口径）─────────────
MC_VERSION = "1.21.1"
PROTOCOL = 767                     # 1.21.1 的协议号（实测自 status 查询）
PORT = 25566                       # 避开现役 25565

def _base() -> Path:
    return Path(os.environ.get("MCVIEW_BASE") or "/home/newqqagent/mcview")

CONTENT = _base() / "store" / f"vanilla-{MC_VERSION}" / "server"
RUNTIME = _base() / "runtime" / "test-server"
LOG = RUNTIME / "server.log"
PIDFILE = RUNTIME / "server.pid"
JAR = CONTENT / "server.jar"
PROPS = CONTENT / "server.properties"
EULA = CONTENT / "eula.txt"

# 固定配置：可预期优先。★ 不写行内注释（properties 里行内 # 不合法，
# 写成 "k=v  # c" 会被解析成值的一部分）
SERVER_PROPERTIES = """\
# 空白测试服（mcfold M0）—— 由 mcfold/server.py 生成，改动会被覆盖
# 目的：验协议与行为，要可预期 => 固定种子 + 超平坦 + 无随机要素
level-seed=mcfold-test-2026
level-name=world
level-type=minecraft\\:flat
online-mode=false
server-port={port}
motd=mcfold test server
max-players=4
view-distance=6
simulation-distance=6
spawn-protection=0
enable-command-block=false
difficulty=peaceful
gamemode=creative
spawn-monsters=false
generate-structures=false
sync-chunk-writes=false
"""

SERVER_URL = ("https://piston-data.mojang.com/v1/objects/"
              "59353fb40c36d304f2035d51e7d6e6baa98dc05c/server.jar")
SERVER_SHA1 = "59353fb40c36d304f2035d51e7d6e6baa98dc05c"


# ── 协议级探活（不依赖任何第三方库）────────────────────────────
def _varint(n: int) -> bytes:
    out = b""
    while True:
        b = n & 0x7F
        n >>= 7
        out += bytes([b | 0x80]) if n else bytes([b])
        if not n:
            return out


def _read_varint(sock) -> int | None:
    n = shift = 0
    while True:
        b = sock.recv(1)
        if not b:
            return None
        n |= (b[0] & 0x7F) << shift
        if not b[0] & 0x80:
            return n
        shift += 7


def status(host: str = "127.0.0.1", port: int = PORT) -> dict | None:
    """Server List Ping（state=1）—— 证明「协议通了」，不只是「端口开着」。

    ★ 这一步是 M0 的验收判据：端口能连 ≠ 服务端是 MC。
      能正确回 version/protocol/motd 才算真起来。
    """
    try:
        s = socket.create_connection((host, port), timeout=8)
    except OSError:
        return None
    try:
        hb = host.encode()
        hs = (_varint(0) + _varint(PROTOCOL) + _varint(len(hb)) + hb
              + struct.pack(">H", port) + _varint(1))
        s.sendall(_varint(len(hs)) + hs)
        s.sendall(_varint(1) + _varint(0))            # status request
        ln = _read_varint(s)
        if not ln:
            return None
        pkt = s.recv(ln)
        i = pkt.find(b"{")
        if i < 0:
            return None
        d = json.loads(pkt[i:].decode("utf-8", "replace"))
        desc = d.get("description")
        if isinstance(desc, dict):
            desc = desc.get("text") or desc.get("translate") or str(desc)
        return {
            "version": d["version"]["name"],
            "protocol": d["version"]["protocol"],
            "motd": desc,
            "players_online": d["players"]["online"],
            "players_max": d["players"]["max"],
        }
    except Exception:
        return None
    finally:
        s.close()


# ── 起停 ────────────────────────────────────────────────────
def _pid() -> int | None:
    if PIDFILE.is_file():
        try:
            pid = int(PIDFILE.read_text().strip())
            os.kill(pid, 0)          # 只探测存活，不发信号
            return pid
        except (ValueError, ProcessLookupError, PermissionError):
            return None
    return None


def write_config() -> None:
    CONTENT.mkdir(parents=True, exist_ok=True)
    RUNTIME.mkdir(parents=True, exist_ok=True)
    EULA.write_text("eula=true\n")
    PROPS.write_text(SERVER_PROPERTIES.format(port=PORT))


def ensure_jar() -> None:
    """确保 server.jar 在位且 sha1 正确。"""
    if JAR.is_file():
        import hashlib
        h = hashlib.sha1(JAR.read_bytes()).hexdigest()
        if h == SERVER_SHA1:
            return
        print(f"  server.jar sha1 不符（{h}）⇒ 重新下载")
    print(f"  下载 server.jar（{MC_VERSION}）…")
    import urllib.request
    urllib.request.urlretrieve(SERVER_URL, JAR)
    import hashlib
    h = hashlib.sha1(JAR.read_bytes()).hexdigest()
    if h != SERVER_SHA1:
        raise RuntimeError(f"sha1 校验失败: {h} != {SERVER_SHA1}")
    print(f"  ✓ sha1 一致（{JAR.stat().st_size // 1024 // 1024} MB）")


def start(wait: float = 90.0) -> dict:
    if (p := _pid()):
        print(f"  已在跑（pid={p}）")
        return {"ok": True, "pid": p, "already": True}
    write_config()
    ensure_jar()
    LOG.parent.mkdir(parents=True, exist_ok=True)
    print(f"  起服（首次会生成世界，约 1-2 分钟）…  port={PORT}")
    with LOG.open("w") as lf:
        proc = subprocess.Popen(
            ["java", "-Xmx2G", "-Xms1G", "-jar", str(JAR), "nogui"],
            cwd=str(CONTENT), stdout=lf, stderr=subprocess.STDOUT,
            start_new_session=True)
    PIDFILE.write_text(str(proc.pid))
    print(f"  pid={proc.pid}  日志={LOG}")

    t0 = time.time()
    while time.time() - t0 < wait:
        st = status()
        if st:
            print(f"  ✓ 就绪（{time.time() - t0:.1f}s）：{st['version']} "
                  f"protocol {st['protocol']}")
            return {"ok": True, "pid": proc.pid, "status": st}
        if proc.poll() is not None:
            print("  ✗ 进程已退出，最后几行：")
            print("".join(LOG.read_text(errors="replace").splitlines(True)[-8:]))
            return {"ok": False, "reason": "进程退出"}
        time.sleep(2)
    print(f"  ⚠ {wait}s 内未就绪（看 {LOG}）")
    return {"ok": False, "reason": "超时"}


def stop(timeout: float = 30.0) -> bool:
    pid = _pid()
    if not pid:
        print("  没在跑")
        return False
    print(f"  停服（pid={pid}）…")
    # 先给 SIGTERM，让它存盘
    try:
        os.kill(pid, 15)
    except ProcessLookupError:
        PIDFILE.unlink(missing_ok=True)
        print("  ✓ 已不在")
        return True
    t0 = time.time()
    while time.time() - t0 < timeout:
        try:
            os.kill(pid, 0)
        except ProcessLookupError:
            PIDFILE.unlink(missing_ok=True)
            print(f"  ✓ 已停（{time.time() - t0:.1f}s）")
            return True
        time.sleep(1)
    print("  ⚠ 超时，强杀")
    try:
        os.kill(pid, 9)
    except ProcessLookupError:
        pass
    PIDFILE.unlink(missing_ok=True)
    return True


def reset() -> None:
    """删世界（配置改了要重生成时用）。"""
    if _pid():
        stop()
    for d in ("world", "world_nether", "world_the_end"):
        p = CONTENT / d
        if p.exists():
            print(f"  删 {d}…")
            subprocess.run(["rm", "-rf", str(p)], check=False)
    LOG.unlink(missing_ok=True)


def info() -> dict:
    return {
        "mc_version": MC_VERSION,
        "protocol": PROTOCOL,
        "port": PORT,
        "content": str(CONTENT),
        "runtime": str(RUNTIME),
        "log": str(LOG),
        "jar_exists": JAR.is_file(),
        "world_exists": (CONTENT / "world").is_dir(),
        "pid": _pid(),
        "status": status(),
    }


def main(argv: list[str]) -> int:
    cmd = argv[1] if len(argv) > 1 else "status"
    if cmd == "start":
        r = start()
        print(json.dumps(r, ensure_ascii=False, indent=2))
        return 0 if r.get("ok") else 1
    if cmd == "stop":
        return 0 if stop() or not _pid() else 1
    if cmd == "reset":
        reset()
        return 0
    if cmd == "status":
        st = status()
        d = info()
        if st:
            print(f"  ✓ 在线  {st['version']}  protocol {st['protocol']}  "
                  f"players {st['players_online']}/{st['players_max']}")
            print(f"    motd: {st['motd']}")
            return 0
        print(f"  ✗ 不在线（port={PORT}）")
        if d["world_exists"]:
            print("    世界已存在 —— 直接 start 即可")
        return 1
    print(__doc__)
    return 2


if __name__ == "__main__":
    sys.exit(main(sys.argv))