"""M4 事件层判据：events.py + session 的事件接线。 ## 为什么这层的判据特别要「反直觉」式断言 事件层最危险的缺陷是**静默**： - 事件丢了 → agent 以为世界没变（比报错糟） - 游标没推进 → agent 反复看同一批（浪费 token 且做出重复动作） - 细分层混进粗粒度 → 每次观察几百条噪声（烧 token） - 事件里字段名与渲染器不一致 → 显示 `None`（看着有事件，其实没信息） 所以判据都对着这些「静默退化」写，且**有反向对照** （负向输入必须报错/拒绝，不能静默通过）。 """ import sys, os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from mcfold.events import EventLog, render_events, COARSE_KINDS from mcfold.state import WorldState from mcfold.chunk import BlockRegistry from mcfold.session import StateSession, _nbt_text def _reg(): from mcfold.state import load_vanilla_registry return load_vanilla_registry( '1.21.1', '/home/newqqagent/plugin_data/mc/runtime/node_modules/minecraft-data/minecraft-data/data') def _bare(): sess = StateSession.__new__(StateSession) sess.state = WorldState(registry=_reg(), num_sections=24) sess.events = EventLog() sent = [] sess.client = type("C", (), { "_serverbound_id": staticmethod(lambda *a: 0x04), "send": staticmethod(lambda b: sent.append(b)), })() sess._listeners = [] sess._sent = sent return sess # ---------- 缓冲与游标 ---------- def test_drain_is_idempotent_and_advances_cursor(): """★ 游标必须幂等：同一时刻反复 drain 不会拿到重复事件。 这是 token 与正确性的双关：重复读会让 agent 重复动作。 """ log = EventLog() for i in range(5): log.emit("chat", {"text": f"m{i}"}) a = log.drain() assert len(a) == 5 b = log.drain() assert b == [], f"第二次排空应为空，实际 {b}" assert log.cursor == log.seq # peek 不消费：★ 判据要点是「peek 前后 cursor 不变」， # 而不是 cursor==seq（emit 新事件后 seq 会比 cursor 大，那是正常的） log.emit("chat", {"text": "later"}) before = log.cursor assert len(log.peek()) == 1 assert len(log.peek()) == 1 # peek 可反复看 assert log.cursor == before # ★ peek 绝不推进游标 assert len(log.drain()) == 1 assert log.cursor == log.seq print(" ★ drain 幂等且推进游标，peek 不消费") def test_drain_limit_keeps_backlog(): """★ limit 只是「本轮取多少」，**不是丢弃**：剩下的留给下次。""" log = EventLog() for i in range(10): log.emit("block_changed", {"x": i}) first = log.drain(limit=4) assert len(first) == 4 second = log.drain(limit=4) assert len(second) == 4 third = log.drain(limit=4) assert len(third) == 2 assert fourth_empty(log) is True # 且序号连续，没有重复或跳过 all_seen = [e["seq"] for e in first + second + third] assert all_seen == list(range(1, 11)), all_seen print(" ★ limit 不丢事件（剩 6 条分两轮取完，序号连续）") def fourth_empty(log): return log.drain() == [] def test_fine_layer_separate_and_queryable(): """★ 细分层（block_changed:x:y:z）不得混进粗粒度流。""" log = EventLog() log.emit_change("block_changed", "block_changed:1:2:3", {"x": 1, "y": 2, "z": 3}) log.emit_change("block_changed", "block_changed:4:5:6", {"x": 4, "y": 5, "z": 6}) coarse = log.drain() assert len(coarse) == 2 and all(e["kind"] == "block_changed" for e in coarse) fine = log.drain_fine("block_changed:") assert len(fine) == 2 assert [e["kind"] for e in fine] == ["block_changed:1:2:3", "block_changed:4:5:6"] # 按坐标前缀筛 one = log.drain_fine("block_changed:4:5:6") assert len(one) == 1 # ★ 查细分层**不推进**粗粒度游标 assert log.drain() == [] print(" ★ 细分层独立可查，不污染粗粒度流") def test_capacity_drops_are_counted_not_hidden(): """★ 环形缓冲满了会丢，但必须**如实计数**。 静默丢 = agent 以为世界没发生事（最坏的撒谎）。 """ log = EventLog(capacity=4) for i in range(10): log.emit("chat", {"text": str(i)}) st = log.stats() assert st["dropped"] == 6, st got = log.drain() assert len(got) == 4 assert got[0]["data"]["text"] == "6", got[0] # 最旧的已丢 print(" ★ 容量溢出时丢最旧且 dropped 如实计数（6）") def test_reset_cursor_skips_history(): log = EventLog() for i in range(3): log.emit("chat", {"text": str(i)}) log.reset_cursor() assert log.drain() == [] log.emit("chat", {"text": "new"}) assert len(log.drain()) == 1 print(" ✓ reset_cursor 跳过历史（agent 明确不关心之前）") # ---------- 渲染 ---------- def test_render_block_change_has_real_names(): """★ 事件里的方块必须是**名字**不是 id，且 from/to 都不能为 None。 回归：先前我发的事件键叫 to_state_id 而渲染器读 to ⇒ 显示 「grass_block → None」——看着有事件，其实没信息。 """ sess = _bare() sess._apply("block_change", { "location": {"x": 9, "y": 64, "z": -3}, "type": 1, # stone }) evs = sess.events.drain() bc = [e for e in evs if e["kind"] == "block_changed"] assert bc, evs d = bc[0]["data"] assert d["x"] == 9 and d["y"] == 64 and d["z"] == -3, d # ★ 名字必须解出来了（不是 id 也不是 None） assert d["to"] == "stone", d assert d["to"] is not None and "unknown" not in str(d["to"]), d txt = render_events(bc) assert "→ stone" in txt, txt assert "None" not in txt, f"渲染里出现 None：{txt}" print(f" ★ 方块变化事件带真实名字（{txt.splitlines()[0]}）") def test_render_unknown_block_is_honest(): """认不出的方块要显示 unknown(id)，不能编个名字。""" sess = _bare() # 用一张几乎空的注册表，逼出未知 sess.state.world.registry = BlockRegistry( [{"name": "stone", "minStateId": 1, "maxStateId": 1}]) sess._apply("block_change", {"location": {"x": 0, "y": 0, "z": 0}, "type": 9999}) d = [e for e in sess.events.drain() if e["kind"] == "block_changed"][0]["data"] assert d["to_state_id"] == 9999 assert d["to"] is None, d # 状态层把它记成未知（供上层告警） assert 9999 in sess.state.world.unknown_state_ids print(" ★ 未知方块不编名字（to=None 且记账）") def test_render_is_compact(): """渲染要省 token：短行、无冗余 JSON 结构。""" log = EventLog() log.emit("block_changed", {"x": 1, "y": 2, "z": 3, "from": "air", "to": "stone"}) log.emit("chat", {"from": "Alice", "text": "hi"}) log.emit("entity_spawned", {"kind": "pig", "entity_id": 7}) log.emit("player_moved", {"x": 1.0, "y": 2.0, "z": 3.0}) txt = render_events(log.drain()) assert "方块变化 @(1,2,3) air → stone" in txt, txt assert "聊天 Alice: hi" in txt, txt assert "出现 pig id=7" in txt, txt assert "我移动到 (1.0,2.0,3.0)" in txt, txt # ★ 断言「不用 JSON 大括号」——这正是省 token 的关键 assert "{" not in txt and '"' not in txt, f"混进了 JSON：{txt}" print(f" ✓ 渲染紧凑（4 事件 {len(txt)} 字符，无 JSON 噪声）") def test_unknown_event_not_silently_swallowed(): """未知事件必须出现在渲染里（agent 能看到有东西发生了）。""" log = EventLog() log.emit("some_new_thing", {"a": 1}) txt = render_events(log.drain()) assert "some_new_thing" in txt, txt print(" ★ 未知事件如实渲染（不静默吞）") def test_coarse_kinds_are_stable_contract(): """COARSE_KINDS 是给 agent 的稳定契约，不该随意变动。""" for k in ("block_changed", "chat", "entity_spawned", "player_moved", "damage", "joined", "chunk_loaded"): assert k in COARSE_KINDS, k # 细分层命名不应混进粗粒度契约 assert not any(":" in k for k in COARSE_KINDS), COARSE_KINDS print(f" ✓ 粗粒度契约稳定（{len(COARSE_KINDS)} 种，名称无冒号）") # ---------- NBT 文本提取 ---------- def test_nbt_text_extracts_or_reports_honestly(): """★ 取不出文本时返回原值，**不返回空串**。 chat 是 agent 判断局势的重要输入；静默变「（空）」比看不懂更糟。 """ assert _nbt_text("hello") == "hello" assert _nbt_text({"text": "hi"}) == "hi" # ★ 真实形状（线上抓的）：translate + with（含空键参数） got = _nbt_text({"translate": "commands.give.success.single", "with": [{"": "1"}, {"translate": "block.minecraft.dirt"}, {"": "mcfold_act"}]}) assert got == "Gave 1 Dirt to mcfold_act", repr(got) # 取不出 ⇒ 原样返回（**不**返回空串：空串会被读成「什么也没发生」） weird = {"weird": {"deep": 1}} assert _nbt_text(weird) == weird, "取不出应原样返回" # 注：曾用 {"raw": "raw!"} 断言过 —— 那是我**凭想象编的**字段， # 线上服务端发的是 text/translate/with/extra/hoverEvent。 # 判据不能建立在想象的输入上（否则测的是我的臆想，不是协议事实）。 assert _nbt_text(None) is None print(" ✓ NBT 文本按真实形状提取，取不出则如实返回原值") # ---------- session 接线 ---------- def test_entity_move_uses_4096_scale(): """★ 实体移动增量的单位是 **1/4096**，不是 1/1。 写错的症状很隐蔽：实体坐标看起来「合理」，但完全错（瞬移到远处）。 """ sess = _bare() from mcfold.state import Entity sess.state.upsert_entity(Entity(entity_id=1, kind="pig", x=0.0, y=0.0, z=0.0)) sess.state.player.x = 0.0 # 4096 * 0.5 = 2048 ⇒ 应移动 0.5 格 sess._apply("rel_entity_move", {"entityId": 1, "deltaX": 2048, "deltaY": 1024, "deltaZ": -2048}) e = sess.state.entities[1] assert (round(e.x, 4), round(e.y, 4), round(e.z, 4)) == (0.5, 0.25, -0.5), \ (e.x, e.y, e.z) # ★ 反向对照：若用 /1 会得到 2048 格（离谱） assert abs(e.x) < 1.0, f"疑似未除 4096：{e.x}" print(f" ★ 实体增量按 1/4096 换算（x={e.x}, y={e.y}, z={e.z}）") def test_damage_reduces_health(): sess = _bare() sess.state.player.entity_id = 7 sess.state.player.health = 20.0 sess._apply("damage_event", {"entityId": 7, "damage": 6, "sourceCauseId": 1}) assert sess.state.player.health == 14.0, sess.state.player.health evs = [e for e in sess.events.drain() if e["kind"] == "damage"] assert evs and evs[0]["data"]["health"] == 14.0 # 血不能扣成负 sess._apply("damage_event", {"entityId": 7, "damage": 100}) assert sess.state.player.health == 0.0, sess.state.player.health print(" ★ 伤害扣血且不会扣成负数") def test_damage_to_others_ignored(): """别人受伤不是「我的伤害」事件（否则 agent 会误判自己挨打）。""" sess = _bare() sess.state.player.entity_id = 7 sess.state.player.health = 20.0 sess._apply("damage_event", {"entityId": 999, "damage": 5}) assert sess.state.player.health == 20.0 assert not [e for e in sess.events.drain() if e["kind"] == "damage"] print(" ✓ 别人受伤不误报为我的伤害") def test_entity_destroy_emits_removal(): from mcfold.state import Entity sess = _bare() sess.state.upsert_entity(Entity(entity_id=5, kind="cow")) sess.state.upsert_entity(Entity(entity_id=6, kind="sheep")) sess._apply("entity_destroy", {"entityIds": [5]}) assert 5 not in sess.state.entities and 6 in sess.state.entities rem = [e for e in sess.events.drain() if e["kind"] == "entity_removed"] assert len(rem) == 1 and rem[0]["data"]["entity_id"] == 5 print(" ✓ 实体消失只报实际消失的那个") def test_both_listeners_and_eventlog_fire(): """两路并存：回调（宿主）与事件日志（LLM）都要收到。""" sess = _bare() got = [] sess.on("block_changed", lambda e, p: got.append(p)) sess._apply("block_change", {"location": {"x": 1, "y": 2, "z": 3}, "type": 1}) assert len(got) == 1 assert [e for e in sess.events.drain() if e["kind"] == "block_changed"] print(" ✓ 回调与事件日志两路并存") if __name__ == "__main__": for fn in [test_drain_is_idempotent_and_advances_cursor, test_drain_limit_keeps_backlog, test_fine_layer_separate_and_queryable, test_capacity_drops_are_counted_not_hidden, test_reset_cursor_skips_history, test_render_block_change_has_real_names, test_render_unknown_block_is_honest, test_render_is_compact, test_unknown_event_not_silently_swallowed, test_coarse_kinds_are_stable_contract, test_nbt_text_extracts_or_reports_honestly, test_entity_move_uses_4096_scale, test_damage_reduces_health, test_damage_to_others_ignored, test_entity_destroy_emits_removal, test_both_listeners_and_eventlog_fire]: fn() print("\n事件层：全部通过")