"""M3 状态层判据：state.py + session.py。 ## 判据原则（沿用本仓纪律） 1. **能失败**：不靠「观察当前输出就算对」，而是**构造**输入并断言语义 （相对位移、总和、范围上限、x/z 顺序这类都必须有反向对照）。 2. **能区分**：「认不出的方块」与「没看见的方块」必须报不同的 source —— 否则 agent 会把「未知」当「空地」而做出错误决策。 3. **解码器的判据用线上真实字节**（见 test_protodef 的 entity_metadata 回归）—— 单元构造的字节可能偏离真实格式。 """ import sys, os, json sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from mcfold.state import ( WorldState, WorldGeometry, PlayerView, Entity, load_vanilla_registry, MAX_SCAN_VOLUME, MAX_BLOCK_KINDS, ) from mcfold.chunk import BlockRegistry DATA_ROOT = '/home/newqqagent/plugin_data/mc/runtime/node_modules/minecraft-data/minecraft-data/data' def _reg(): return load_vanilla_registry('1.21.1', DATA_ROOT) def _state(reg=None): return WorldState(registry=reg if reg is not None else _reg(), num_sections=24) def _bare_session(): """造一个不连网的 StateSession，只测「包 → 状态」的落状态逻辑。 ★ 为什么不用 __new__：它跳过 __init__ ⇒ _listeners 不存在， 调 on() 就 AttributeError。这里显式补上必需字段。 """ from mcfold.session import StateSession sess = StateSession.__new__(StateSession) sess.state = _state() sess.client = type("C", (), { "_serverbound_id": staticmethod(lambda *a: 0x26), "send": staticmethod(lambda b: None), })() sess._listeners = [] sess._sent = [] return sess # ---------- 几何 ---------- def test_geometry_not_hardcoded(): """几何来自服务端，不是写死 24/384。""" ws = _state() assert ws.geometry.height == 384 and ws.geometry.min_y == -64 # 改成别的维度（如 y=-128 起、512 高）后应如实反映 ws.geometry.min_y, ws.geometry.height = -128, 512 assert ws.geometry.in_world(-128) and not ws.geometry.in_world(-129) assert ws.geometry.in_world(383) and not ws.geometry.in_world(384) assert "y∈[-128, 384)" in ws.geometry.describe() print(" ✓ 几何不写死（可随服务端维度变）") # ---------- position：绝对 vs 相对 ---------- def test_position_absolute_and_relative(): """★ position 包的 flags 位决定绝对/相对 —— 必须分别处理。 这是最容易出的错：服务端做小步纠正时发**相对增量**， 若照绝对赋值，位置会瞬间跳回原点。 """ ws = _state() ws.update_from_position({"x": 100.0, "y": 64.0, "z": -50.0, "yaw": 90.0, "pitch": 10.0}) assert (ws.player.x, ws.player.y, ws.player.z) == (100.0, 64.0, -50.0) # 全相对：应累加 ws.update_from_position({"x": 1.0, "y": -1.0, "z": 2.0, "yaw": 15.0, "pitch": 5.0, "flags": {"_value": 0x1F, "x": True, "y": True, "z": True, "yaw": True, "pitch": True}}) assert (ws.player.x, ws.player.y, ws.player.z) == (101.0, 63.0, -48.0), ws.player.as_dict() assert ws.player.yaw == 105.0, ws.player.yaw assert ws.player.pitch == 15.0, ws.player.pitch # ★ 混合：只有 x 相对 ⇒ 另两个轴必须是绝对值 ws.update_from_position({"x": 5.0, "y": 70.0, "z": 999.0, "flags": {"_value": 0x01, "x": True, "y": False, "z": False, "yaw": False, "pitch": False}}) assert ws.player.x == 106.0, ws.player.x assert ws.player.y == 70.0, ws.player.y # 绝对覆盖，不是累加 assert ws.player.z == 999.0, ws.player.z print(" ✓ position 绝对/相对/混合三种都正确") def test_pitch_clamped(): """相对 pitch 必须夹在 ±90（否则视野会翻转/穿模）。""" ws = _state() ws.update_from_position({"x": 0, "y": 0, "z": 0, "yaw": 0, "pitch": 0}) ws.update_from_position({"x": 0, "y": 0, "z": 0, "yaw": 0, "pitch": 200, "flags": {"_value": 0x10, "x": False, "y": False, "z": False, "yaw": False, "pitch": True}}) assert ws.player.pitch == 90.0, ws.player.pitch print(" ✓ 相对 pitch 夹紧到 ±90") # ---------- 方块查询：未知 ≠ 没看见 ---------- def test_unknown_state_is_not_silently_mapped(): """★ 认不出的 state id 必须报 unknown-state，**不能**映射成别的方块。 静默映射是这类工具最危险的缺陷：agent 会基于「虚构的世界」做决策。 """ # 故意用一张缺 id=9 的注册表，模拟「本机没有这个方块」 partial = BlockRegistry([{"name": "stone", "minStateId": 1, "maxStateId": 1}]) ws = _state(partial) # 手工塞一个 state_id=9 的区块 class FakeChunk: def state_id_at(self, x, y, z): return 9 ws.world.chunks[(0, 0)] = FakeChunk() hit = ws.block_at(0, 0, 0) assert hit["name"] is None, hit assert hit["source"] == "unknown-state", hit["source"] assert hit["state_id"] == 9 # 且必须被记账（供上层告警） assert 9 in ws.world.unknown_state_ids assert ws.summary()["has_unresolved_unknown"] is True print(" ✓ 未知方块显式记账，不静默映射") def test_unloaded_vs_out_of_world_distinguished(): """「区块没加载」与「在世界外」必须报不同 source。""" ws = _state() assert ws.block_at(0, 0, 0)["source"] == "unloaded" assert ws.block_at(0, 9999, 0)["source"] == "out-of-world" # 负坐标也要成立（Python 的 >> 对负数是 floor 除，正是 MC 的语义） assert ws.block_at(-1, 0, -1)["source"] == "unloaded" print(" ✓ unloaded / out-of-world 可区分（负坐标亦然）") # ---------- 受控范围 ---------- def test_nearby_blocks_refuses_oversized_scan(): """★ 超范围要**拒绝并说明**，不能静默缩小。 静默缩小会让 agent 以为扫全了，进而做出错误规划。 """ ws = _state() big = ws.nearby_blocks(0, 0, 0, radius=100) assert big["scanned"] is False assert "超过上限" in big["error"], big assert big["requested_radius"] == 100 # 边界内则正常（即便空） ok = ws.nearby_blocks(0, 0, 0, radius=4) assert ok["scanned"] is True assert (2 * 4 + 1) ** 3 <= MAX_SCAN_VOLUME print(" ★ 超大扫描被拒（不静默缩小）") print(f" ✓ 正常范围扫描可用（{ok['kinds']} 种方块，0 个区块时全空）") def test_summary_is_bounded_and_honest(): """summary 要有界、且不谎报。""" ws = _state() s = ws.summary() assert s["chunks_loaded"] == 0 assert s["entities"] == 0 # 没识别出东西时 note 应为 None（不啰嗦） assert s["note"] is None # 块数上限生效 assert MAX_BLOCK_KINDS <= 20 print(" ✓ summary 有界且不谎报") # ---------- 实体 ---------- def test_nearby_entities_sorted_by_distance(): ws = _state() ws.player.x, ws.player.y, ws.player.z = 0.0, 0.0, 0.0 for eid, d in ((1, 10.0), (2, 3.0), (3, 6.0)): ws.upsert_entity(Entity(entity_id=eid, kind="pig", x=d, y=0.0, z=0.0)) got = ws.nearby_entities(radius=16) assert got["count"] == 3 dists = [e["distance"] for e in got["entities"]] assert dists == sorted(dists), dists assert got["entities"][0]["entity_id"] == 2 # 半径外的不算 assert ws.nearby_entities(radius=5)["count"] == 1 # 移除 ws.remove_entity(2) assert ws.nearby_entities(radius=16)["count"] == 2 print(" ✓ 实体按距离排序、半径过滤、可移除") # ---------- 注册表（从服务端来） ---------- def test_server_registries_kept_verbatim(): """服务端注册表原样保留（权威来源，不加工）。""" ws = _state() dt = [{"key": "minecraft:overworld", "value": {"height": {"height": 384, "min_y": -64}}}] ws.apply_registry("minecraft:dimension_type", dt) assert ws.registries["minecraft:dimension_type"] is dt s = ws.summary() assert "minecraft:dimension_type" in s["registries_from_server"] assert s["registry_entry_counts"]["minecraft:dimension_type"] == 1 print(" ✓ 服务端注册表原样保留并计数") def test_geometry_absorbed_from_dimension_registry(): """从 dimension_type 注册表吸收几何（不再写死）。""" from mcfold.state import ingest_configuration ws = _state() ws.geometry = WorldGeometry(height=384, min_y=-64) # 造一个不同高度的 overworld 条目 fake = {"minecraft:dimension_type": [ {"key": "minecraft:overworld", "value": {"height": {"height": 512, "min_y": -128}}}]} ingest_configuration(ws, {}, fake) assert ws.geometry.height == 512 and ws.geometry.min_y == -128 print(" ✓ 几何可从服务端维度注册表覆盖") # ---------- session：包 → 状态（x/z 顺序这个真 bug 的回归） ---------- def test_unload_chunk_xz_order(): """★ unload_chunk 协议字段顺序是 **chunkZ 在前**、chunkX 在后。 回归判据：我先前把 chunkZ 塞给 cx ⇒ x/z 互换 ⇒ 卸载永不命中， 表现为区块越积越多（状态在撒谎）。 """ sess = _bare_session() ws = sess.state ws.world.chunks[(5, -9)] = "sentinel" # 协议给的是 {"chunkZ": -9, "chunkX": 5} sess._apply("unload_chunk", {"chunkZ": -9, "chunkX": 5}) assert (5, -9) not in ws.world.chunks, f"卸载未命中：{list(ws.world.chunks)}" # 反向对照：坐标顺序传错时**不能**误删 ws.world.chunks[(5, -9)] = "sentinel" ws.world.chunks[(-9, 5)] = "other" sess._apply("unload_chunk", {"chunkZ": 5, "chunkX": -9}) # 故意交换 assert (-9, 5) not in ws.world.chunks assert (5, -9) in ws.world.chunks print(" ★ unload_chunk 按协议顺序卸载命中（不交换）") print(" ✓ 反向对照：坐标错位时不误删") def test_keepalive_is_answered(): """★ keep_alive 必须回，否则 15s 后被踢 —— 状态层的头号责任。""" sent = [] sess = _bare_session() sess.client.send = staticmethod(lambda b: sent.append(b)) sess._apply("keep_alive", {"keepAliveId": 12345678901234}) assert len(sent) == 1, sent # 载荷：varint(0x26) + int64 assert sent[0][0] == 0x26, sent[0][:3].hex() import struct assert struct.unpack(">q", sent[0][1:9])[0] == 12345678901234 print(" ★ keep_alive 必被回应（包 id + int64 载荷）") print(" ✓ keepAliveId 原样回填") def test_health_and_gamemode_landed(): sess = _bare_session() sess._apply("update_health", {"health": 7.5, "food": 3}) assert sess.state.player.health == 7.5 assert sess.state.player.food == 3 sess._apply("game_state_change", {"reason": 3, "gameMode": 1}) assert sess.state.player.gamemode == "1", sess.state.player.gamemode print(" ✓ 生命/饥饿/gamemode 落入状态") def test_events_emitted(): """M4 的机制留口：事件要真发出去。""" got = [] sess = _bare_session() sess.on("player_moved", lambda e, p: got.append(e)) sess.on("joined", lambda e, p: got.append(e)) sess._apply("position", {"x": 1, "y": 2, "z": 3, "yaw": 0, "pitch": 0}) sess._apply("login", {"entityId": 7}) assert got == ["player_moved", "joined"], got print(" ✓ 事件回调真的被调用（M4 机制就位）") if __name__ == "__main__": for fn in [test_geometry_not_hardcoded, test_position_absolute_and_relative, test_pitch_clamped, test_unknown_state_is_not_silently_mapped, test_unloaded_vs_out_of_world_distinguished, test_nearby_blocks_refuses_oversized_scan, test_summary_is_bounded_and_honest, test_nearby_entities_sorted_by_distance, test_server_registries_kept_verbatim, test_geometry_absorbed_from_dimension_registry, test_unload_chunk_xz_order, test_keepalive_is_answered, test_health_and_gamemode_landed, test_events_emitted]: fn() print("\n状态层：全部通过")