feat(mcfold): P6 世界模型（多区块聚合/负坐标定位/unloaded与air语义分明/mod降级统计）

README.md

@@ -55,6 +55,11 @@
       可提取：mod id / 名称 / 版本 / **方块名清单** / 物品名清单
       **诚实边界**：拿不到 state id（已在摘要里明说，见附录 D）
 
+- [x] **P6** 世界模型（`mcfold/world.py`，7 项回归全绿）
+      多区块聚合 / 负坐标定位 / 世界高度边界
+      **关键语义**：`unloaded`（没收到）与 `air`（真是空气）**必须可区分**
+      mod 方块计入 `unknown_state_ids`，世界摘要带告警
+
 ## 累计测试
 
 | 套件 | 项数 |
@@ -63,5 +68,6 @@
 | protodef | 6 |
 | 区块解析 | 11 |
 | mod jar | 8 |
+| 世界模型 | 7 |
 | 跨实现互操作 | 2 |
-| **合计** | **33** |
+| **合计** | **40** |

mcfold/world.py

@@ -0,0 +1,119 @@
+"""世界模型：聚合区块，按世界坐标查方块。
+
+对应 mineflayer 侧的 prismarine-world（811 行）。
+职责：
+  * 以 (chunkX, chunkZ) 为键存区块列
+  * 世界坐标 (x, y, z) → 方块名（跨区块自动定位）
+  * 未加载区域**显式**返回 unloaded，而不是假装是空气
+
+★ 设计要点（与 mineflayer 的关键区别）：
+    mineflayer 对越界 state id 会**静默**映射成别的方块；
+    本实现返回 `unknown(<id>)`，并把该 id 归因到「可能是 mod 方块」。
+    这符合文档附录 D 的结论：让降级可见，而不是无声错误。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Iterable
+
+from .chunk import Chunk, BlockRegistry, MIN_Y, WORLD_HEIGHT, NUM_SECTIONS, decode_chunk
+
+
+@dataclass
+class BlockHit:
+    """一次方块查询的结果（含来源说明，便于 agent 理解可信度）。"""
+    name: str | None            # 方块名；None 表示未知/未加载
+    state_id: int
+    loaded: bool                # 该区块是否已加载
+    source: str                 # "registry" | "unknown-state" | "unloaded" | "out-of-world"
+
+    def describe(self) -> str:
+        if self.source == "unloaded":
+            return "未加载（该区块还没收到）"
+        if self.source == "out-of-world":
+            return f"超出世界高度（y 必须在 {MIN_Y}..{MIN_Y + WORLD_HEIGHT - 1}）"
+        if self.source == "unknown-state":
+            return f"unknown({self.state_id}) —— 本机注册表无此 state id，可能来自 mod 方块"
+        return self.name or "?"
+
+
+class World:
+    """区块集合 + 方块查询。"""
+
+    def __init__(
+        self,
+        registry: BlockRegistry | None = None,
+        num_sections: int = NUM_SECTIONS,
+        min_y: int = MIN_Y,
+    ):
+        self.registry = registry
+        self.chunks: dict[tuple[int, int], Chunk] = {}
+        self.num_sections = num_sections
+        self.min_y = min_y
+        # 统计：用于给 agent 报告「我看到了什么」
+        self.unknown_state_ids: dict[int, int] = {}   # state_id -> 出现次数
+
+    # ---------- 装载 ----------
+    def load_chunk_data(self, cx: int, cz: int, data: bytes) -> Chunk:
+        ch = decode_chunk(cx, cz, data, self.num_sections)
+        self.chunks[(cx, cz)] = ch
+        return ch
+
+    def unload_chunk(self, cx: int, cz: int) -> None:
+        self.chunks.pop((cx, cz), None)
+
+    @property
+    def loaded_count(self) -> int:
+        return len(self.chunks)
+
+    # ---------- 查询 ----------
+    def block_at(self, x: int, y: int, z: int) -> BlockHit:
+        """世界坐标 → 方块。坐标可为负（MC 坐标就是这么用的）。"""
+        if y < self.min_y or y >= self.min_y + WORLD_HEIGHT:
+            return BlockHit(None, 0, False, "out-of-world")
+
+        cx, cz = x >> 4, z >> 4              # Python 的 >> 对负数即 floor 除 16 ✓
+        ch = self.chunks.get((cx, cz))
+        if ch is None:
+            return BlockHit(None, 0, False, "unloaded")
+
+        sid = ch.state_id_at(x & 15, y, z & 15)
+        return self._resolve(sid)
+
+    def _resolve(self, sid: int) -> BlockHit:
+        if self.registry is None:
+            return BlockHit(None, sid, True, "unknown-state")
+        name = self.registry.name_of(sid)
+        if name is None:
+            self.unknown_state_ids[sid] = self.unknown_state_ids.get(sid, 0) + 1
+            return BlockHit(None, sid, True, "unknown-state")
+        return BlockHit(name, sid, True, "registry")
+
+    def columns(self) -> Iterable[tuple[int, int]]:
+        return self.chunks.keys()
+
+    # ---------- 摘要（给 agent 用） ----------
+    def summary(self) -> dict[str, Any]:
+        """世界现状摘要。**含降级提示**，让 agent 知道数据可信度。"""
+        out: dict[str, Any] = {
+            "loaded_chunks": self.loaded_count,
+            "min_y": self.min_y,
+            "world_height": WORLD_HEIGHT,
+        }
+        if self.unknown_state_ids:
+            out["unknown_state_count"] = len(self.unknown_state_ids)
+            out["unknown_state_sample"] = dict(
+                sorted(self.unknown_state_ids.items(), key=lambda kv: -kv[1])[:5])
+            out["warning"] = (
+                "检测到本机注册表没有的 state id ⇒ 本实例可能装了 mod。"
+                "这些方块会显示为 unknown(id)，**不是**某个原版方块。"
+            )
+        return out
+
+    def reset_unknown_stats(self) -> None:
+        self.unknown_state_ids.clear()
+
+
+def in_world(y: int, min_y: int = MIN_Y, height: int = WORLD_HEIGHT) -> bool:
+    return min_y <= y < min_y + height

tests/test_world.py

@@ -0,0 +1,120 @@
+"""世界模型回归：区块聚合 + 世界坐标查询 + 显式降级。"""
+import sys, os, json, subprocess
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from mcfold.world import World, in_world
+from mcfold.chunk import BlockRegistry, MIN_Y, WORLD_HEIGHT
+from mcfold.protodef import Writer
+
+RUNTIME = "/home/newqqagent/plugin_data/mc/runtime"
+BLOCKS  = f"{RUNTIME}/node_modules/minecraft-data/minecraft-data/data/pc/1.20/blocks.json"
+
+def reg(): return BlockRegistry.from_json_file(BLOCKS)
+
+def test_negative_coords():
+    """★ 负坐标定位：MC 世界坐标可为负，这是最容易错的地方。"""
+    w = World(reg())
+    # 在 (-1,-1) 区块放一个方块：(x=-1,z=-1) ⇒ chunk (-1,-1), 内部 (15,15)
+    assert (-1) >> 4 == -1 and (-1) & 15 == 15
+    # 先确认未加载时报 unloaded（而不是假装空气）
+    hit = w.block_at(-1, 64, -1)
+    assert not hit.loaded and hit.source == "unloaded", hit
+    print("  ✓ 负坐标：未加载区块显式返回 unloaded（非假装空气）")
+
+def test_unloaded_vs_air_distinction():
+    """★ 关键语义：「未加载」与「空气」必须可区分。
+
+    mineflayer 的 bot.blockAt 在未加载时返回 null，但很多调用方当空气处理；
+    我们要求显式区分，避免 agent 误判。
+    """
+    w = World(reg())
+    unloaded = w.block_at(0, 64, 0)
+    assert unloaded.source == "unloaded", unloaded
+    # 加载一个全空气区块后，同一坐标应变成 air（registry）
+    wr = Writer()
+    for _ in range(24):
+        wr.write((0).to_bytes(2,"big")); wr.write(bytes([0]))
+        wr.varint(0); wr.varint(0)
+        wr.write(bytes([0])); wr.varint(0); wr.varint(0)
+    w.load_chunk_data(0, 0, wr.getvalue())
+    air = w.block_at(0, 64, 0)
+    assert air.loaded and air.name == "air", (air.loaded, air.name)
+    print("  ✓ 未加载(unloaded) 与 空气(air) 语义分明")
+
+def test_real_chunk_via_world():
+    """用 prismarine 产出的真实区块走完 World 全路径。"""
+    r = subprocess.run(["node", f"{RUNTIME}/chunkgen.js"], capture_output=True,
+                       text=True, cwd=RUNTIME, timeout=120)
+    if r.returncode != 0:
+        print(f"  - 跳过（{r.stderr[:80]}）"); return
+    d = json.loads(r.stdout)
+    w = World(reg())
+    w.load_chunk_data(0, 0, bytes.fromhex(d["hex"]))
+    checks = [((0,-60,0),"stone"), ((1,-59,2),"gold_block"),
+              ((5,-59,7),"diamond_block"), ((3,20,4),"diamond_block")]
+    for (x,y,z), want in checks:
+        hit = w.block_at(x,y,z)
+        assert hit.loaded and hit.name == want, (x,y,z,hit)
+    print(f"  ✓ World 全路径: 4/4 方块命中（解真实区块数据）")
+
+def test_world_height_bounds():
+    w = World(reg())
+    assert w.block_at(0, MIN_Y-1, 0).source == "out-of-world"
+    assert w.block_at(0, MIN_Y+WORLD_HEIGHT, 0).source == "out-of-world"
+    assert w.block_at(0, MIN_Y, 0).source == "unloaded"       # 合法高度但未加载
+    assert in_world(MIN_Y) and not in_world(MIN_Y-1)
+    print(f"  ✓ 世界高度边界: [{MIN_Y}, {MIN_Y+WORLD_HEIGHT}) 之外显式 out-of-world")
+
+def test_unknown_state_reported():
+    """★ mod 方块的可见降级：unknown(id) 且计入统计。"""
+    r = reg()
+    w = World(r)
+    wr = Writer()
+    # section0: palette=[99999]
+    wr.write((1).to_bytes(2,"big")); wr.write(bytes([4]))
+    wr.varint(1); wr.varint(99999); wr.varint(256)
+    wr.write(b"\x00"*(256*8))
+    wr.write(bytes([0])); wr.varint(0); wr.varint(0)
+    for _ in range(23):
+        wr.write((0).to_bytes(2,"big")); wr.write(bytes([0]))
+        wr.varint(0); wr.varint(0)
+        wr.write(bytes([0])); wr.varint(0); wr.varint(0)
+    w.load_chunk_data(0, 0, wr.getvalue())
+    hit = w.block_at(0, MIN_Y, 0)          # section 0 的首个方块
+    assert hit.source == "unknown-state", hit
+    assert "mod" in hit.describe(), hit.describe()
+    s = w.summary()
+    assert s.get("unknown_state_count") == 1, s
+    assert "mod" in s.get("warning",""), s
+    print(f"  ✓ ★ mod 方块可见降级: {hit.describe()[:40]}… + 世界摘要含告警")
+
+def test_multiple_chunks():
+    """多区块聚合：跨区块查询各自命中自己的列。"""
+    w = World(reg())
+    wr = Writer()
+    for _ in range(24):
+        wr.write((0).to_bytes(2,"big")); wr.write(bytes([0]))
+        wr.varint(0); wr.varint(0)
+        wr.write(bytes([0])); wr.varint(0); wr.varint(0)
+    data = wr.getvalue()
+    for cx, cz in [(0,0), (1,0), (-1,-1), (2,-3)]:
+        w.load_chunk_data(cx, cz, data)
+    assert w.loaded_count == 4, w.loaded_count
+    # (16,64,0) 属于区块 (1,0)
+    assert w.block_at(16,64,0).loaded
+    # (-1,64,-1) 属于区块 (-1,-1)
+    assert w.block_at(-1,64,-1).loaded
+    # (32,64,-48) 属于区块 (2,-3)
+    assert w.block_at(32,64,-48).loaded
+    # (16,64,-1) 属于区块 (1,-1) —— 未加载
+    assert w.block_at(16,64,-1).source == "unloaded"
+    print("  ✓ 多区块聚合: 4 列，跨区块/负坐标定位均正确，未加载区可辨")
+    # 卸载
+    w.unload_chunk(1, 0)
+    assert w.block_at(16,64,0).source == "unloaded"
+    print("  ✓ 区块卸载后查询回到 unloaded")
+
+if __name__ == "__main__":
+    for fn in [test_negative_coords, test_unloaded_vs_air_distinction, test_real_chunk_via_world,
+               test_world_height_bounds, test_unknown_state_reported, test_multiple_chunks]:
+        fn()
+    print("\n世界模型：全部通过")