feat(apilot): real network stubs with cross-tick buffer

NetClientStub + NetworkCtlStub 真实网络: 环回 socket 测试覆盖 send/dictch
- netclient.py: 持 Wire 收发 Frame, set_sdata/get_sdata/is_connected
- networkctl.py: send_msg 组 Frame, register_lua_handler 注册 Lua cb,
  install_to_lua 暴露冒号调用的 networkCtl 闭包 + mkmsg 工厂(IIFE 形式,
  避免 Lupa eval 对 'return function' 返回 None 的坑),
  dispatch_frame 懒安装
- wire.py: Wire 自持跨 recv 调用的 _buf, 半包 rest 留存到下次
  (派发附加约束: 帧跨两次 tick 到达不丢)
- test_network_stubs.py: 4 用例 (roundtrip/dispatch/register/lifecycle)
- test_wire.py: 加 test_recv_buffered_across_calls 覆盖跨 tick 半包

apilot/stubs/netclient.py

@@ -0,0 +1,40 @@
+"""netClient C++ 单例的 stub: 直接持有 Wire,负责收发 Frame。"""
+from __future__ import annotations
+
+from apilot.wire import Frame, Wire
+
+
+class NetClientStub:
+    def __init__(self) -> None:
+        self.wire: Wire | None = None
+        self._sdata: int = 0
+
+    def connect(self, host: str, port: int, timeout: float = 10) -> None:
+        self.wire = Wire(host, port, timeout)
+
+    def disconnect(self) -> None:
+        if self.wire is not None:
+            self.wire.close()
+        self.wire = None
+
+    def is_connected(self) -> bool:
+        return self.wire is not None
+
+    def send_raw(self, data: bytes) -> None:
+        assert self.wire is not None, "netClient not connected"
+        self.wire.sock.sendall(data)
+
+    def send_frame(self, f: Frame) -> None:
+        assert self.wire is not None, "netClient not connected"
+        self.wire.send(f)
+
+    def tick(self, deadline_s: float) -> list[Frame]:
+        if self.wire is None:
+            return []
+        return self.wire.recv(deadline_s)
+
+    def set_sdata(self, v: int) -> None:
+        self._sdata = int(v)
+
+    def get_sdata(self) -> int:
+        return self._sdata

apilot/stubs/networkctl.py

@@ -0,0 +1,91 @@
+"""LuaNetworkController stub + msg 对象构造 + 分发。"""
+from __future__ import annotations
+
+from typing import Any
+
+from apilot.wire import Frame
+from apilot.stubs.netclient import NetClientStub
+
+MSG_LUA = """
+local function mkmsg(recog, msgId, p1, p2, p3, body)
+    local pos = 1
+    local reader = {}
+    function reader:ReadString(n)
+        local s = string.sub(body, pos, pos + n - 1)
+        pos = pos + n
+        return s
+    end
+    local m = {}
+    function m:GetMsgId() return msgId end
+    function m:GetHeader()
+        return {recog=recog, param1=p1, param2=p2, param3=p3, msgId=msgId}
+    end
+    function m:GetDataLength() return #body end
+    function m:GetData() return reader end
+    return m
+end
+return mkmsg
+"""
+
+
+class NetworkCtlStub:
+    def __init__(self, nc: NetClientStub) -> None:
+        self.nc = nc
+        self._tail = 0
+        self.handlers: dict[int, Any] = {}
+        self._installed: set[int] = set()  # 已 install 的 lua runtime id
+
+    def _next_tail(self) -> int:
+        self._tail += 1
+        return self._tail
+
+    def send_msg(self, msg_id: int, recog: int, p1: int, p2: int, p3: int,
+                 data_str: "bytes | str | None", data_len: int) -> None:
+        if data_str is None:
+            body = b""
+        elif isinstance(data_str, str):
+            body = data_str.encode("utf-8")
+        else:
+            body = bytes(data_str)
+        self.nc.send_frame(Frame(recog=recog, msg_id=msg_id, p1=p1, p2=p2, p3=p3,
+                                 tail=self._next_tail(), body=body))
+
+    def register_lua_handler(self, msg_id: int, cb: Any) -> None:
+        self.handlers[int(msg_id)] = cb
+
+    def unregister_all(self) -> None:
+        self.handlers.clear()
+
+    def install_to_lua(self, lua: Any) -> None:
+        g = lua.globals()
+        g["AP_networkCtl"] = self
+        # 冒号调用 networkCtl:SendMsg(...) 兼容:__index 返回 (self, ...) 闭包
+        lua.execute("""
+networkCtl = networkCtl or {}
+setmetatable(networkCtl, {__index = function(_, k)
+    local self = AP_networkCtl
+    local map = {SendMsg="send_msg",
+                 RegisterLuaHandler="register_lua_handler",
+                 UnregisterAllLuaHandler="unregister_all"}
+    local pyname = map[k]
+    if pyname then
+        return function(_, ...) return self[pyname](self, ...) end
+    end
+    return function(...) return nil end
+end})
+LuaNetworkController = {Inst = function() return networkCtl end}
+""")
+        lua.execute("AP_mkmsg = (function() " + MSG_LUA + " end)()")
+        self._installed.add(id(lua))
+
+
+def dispatch_frame(lua: Any, ctl: NetworkCtlStub, f: Frame) -> None:
+    cb = ctl.handlers.get(f.msg_id)
+    if cb is None:
+        return
+    if id(lua) not in ctl._installed:
+        ctl.install_to_lua(lua)
+    mkmsg = lua.globals()["AP_mkmsg"]
+    body = f.body.decode("latin-1")
+    msg = mkmsg(f.recog, f.msg_id, f.p1, f.p2, f.p3, body)
+    cb(msg)

apilot/wire.py

@@ -45,27 +45,27 @@ def unpack_stream(buf: bytes) -> tuple[list[Frame], bytes]:
 
 
 class Wire:
-    """同步 TCP 线客户端。"""
+    """同步 TCP 线客户端。持跨 recv 调用的内部 buffer:半包 rest 留存到下次。"""
 
     def __init__(self, host: str, port: int, timeout: float = 10) -> None:
         self.sock = socket.create_connection((host, port), timeout=timeout)
+        self._buf: bytes = b""
 
     def send(self, f: Frame) -> None:
         self.sock.sendall(pack_frame(f))
 
     def recv(self, deadline_s: float) -> list[Frame]:
         deadline = time.time() + deadline_s
-        buf = b""
         while time.time() < deadline:
             self.sock.settimeout(max(0.2, deadline - time.time()))
             try:
                 d = self.sock.recv(65536)
                 if not d:
                     break
-                buf += d
+                self._buf += d
             except (socket.timeout, ConnectionResetError, OSError):
                 break
-        frames, _ = unpack_stream(buf)
+        frames, self._buf = unpack_stream(self._buf)
         return frames
 
     def close(self) -> None:

tests/test_network_stubs.py

@@ -0,0 +1,101 @@
+# tests/test_network_stubs.py
+"""NetClientStub + NetworkCtlStub: 真实环回 socket 模拟网关。"""
+import socket
+import struct
+import sys
+import threading
+
+import lupa.luajit21 as lj
+
+sys.path.insert(0, "protocol_bot")
+from encode6bitbuf_final import encode6  # type: ignore  # noqa: E402
+
+from apilot.stubs.netclient import NetClientStub  # noqa: E402
+from apilot.stubs.networkctl import NetworkCtlStub, dispatch_frame  # noqa: E402
+
+
+def _build_frame(msg_id: int, body: bytes) -> bytes:
+    """encode6(9×i32 header) + body。字段序对齐 wire.pack_frame。"""
+    head = struct.pack("<9i", 0, msg_id, 1, 0, 0, 0, 0, len(body), 1)
+    return encode6(head) + body
+
+
+def _spawn_gateway(handler):
+    """启一个临时监听,handler(server_sock) 在调用线程同步完成一次会话。"""
+    srv = socket.socket()
+    srv.bind(("127.0.0.1", 0))
+    srv.listen(1)
+    port = srv.getsockname()[1]
+    threading.Thread(target=handler, args=(srv,), daemon=True).start()
+    return port
+
+
+def _fake_gateway_echo529(server_sock):
+    """收一帧, 回一个 529 帧。body 在线缆上是 raw bytes(仅 header 编码)。"""
+    conn, _ = server_sock.accept()
+    try:
+        conn.recv(65536)  # drain client frame
+        conn.sendall(_build_frame(529, b"key32/session123"))
+    finally:
+        conn.close()
+        server_sock.close()
+
+
+def test_sendmsg_roundtrip_and_dispatch():
+    port = _spawn_gateway(_fake_gateway_echo529)
+
+    nc = NetClientStub()
+    nc.connect("127.0.0.1", port)
+    ctl = NetworkCtlStub(nc)
+    ctl.send_msg(2001, 123456, 1, 2, 3, b"uid/key/ch/1/0", 19)
+
+    frames = nc.tick(1.0)
+    assert any(f.msg_id == 529 for f in frames), f"expected 529 in {frames}"
+
+    rt = lj.LuaRuntime()
+    rt.execute("""
+got = nil
+function on_529(msg)
+    local h = msg:GetHeader()
+    local data = msg:GetData()
+    got = h.msgId .. ":" .. data:ReadString(5)
+end
+""")
+    ctl.register_lua_handler(529, rt.eval("on_529"))
+    target = next(f for f in frames if f.msg_id == 529)
+    dispatch_frame(rt, ctl, target)
+    assert rt.eval("got") == "529:key32"
+    nc.disconnect()
+
+
+def test_register_unregister():
+    """register_lua_handler 覆盖, unregister_all 清空。"""
+    nc = NetClientStub()
+    ctl = NetworkCtlStub(nc)
+    rt = lj.LuaRuntime()
+    rt.execute("function h(msg) end")
+    cb = rt.eval("h")
+    ctl.register_lua_handler(10, cb)
+    ctl.register_lua_handler(10, cb)  # 覆盖
+    assert ctl.handlers[10] is cb
+    ctl.unregister_all()
+    assert ctl.handlers == {}
+
+
+def test_dispatch_no_handler_is_noop():
+    """无 handler 的帧不抛异常。"""
+    nc = NetClientStub()
+    ctl = NetworkCtlStub(nc)
+    from apilot.wire import Frame
+    rt = lj.LuaRuntime()
+    dispatch_frame(rt, ctl, Frame(recog=0, msg_id=99999, p1=0, p2=0, p3=0, tail=1, body=b""))
+
+
+def test_is_connected_and_disconnect():
+    nc = NetClientStub()
+    assert not nc.is_connected()
+    port = _spawn_gateway(_fake_gateway_echo529)
+    nc.connect("127.0.0.1", port)
+    assert nc.is_connected()
+    nc.disconnect()
+    assert not nc.is_connected()

tests/test_wire.py

@@ -29,3 +29,46 @@ def test_pack_matches_reference_impl():
     ref = encode6(struct.pack("<9i", 0, 20, 900000, 111, 14481607, 8043, 0, len(body), 9)) + body
     mine = pack_frame(Frame(recog=900000, msg_id=20, p1=111, p2=14481607, p3=8043, tail=9, body=body))
     assert mine == ref
+
+def test_recv_buffered_across_calls():
+    """帧跨两次 recv() 到达不能丢(Wire 自持 buffer,rest 留存到下次)。"""
+    import socket, struct, threading, time
+    sys_path_hack = __import__("sys"); sys_path_hack.path.insert(0, "protocol_bot")
+    from encode6bitbuf_final import encode6
+    from apilot.wire import Wire
+
+    srv = socket.socket(); srv.bind(("127.0.0.1", 0)); srv.listen(1)
+    port = srv.getsockname()[1]
+
+    body = b"key32/session123"
+    head = struct.pack("<9i", 0, 529, 1, 0, 0, 0, 0, len(body), 1)
+    wire_bytes = encode6(head) + body
+    mid = len(wire_bytes) // 2
+
+    def gateway():
+        conn, _ = srv.accept()
+        try:
+            conn.setblocking(False)
+            try:
+                conn.recv(65536)
+            except (BlockingIOError, OSError):
+                pass
+            conn.setblocking(True)
+            conn.sendall(wire_bytes[:mid])  # 第一半
+            time.sleep(0.2)
+            conn.sendall(wire_bytes[mid:])  # 第二半
+            time.sleep(0.2)
+        finally:
+            conn.close()
+            srv.close()
+
+    threading.Thread(target=gateway, daemon=True).start()
+
+    w = Wire("127.0.0.1", port, timeout=2.0)
+    f1 = w.recv(0.3)            # 只能拿到第一半
+    f2 = w.recv(0.3)            # 第二半在此次补齐
+    w.close()
+    all_frames = f1 + f2
+    assert len([f for f in all_frames if f.msg_id == 529]) == 1, f"got {all_frames}"
+    g = next(f for f in all_frames if f.msg_id == 529)
+    assert g.body == body