"""M5 操作层判据：actions.py + textcomp.py。 ## 这层判据的判别标准 操作层的缺陷几乎都是**静默**的： - 包发错 ⇒ 服务端默默忽略（不报错、不断连） - 少了某个前置包 ⇒ 所有操作集体失效 - 字段名写错 ⇒ 一切都「没反应」 所以判据分两类，缺一不可： 1. **字节级自洽**：构造的包必须能被**官方 protodef** 解回原值 （这是唯一能离线验证「包写对了」的办法）； 2. **结构级**：预置条件、夹紧、不编造 —— 这些纯本地可判。 """ import sys, os, json, struct sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from mcfold.actions import Actions, _packed_position, _direction, _f32, _f64 from mcfold.textcomp import render_text, _humanize, flatten_with from mcfold.protodef import Reader, Writer, for_phase from mcfold.state import WorldState, item_name from mcfold.client import Client from mcfold.session import StateSession from mcfold.events import EventLog DATA_ROOT = ('/home/newqqagent/plugin_data/mc/runtime/node_modules/' 'minecraft-data/minecraft-data/data') V = '1.21.1' def _client(): c = Client(host='h', port=1, username='u', version=V) c.load_protocol() return c class _FakeChunk: """占位区块：让 dig 的「已加载」前置检查通过。""" def state_id_at(self, x, y, z): return 9 def _bare(): """一个不联网的 StateSession，能捕获发出的包。""" from mcfold.state import load_vanilla_registry s = StateSession.__new__(StateSession) s.state = WorldState(registry=load_vanilla_registry(V, DATA_ROOT), num_sections=24) s.events = EventLog() s.sent = [] s.client = _client() s.client.send = lambda b: s.sent.append(b) s._listeners = [] s.teleports_confirmed = 0 s.pending_corrections = 0 return s # ---------- 字节级：包必须能被官方 protodef 解回 ---------- def test_every_packet_decodes_with_official_protodef(): """★ 用官方 protodef 解回我构造的每个包 —— 唯一能离线证明「包写对了」的办法。 这条判据抓过的真 bug： - dig 原本写成 packet_block_place + 2 个多余 varint（应为独立 packet_block_dig） - interact/swing 用了 1.21.1 **不存在**的包名（packet_interact / packet_swing） """ s = _bare() s.state.player.entity_id = 42 s.state.player.held_slot = 0 # ★ 9 格都放上物品：前面的 select_slot 用例会切换 held_slot， # 若只填 0 号格，后面 place 会因「手上是空槽」被前置检查拦下 # —— 那时测的是「被拦下」，而不是「包写对了」。 for _i in range(9): s.state.hotbar[_i] = {"itemCount": 1, "itemId": 1} # ★ dig 有「区块未加载就拒发」的前置检查 ⇒ 测试里要放一个假区块， # 否则测的是「被拦下」而不是「包写对了」。 class _FakeChunk: def state_id_at(self, x, y, z): return 9 s.state.world.chunks[(0, 0)] = _FakeChunk() s.state.world.chunks[(7 >> 4, 10 >> 4)] = _FakeChunk() a = Actions(s) pd = for_phase(s.client.proto, "play", "toServer") cases = [ ("look", lambda: a.look(90, 45)), ("move", lambda: a.move(1.5, -60.0, 2.5)), ("move_look", lambda: a.move_and_look(1, 2, 3, 10, 20)), ("jump", lambda: a.jump()), ("sneak", lambda: a.sneak(True)), ("sprint", lambda: a.sprint(True)), ("select_slot", lambda: a.select_slot(3)), ("use_item", lambda: a.use_item()), ("swing", lambda: a.swing()), ("dig", lambda: a.dig(7, -61, 10, face="up")), ("dig_finish", lambda: a.dig(7, -61, 10, face="up", finish=True)), ("place", lambda: a.place(7, -61, 10, face="up")), ("interact_blk",lambda: a.interact_block(7, -61, 10)), ("attack", lambda: a.attack(7)), ("use_entity", lambda: a.use_entity(7)), ("interact_at", lambda: a.interact_entity_at(7, 1, 2, 3)), ] for label, fn in cases: s.sent.clear() fn() assert len(s.sent) == 1, f"{label}: 期望发 1 个包，实发 {len(s.sent)}" # ★ 关键：用**同一个 Reader** 解完并检查是否正好读完 # （另起一个 Reader 量 remaining 是我先前写错的地方： # 那个 Reader 从未读过，pos 恒为 0，left 恒等于包长） rr = Reader(s.sent[0]) dec = pd.read("packet", rr, {}) assert dec.get("name"), f"{label}: 解不出包名 → {dec}" left = rr.remaining() assert left == 0, f"{label}: 解完还剩 {left} 字节（多写字段或包体超长）" print(f" ★ {len(cases)} 个操作的包都能被官方 protodef 精确解回（0 余字节）") def test_packed_position_roundtrip(): """★ 打包坐标必须与官方 position 解码**互逆**（含负数与极值）。""" pd = for_phase(_client().proto, "play", "toServer") for (x, y, z) in [(0, 0, 0), (6, -60, 10), (-2, -61, -5), (1000, 320, -1000), (-2_000_000, 319, 2_000_000)]: b = _packed_position(x, y, z) assert len(b) == 8, len(b) got = pd.read("position", Reader(b), {}) assert (got["x"], got["y"], got["z"]) == (x, y, z), \ f"({x},{y},{z}) 往返不一致 → {got}" print(" ★ 打包坐标与官方解码互逆（含负数/极值）") def test_direction_mapping(): """面名 → 协议编号：0=down 1=up 2=north 3=south 4=west 5=east。""" c = _client() expect = {"down": 0, "up": 1, "north": 2, "south": 3, "west": 4, "east": 5} for face, n in expect.items(): assert _direction(c, face) == n, f"{face} → {_direction(c, face)}，期望 {n}" # 别名 assert _direction(c, "top") == 1 and _direction(c, "bottom") == 0 print(f" ★ 六个面映射正确（{expect}）") # ---------- 行为级：不编造、夹紧、预置条件 ---------- def test_pitch_is_clamped_and_reported(): """★ 超界 pitch 必须夹住**并告知**（不是静默改）。""" s = _bare() a = Actions(s) r = a.look(0, 200) assert r["pitch"] == 90.0 and "clamped" in r, r r2 = a.look(0, -200) assert r2["pitch"] == -90.0 and "clamped" in r2, r2 r3 = a.look(0, 30) assert "clamped" not in r3, "正常值不该报 clamped" # yaw 取模 assert a.look(370, 0)["yaw"] == 10.0 print(" ★ pitch 夹紧并如实报告，yaw 正确取模") def test_invalid_slot_rejected_before_sending(): """★ 非法槽位**不发送**（预置条件检查在本地拦下）。""" s = _bare() a = Actions(s) s.sent.clear() r = a.select_slot(99) assert r["ok"] is False and "0..8" in r["reason"], r assert s.sent == [], "预置条件失败时不该发包" a.select_slot(8) assert len(s.sent) == 1, "合法槽位应发送" print(" ★ 非法槽位本地拦下（不发包）") def test_dig_refuses_on_unloaded_chunk(): """★ 区块没加载就别发挖掘意图（发了也无效，且浪费往返）。""" s = _bare() a = Actions(s) s.sent.clear() r = a.dig(1000, 60, 1000) # 无区块 assert r["ok"] is False, r assert "未加载" in r["reason"], r assert s.sent == [], "不应发包" print(" ★ 未加载区块的挖掘被本地拒绝") def test_place_warns_when_held_unknown(): """★ 手上是**未知**时必须 warning —— 因为症状与「服务端拒绝」无法区分。""" s = _bare() a = Actions(s) s.sent.clear() r = a.place(0, 0, 0) assert r["ok"] is False and "set_slot" in r["reason"], r # 已知为空槽 s.state.hotbar[0] = None s.state.slots[36] = {"itemCount": 0, "itemId": 0} r2 = a.place(0, 0, 0) assert "warning" in r2, r2 # 已知有东西（★ 同时要把 held_slot 指回 0：上一例把它留在了空槽 1） s.state.player.held_slot = 0 s.state.hotbar[0] = {"itemCount": 64, "itemId": 1} r3 = a.place(0, 0, 0) assert r3["ok"] and "warning" not in r3 and r3["held"], r3 print(" ★ 未知/空/有物 三种状态都如实报告") def test_move_updates_local_state_because_client_is_authoritative(): """★★ move **必须**更新本地坐标 —— 自身位置是**客户端权威**。 这条判据我第一版写反了（写成「不乐观改」），是**错的**。 实测证据（2026-09-26，M5 之后才发现）： 本地以为: (0.5, -60.0, -3.5) ← 连发 5 次 move 后仍是旧值 服务端认为: (5.5, -60.0, -3.5) ← 真的动了 5 格 ⇒ MC 里自身位置由客户端自己算、自己报；服务端只做校验， **不会**在每次普通移动后回发 position（那包只在传送/纠正时发）。 不更新本地状态 ⇒ 状态层永久陈旧 ⇒ 寻路、碰撞判断全建立在错位置上。 ★ 但**改世界**（dig/place）恰好相反：服务端权威，那里**不能**本地改。 """ s = _bare() a = Actions(s) r = a.move(100.0, -60.0, 100.0) after = (s.state.player.x, s.state.player.y, s.state.player.z) assert after == (100.0, -60.0, 100.0), \ f"move 后本地坐标未更新：{after}（服务端会发 position 纠正）" assert r["pending_server_correction"] is True, r assert "客户端权威" in r["note"], r # 待确认计数增加（尚未收到服务端 position） assert s.pending_corrections == 1, s.pending_corrections print(" ★★ move 更新本地状态（客户端权威）且标记待确认") def test_dig_place_do_not_touch_local_world_state(): """★ 与 move 相反：**改世界**等服务端权威，本地不动。 本地擅自改方块 = 撒谎：服务端可能拒绝（没权限/放不下）， 而 agent 会以为已经成功。 """ s = _bare() s.state.player.entity_id = 42 s.state.player.held_slot = 0 for _i in range(9): s.state.hotbar[_i] = {"itemCount": 1, "itemId": 1} s.state.world.chunks[(0, 0)] = _FakeChunk() a = Actions(s) before = dict(s.state.overrides) a.dig(0, 0, 0) a.place(0, 0, 0) assert s.state.overrides == before, "dig/place 不该改本地方块状态" print(" ★ dig/place 不改本地方块（等服务端 block_changed）") # ---------- 背包与手持物 ---------- def test_item_name_handles_real_slot_shape(): """★ 实测 1.21.1 字段是 **itemId**（驼峰）；写错就永远判「空手」。""" assert item_name({"itemCount": 31, "itemId": 1}) == "item#1" assert item_name({"itemCount": 1, "item_id": "minecraft:stone"}) == "minecraft:stone" assert item_name({"itemCount": 0, "itemId": 0}) is None, "空槽应是 None" assert item_name(None) is None assert item_name({}) is None, "无 itemId 时不编名字" print(" ★ itemId / item_id / 空槽 / 缺失 四种都对") def test_hotbar_index_mapping(): """★ window0 的 slot 36..44 才是热键栏（0..8）。搞错就看错物品。""" s = _bare() s.state.set_held_slot_item(36, {"itemCount": 5, "itemId": 2}) s.state.set_held_slot_item(44, {"itemCount": 7, "itemId": 3}) s.state.player.held_slot = 0 assert s.state.held_item_name() == "item#2" s.state.player.held_slot = 8 assert s.state.held_item_name() == "item#3" assert s.state.hotbar_item_name(0) == "item#2" assert s.state.hotbar_item_name(3) is None print(" ★ 热键栏 36..44 ↔ 0..8 映射正确") # ---------- 文本组件（1.20.3+ 聊天） ---------- def test_render_uses_real_server_shapes(): """★ 用**线上真实抓到的** NBT 形状（不是我想象的）。 三个真实 bug 都出在这： 1. 只取 `text` ⇒ 拿到 ""（组件的 text 常为空串） 2. translate 原样吐 ⇒ agent 看到 commands.give.success.single 这种键名 3. with 里数字存成 {"": 1}（空键）没读 ⇒ “Gave Dirt to ” """ assert render_text({"color": "red", "extra": [{"translate": "command.unknown.command"}], "text": ""}) \ == "Unknown or incomplete command, see below for error" got = render_text({"translate": "commands.give.success.single", "with": [ {"": "1"}, {"translate": "block.minecraft.dirt"}, {"": "mcfold_act"}]}) assert got == "Gave 1 Dirt to mcfold_act", repr(got) got2 = render_text({"translate": "commands.setblock.success", "with": [ {"": -2}, {"": -61}, {"": -5}]}) assert got2 == "Changed the block at -2, -61, -5", repr(got2) print(f" ★ 真实形状渲染正确（{got!r}）") def test_render_never_returns_empty_for_unknown(): """★ 取不到就保留键名，**绝不**返回空串（空串=「什么也没发生」）。""" assert render_text({"translate": "no.such.key.xyz"}) == "⟨no.such.key.xyz⟩" assert render_text(None) == "" assert render_text("bare") == "bare" assert render_text({"hoverEvent": {"contents": {"id": "minecraft:stone"}}}) == "Stone" print(" ★ 未知键保留原样（不静默变空）") def test_render_survives_deep_nesting(): """深嵌套不能栈溢出（服务端数据不可信）。""" comp = {"text": "x"} for _ in range(40): comp = {"extra": [comp]} out = render_text(comp) assert isinstance(out, str) and ("⟨" in out or "x" in out) print(" ✓ 深嵌套安全降级（不崩溃）") def test_humanize(): assert _humanize("minecraft:stone") == "Stone" assert _humanize("item.minecraft:diamond_sword") == "Diamond sword" assert _humanize("block.minecraft:red_flower") == "Red flower" print(" ✓ 物品 id 人性化") def test_flatten_with(): args = [{"": 1}, {"translate": "block.minecraft.stone"}] assert flatten_with(args) == ["1", "Stone"], flatten_with(args) assert flatten_with(None) == [] print(" ✓ with 参数摊平（含空键数字与 translate）") # ---------- 决定性：teleport_confirm（这层最隐蔽的必需前置包） ---------- def test_teleport_confirm_sent_on_position(): """★★ 1.19+ 必须在 position 后回 teleport_confirm。 不回 ⇒ 服务端**默默忽略所有方块操作**（dig/place 全无效、不报错）。 症状与「权限不足」完全无法区分，是本层最隐蔽的必需前置包。 回归：手工补发 confirm 后 place 立刻成功（air → stone）。 """ s = _bare() s._apply("position", {"x": 1.0, "y": 2.0, "z": 3.0, "yaw": 0, "pitch": 0, "teleportId": 7}) assert len(s.sent) == 1, s.sent raw = s.sent[0] assert raw[0] == s.client._serverbound_id("play", "packet_teleport_confirm") assert raw[1] == 7, raw[:4].hex() assert s.teleports_confirmed == 1 # ★ 反向对照：teleportId 缺失（老版本）时不得乱发 s.sent.clear() s._apply("position", {"x": 1.0, "y": 2.0, "z": 3.0, "yaw": 0, "pitch": 0}) assert s.sent == [], "无 teleportId 时不该发 confirm" print(" ★★ position 必回 teleport_confirm；无 id 时不乱发") if __name__ == "__main__": for fn in [test_every_packet_decodes_with_official_protodef, test_packed_position_roundtrip, test_direction_mapping, test_pitch_is_clamped_and_reported, test_invalid_slot_rejected_before_sending, test_dig_refuses_on_unloaded_chunk, test_place_warns_when_held_unknown, test_move_updates_local_state_because_client_is_authoritative, test_dig_place_do_not_touch_local_world_state, test_item_name_handles_real_slot_shape, test_hotbar_index_mapping, test_render_uses_real_server_shapes, test_render_never_returns_empty_for_unknown, test_render_survives_deep_nesting, test_humanize, test_flatten_with, test_teleport_confirm_sent_on_position]: fn() print("\n操作层：全部通过")