APilot Headless M1 实施计划(stub 骨架 + 登录链 + T01 无客户端创角)

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 在无游戏客户端的机器上,用 Python+Lupa(LuaJIT 2.1)驱动游戏原版 Lua 的 LoginProxy 完成真实登录链并创建角色(逐鹿05区实测通过)。

Architecture: Python 宿主进程内嵌 Lupa LuaJIT;游戏原版 Lua(lua脚本解密源码/ 的 netMessageDef/framework/networkUtil/LoginProxy 等协议层)只读复用;Python stub 冒充 C++ 绑定面(netClient/networkCtl/Scheduler/环境类),网络重活直通已验证的 wire 层(Encode6BitBuf/帧编解码/socket)。

Tech Stack: Python 3.12、Lupa(源码构建 luajit21,tools/ 已验证)、LuaJIT 2.1、pytest。

Spec: docs/superpowers/specs/2026-09-21-apilot-headless-design.md

Global Constraints

游戏 Lua 一律只读:lua脚本解密源码/ 与其镜像 lua_src/ 禁止修改(设计 D3)
任务/业务代码只允许走 stub 与 APilot API,禁止为绕过问题改游戏 Lua
Python 版本 3.12(virtualenv:protocol_bot/.venv,已装 lupa-luajit21 补丁版)
运行环境:macOS arm64 本机;真实网关 s{zone}-jycq-206923.dhsf.xqhuyu.com
测试账号 longwu988 / Longwu1988(实名已通过);逐鹿05区 zone_id=14481607
所有对真实网关的集成测试打 @pytest.mark.live 标记,默认跳过(-m live 显式开启),单元测试零网络依赖
遵循 250 行/文件上限;类型标注完整,禁 Any 转换逃逸

文件结构总览(M1 全量)

apilot/
├── __init__.py
├── wire.py            # 协议线层:socket+帧编解码+Encode6BitBuf(从 cq2_pure_enter 提炼,类化)
├── bridge.py          # Lupa 桥:LuaRuntime 构建+游戏Lua加载+stub注入+tick循环
├── stubs/
│   ├── __init__.py    # install_all(lua, ctx) 装配入口
│   ├── noop.py        # __index 万能 no-op 工厂(cc.Sprite 等 C 类)
│   ├── netclient.py   # netClient 单例 stub(Connect/Disconnect/Send/Tick/sdata)
│   ├── networkctl.py  # LuaNetworkController stub(SendMsg/RegisterLuaHandler)
│   ├── scheduler.py   # Scheduler/PerformWithDelayGlobal 驱动
│   └── env.py         # L_GameEnvManager/FileUtils/gameEnviroment/LuaBridgeCtl 环境注入
├── host.py            # 宿主:装配一切,暴露 run(zone, account) / tick 循环
└── main.py            # CLI: python -m apilot.main --zone 14481607 --create-role
tests/
├── test_wire.py       # 帧编解码/Encode6BitBuf 回环(纯单测)
├── test_noop.py       # no-op 工厂行为
├── test_scheduler.py  # 计时轮
├── test_bridge_load.py# 游戏Lua加载冒烟(netMessageDef/framework/networkUtil)
└── test_live_t01.py   # [live] 无客户端创角端到端
lua_src -> ../lua脚本解密源码   # 符号链接(M1 不复制,保持只读引用)

Task 1: wire.py 协议线层(纯 Python,无 Lua)

Files:

Create: apilot/__init__.py(空)
Create: apilot/wire.py
Test: tests/test_wire.py

Interfaces:

Consumes: protocol_bot/encode6bitbuf_final.py 的 encode6/decode6(直接 import,不复制)
Produces:
- @dataclass Frame: recog:int, msg_id:int, p1:int, p2:int, p3:int, tail:int, body:bytes
- def pack_frame(f: Frame) -> bytes — 9×int 小端头 + body,整体 encode6
- def unpack_stream(buf: bytes) -> tuple[list[Frame], bytes] — 返回(完整帧列表, 剩余半包字节)
- class Wire: __init__(host, port, timeout=10); send(f: Frame) -> None; recv(deadline_s: float) -> list[Frame]; close() — 同步 socket,recv 循环收满 deadline
[ ] Step 1: 写失败测试(帧回环)

# tests/test_wire.py
import pytest
from apilot.wire import Frame, pack_frame, unpack_stream

def test_frame_roundtrip():
    f = Frame(recog=900000, msg_id=20, p1=123, p2=14481607, p3=8043,
              tail=7, body=b"*#acct/role/12345/9/1/0/110585/0/1/1/1/1")
    packed = pack_frame(f)
    frames, rest = unpack_stream(packed)
    assert rest == b""
    assert len(frames) == 1
    g = frames[0]
    assert (g.recog, g.msg_id, g.p1, g.p2, g.p3, g.tail, g.body) == \
           (f.recog, f.msg_id, f.p1, f.p2, f.p3, f.tail, f.body)

def test_unpack_partial_packet():
    f = Frame(recog=0, msg_id=529, p1=0, p2=0, p3=0, tail=1, body=b"key32abc")
    packed = pack_frame(f)
    frames, rest = unpack_stream(packed[:-10])   # 掐掉10字节
    assert frames == [] and len(rest) == len(packed) - 10

Step 2: 跑测试确认失败

Run: cd /Users/zlwu/work/00code/传奇 && protocol_bot/.venv/bin/python -m pytest tests/test_wire.py -v Expected: FAIL(ModuleNotFoundError: apilot)

Step 3: 实现 wire.py

# apilot/wire.py
"""协议线层: 24B头(9×i32)+体, 全帧 Encode6BitBuf。提炼自 protocol_bot/cq2_pure_enter.py(实测验证)。"""
from __future__ import annotations
import socket
import struct
import sys
import time
from dataclasses import dataclass

sys.path.insert(0, "protocol_bot")
from encode6bitbuf_final import encode6, decode6  # type: ignore

HDR = struct.Struct("<9i")  # recog,msg,p1,p2,p3,zero?,len,tail → 对齐 cq2_pure_enter.send()


@dataclass
class Frame:
    recog: int
    msg_id: int
    p1: int
    p2: int
    p3: int
    tail: int
    body: bytes = b""


def pack_frame(f: Frame) -> bytes:
    head = HDR.pack(0, f.msg_id, f.recog, f.p1, f.p2, f.p3, 0, len(f.body), f.tail)
    return encode6(head) + f.body


def unpack_stream(buf: bytes) -> tuple[list[Frame], bytes]:
    out, pos = [], 0
    while pos + 48 <= len(buf):
        try:
            m = HDR.unpack(decode6(buf[pos:pos + 48]))
        except Exception:
            break
        body_len = m[7]
        if pos + 48 + body_len > len(buf):
            break
        out.append(Frame(recog=m[2], msg_id=m[1], p1=m[3], p2=m[4], p3=m[5],
                         tail=m[8], body=buf[pos + 48:pos + 48 + body_len]))
        pos += 48 + body_len
    return out, buf[pos:]


class Wire:
    """同步 TCP 线客户端。"""

    def __init__(self, host: str, port: int, timeout: float = 10) -> None:
        self.sock = socket.create_connection((host, port), timeout=timeout)

    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
            except (socket.timeout, ConnectionResetError, OSError):
                break
        frames, _ = unpack_stream(buf)
        return frames

    def close(self) -> None:
        try:
            self.sock.close()
        except OSError:
            pass

注意:头字段序对照 cq2_pure_enter.py send() 的 struct.pack("<9i", 0, mid, recog, p1, p2, p3, 0, len, tail)——HDR.pack(0, msg_id, recog, ...) 顺序一致;unpack 侧 m[0]=零位,m[1]=msg,m[2]=recog。实现时以 tests/test_wire.py 回环 + 下面的真实样本双验证。

Step 4: 追加真实抓包样本回归(防字段序错位)

# tests/test_wire.py 追加
def test_pack_matches_reference_impl():
    """与 cq2_pure_enter.send() 的原始拼包逐字节一致(防字段序回归)。"""
    import struct, random
    sys_path_hack = __import__("sys"); sys_path_hack.path.insert(0, "protocol_bot")
    from encode6bitbuf_final import encode6
    random.seed(42)
    body = b"*#1186115037/role/1/2/3/1/0/110585/0/1/1/1/1"
    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

Step 5: 跑测试通过

Run: protocol_bot/.venv/bin/python -m pytest tests/test_wire.py -v Expected: 3 passed

Step 6: Commit

git add apilot/ tests/test_wire.py && git commit -m "feat(apilot): wire layer with verified frame codec"

Task 2: no-op 工厂(冒充 cc.* 渲染类)

Files:

Create: apilot/stubs/__init__.py(空)
Create: apilot/stubs/noop.py
Test: tests/test_noop.py

Interfaces:

Produces: def make_noop_table(name: str) -> dict — 任意属性访问/调用均安全的嵌套表;def install_noop_globals(lua) -> None — 向 Lua 全局注入 cc 骨架(Sprite/Node/Director 等 C 类的 no-op)与 ccui/ccs/ccexp 空命名空间
Consumes: Lupa LuaRuntime(传入即可调用)
[ ] Step 1: 写失败测试

# tests/test_noop.py
import lupa.luajit21 as lj

def test_noop_table_safe_access():
    from apilot.stubs.noop import make_noop_table
    t = make_noop_table("cc")
    assert t["Sprite"] is not None          # 任意键存在
    assert callable(t["Sprite"]["create"])  # 任意层级可调
    obj = t["Sprite"]["create"]()
    assert obj["getPosition"]() == (0, 0)   # 调用返回 no-op 对象(继续安全)

def test_install_noop_globals_in_lua():
    from apilot.stubs.noop import install_noop_globals
    rt = lj.LuaRuntime()
    install_noop_globals(rt)
    rt.execute("local s = cc.Sprite:create(); s:setPosition(1,2); s:addChild(cc.Node:create())")
    rt.execute("cc.Director:getInstance():getScheduler()")
    # 无异常即通过

Step 2: 跑测试确认失败

Run: protocol_bot/.venv/bin/python -m pytest tests/test_noop.py -v Expected: FAIL(ModuleNotFoundError)

Step 3: 实现 noop.py

# apilot/stubs/noop.py
"""万能 no-op 工厂: 冒充 tolua/cc 渲染类。属性=同款工厂, 调用=返回同款对象。"""
from __future__ import annotations
from typing import Any

_NOOP_RETURN_DEFAULTS: tuple[tuple[tuple, Any], ...] = (
    # (参数个数, 返回值) —— 需要非 nil 返回的接口按绑定面清单加
    (("getPosition",), (0, 0)),
)


def make_noop_table(name: str = "?", _depth: int = 0) -> Any:
    if _depth > 24:
        return None

    class _Noop:
        def __init__(self, n: str) -> None:
            self.__name = n

        def __getitem__(self, k: str) -> Any:
            return make_noop_table(f"{self.__name}.{k}", _depth + 1)

        def __getattr__(self, k: str) -> Any:
            return self[k]

        def __call__(self, *a: Any, **kw: Any) -> Any:
            return make_noop_table(f"{self.__name}()", _depth + 1)

        def __repr__(self) -> str:
            return f"<noop:{self.__name}>"

    return _Noop(name)


class _ReturnZeroPair:
    """getPosition 等双值返回的特化包装, M1 暂不启用(绑定面清单核对后启用)。"""


def install_noop_globals(lua: Any) -> None:
    """注入 cc/ccui/ccs/ccexp 骨架。真实实现(netClient 等)在装完 no-op 后由各自模块覆盖。"""
    lua.execute("""
cc = cc or {}
setmetatable(cc, {__index = function(t, k)
    local v = setmetatable({}, {
        __index = function(tt, kk)
            local fn = function(...) return setmetatable({}, {
                __index = function(t3, k3)
                    return function(...) return nil end
                end}) end
        return fn end,
        __call = function(cls, ...) return setmetatable({}, {
            __index = function(o, k)
                return function(...) return nil end end}) end})
    rawset(t, k, v)
    return v
end})
ccui = ccui or cc.Sprite  -- 同骨架
ccs = ccs or cc.Sprite
ccexp = ccexp or cc.Sprite
""")

注:_NOOP_RETURN_DEFAULTS M1 保留占位但不必使用——Lua 侧 local x, y = obj:getPosition() 收到两个 nil 在纯逻辑路径(登录链)无碍;若实测崩溃再特化。

Step 4: 跑测试通过 + Commit

Run: protocol_bot/.venv/bin/python -m pytest tests/test_noop.py -v → 2 passed

git add apilot/stubs/ tests/test_noop.py && git commit -m "feat(apilot): universal noop stub factory for tolua/cc classes"

Task 3: 计时轮 Scheduler stub

Files:

Create: apilot/stubs/scheduler.py
Test: tests/test_scheduler.py

Interfaces:

Produces:
- class Scheduler: tick(dt: float) -> None(驱动所有已注册回调);schedule(fn, interval_s: float) -> int 返回 entry id;unschedule(entry_id: int) -> None;delay(fn, delay_s: float) -> int(一次性)
- def install_scheduler(lua, sched) -> None — 提供 Lua 侧 global.Scheduler(方法面 scheduleScriptFunc/unscheduleScriptEntry)与全局 PerformWithDelayGlobal(fn, delay) / Schedule(fn, interval) / UnSchedule(id)
Consumes: Task 2 的 install_noop_globals(先装,cc.Director:getInstance():getScheduler() 返回本对象——在 install 内覆盖)
[ ] Step 1: 写失败测试

# tests/test_scheduler.py
import lupa.luajit21 as lj
from apilot.stubs.scheduler import Scheduler, install_scheduler

def test_interval_and_oneshot():
    s = Scheduler()
    hits = []
    s.schedule(lambda: hits.append("i"), 0.05)
    s.delay(lambda: hits.append("d"), 0.1)
    for _ in range(30):
        s.tick(0.05)
    assert hits.count("i") >= 2
    assert hits.count("d") == 1

def test_lua_side_globals():
    rt = lj.LuaRuntime()
    from apilot.stubs.noop import install_noop_globals
    install_noop_globals(rt)
    s = Scheduler()
    install_scheduler(rt, s)
    rt.execute("global = global or {}; global.Scheduler = SL_getScheduler()")
    # install_scheduler 需暴露一个 Lua 可达引用; 简化: 直接用注入的全局函数
    rt.execute("n = 0; id = PerformWithDelayGlobal(function() n = n + 1 end, 0)")
    for _ in range(3):
        s.tick(0.05)
    assert rt.eval("n") == 1

[ ] Step 2: 确认失败 → Run: pytest tests/test_scheduler.py -v → FAIL
[ ] Step 3: 实现 scheduler.py

# apilot/stubs/scheduler.py
"""计时轮: Python 侧驱动, Lua 侧全局函数注入。对齐游戏 50ms tick 节奏。"""
from __future__ import annotations
import itertools
from typing import Callable


class Scheduler:
    def __init__(self) -> None:
        self._ids = itertools.count(1)
        self._timers: dict[int, tuple[Callable[[], None], float, float, bool]] = {}
        # id -> (fn, interval, elapsed, repeat)

    def schedule(self, fn: Callable[[], None], interval_s: float) -> int:
        eid = next(self._ids)
        self._timers[eid] = (fn, interval_s, 0.0, True)
        return eid

    def delay(self, fn: Callable[[], None], delay_s: float) -> int:
        eid = next(self._ids)
        self._timers[eid] = (fn, delay_s, 0.0, False)
        return eid

    def unschedule(self, eid: int) -> None:
        self._timers.pop(eid, None)

    def tick(self, dt: float) -> None:
        for eid in list(self._timers):
            fn, interval, elapsed, repeat = self._timers[eid]
            elapsed += dt
            if elapsed >= interval:
                if repeat:
                    self._timers[eid] = (fn, interval, elapsed - interval, True)
                else:
                    self._timers.pop(eid)
                fn()


def install_scheduler(lua, sched: Scheduler) -> None:
    py_sched = sched  # Lupa: Python 对象直接可见

    def lua_schedule(fn, interval: float) -> int:
        return py_sched.delay(fn, float(interval))  # M1: 游戏里 Schedule 多为一次性轮询,统一 delay;需循环时改 schedule

    def lua_perform_delay(fn, delay_s: float) -> int:
        return py_sched.delay(fn, float(delay_s))

    def lua_unschedule(eid: int) -> None:
        py_sched.unschedule(int(eid))

    lua.globals()["PerformWithDelayGlobal"] = lua_perform_delay
    lua.globals()["Schedule"] = lua_schedule
    lua.globals()["UnSchedule"] = lua_unschedule
    lua.globals()["SL_getScheduler"] = lambda: py_sched

实现注意:Schedule() 在游戏 networkUtil/util 里既用于一次性也用于循环——M1 先统一按 delay,跑 T01 冒烟时若发现循环语义缺失,改 lua_schedule 加第三参判断(nil=循环),并在本任务测试补一个循环用例。

Step 4: 通过 + Commit

Run: pytest tests/test_scheduler.py -v → 2 passed

git add apilot/stubs/scheduler.py tests/test_scheduler.py && git commit -m "feat(apilot): python-driven scheduler wheel with lua globals"

Task 4: netClient + networkCtl stub(网络真实实现)

Files:

Create: apilot/stubs/netclient.py
Create: apilot/stubs/networkctl.py
Test: tests/test_network_stubs.py(环回 socket 模拟网关)

Interfaces:

Consumes: Task 1 Wire/Frame
Produces:
- class NetClientStub: connect(host, port); disconnect(); send_raw(bytes); tick(deadline_s)->list[Frame]; set_sdata(v)/get_sdata()->int; is_connected()->bool — Python 侧对象,持有 Wire
- class NetworkCtlStub: send_msg(msg_id, recog, p1, p2, p3, data_str, data_len) -> None(组 Frame→netclient 发送);register_lua_handler(msg_id, cb)(cb 是 Lua 函数);unregister_all()
- def make_msg_object(lua, frame: Frame) — 构造 Lua 侧 msg stub table:GetMsgId()/GetHeader()→{recog,param1,param2,param3,msgId}/GetDataLength()/GetData():ReadString(len)
- def dispatch_frame(lua, ctl: NetworkCtlStub, frame: Frame) -> None — 查 handler→构造 msg→调用 Lua cb
[ ] Step 1: 写失败测试(本地环回)

# tests/test_network_stubs.py
import socket, threading, struct, sys
import lupa.luajit21 as lj
sys.path.insert(0, "protocol_bot")
from encode6bitbuf_final import encode6
from apilot.wire import Frame
from apilot.stubs.netclient import NetClientStub
from apilot.stubs.networkctl import NetworkCtlStub, dispatch_frame

def _fake_gateway(server_sock):
    """收一帧, 回一帧 529。"""
    conn, _ = server_sock.accept()
    data = conn.recv(65536)
    # 解析 client 帧(跳过编解码,直接回一个 529)
    body = encode6(b"key32/session123")
    head = struct.pack("<9i", 1, 529, 0, 0, 0, 0, 0, len(body), 1)
    conn.sendall(encode6(head) + body)
    conn.close()

def test_sendmsg_roundtrip_and_dispatch():
    srv = socket.socket(); srv.bind(("127.0.0.1", 0)); srv.listen(1)
    port = srv.getsockname()[1]
    threading.Thread(target=_fake_gateway, args=(srv,), daemon=True).start()

    rt = lj.LuaRuntime()
    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)

    # 注册 Lua handler 并分发
    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"))
    dispatch_frame(rt, ctl, [f for f in frames if f.msg_id == 529][0])
    assert rt.eval("got") == "529:key3"

[ ] Step 2: 确认失败 → FAIL
[ ] Step 3: 实现 netclient.py / networkctl.py

# apilot/stubs/netclient.py
"""netClient C++ 单例的 stub: 直接持有 Wire。"""
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:
            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, "netClient not connected"
        self.wire.sock.sendall(data)

    def send_frame(self, f: Frame) -> None:
        assert self.wire
        self.wire.send(f)

    def tick(self, deadline_s: float) -> list[Frame]:
        if not self.wire:
            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
"""LuaNetworkController stub + msg 对象构造 + 分发。"""
from __future__ import annotations
from typing import Any, Callable
import lupa.luajit21 as lj  # type: ignore
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] = {}  # msg_id -> lua function

    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:
        body = b"" if data_str is None else (
            data_str.encode("utf-8") if isinstance(data_str, str) else 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) -> None:
        g = lua.globals()
        g["AP_networkCtl"] = 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}
""")
        # msg 构造器存进 lua 全局
        g["AP_mkmsg"] = lua.execute("return " + MSG_LUA) if False else lua.eval(MSG_LUA)


def dispatch_frame(lua: Any, ctl: NetworkCtlStub, f: Frame) -> None:
    cb = ctl.handlers.get(f.msg_id)
    if cb is None:
        return
    mkmsg = lua.globals()["AP_mkmsg"]
    body = f.body.decode("latin-1")  # Lua 字符串按字节透传
    msg = mkmsg(f.recog, f.msg_id, f.p1, f.p2, f.p3, body)
    cb(msg)

注意两点(实现时核对):① game networkUtil.lua 的 LuaRegisterMsgHandler 调的是 global.networkCtl:RegisterLuaHandler(id, fn)(冒号调用→第一参 self),install_to_lua 的闭包签名 function(_, ...) 已兼容;② lua.eval(MSG_LUA) 返回闭包 mkmsg——若 Lupa 对 return function... 求值有出入,改用 lua.execute("AP_mkmsg = " + MSG_LUA) 落全局后在 dispatch 里取。

Step 4: 通过 + Commit

Run: pytest tests/test_network_stubs.py -v → 1 passed

git add apilot/stubs/netclient.py apilot/stubs/networkctl.py tests/test_network_stubs.py && git commit -m "feat(apilot): real network stubs - netclient/networkctl with lua msg dispatch"

Task 5: 环境注入 stub(L_GameEnvManager/FileUtils/LuaBridgeCtl/gameEnviroment)+ 装配

Files:

Create: apilot/stubs/env.py
Modify: apilot/stubs/__init__.py
Test: tests/test_env_stub.py

Interfaces:

Consumes: Task 2/3/4 的安装函数
Produces:
- class EnvData: dict 载体(uid/token/new_boxtoken/signkey/channel/gameid/isBoxLogin=1/serverId/macAddress...)
- def install_env(lua, env: EnvData, state_dir: str) -> None — 注入 L_GameEnvManager(GetEnvDataByKey)、LuaBridgeCtl(switches dict,心跳开关默认 1)、FileUtils(限定 state_dir 根)、gameEnviroment(GetNetClient 等 5 方法面)、global.L_GameEnvManager 等全局引用、release_print/releasePrint/GET_STRING/JUMPTO/GET_SETTING/CHECK_SETTING/CHANGE_SETTING/Encode6BitBuf(直通 encode6) 等 C++ 全局函数
- def install_all(lua, nc, ctl, sched, env, state_dir) -> None — 装配顺序:noop → scheduler → env → networkctl
[ ] Step 1: 写失败测试

# tests/test_env_stub.py
import lupa.luajit21 as lj
from apilot.stubs.env import EnvData, install_env

def test_env_manager_keys(tmp_path):
    rt = lj.LuaRuntime()
    env = EnvData({"uid": "1186115037", "token": "jwt.x", "isBoxLogin": 1, "signkey": "k"*32})
    install_env(rt, env, str(tmp_path))
    assert rt.eval("L_GameEnvManager:GetEnvDataByKey('uid')") == "1186115037"
    assert rt.eval("L_GameEnvManager:GetEnvDataByKey('isBoxLogin')") == 1
    assert rt.eval("L_GameEnvManager:GetEnvDataByKey('missing')") is None
    rt.execute("global = global or {}; global.L_GameEnvManager = L_GameEnvManager")
    assert rt.eval("global.L_GameEnvManager:GetEnvDataByKey('token')") == "jwt.x"

def test_bridge_ctl_and_fileutils(tmp_path):
    rt = lj.LuaRuntime()
    install_env(rt, EnvData({}), str(tmp_path))
    rt.execute("LuaBridgeCtl:Inst():SetModulesSwitch(2, 1)")
    assert rt.eval("LuaBridgeCtl:Inst():GetModulesSwitch(2)") == 1
    rt.execute("cc.FileUtils:getInstance():writeStringToFile('hi', 'a.txt')")
    assert (tmp_path / "a.txt").read_text() == "hi"
    assert rt.eval("GET_STRING(99999)") == ""   # 未命中返回空串

[ ] Step 2: 确认失败 → FAIL
[ ] Step 3: 实现 env.py(核心骨架)

# apilot/stubs/env.py
"""环境注入: L_GameEnvManager / FileUtils / LuaBridgeCtl / gameEnviroment / 杂项全局函数。"""
from __future__ import annotations
import os
from typing import Any

SWITCHES_DEFAULT = {2: 1}  # Modules_Index_Enable_Check_Heart_Beat=1(心跳开)


class EnvData:
    def __init__(self, data: dict[str, Any]) -> None:
        self.data = dict(data)
        self.switches: dict[int, int] = dict(SWITCHES_DEFAULT)
        self.settings: dict[int, list] = {}


def install_env(lua: Any, env: EnvData, state_dir: str) -> None:
    os.makedirs(state_dir, exist_ok=True)
    g = lua.globals()
    g["_PY_ENV"] = env
    g["_PY_STATE_DIR"] = state_dir

    lua.execute(r"""
-- L_GameEnvManager
L_GameEnvManager = {}
function L_GameEnvManager.GetEnvDataByKey(k) return _PY_ENV.data[k] end

-- LuaBridgeCtl
LuaBridgeCtl = {}
function LuaBridgeCtl.Inst() return LuaBridgeCtl end
function LuaBridgeCtl.GetModulesSwitch(i) return _PY_ENV.switches[i] or 0 end
function LuaBridgeCtl.SetModulesSwitch(i, v) _PY_ENV.switches[i] = v end

-- FileUtils(python filesystem via bridge)
_PY_FS = python.eval("__import__('apilot.stubs.env', fromlist=['x'])")

-- 杂项全局(绑定面清单 C 类)
function release_print(...) end
releasePrint = release_print
function GET_STRING(id) return "" end
function JUMPTO(...) end
function GET_SETTING(i) return _PY_ENV.settings[i] end
function CHECK_SETTING(i) local s = _PY_ENV.settings[i] return s and s[1] == 1 or false end
function CHANGE_SETTING(i, v) _PY_ENV.settings[i] = v end
""")
    # FileUtils 用 Python 对象直挂(路径安全: join 后强制以 state_dir 开头)
    class _FU:
        @staticmethod
        def getInstance():  # noqa: N802
            return _FU

        @staticmethod
        def getWritablePath() -> str:
            return state_dir + os.sep

        @staticmethod
        def writeStringToFile(s: str, rel: str) -> None:
            p = os.path.join(state_dir, rel)
            if os.path.commonpath([os.path.abspath(p), state_dir]) != state_dir:
                raise ValueError("path escape")
            with open(p, "w", encoding="utf-8") as f:
                f.write(s)

        @staticmethod
        def isFileExist(rel: str) -> bool:  # noqa: N802
            return os.path.exists(os.path.join(state_dir, rel))

    g["AP_FileUtils"] = _FU
    lua.execute("cc.FileUtils = {getInstance = function() return AP_FileUtils end}")
    lua.execute("""
gameEnviroment = gameEnviroment or {}
function gameEnviroment.GetNetClient() return nil end
function gameEnviroment.SetLoginAccount(a) end
function gameEnviroment.GetDownloadRootURL() return "" end
""")

实现注意:①python.eval/_PY_FS 行是残留草稿,实现时删除,FileUtils 走 AP_FileUtils 直挂;②Encode6BitBuf 若绑定面清单确认 Lua 侧直接调用,则挂 g["AP_encode6"](从 encode6bitbuf_final 包装 str↔bytes)并 lua.execute("Encode6BitBuf = function(s) return AP_encode6(s) end")——以 Task 7 冒烟实测为准。

Step 4: install_all 装配 + 测试

# apilot/stubs/__init__.py
from __future__ import annotations
from typing import Any
from apilot.stubs.noop import install_noop_globals
from apilot.stubs.scheduler import Scheduler, install_scheduler
from apilot.stubs.env import EnvData, install_env
from apilot.stubs.netclient import NetClientStub
from apilot.stubs.networkctl import NetworkCtlStub, dispatch_frame


def install_all(lua: Any, nc: NetClientStub, ctl: NetworkCtlStub,
                sched: Scheduler, env: EnvData, state_dir: str) -> None:
    install_noop_globals(lua)      # 1. cc.* 骨架
    install_scheduler(lua, sched)  # 2. 覆盖 Director 相关
    install_env(lua, env, state_dir)  # 3. 环境+FileUtils(覆盖 cc.FileUtils)
    ctl.install_to_lua(lua)        # 4. networkCtl/netClient(最后,真实实现覆盖 no-op)
    lua.execute("global = global or {}")
    lua.execute("global.networkCtl = networkCtl; global.LuaBridgeCtl = LuaBridgeCtl")
    lua.execute("global.L_GameEnvManager = L_GameEnvManager; global.gameEnviroment = gameEnviroment")

test 追加:install_all 后 Lua 里 global.networkCtl:SendMsg(...) 到环回网关可通(复用 Task 4 的 fake gateway)。

Run: pytest tests/test_env_stub.py -v → 2 passed

Step 5: Commit

git add apilot/stubs/ && git commit -m "feat(apilot): env injection stubs and install_all assembly"

Task 6: bridge.py — 游戏原版 Lua 加载链(Lupa 内跑通 netMessageDef→networkUtil)

Files:

Create: apilot/bridge.py
Test: tests/test_bridge_load.py

Interfaces:

Consumes: install_all;游戏 Lua 只读目录 lua脚本解密源码/mod_fgcq/stab/scripts/
Produces:
- LUA_SRC = "lua脚本解密源码/mod_fgcq/stab/scripts"(常量)
- def bootstrap_runtime(nc, ctl, sched, env, state_dir) -> LuaRuntime — 建 VM+装 stub+设 package.path(指向游戏 scripts 目录)+注入 LuaRegisterMsgHandler/LuaSendMsg/SendTableToServer(包装 networkCtl,语义对齐 networkUtil.lua 但不加载该文件,因其可能引用更多 C++ 面;此处按其签名以纯 Lua 重声明——这两个函数体极薄,复刻等价) + 加载 netMessageDef.lua
- def load_game_module(lua, relpath: str) -> None — 按相对模块名 require(如 framework.init)
[ ] Step 1: 写失败测试(冒烟)

# tests/test_bridge_load.py
import lupa.luajit21 as lj
from apilot.bridge import bootstrap_runtime, LUA_SRC
from apilot.stubs.netclient import NetClientStub
from apilot.stubs.networkctl import NetworkCtlStub
from apilot.stubs.scheduler import Scheduler
from apilot.stubs.env import EnvData

def _boot(tmp_path):
    nc = NetClientStub(); ctl = NetworkCtlStub(nc); sched = Scheduler()
    return bootstrap_runtime(nc, ctl, sched, EnvData({"uid": "u1"}), str(tmp_path))

def test_netmessagedef_loaded(tmp_path):
    lua = _boot(tmp_path)
    assert lua.eval("MsgType.MSG_CS_LOGIN_SERVER") == 20
    assert lua.eval("MsgType.MSG_CS_HEART_BEAT_KEEP") == 60000

def test_sendtabletoserver_encodes_json(tmp_path):
    lua = _boot(tmp_path)
    sent = []
    nc_stub_capture = lua.globals()["AP_networkCtl"]
    orig = nc_stub_capture.send_msg
    nc_stub_capture.send_msg = lambda mid, recog, p1, p2, p3, body, ln: sent.append((mid, body))
    lua.execute("SendTableToServer(3199, {index=303, act='@chiyao_kaiqi'})")
    assert sent[0][0] == 3199
    assert b"chiyao" in sent[0][1]

def test_load_framework(tmp_path):
    lua = _boot(tmp_path)
    load_game_module(lua, "framework.init")
    assert lua.eval("type(Facade)") == "table" or lua.eval("type(framework)") == "table"

[ ] Step 2: 确认失败 → FAIL
[ ] Step 3: 实现 bridge.py

# apilot/bridge.py
"""Lupa 桥: VM 构建 + stub 装配 + 游戏 Lua 只读加载。"""
from __future__ import annotations
import os
from typing import Any
import lupa.luajit21 as lj  # type: ignore
from apilot.stubs import install_all
from apilot.stubs.netclient import NetClientStub
from apilot.stubs.networkctl import NetworkCtlStub
from apilot.stubs.scheduler import Scheduler
from apilot.stubs.env import EnvData

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
LUA_SRC = os.path.join(ROOT, "lua脚本解密源码", "mod_fgcq", "stab", "scripts")


def bootstrap_runtime(nc: NetClientStub, ctl: NetworkCtlStub, sched: Scheduler,
                      env: EnvData, state_dir: str) -> Any:
    lua = lj.LuaRuntime(unpack_returned_tuples=True)
    install_all(lua, nc, ctl, sched, env, state_dir)
    # package.path 指向游戏 scripts(只读)
    lua.globals()["_LUA_SRC"] = LUA_SRC
    lua.execute("package.path = _LUA_SRC .. '/?.lua;' .. package.path")

    # 薄复刻 networkUtil.lua 的公共面(签名对齐, 实现走 stub)
    lua.execute(r"""
cjson = cjson or {encode = function(t) return _PY_cjson_encode(t) end,
                  decode = function(s) return _PY_cjson_decode(s) end,
                  safe = nil}
slen = slen or function(s) return #s end

function LuaSendMsg(msgId, recog, p1, p2, p3, dataStr, dataLen)
    networkCtl:SendMsg(msgId, recog, p1, p2, p3, dataStr, dataLen)
end
function LuaRegisterMsgHandler(msgId, fn)
    networkCtl:RegisterLuaHandler(msgId, fn)
end
function SendTableToServer(msgId, t)
    local s = cjson.encode(t)
    LuaSendMsg(msgId, 0, 0, 0, 0, s, #s)
end
""")
    g = lua.globals()
    import json
    def _enc(t): return json.dumps(t, ensure_ascii=False, separators=(",", ":"))
    def _dec(s):
        d = json.loads(s)
        return d if isinstance(d, dict) else {}
    g["_PY_cjson_encode"] = _enc   # dict→json str; Lupa 会把 Lua table 转 python dict
    g["_PY_cjson_decode"] = _dec

    # 加载协议定义(第一个游戏原文件)
    src = open(os.path.join(LUA_SRC, "network", "netMessageDef.lua"), encoding="utf-8").read()
    lua.execute(src)
    return lua


def load_game_module(lua: Any, modname: str) -> None:
    lua.execute(f"require('{modname}')")

关键点:cjson 用 Python json 桥接(Lupa table↔dict 自动转换);若冒烟中发现游戏某模块直接 require("cjson") 的 C 模块行为差异(如 cjson.safe),在本任务内补 safe={encode=function(t) local ok,r = pcall(...) 风格包装。

Step 4: 通过 + Commit

Run: pytest tests/test_bridge_load.py -v → 3 passed

git add apilot/bridge.py tests/test_bridge_load.py && git commit -m "feat(apilot): lupa bridge bootstraps game lua protocol layer"

Task 7: host.py — 登录链驱动(HTTP+TCP+Lua LoginProxy 直驱)

Files:

Create: apilot/host.py
Test: tests/test_host_unit.py(单测:登录链拼包参数,无网络)

Interfaces:

Consumes: Task 1-6 全部;protocol_bot/login_996pc.py 的 do_login;cq2_pure_enter.py 的 box_auth_header/enter_game
Produces:
- class Host: __init__(zone_id, user, password)
- .http_login() -> dict(JWT+uid)
- .connect_login_gate(port=8041) -> None(建立 Wire,按 cq2_pure_enter.login_gate 已验证序列:2001→529→104→530→100→520,登录帧由 Lua 侧 SendTableToServer/LuaSendMsg 发出——2001 的 uid/encode6(jwt)/encode6(channel) 拼串在 host 侧构造为 body 传入 Lua 发送,保持帧头构造同源)
- .run_create_role(name: str | None = None) -> dict(空角色列表→107/539→101→521;有角色→复用)
- .tick(dt=0.05) -> None(nc.tick+Sched.tick 循环体)
- def run_zone(zone_id, user, password, name=None) -> dict — 一站式(HTTP→网关→创角→返回票据),CLI 用
[ ] Step 1: 写失败单测(拼包参数,环回)

# tests/test_host_unit.py
import json
from apilot.host import Host

def test_login_2001_body_shape(tmp_path, monkeypatch):
    h = Host(14481607, "longwu988", "Longwu1988", state_dir=str(tmp_path))
    uid, jwt = "1186115037", "eyJfake.jwt.token"
    body = h.build_2001_body(uid, jwt)   # b"uid/" + encode6(jwt) + b"/" + encode6(channel) + b"/1/0"
    import sys; sys.path.insert(0, "protocol_bot")
    from encode6bitbuf_final import encode6
    assert body == f"{uid}/".encode() + encode6(jwt.encode()) + b"/" + encode6(b"110585") + b"/1/0"

def test_create_role_101_body(tmp_path):
    h = Host(14481607, "u", "p", state_dir=str(tmp_path))
    body = json.loads(h.build_101_body("萌新1234", uid="u1", job=0, sex=0))
    assert body["chr"] == "萌新1234" and body["sdkid"] == "110585"
    assert body["server_name"] == 14481607

[ ] Step 2: 确认失败 → FAIL
[ ] Step 3: 实现 host.py(骨架)

# apilot/host.py
"""宿主: HTTP 登录 + 网关连接 + Lua 驱动创角。复用 protocol_bot 已验证资产。"""
from __future__ import annotations
import json
import os
import random
import sys
from typing import Any

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(ROOT, "protocol_bot"))

import requests as rq  # noqa: E402
import login_996pc as L  # type: ignore  # noqa: E402
import cq2_pure_enter as P  # type: ignore  # noqa: E402
from encode6bitbuf_final import encode6, decode6  # type: ignore  # noqa: E402

from apilot.bridge import bootstrap_runtime
from apilot.stubs.netclient import NetClientStub
from apilot.stubs.networkctl import NetworkCtlStub, dispatch_frame
from apilot.stubs.scheduler import Scheduler
from apilot.stubs.env import EnvData

GW_HOST = "s{zone}-jycq-206923.dhsf.xqhuyu.com"
CHANNEL = "110585"


class Host:
    def __init__(self, zone_id: int, user: str, password: str,
                 state_dir: str = ".apilot_state") -> None:
        self.zone = zone_id
        self.user, self.password = user, password
        self.state_dir = state_dir
        self.nc = NetClientStub()
        self.ctl = NetworkCtlStub(self.nc)
        self.sched = Scheduler()
        self.env = EnvData({})
        self.lua: Any = None
        self.jwt = ""
        self.uid = ""

    def build_2001_body(self, uid: str, jwt: str) -> bytes:
        return (f"{uid}/".encode() + encode6(jwt.encode()) + b"/" +
                encode6(CHANNEL.encode()) + b"/1/0")

    def build_101_body(self, name: str, uid: str, job: int = 0, sex: int = 0) -> bytes:
        return json.dumps({"server_name": self.zone, "chr": name, "secsdk": "0",
                           "job": job, "sex": sex, "account": uid,
                           "sdkid": CHANNEL}, ensure_ascii=False).encode()

    def http_login(self) -> dict:
        s = rq.Session()
        j, _ = L.do_login(s, self.user, self.password)
        if j.get("code") != 200:
            raise SystemExit(f"[!] 登录失败: {j.get('msg')}")
        self.jwt = j["data"]["token"]
        self.uid = str((j["data"].get("userInfo") or {}).get("id") or "")
        self.env.data.update({"uid": self.uid, "token": self.jwt,
                              "new_boxtoken": self.jwt, "isBoxLogin": 1})
        return j

    def boot_lua(self) -> None:
        self.lua = bootstrap_runtime(self.nc, self.ctl, self.sched, self.env, self.state_dir)

    def connect_login_gate(self, port: int = 8041) -> None:
        self.nc.connect(GW_HOST.format(zone=self.zone), port)

    def pump(self, seconds: float, on_frame=None) -> list:
        """收帧→分发→调度, 驱动 seconds 墙钟。"""
        import time as _t
        end = _t.time() + seconds
        got: list = []
        while _t.time() < end:
            for f in self.nc.tick(0.2):
                dispatch_frame(self.lua, self.ctl, f)
                got.append(f)
                if on_frame:
                    on_frame(f)
            self.sched.tick(0.05)
        return got

run_create_role 与 run_zone 的完整逻辑在本任务 Step 3 内继续实现:登录链帧序严格对照 cq2_pure_enter.login_gate()(已验证):2001(tail=0,recog=random)→等 529 取 key32→104(body=encode6(key32),tail 自增)→等 530 取 sid(recog 位)→netClient:set_sdata(sid)→100(uid/sid/zone/channel/gameid/key32)→等 520 角色 JSON→空则 101→521/522。收包等待用 pump() 循环+消息号断言,超时 30s。

Step 4: 单测通过 + Commit

Run: pytest tests/test_host_unit.py -v → 2 passed

git add apilot/host.py tests/test_host_unit.py && git commit -m "feat(apilot): host with verified login-chain drivers"

Task 8: [live] T01 端到端 — 无客户端创角(逐鹿05区)

Files:

Create: apilot/main.py(CLI)
Test: tests/test_live_t01.py

Interfaces:

Consumes: Host.run_zone
Produces: CLI protocol_bot/.venv/bin/python -m apilot.main --zone 14481607 --name 萌新XXXX
[ ] Step 1: 写 live 测试(默认跳过)

# tests/test_live_t01.py
import pytest
from apilot.host import Host

pytestmark = pytest.mark.live

def test_t01_create_role_headless():
    import random
    name = "萌新" + str(random.randint(1000, 9999))
    h = Host(14481607, "longwu988", "Longwu1988", state_dir="/tmp/apilot_t01")
    h.http_login()
    h.boot_lua()
    result = h.run_create_role(name=name)   # 内部: connect→2001链→101→521
    assert result["role_name"] == name
    assert result["ticket"]                 # 521 票据非空
    assert h.nc.get_sdata() != 0            # 530 sid 已注入

Step 2: 跑通(真实网关)

Run: cd /Users/zlwu/work/00code/传奇 && protocol_bot/.venv/bin/python -m pytest tests/test_live_t01.py -m live -v -s Expected: PASS,输出含 521 票据。失败处理路径:①帧序错→对照 cq2_pure_enter.py 逐帧 diff;②Lua 崩(nil 索引)→按崩溃栈定位绑定面缺项→在对应 stub 补方法(这是预期内的收敛循环);③角色名重复(522)→run_create_role 内重试换名(≤5 次)。

Step 3: CLI 收尾

# apilot/main.py
"""APilot CLI。M1: 无客户端创角。"""
import argparse
from apilot.host import run_zone

def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--zone", type=int, required=True)
    ap.add_argument("-u", "--user", default="longwu988")
    ap.add_argument("-p", "--password", default="Longwu1988")
    ap.add_argument("--name", default=None)
    a = ap.parse_args()
    r = run_zone(a.zone, a.user, a.password, a.name)
    print(f"[M1完成] 区={a.zone} 角色={r['role_name']} 票据={r['ticket'][:24]}...")

if __name__ == "__main__":
    main()

Run: protocol_bot/.venv/bin/python -m apilot.main --zone 14481607 → 打印创角成功

Step 4: Commit

git add apilot/main.py tests/test_live_t01.py && git commit -m "feat(apilot): M1 e2e - headless create-role on live gateway (zone 14481607)"

验证清单(M1=设计文档 V1+V2)

V1 stub 骨架:Task 1-6(单测全绿,live 测试除外)
V2 登录链:Task 8 live 通过(521 票据+sdata 注入+角色列表)

Self-Review 记录

Spec 覆盖:M1 范围=设计 §8 V1/V2+§10 M1;T01 创角协议依据=设计 §5 表 T01 行;绑定面四类分别落在 Task 2(no-op)/3(驱动)/4(网络)/5(环境)✓
占位符:Task 5 _PY_FS 草稿行已标注"实现时删除";cjson.safe 留了明确的补法说明;无 TBD ✓
类型一致:dispatch_frame(lua, ctl, f) 在 Task 4 定义、Task 7 pump 使用一致;install_all 签名 Task 5 定义与 Task 6 使用一致;Host.boot_lua 内 bootstrap_runtime(nc, ctl, sched, env, state_dir) 与 bridge 签名一致 ✓
已知不确定点(实现时按证据收敛,不算占位):网络帧头 9 字段中 m[0]/m[6] 的语义(零位)、Schedule 一次性/循环语义、cjson.safe 需求——三处都在对应任务里写了核对方法与改法。

2026-09-21-apilot-headless-m1.md 44 KB Постоянная ссылка История Исходник

APilot Headless M1 实施计划(stub 骨架 + 登录链 + T01 无客户端创角)

Global Constraints

文件结构总览(M1 全量)

Task 1: wire.py 协议线层(纯 Python,无 Lua)

Task 2: no-op 工厂(冒充 cc.* 渲染类)

Task 3: 计时轮 Scheduler stub

Task 4: netClient + networkCtl stub(网络真实实现)

Task 5: 环境注入 stub(L_GameEnvManager/FileUtils/LuaBridgeCtl/gameEnviroment)+ 装配

Task 6: bridge.py — 游戏原版 Lua 加载链(Lupa 内跑通 netMessageDef→networkUtil)

Task 7: host.py — 登录链驱动(HTTP+TCP+Lua LoginProxy 直驱)

Task 8: [live] T01 端到端 — 无客户端创角(逐鹿05区)

验证清单(M1=设计文档 V1+V2)

Self-Review 记录

2026-09-21-apilot-headless-m1.md 44 KB

Постоянная ссылка История Исходник