"""把 M2 的包流灌进 M3 的状态层（M3 的「线」）。 ## 为什么单独一个模块 M2（协议）与 M3（状态）关心的事不同： - 协议层：**这串字节是什么包** - 状态层：**世界现在是什么样** 把「收到包 ⇒ 更新状态」单独放一处，好处： - 状态层的每个字段都能追到它来自哪个包（可核查） - 新增状态字段时不必碰协议解析 - 线上的坏包（解不开的）在此处统一记账，不污染状态 ## 刻意的克制 只做「落状态」，**不做**： - 预测（不打提前量、不猜实体下一步） - 决策（不去找目标、不规划路径） - 猜测（认不出的 state id 照实记 unknown，不映射成别的方块） """ from __future__ import annotations from typing import Any, Callable from .client import Client from .state import Entity, WorldState, ingest_configuration from .protodef import Writer from .events import EventLog from .actions import Actions def _nbt_text(nbt: Any) -> Any: """把 NBT 文本组件渲染成可读文本（1.20.3+ 聊天走组件树，不是纯字符串）。 ★ 之前的三个错（都是“看着像服务端没反应”的静默失败）： 1. 只取 `text` 字段 ⇒ 拿到 `""`（组件的 text 常常是空串） 2. 就算取到 `translate` 也只当字符串吐 ⇒ agent 读到的是 `commands.give.success.single` 这种键名，不是人话 3. 就算翻译了，`with` 参数里数字存成 `{"": 1}`（空键）没读 ⇒ “Gave Dirt to ”（参数全空） 现在统一走 textcomp.render_text（查 language.json + 填 %s）。 ★ 失败时**保留原文/键名**而不是空串：空串会让 agent 以为「什么也没发生」。 """ try: from .textcomp import render_text out = render_text(nbt) return out if out else nbt except Exception: return nbt class StateSession: """一条「客户端 + 状态」的可运行会话。 ## 用法 ```python with StateSession("127.0.0.1", 25566, "mcfold_test", "1.21.1") as s: s.connect() # handshake → login → configuration → play s.pump(2.0) # 收 2 秒，把区块/位置/实体落进状态 print(s.state.summary()) ``` """ def __init__( self, host: str, port: int, username: str, version: str, registry: Any = None, data_root: str = "/home/newqqagent/plugin_data/mc/runtime/node_modules/" "minecraft-data/minecraft-data/data", ): self.client = Client(host=host, port=port, username=username, version=version) self.state = WorldState(registry=registry, num_sections=24) if registry is None: from .state import load_vanilla_registry self.state.world.registry = load_vanilla_registry(version, data_root) # ★ M4 事件层：给 LLM 用的**可按需排空**的事件日志（非回调推送） self.events = EventLog() # ★ M5 操作层：让 agent 能改变世界（chat/look/move/dig/place…） self.actions = Actions(self) # ★ 服务端传送确认计数（不 confirm 就会导致方块操作被静默忽略） self.teleports_confirmed = 0 # ★ 自身移动后等不到「纠正」的次数 —— 用于判断本地状态是否可信。 # 普通移动服务端**不会**回 position（那不是它的职责）， # 所以本地位置始终是「预期值」；只有收到 position 才算被确认。 self.pending_corrections = 0 # 兼容旧接口：回调式监听（给非 LLM 的宿主用），事件仍会同时入 EventLog self._listeners: list[tuple[str, Callable[[str, Any], None]]] = [] # ---------- 事件（M4 的落点，本轮只留机制） ---------- def on(self, event: str, fn: Callable[[str, Any], None]) -> None: self._listeners.append((event, fn)) def _emit(self, event: str, payload: Any) -> None: """同时进事件日志（LLM 路径）与回调（宿主路径）。 ★ 两路并存的原因：LLM 是回合制的，需要 `drain()` 式拉取； 而某些宿主（如测试/工具）想要即时回调。 只做其中一路都会让另一类使用者难用。 """ if isinstance(payload, dict): self.events.emit(event, payload) for name, fn in self._listeners: if name == event: fn(event, payload) # ---------- 连接 ---------- def connect(self, timeout: float = 15.0) -> dict[str, Any]: """走完 login → configuration → play，返回各阶段摘要。""" import time as _t c = self.client c.load_protocol() c.connect() c.sock.settimeout(timeout) c.handshake(state=2) c.login_start() # login 阶段：读到 success（途中处理压缩阈值） login_seen: list[str] = [] deadline = _t.time() + timeout while _t.time() < deadline: p = c._recv("login") nm = p["name"] login_seen.append(nm) if nm == "compress": c.compression = p["params"].get("threshold", 256) continue if nm == "success": break if nm == "disconnect": raise RuntimeError(f"服务端拒绝: {p['params'].get('reason')}") cfg = c.configuration_handshake() ingest_configuration(self.state, cfg, c.registries) self._emit("connected", {"cfg": cfg}) return { "login_packets": login_seen, "registries": cfg.get("registryCount"), "phase": c.phase, "geometry": self.state.geometry.describe(), } # ---------- 收包并落状态 ---------- def pump(self, seconds: float = 1.0) -> dict[str, int]: """收 N 秒包，把能落地的状态落地，返回各包计数。""" import time as _t c = self.client c.sock.settimeout(0.5) deadline = _t.time() + seconds counts: dict[str, int] = {} while _t.time() < deadline: try: p = c._recv(c.phase) except TimeoutError: continue except Exception as e: counts["_error"] = counts.get("_error", 0) + 1 self.state.note_packet(f"_error:{type(e).__name__}") continue nm = p["name"] counts[nm] = counts.get(nm, 0) + 1 self.state.note_packet(nm) if p.get("_undecoded"): # 上游数据缺口（当前只有 declare_recipes）：不落状态，但记账 continue try: self._apply(nm, p["params"]) except Exception as e: counts["_apply_error"] = counts.get("_apply_error", 0) + 1 self._emit("state_error", {"packet": nm, "error": str(e)[:120]}) return counts def _apply(self, name: str, params: dict[str, Any]) -> None: st = self.state c = self.client if name == "position": # ★ 收到服务端的权威位置 ⇒ 本地「预期值」得到确认/纠正 self.pending_corrections = 0 st.update_from_position(params) # ★★ M5 的关键一环（实测 2026-09-26，卡了很久）： # 1.19+ 的 position 包带 `teleportId`，客户端**必须**回 # teleport_confirm，**否则服务端会忽略后续的方块操作** # （dig/place/use 全部默默无效，且不给任何错误）。 # 症状：“操作没反应”，与“权限不足/空手”完全无法区分。 # 实测：补发 confirm 后 place 立刻成功（air → stone）。 tid = params.get("teleportId") if isinstance(tid, int) and tid >= 0: w = Writer() w.varint(c._serverbound_id("play", "packet_teleport_confirm")) w.varint(tid) c.send(w.getvalue()) self.teleports_confirmed += 1 self._emit("player_moved", st.player.as_dict()) elif name == "login": st.update_from_login(params) self._emit("joined", st.player.as_dict()) elif name == "map_chunk": ch = st.load_chunk(params["x"], params["z"], params["chunkData"]) self._emit("chunk_loaded", {"x": params["x"], "z": params["z"], "sections": len(ch.sections)}) elif name == "unload_chunk": # ★ 协议字段顺序是 chunkZ **在前**、chunkX 在后 # （1.21.1 packet_unload_chunk: [chunkZ:i32, chunkX:i32]） # 我先前写成 params.get("chunkZ", …) 塞给 cx ⇒ x/z 互换， # 卸载永远不命中 ⇒ 区块越积越多（内存缓慢泄漏 + 状态撒谎）。 cx = params.get("chunkX") cz = params.get("chunkZ") if isinstance(cx, int) and isinstance(cz, int): st.unload_chunk(cx, cz) self._emit("chunk_unloaded", {"x": cx, "z": cz}) elif name == "keep_alive": # ★ 必须回，否则服务端 15s 后踢我们（状态层的第一责任） w = Writer() w.varint(c._serverbound_id("play", "packet_keep_alive")) w.int64(int(params.get("keepAliveId", 0))) c.send(w.getvalue()) elif name == "game_state_change": reason = params.get("reason") # 0=开始游戏 1=结束游戏 3=改变gamemode 5=改变难度 6=改变生命 if reason == 3: st.player.gamemode = str(params.get("gameMode")) self._emit("game_state", params) elif name == "update_health": v = params.get("health") if isinstance(v, (int, float)): st.player.health = float(v) f = params.get("food") if isinstance(f, (int, float)): st.player.food = int(f) self._emit("health_changed", params) elif name == "entity_position": eid = params.get("entityId") if eid in st.entities: e = st.entities[eid] if params.get("deltaX") is not None: e.x += float(params["deltaX"]) e.y += float(params["deltaY"]) e.z += float(params["deltaZ"]) self._emit("entity_moved", e.as_dict()) elif name == "remove_entities": ids = params.get("entityIds") or params.get("entities") or [] if isinstance(ids, (list, tuple)): for eid in ids: st.remove_entity(int(eid)) self._emit("entities_removed", ids) elif name == "add_entity": eid = params.get("entityId") if isinstance(eid, int): st.upsert_entity(Entity( entity_id=eid, kind=str(params.get("type", "unknown")), x=float(params.get("x", 0.0)), y=float(params.get("y", 0.0)), z=float(params.get("z", 0.0)), )) self._emit("entity_spawned", eid) elif name == "update_view_position": self._emit("view_moved", params) # ---------- M4：世界变化类事件 ---------- elif name == "block_change": # 1.21.1 的 position 是 **bitfield 打包坐标**（x 26 / z 26 / y 12，有符号） loc = params.get("location") or {} bx, by, bz = int(loc.get("x", 0)), int(loc.get("y", 0)), int(loc.get("z", 0)) before = st.block_at(bx, by, bz) # ★ from 要能区分「之前是空气」与「之前我没见过」： # unloaded → 写「未加载」，而不是 None（None 看着像“没有值”）。 # agent 若把 unloaded 读成 air，会以为脚下是空气而做出错误动作。 from_name = before.get("name") if from_name is None: from_name = {"unloaded": "未加载(无原值)", "out-of-world": "界外"}.get( before.get("source"), f"未知({before.get('source')})") st.on_block_change(bx, by, bz, params.get("type")) # ★ 把 state_id 解成**方块名**再进事件： # agent 读「grass_block → stone」比读「9 → 1」有用得多； # 而且我先前写的事件键是 to_state_id、渲染器读 to ⇒ 显示 None # （「变了个不知道是什么的东西」，比不报还糟）。 now = st.block_at(bx, by, bz) self._emit("block_changed", { "x": bx, "y": by, "z": bz, "from": from_name, "to": now.get("name"), "to_state_id": params.get("type"), }) self.events.emit_fine(f"block_changed:{bx}:{by}:{bz}", {"to": now.get("name"), "to_state_id": params.get("type")}) elif name == "multi_block_change": sec = params.get("chunkCoordinates") or {} recs = params.get("records") or [] st.on_multi_block_change(int(sec.get("x", 0)), int(sec.get("z", 0)), recs) self._emit("block_changed", { "chunk": [int(sec.get("x", 0)), int(sec.get("z", 0))], "count": len(recs), "bulk": True, }) elif name in ("system_chat", "profileless_chat"): self._emit("chat", { "text": _nbt_text(params.get("content")), "action_bar": bool(params.get("isActionBar")), "system": True, }) elif name == "player_chat": self._emit("chat", { "from": params.get("senderName") or str(params.get("senderUuid")), "text": params.get("plainMessage"), "system": False, }) elif name == "damage_event": eid = params.get("entityId") if eid == st.player.entity_id: amt = params.get("damage") if amt is not None and st.player.health is not None: st.player.health = max(0.0, st.player.health - float(amt)) self._emit("damage", {"amount": amt, "health": st.player.health}) elif name == "entity_destroy": for eid in (params.get("entityIds") or []): e = st.entities.pop(int(eid), None) if e is not None: self._emit("entity_removed", e.as_dict()) elif name == "spawn_entity": eid = params.get("entityId") if isinstance(eid, int): st.upsert_entity(Entity( entity_id=eid, kind=str(params.get("type", "unknown")), x=float(params.get("x", 0.0)), y=float(params.get("y", 0.0)), z=float(params.get("z", 0.0)), )) self._emit("entity_spawned", st.entities[eid].as_dict()) elif name in ("rel_entity_move", "entity_move_look"): eid = params.get("entityId") e = st.entities.get(eid) if isinstance(eid, int) else None if e is not None: # ★ 增量单位是 1/4096 方块（协议定义），不是 1/1 —— 写错会得到 # 看似合理但完全错误的坐标（实体“瞬移”到远处）。 e.x += float(params.get("deltaX", 0)) / 4096.0 e.y += float(params.get("deltaY", 0)) / 4096.0 e.z += float(params.get("deltaZ", 0)) / 4096.0 p = st.player dist = ((e.x - p.x) ** 2 + (e.y - p.y) ** 2 + (e.z - p.z) ** 2) ** 0.5 self._emit("entity_moved", e.as_dict() | {"distance": round(dist, 2)}) elif name == "command_suggestions": self._emit("chat", {"text": params, "system": True, "suggestions": True}) elif name == "set_slot": # ★ 背包变化：没有它就不知道手上是什么 ⇒ place 会“发了但没反应” # 而你完全看不出原因（症状与“服务端拒绝”一模一样）。 wid = params.get("windowId") if wid == 0 or wid is None: slot = params.get("slot") if isinstance(slot, int): st.set_held_slot_item(slot, params.get("item")) self._emit("held_item", {"slot": slot, "item": st.held_item_name()}) elif name == "held_item_slot": self._emit("held_item", {"slot": params.get("slot")}) elif name == "declare_recipes": # 上游数据缺口解不开，但**包名已知**；这里不做任何事 self._emit("recipes_declared", {"decoded": False}) # ---------- 发（M5 的第一块，先把 chat/command 铺上） ---------- def send(self, payload: bytes) -> None: self.client.send(payload) def chat(self, text: str) -> None: """发聊天（M5）。结构随版本变化，**从协议数据取**而非写死。""" fields = self._serverbound_fields("play", "packet_chat_message") w = Writer() w.varint(self.client._serverbound_id("play", "packet_chat_message")) w.pstring(text) # 1.19.1+ 有 timestamp / salt / signature / etc for name, ftype in fields: if name == "message": continue if name in ("timestamp", "salt"): w.int64(0) elif isinstance(ftype, list) and ftype and ftype[0] == "option": w.boolean(False) # signature / unsignedChatContent：无 elif isinstance(ftype, list) and ftype and ftype[0] in ("varint", "i8", "u8", "bool"): w.varint(0) else: w.pstring("") # 未知可选字段：保守发空 self.send(w.getvalue()) def command(self, cmd: str) -> None: """发命令（**不带**前导 /）。1.19+ 走 chat_command 独立包。""" cmd = cmd.lstrip("/") pid = self.client._serverbound_id("play", "packet_chat_command") w = Writer() w.varint(pid) w.pstring(cmd) self.send(w.getvalue()) @staticmethod def _serverbound_fields(phase: str, pname: str) -> list[tuple[str, Any]]: """从 protocol.json 读 serverbound 包的字段（顺序敏感）。""" import json as _json try: types = self.client.proto[phase]["toServer"]["types"] c = types.get(pname) if c and isinstance(c, list) and len(c) > 1: return [(f["name"], f["type"]) for f in c[1]] except Exception: pass return [] # ---------- 便捷 ---------- def close(self) -> None: try: self.client.close() except Exception: pass def __enter__(self) -> "StateSession": return self def __exit__(self, *exc: Any) -> None: self.close()