Files
agentmc/mcfold/servers.py
JianFeeeee 554ebc239a M10c:mod 环境下寻路 —— 问运行时而非猜名字(+ 语义表与覆盖率报告)
用户要求:Create 与 Aeronautics 会引入大量原版没有的方块与危险区,
寻路必须一起优化。用户裁定:危险区用「表 + 命名推测并标记」,
动态机械用「警告 + 提高代价」。

## ★ 关键实测结论(本轮最重要)

我先按「给 mod 方块建语义表」做,实测覆盖率:

    Create 643 个方块 → 显式表命中 9、命名规则推测 21、完全不识 613
    覆盖率 4.7%

⇒ **手写表是死路**(64 mod / 1465 方块永远追不上)。
⇒ 但寻路真正需要的属性,**运行时全都有**:
   Java 侧 blockinfo 已在读 BuiltInRegistries.BLOCK(**含 mod 运行时注册的方块**),
   并返回 isSolid / destroySpeed / lightEmission / isAir / hasBlockEntity。
⇒ 正确做法是**问**,不是猜。

## 改动

新增 mcfold/solid.py:
- SolidSource 抽象:RegistrySolidSource(本机,快但无 mod)
  vs RuntimeSolidSource(问 JVM,**含 mod**,准但要 JVM)
- 按 **state_id** 缓存(属性只依赖 state,不依赖坐标)
- attach_runtime() 批量探测(一次问一格要一次往返,批量才可用)

mcfold/pathfind.py:Terrain 改为「**先运行时缓存,再本机注册表**」;
  can_dig 用 destroySpeed 判可挖(基岩官方约定是 -1 = 不可破坏),
  不再靠名字猜「基岩挖不动」

新增 mcfold/modblocks.py(危险/会动这部分的表,配套覆盖率报告):
- 语义取���:ground/passable/hazard/breakable/solid/dynamic/carrier/unknown
- 三级来源:显式表 → 命名规则(**标记 inferred**)→ **如实说不认识**
- ★ vanilla 与 mod **分开算覆盖率**:vanilla 语义由注册表给,
  把它算成「未知」会让人误以为「连石头都认不出」

WorldState:加 solid_cache 字段

## 判据

test_pathfind 19 → 24 项,新增 5 项:
- mod 方块优先用运行时 isSolid(本机注册表确实认不出)
- 不可破坏吃 destroySpeed(含基岩 -1 约定)
- 两边都无数据 ⇒ 不知道(**不谎称可走**)
- 覆盖率报告如实(且区分 mod/vanilla)
- 语义查询永远返回带 inferred 标记的结构

★ 写判据时踩了两个坑(都是 fixture 形状错,不是代码错):
  1. `Chunk.sections[i]` 对应 y = MIN_Y + 16*i,MIN_Y=-64
     ⇒ 我按「第 0 个 section = y=0」写,查 y=0 拿到的是空气
  2. 同一 section 内 16 格**只能一个值** ⇒ mod 方块与空气落同一 section 时,
     空气格会变成 mod 方块(看着像代码 bug,其实是 fixture 形状问题)
  ⇒ 教训:**判据的 fixture 形状必须与被测代码的真实输入结构一致**。

226 项全通过。
2026-09-26 14:29:58 +08:00

428 lines
18 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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_libraries(version: str, loader_version: str = "") -> dict[str, Any]:
"""取 Fabric 服务端要的库(loader / intermediary / asm / sponge-mixin)。
★ **优先复用本地资产**:客户端路线(M8 之前)已经把同一批库装进
`store/<loader>-<version>/.mcfold/fabric/`,那正是「内容根共享、只读」
的用途所在。重新下载既慢又会多一份副本。
⇒ 这也是「同一版本多实例」在资产层面的体现:库只存一份,实例各自持有引用。
"""
out: list[str] = []
missing: list[str] = []
shared = STORE / f"fabric-{version}" / ".mcfold" / "fabric"
if not shared.exists():
# 退化:vanilla-<v> 内容根里也放过同一批(历史上这么建的)
shared = STORE / f"vanilla-{version}" / ".mcfold" / "fabric"
if shared.is_dir():
out += [str(p) for p in sorted(shared.glob("*.jar"))]
if out:
return {"ok": True, "jars": out, "source": str(shared), "cached": True}
return {"ok": False, "why": f"本地没有 {version} 的 Fabric 库(找过 {shared})",
"how_to": "用 fabric_installer() 下载安装器后安装,或先把客户端路线的"
"fabric 库准备好",
"missing": missing}
def fabric_server_launcher(version: str) -> dict[str, Any]:
"""取可起 Fabric 服务端的启动件。
★ 实测:`fabric-server-launch` 在 maven 上**不存在**(404)——
Fabric 服务端的正确起法是把 loader + intermediary + asm + mixin
放上 classpath,然后跑官方 `net.fabricmc.loader.impl.launch.knot.KnotServer`。
我先前按「有个 server-launch jar」的假设拼 URL,拿到的是 404。
"""
libs = fabric_libraries(version)
if not libs.get("ok"):
return libs
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}
return {"ok": False,
"why": "没有 fabric-server-launch 这种产物(实测 404)",
"how_to": "改用 classpath 方式:loader+intermediary+asm+mixin 走 "
"KnotServer(见 fabric_classpath_launch)",
"classpath": libs["jars"]}
def fabric_classpath_launch(version: str) -> dict[str, Any]:
"""Fabric 服务端的正确启动方式:classpath + 官方 KnotServer 主类。"""
libs = fabric_libraries(version)
if not libs.get("ok"):
return libs
cp = ":".join(libs["jars"])
return {
"ok": True,
"classpath": cp,
"main_class": "net.fabricmc.loader.impl.launch.knot.KnotServer",
"n_jars": len(libs["jars"]),
"source": libs.get("source"),
"how_to": "java -cp <上面那串> net.fabricmc.loader.impl.launch.knot.KnotServer",
}
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]}
# ---------- NeoForge(生产服用的那条路) ----------
#: MC 版本 → NeoForge 版本前缀。1.21.1 → "21.1.",1.20.1 → "20.1."
_NEOFORGE_PREFIX = {
"1.21.1": "21.1.", "1.21": "21.1.", "1.20.1": "20.1.", "1.20.4": "20.4.",
"1.20.6": "20.6.", "1.20.2": "20.2.", "1.20.3": "20.3.", "1.19.4": "19.4.",
}
def neoforge_versions(mc_version: str) -> dict[str, Any]:
"""查某 MC 版本可用的 NeoForge 版本列表(从官方 maven 读)。"""
import urllib.request
import json as _json
url = "https://maven.neoforged.net/api/maven/versions/releases/net/neoforged/neoforge"
try:
with urllib.request.urlopen(url, timeout=60) as r:
all_v = _json.loads(r.read())["versions"]
except Exception as e:
return {"ok": False, "why": f"查 NeoForge 版本失败:{type(e).__name__}: {e}",
"how_to": "检查能否访问 maven.neoforged.net"}
pre = _NEOFORGE_PREFIX.get(mc_version)
if pre is None:
return {"ok": False, "why": f"不知道 {mc_version} 对应的 NeoForge 版本前缀",
"how_to": f"在 _NEOFORGE_PREFIX 里补一条,或直接给 neoforge_version",
"supported": sorted(_NEOFORGE_PREFIX)}
vs = [v for v in all_v if v.startswith(pre)]
if not vs:
return {"ok": False, "why": f"NeoForge 没有匹配 {mc_version} 的版本(前缀 {pre})"}
return {"ok": True, "versions": vs, "prefix": pre}
def neoforge_installer(mc_version: str, neoforge_version: str = "",
latest: bool = True) -> dict[str, Any]:
"""取 NeoForge installer(显式给版本,或取该 MC 版本的最新版)。"""
if not neoforge_version:
if not latest:
return {"ok": False, "why": "需要 neoforge_version(或 latest=True)",
"how_to": f"可选:{neoforge_versions(mc_version).get('versions', [])[:5]}"}
got = neoforge_versions(mc_version)
if not got.get("ok"):
return got
neoforge_version = got["versions"][0]
p = LOADER_CACHE / f"neoforge-installer-{neoforge_version}.jar"
if p.exists() and p.stat().st_size > 100_000:
return {"ok": True, "path": str(p), "cached": True, "version": neoforge_version}
url = (f"https://maven.neoforged.net/releases/net/neoforged/neoforge/"
f"{neoforge_version}/neoforge-{neoforge_version}-installer.jar")
r = _download(url, p, timeout=900)
if r.get("ok"):
r["version"] = neoforge_version
return r
def run_installer(installer: str, cwd: str | os.PathLike[str],
*, kind: str, timeout: int = 1800) -> dict[str, Any]:
"""跑加载器安装器(NeoForge/Fabric 都用这个形态)。
★ 安装器是**执行第三方 jar** ⇒ 失败必须把 stderr 尾部带回来
(否则只看到「退出码 1」,无法判断是网络、Java 版本还是 mod 冲突)。
"""
cwd = Path(cwd)
cmd = ["java", "-jar", str(installer)]
if kind == "neoforge":
cmd.append("--installServer")
elif kind == "fabric":
cmd.append("server")
try:
p = subprocess.run(cmd, cwd=str(cwd), capture_output=True,
text=True, timeout=timeout)
except subprocess.TimeoutExpired:
return {"ok": False, "why": f"安装器超时(>{timeout}s)",
"how_to": "网络慢或 mod 冲突;可加大 timeout 重试"}
except Exception as e:
return {"ok": False, "why": f"跑安装器失败 {type(e).__name__}: {e}"}
if p.returncode != 0:
tail = ((p.stderr or "") + (p.stdout or ""))[-800:]
return {"ok": False, "why": f"安装器退出码 {p.returncode}",
"how_to": "看下面的输出判断(网络/Java 版本/mod 冲突)",
"output_tail": tail}
return {"ok": True, "returncode": 0}
def neoforge_run_cmd(nf_version: str) -> list[str]:
"""NeoForge 服务端的启动命令(官方 userdev/jars 形态)。"""
return ["java", "@userdev/java.args", "nogui"]
# ---------- 装一个「带 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",
neoforge_version: str = "",
install: bool = False,
timeout: int = 900,
) -> dict[str, Any]:
"""在**实例层**里装一个 modded 服务端(内容根只读共享)。
参数
----
loader vanilla / fabric / neoforge
neoforge_version 显式给 NeoForge 版本;缺省取该 MC 版本的最新
install NeoForge 必须为 True(要跑官方 installer 生成服务端)
返回 `{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", "neoforge"):
return {
"ok": False,
"why": f"加载器 {loader!r} 的服务端还没接"
f"(当前支持 vanilla / fabric / neoforge)",
"how_to": "Forge/Quilt 需各自的 installer 流程;"
"客户端渲染路线见 scaffold/LOADERS.md",
"left_nothing": True,
}
# ★ NeoForge 缺 install=True 要在**建目录之前**就拒:
# 否则会留下一个「看着存在但没配好」的实例目录
# (list_instances 会把它列出来,使用者以为装好了)。
if loader == "neoforge" and not install:
return {
"ok": False,
"why": "NeoForge 需要先跑官方 installer(--installServer)"
"才能得到服务端运行时",
"how_to": "传 install=True;首次约 1-3 分钟(要下依赖)",
"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_classpath_launch(version)
if not got.get("ok"):
return {"ok": False, "why": got.get("why"), "how_to": got.get("how_to"),
"installed": installed}
# 把库复制进实例(实例独占,不依赖内容根的相对路径)
libs_dir = inst / "fabric-libs"
libs_dir.mkdir(exist_ok=True)
copied = []
for j in got["classpath"].split(":"):
if Path(j).exists():
shutil.copy2(j, libs_dir / Path(j).name)
copied.append(Path(j).name)
cp = ":".join(f"fabric-libs/{n}" for n in copied)
launch = {"cmd": ["java", f"-Xmx{java_heap}", f"-Xms{java_heap}",
"-cp", cp, "net.fabricmc.loader.impl.launch.knot.KnotServer"]}
elif loader == "neoforge":
got = neoforge_installer(version, neoforge_version)
if not got.get("ok"):
return {"ok": False, "why": got.get("why"), "how_to": got.get("how_to"),
"installed": installed}
run = run_installer(got["path"], inst, kind="neoforge", timeout=1800)
if not run.get("ok"):
return {"ok": False, "why": run.get("why"), "how_to": run.get("how_to"),
"output_tail": run.get("output_tail"),
"installed": installed}
args = inst / "userdev" / "java.args"
if not args.exists():
return {"ok": False,
"why": f"installer 跑完但没产出 {args}",
"how_to": "检查 neoforge 版本是否支持该 MC 版本;或改用 fabric",
"installed": installed, "neoforge_version": got.get("version")}
launch = {"cmd": ["java", f"-Xmx{java_heap}", f"-Xms{java_heap}",
"@userdev/java.args", "nogui"]}
meta["neoforge_version"] = got.get("version")
(inst / "instance.json").write_text(
json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")
return {
"ok": True, "instance": str(inst), "runtime": str(rt),
"loader": "neoforge", "version": version,
"neoforge_version": got.get("version"),
"mods_installed": installed, "mods_dir": str(mods_dir),
"launch_cmd": launch["cmd"],
"next": (f"cd {inst} && java {launch['cmd'][1]} "
f"(mod 已就位,NeoForge 启动时加载)"),
"note": "NeoForge 服务端由官方 installer 生成;mods 在实例层",
}
# 写启动脚本(内容根零写入)
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 分钟)",
}