AIP 协议:设备互联的 WebSocket 通信与容错层
30 秒导读: 第 05 章 的 Galaxy 把一个请求拆成了跨设备的 DAG 星座——但"编排器"和"某台 Windows 机器上的 agent"之间,隔着一整个网络。AIP(Agent Interaction Protocol)就是填这条缝的通信底座: 一条持久的 WebSocket、一套强类型消息、 一组连接生命周期(注册→心跳→派单→回报),外加自动重连与超时兜底。本章只讲"消息怎么在 线上流动、连接怎么活着",不讲"派什么任务"(那是 05)、也不讲"任务在设备上怎么执行"(那是 01–04)。
1. 这是什么(零基础也能懂)
一句话定义: AIP 是 UFO³ 里"设备之间说话的语言 + 电话线",定义了编排器和设备 agent 之间发什么消息、按什么顺序、断线了怎么办。
它解决什么问题。 想象你是 Galaxy 编排器,手上一个任务要"让笔记本查日历,再让台式机改 PPT"。这两台机器不在你进程里,它们是网络另一头的独立 agent。你需要一条长期在线的通道:
- 能双向说话——你派活给它,它把结果回给你,中途还要能推进度。
- 能活很久——一个任务可能跑几分钟,期间连接不能随便断;断 了要能自己接回来。
- 两边说的是同一套话——消息字段、类型、状态码都对得上,不能一边发 JSON 一边猜。
AIP 就是把这三件事标准化下来的那一层。
一句话直觉/类比。 把 AIP 想成设备版的 HTTP + 心跳起搏器:HTTP 定义"请求/响应长什么 样",AIP 定义"注册/派单/回报长什么样";但 HTTP 是一问一答就挂,AIP 是一条**始终在线、会自己 喘气(心跳)、断了会自己拨回来(重连)**的长连接。
用起来什么样。 三种角色各起一个"端点",几行就连上(源自 aip/__init__.py:27-41 的用法示例):
# 示意,非源码:三种端点的最小用法
# ① 设备侧的服务器(承接连接、广播任务)
endpoint = DeviceServerEndpoint(ws_manager, session_manager)
await endpoint.handle_websocket(websocket)
# ② 设备侧的 agent 客户端(连上自己的服务器、登记自己)
endpoint = DeviceClientEndpoint(ws_url, ufo_client)
await endpoint.start()
# ③ 编排器侧的星座客户端(连到目标设备的服务器)
endpoint = ConstellationEndpoint(task_name, message_processor)
await endpoint.connect_to_device(device_info, message_processor)
本节不碰底层。你只要记住:AIP = 设备互联的持久 WebSocket 通道 + 一套约定好的消息与生命周期。
2. 顶层全景(它大概怎么转)
AIP 是一个分五层的栈,每层只干一件事,上层调下层。下图从上到下就是 aip/ 目录的结构,
也是一条消息"从对象到网线"要穿过的层。
怎么读这张图: 从上往下是"越来越贴近网线";虚线框是抽象基类,实线是具体实现。
┌──────────────────────────────────────────────────────────────┐
│ endpoints/ 端点:三种角色,把下面所有层组装成"能用的连接" │
│ DeviceServerEndpoint · DeviceClientEndpoint · Constellation… │
└───────────────┬───────────────────────────┬──────────────────┘
│ 用 │ 用
┌───────▼────────┐ ┌────────▼─────────┐
│ resilience/ │ │ protocol/ │
│ 容错三件套 │ │ 协议:消息语义 │
│ · 重连(退避) │ │ Registration/ │
│ · 心跳保活 │ │ Heartbeat/Command/│
│ · 超时 │ │ TaskExecution/ │
└────────────────┘ │ DeviceInfo │
└────────┬──────────┘
│ 继承
┌────────▼──────────┐
│ protocol/base.py │
│ AIPProtocol │
│ 收/发·序列化·中间件 │
└────────┬──────────┘
│ 用
┌────────▼──────────┐
│ transport/ │
│ WebSocketTransport │
│ + Adapter(收发帧) │
└────────┬──────────┘
┌─────────────────────────┼─────────────────────┐
│ 贯穿所有层的数据契约:messages.py(强类型消息) │
│ extensions/:横切扩展(日志/指标),可选旁挂 │
└───────────────────────────────────────────────┘
各层一句话职责:
| 层 | 干什么 | 关键文件 |
|---|---|---|
| messages | 定义线上所有消息的强类型结构(Pydantic) | aip/messages.py |
| protocol | 给消息赋予语义:注册/心跳/派单/回报各一套方法 | aip/protocol/* |
| transport | 把字节收发到 WebSocket 帧上,屏蔽库差异 | aip/transport/* |
| endpoints | 把上面全部组装成三种角色的可用连接 | aip/endpoints/* |
| resilience | 断线自动重连、心跳保活、操作超时 | aip/resilience/* |
| extensions | 旁挂横切逻辑(日志、指标),不改主链路 | aip/extensions/* |
主线走一遍(高层,不进代码): 一个 Pydantic 消息对象 → AIPProtocol.send_message 过一遍
中间件、序列化成 UTF-8 JSON → WebSocketTransport.send 经 Adapter 写成 WebSocket 文本帧 → 网线
→ 对端 Adapter 收帧 → receive_message 反序列化回强类型对象 → 分发给处理器。容错层在旁边盯着:
心跳定时喘气,断了就按退避重连。
3. 谁连谁:星型拓扑与角色(先搞清方向)
这是全章最容易搞混的一点,先讲清连接方向,后面的时序才看得懂。
关键事实:每台设备自己跑一个 UFO 服务器(WebSocket 服务端),它是这台设备的"总机"。 两类客户端都连到这个总机,总机在两者之间转发消息:
- 设备自己的 agent 作为
DEVICE客户端,连到本机总机、登记自己 (ufo/client/websocket.py:192register_as_device)。 - Galaxy 编排器 作为
CONSTELLATION客户端,连到目标设备的总机 (galaxy/client/components/connection_manager.py:92transport.connect(device_info.server_url)), 注册时用target_id声明"我要控哪台设备"(aip/protocol/registration.py:99register_as_constellation)。
怎么读这张图: 中间是设备总机(server);左右两个都是客户端,箭头是"谁主动连谁"; 业务消息由总机居中转发。
编排器侧(Galaxy) 设备侧(一台 Windows 机)
┌────────────────────┐ ┌───────────────────────────────┐
│ Constellation │ │ UFO 设备总机(server) │
│ Endpoint │ ws │ DeviceServerEndpoint │
│ (CONSTELLATION 客户端)├──────►│ ┌─────────────────────────┐ │
│ target_id=该设备 │ │ │ 转发:requester↔target │ │
└────────────────────┘ │ └─────────────────────────┘ │
│ ▲ │
│ ws │ (DEVICE 客户端) │
│ ┌──────┴───────┐ │
│ │ Device Agent │ │
│ │ (UFO² 主循环) │ │
│ └──────────────┘ │
└───────────────────────────────┘
服务端确实是"转发者"而非"终点":它收到编排器的 TASK 后,用
client_manager.get_task_protocol(...) 找到目标设备那条连接的协议对象再转发过去
(ufo/server/ws/handler.py:874)。注册时服务端还会把连接的身份钉死
(registered_client_id / registered_client_type),防止一条注册为 DEVICE 的连接后面冒充
CONSTELLATION(ufo/server/ws/handler.py:178-179)。
说明:本章聚焦
aip/这套协议库本身;真正的生产收发循环在ufo/server/ws/handler.py(服务端)和ufo/client/websocket.py(设备端),它们直接复用aip/里的协议类。ConstellationEndpoint 则包了 Galaxy 的WebSocketConnectionManager(aip/endpoints/constellation_endpoint.py:62)。所以aip/提供"词汇表 + 传输 + 容错", 三个宿主把它拼成实际链路。
4. 消息模型(线上说什么话)
这节讲数据契约:两个方向各一个消息类,加一组枚举把"谁、干什么、什么状态"编码清楚。
4.1 两个方向,两个消息类
所有消息就两大类,方向写死在类名里:
| 类 | 方向 | 核心字段 | 定义 |
|---|---|---|---|
ClientMessage | 客户端 → 服务端 | type client_type client_id target_id request action_results | aip/messages.py:343 |
ServerMessage | 服务端 → 客户端 | type status actions(命令列表) user_request result response_id | aip/messages.py:299 |
两类都是 Pydantic BaseModel,字段大多 Optional——同一个类复用于多种消息类型,靠 type
枚举区分用途。相关性由响应关联字段串起来:客户端带 request_id / prev_response_id,
服务端带 response_id,一来一回对得上号。
4.2 消息类型枚举:一张"报文菜单"
type 字段的取值就是这套协议的"动词表"。客户端能发的和服务端能发的各一套:
| 阶段 | 客户端发(ClientMessageType) | 服务端发(ServerMessageType) |
|---|---|---|
| 注册 | REGISTER | (用 HEARTBEAT+OK 当确认) |
| 保活 | HEARTBEAT | HEARTBEAT |
| 派单 | TASK | TASK / COMMAND |
| 回报 | COMMAND_RESULTS / TASK_END | TASK_END |
| 查设备 | DEVICE_INFO_REQUEST / _RESPONSE | DEVICE_INFO_REQUEST / _RESPONSE |
| 出错 | ERROR | ERROR |
定义见 aip/messages.py:224(ClientMessageType)与 :255(ServerMessageType)。另有两个
"名词"枚举:ClientType(DEVICE vs CONSTELLATION,:282)标明连接身份;TaskStatus
(CONTINUE/COMPLETED/FAILED/OK/ERROR,:200)贯穿所有消息表达"当前状态"。
4.3 命令与工具:Command / MCPToolInfo
派单的最小单位是 Command——一次工具调用(aip/messages.py:149):
# 示意,非源码:Command 的形状
Command(
tool_name="click_button", # 调哪个工具
tool_type="action", # "action" 或 "data_collection"
parameters={"control_id": "42"}, # 工具入参
call_id="c-001", # 回报时用它对上号
)
服务端把一批 Command 塞进 ServerMessage.actions 下发;设备执行完,把每个 Command.call_id
对应的 Result(:183,带 status/error/result/call_id)装进
ClientMessage.action_results 回传——call_id 是命令与结果的对齐键。
工具本身怎么描述?MCPToolInfo(:92)是可上线传输的工具元数据(名字、命名空间、输入/
输出 schema)。它和 MCPToolCall(:109)是一对:后者额外持有一个活的 BaseMCPServer
实例引用(不可序列化,故 arbitrary_types_allowed=True),要上线时用 .tool_info 属性
(:129)降解成纯数据的 MCPToolInfo。这是一个典型的"传输态 vs 运行态"分离。
Command/MCPToolInfo只是载荷的形状;"派什么命令、命令在设备上怎么跑"分别属于 05 和 04。
4.4 五套协议:给消息装上语义
messages.py 只是数据结构;要"按协议规矩收发",靠 protocol/ 下五个类。它们都继承同一个
AIPProtocol(aip/protocol/base.py:22),各自封装一组语义方法:
| 协议类 | 管什么 | 代表方法 | 文件 |
|---|---|---|---|
RegistrationProtocol | 登记身份、能力广告 | register_as_device / register_as_constellation | aip/protocol/registration.py:25 |
HeartbeatProtocol | 周期保活 | send_heartbeat / start_heartbeat | aip/protocol/heartbeat.py:26 |
TaskExecutionProtocol | 派单/回报/结束 | send_task_assignment / send_commands / send_command_results / send_task_end | aip/protocol/task_execution.py:28 |
CommandProtocol | 命令/结果校验 | validate_command / validate_result | aip/protocol/command.py:17 |
DeviceInfoProtocol | 查设备信息 | request_device_info / send_device_info_response | aip/protocol/device_info.py:26 |
这些方法基本都是"造一个强类型消息 → 调 send_message 发出去",偶尔配一个"发完等回执"的
往返(如 register_as_device 发完 REGISTER 就 receive_message(ServerMessage) 等确认,
aip/protocol/registration.py:80-93)。
5. 底座:AIPProtocol 收发一条消息
所有协议类的公共父类 AIPProtocol(aip/protocol/base.py:22)是"消息进出 的唯一关口"。
两个核心方法把"对象 ↔ 字节"这段路走完。
发(send_message,aip/protocol/base.py:52):
msg(Pydantic) → 逐个中间件 process_outgoing → model_dump_json().encode()
→ transport.send(字节)
收(receive_message,aip/protocol/base.py:95):
transport.receive()(字节) → decode → message_type.model_validate_json()
→ 逆序过中间件 process_incoming → 返回强类型对象
两点设计值得记:
- 中间件方向相反。 出站按加入顺序、入站按逆序(
for middleware in reversed(...),:121),这样"最先包的最后拆",像洋葱一样对称。中间件基类是ProtocolMiddleware(:546),内置一个LoggingMiddleware(:575)。 - 断线是常态,不是错误。 收发里对
ConnectionError/IOError特判:消息含 "closed" / "not connected" 就降到DEBUG日志(:82-89),避免正常关连接时刷一屏 ERROR。
AIPProtocol 还内建一套二进制/文件传输能力(截图、大文件):send_binary_message
(:239)用"先发一帧 JSON 元数据、再发一帧二进制"的两帧法;send_file(:368)在此之上
做分块 + MD5 校验(file_transfer_start → 若干块 → file_transfer_complete)。对应的消息
壳在 messages.py:484 起(BinaryMetadata/FileTransferStart/ChunkMetadata 等)。这让同一条
连接既走 JSON 控制消息、也能高效搬运截图字节。
6. 传输层(把消息落到 WebSocket 帧上)
传输层回答一个很具体的问题:"发字节"这个动作,在服务端(FastAPI)和客户端(websockets 库) 里 API 根本不一样,怎么让上层无感? 答案是接口 + 适配器。
6.1 抽象接口:Transport
Transport(aip/transport/base.py:34)是一个 ABC,只规定五个异步动作:connect / send /
receive / close / wait_closed,外加一个 TransportState 状态机
(DISCONNECTED→CONNECTING→CONNECTED→DISCONNECTING→ERROR,:16)。上层(AIPProtocol)只依赖
这个接口,不知道底下是 WebSocket 还是别的——注释里明说未来可换 HTTP/3、gRPC。
6.2 唯一实现:WebSocketTransport
WebSocketTransport(aip/transport/websocket.py:24)是当前唯一实现。几个要点:
- 一个类兼两种角色。 构造时传入已存在的 websocket 就是服务端模式(直接标
CONNECTED,:79-86);不传、事后调connect(url)就是客户端模式(:88)。 - 参数为长连接调优。 默认
ping_interval=30、ping_timeout=180、max_size=100MB(:52-59)——WebSocket 自带的 ping/pong 是传输级保活,和第 7 节的应用级心跳是两层 独立的活性检测。 - 发送统一走文本帧。
send把字节 decode 成文本再发(:148-155),因为 JSON 消息本就是 文本;二进制走单独的send_binary(:248)。 - 收帧可自动辨型。
receive_auto(:350)能自动区分文本帧(返回str)和二进制帧 (返回bytes),给上层的二进制/文件流程用。
6.3 适配器:抹平两套 WebSocket 库
真正"抹平差异"的是 transport/adapters.py。WebSocketAdapter(:21)是抽象接口,两个实现:
| 适配器 | 用在哪 | 底层 API 例子 |
|---|---|---|
FastAPIWebSocketAdapter | 服务端 | send_text / receive_text / send_bytes(:105) |
WebSocketsLibAdapter | 客户端 | ws.send / ws.recv(:174) |
选哪个不用上层操心:工厂函数 create_adapter(:243)靠鸭子类型判断——对象有
client_state/application_state 属性就当 FastAPI,否则当 websockets 库(:253)。于是
WebSocketTransport.connect 里一句 create_adapter(self._ws) 就自动挑对了实现(:114)。
7. 报文时序:注册 → 心跳 → 派单 → 回报
前面把零件讲完了,现在把它们按时间串起来——这是本章的核心。下图是一次完整会话的报文 往返(以设备 agent 上线、编排器派活为例)。
怎么读这张图: 竖线是三方,横箭头是一次消息(标了 type/status);时间自上而下。
编排器(CONSTELLATION) 设备总机(server) 设备 Agent(DEVICE)
│ │ │
│ │ ① REGISTER │ 连上 & 登记
│ │◄───(client_type=DEVICE)─┤
│ │──HEARTBEAT/OK(确认)────►│
│ │ │
│ ①' REGISTER │ │
│ (client_type= │ │
│ CONSTELLATION, │ │
│ target_id=设备) │ │
├───────── ──────────────►│ │
│◄──HEARTBEAT/OK(确认)───┤ │
│ │ │
│ ② HEARTBEAT ⇄ HEARTBEAT/OK(双方各自周期喘气,贯穿全程)
│ │ │
│ ③ TASK │ │
│ (request=…, │ 转发 TASK │
│ target_id=设备) ├───────────────────────►│ UFO² 主循环启动
├───────────────────────►│ │ (见 01 章)
│ │ │
│ │◄── COMMAND_RESULTS ─────┤ 每步回报
│◄── (转发结果)──────────┤ (action_results, │
│ │ prev_response_id) │
│ │ …循环… │
│ │ │
│ │◄── TASK_END ────────────┤ status=COMPLETED
│◄── (转发 TASK_END)─────┤ │
│ │ │
分阶段读:
-
注册(建立身份)。 客户端发
REGISTER,带client_id、client_type、metadata(平台、能力、时间戳)。设备侧见register_as_device(aip/protocol/registration.py:41), 编排器侧见register_as_constellation(:99,多带target_id)。服务端校验并钉死身份后, 用一条HEARTBEAT+status=OK当注册确认(send_registration_confirmation,:171)。 失败则回ERROR(:187)。 -
心跳(保活)。 客户端每 30 秒发一个
HEARTBEAT(send_heartbeat,aip/protocol/heartbeat.py:43);服务端回send_heartbeat_ack(:62)。这是应用级活性 检测,和 WebSocket 传输级 ping/pong 并行。生 产服务端收到心跳就回执,见ufo/server/ws/handler.py:633。 -
派单(下发工作)。 编排器发
TASK(send_task_request,aip/protocol/task_execution.py:44), 服务端转发给目标设备;服务端也可用send_commands(:127)下发一批Command(ServerMessage.type=COMMAND,actions=[...])。 -
回报(交结果)。 设备每步用
send_command_results(:166)回传action_results(带prev_response_id对齐上一条服务端响应);任务终结用send_task_end(:228,status=COMPLETED/FAILED)。
8. 端点:把五层组装成三种角色
前面的层都是零件;端点(endpoint)是把它们装成"开箱即用的角色"的门面。 抽象基类
AIPEndpoint(aip/endpoints/base.py:18)规定:一个端点 = 协议 + 会话跟踪 + 容错(重连/心跳/
超时),并提供 send_with_timeout / receive_with_timeout 这类带超时的封装(:93/:106)。
三个具体端点各对应第 3 节的一个角色:
| 端点 | 角色 | 组装了什么 | 委托给谁执行 |
|---|---|---|---|
DeviceServerEndpoint | 设备总机(服务端) | 协议 + 重连策略 | ufo.server.ws.handler.UFOWebSocketHandler(server_endpoint.py:62) |
DeviceClientEndpoint | 设备 agent(客户端) | 传输 + 注册/心跳协议 + 心跳管理器 + 重连 | ufo.client.websocket.UFOWebSocketClient(client_endpoint.py:69) |
ConstellationEndpoint | 编排器(客户端) | 传输 + 注册协议 + 重连 | galaxy…WebSocketConnectionManager(constellation_endpoint.py:62) |
共同套路:端点是"门面 + 委托"。 每个端点在 __init__ 里搭好 AIP 的传输/协议/容错组件,但
真正的连接循环委托给已存在的宿主类(通过延迟 import 避免循环依赖)。例如
DeviceClientEndpoint.__init__ 建好 WebSocketTransport、RegistrationProtocol、
HeartbeatProtocol、HeartbeatManager(aip/endpoints/client_endpoint.py:46-74),但
start() 只是起个 task 跑 self.client.connect_and_listen()(:87)。这让 AIP 能逐步接管
既有代码,而不是推倒重来。
DeviceServerEndpoint 则把 handle_websocket 直接转交给老 handler(server_endpoint.py:86),
并实现 cancel_device_tasks——按 device_id 找出会话逐个取消(:107-119),这在设备掉线时被
容错层调用。
9. 容错:让连接"活着"
网络会抖、进程会崩。resilience/ 三个组件让连接尽量自愈。
9.1 自动重连(退避)
ReconnectionStrategy(aip/resilience/reconnection.py:29)在设备掉线时按四步走
(handle_disconnection,:67):
怎么读: 从左到右是掉线后的处理顺序,重连按退避节奏重试到成功或耗尽。
掉线 → ① 取消该设备待办任务 → ② 通知上层 → ③ 退避重连(循环) → ④ 成功则回调
│
等 backoff → 试 reconnect_device
失败 → retry_count++ → 再等更久
达 max_retries → 放弃
退避时长由 _calculate_backoff(:196)按策略算:EXPONENTIAL_BACKOFF 是
initial * multiplier^retry,再用 min(..., max_backoff) 封顶(:213-214)。还支持
LINEAR/IMMEDIATE/NONE(ReconnectionPolicy,:20)。不同端点给不同参数:设备客户端
max_retries=3, initial=2.0(client_endpoint.py:56),编排器 max_retries=5, initial=1.0
(constellation_endpoint.py:47)。
一个稳健细节:_cancel_pending_tasks / _notify_disconnection 都用 hasattr 探测再调
(:168/:188),端点没实现对应钩子也不会炸。
9.2 心跳管理器(多客户端保活)
第 7 节的 HeartbeatProtocol 只管"发一个心跳";HeartbeatManager
(aip/resilience/heartbeat_manager.py:18)管"给一堆客户端各开一个周期心跳任务"——每个
client_id 一个 asyncio.Task,存在字典里(:45)。循环里每隔 interval 检查
protocol.is_connected() 再发,发失败只记日志不中断循环,把断线判定留给连接管理器
(_heartbeat_loop,:114-138)。停止时 stop_all 逐个 cancel(:88)。
9.3 超时
TimeoutManager(aip/resilience/timeout.py:17)用 asyncio.wait_for 给任意协程套超时
(with_timeout,:37),超时抛 TimeoutError;另有 with_timeout_or_none(:67)超时返回
None 而不抛。端点的 send_with_timeout 就走它,默认 120 秒(endpoints/base.py:53)。
10. extensions:旁挂横切逻辑
extensions/ 和第 5 节的"中间件"是两套不同东西,别混:
| 机制 | 挂在哪 | 能改消息吗 | 用途 |
|---|---|---|---|
ProtocolMiddleware | AIPProtocol 收发链内 | 能(返回改写后的 msg) | 加密、改写、内联日志 |
AIPExtension | 端点事件旁路 | 不能(只观察) | 日志、指标、监控 |
AIPExtension(aip/extensions/base.py:14)是纯观察者接口,定义五个事件钩子:
on_message_sent / on_message_received / on_connection_established / on_connection_closed
/ on_error。两个开箱实现在 extensions/middleware.py:LoggingExtension(:17,把事件打日志)
和 MetricsExtension(:54,统计收发条数、按类型分桶、并用 request_id/response_id 配对算
往返延迟,:82-95)。这类扩展是"可选旁挂",不接也不影响主链路。
11. 巧妙之处(可借鉴的技术)
-
传输态 vs 运行态分离。
MCPToolCall持有不可序列化的BaseMCPServer活对象,靠.tool_info属性降解成纯数据的MCPToolInfo再上线(aip/messages.py:129)。想传"能力清单" 又不想传"活连接",这是干净解法。 -
鸭子类型选适配器。
create_adapter不做isinstance硬判,而是看有没有client_state属性来区分 FastAPI / websockets 库(aip/transport/adapters.py:253)——一个类 两种角色,连接代码零分支。 -
两级独立保活。 WebSocket 传输级 ping/pong(
ping_interval)+ AIP 应用级HEARTBEAT报文,两层都断才判死,降低误杀长任务的概率(transport/websocket.py:70对protocol/heartbeat.py:43)。 -
断线降噪。 收发路径把"connection closed"类异常从 ERROR 降到 DEBUG (
aip/protocol/base.py:82-89),正常关连接不再刷一屏假告警——运维友好。 -
门面+委托的渐进接管。 端点搭好 AIP 组件却把连接循环委托给既有宿主类 (
aip/endpoints/client_endpoint.py:87),让协议层能一点点接管遗留代码而非重写。 -
中间件对称洋葱。 出站顺序、入站逆序(
aip/protocol/base.py:121),保证"先包的后拆"。
12. 边界与局限(诚实)
-
只有 WebSocket 一种传输。
Transport抽象为未来的 HTTP/3、gRPC 留了口 (transport/base.py类注释),但当前仅WebSocketTransport一个实现。 -
aip/本身不是完整运行时。 三个端点大量委托给ufo.server.ws.handler/ufo.client.websocket/galaxy…WebSocketConnectionManager;aip/提供协议词汇、传输与容错, 真实收发循环在那些宿主里(endpoints/*的延迟 import 处可见)。 -
部分方法是占位/未完全实现。 例如
ConstellationEndpoint.reconnect_device直接记 "not fully implemented" 返回False(constellation_endpoint.py:147);send_task_result在缺client_id时回退成字面量"unknown_client"(task_execution.py:217)。 -
重连状态是每策略实例一份。
ReconnectionStrategy._retry_count是实例字段 (reconnection.py:64),多设备共享一个策略实例时计数会相互影响;当前每个端点各持一个实例 规避了这点。 -
心跳发送失败不主动断连。
_heartbeat_loop发失败只记日志、继续循环 (heartbeat_manager.py:130-134),真正的死连接判定依赖底层传输/连接管理器,不由心跳直接触发。
13. 代码地图(导航索引)
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 客户端消息(上行) | aip/messages.py | ClientMessage |
| 服务端消息(下行) | aip/messages.py | ServerMessage |
| 消息类型枚举 | aip/messages.py | ClientMessageType / ServerMessageType |
| 连接身份 / 任务状态 | aip/messages.py | ClientType / TaskStatus |
| 命令 / 结果载荷 | aip/messages.py | Command / Result |
| 工具元数据(传输/运行态) | aip/messages.py | MCPToolInfo / MCPToolCall.tool_info |
| 二进制/文件传输壳 | aip/messages.py | BinaryMetadata / FileTransferStart |
| 协议底座(收发/中间件) | aip/protocol/base.py | AIPProtocol / ProtocolMiddleware |
| 分块文件传输 | aip/protocol/base.py | AIPProtocol.send_file / receive_file |
| 注册协议 | aip/protocol/registration.py | RegistrationProtocol.register_as_device / register_as_constellation |
| 心跳协议 | aip/protocol/heartbeat.py | HeartbeatProtocol.send_heartbeat / start_heartbeat |
| 派单/回报协议 | aip/protocol/task_execution.py | TaskExecutionProtocol.send_commands / send_command_results / send_task_end |
| 命令校验 | aip/protocol/command.py | CommandProtocol.validate_command |
| 设备信息协议 | aip/protocol/device_info.py | DeviceInfoProtocol.request_device_info |
| 传输抽象 + 状态机 | aip/transport/base.py | Transport / TransportState |
| WebSocket 传输 | aip/transport/websocket.py | WebSocketTransport.connect / send / receive_auto |
| 收发帧适配器 + 工厂 | aip/transport/adapters.py | WebSocketAdapter / create_adapter |
| 端点抽象 | aip/endpoints/base.py | AIPEndpoint |
| 设备总机端点 | aip/endpoints/server_endpoint.py | DeviceServerEndpoint.handle_websocket |
| 设备客户端端点 | aip/endpoints/client_endpoint.py | DeviceClientEndpoint.start |
| 编排器端点 | aip/endpoints/constellation_endpoint.py | ConstellationEndpoint.connect_to_device |
| 自动重连(退避) | aip/resilience/reconnection.py | ReconnectionStrategy.handle_disconnection / _calculate_backoff |
| 心跳管理器 | aip/resilience/heartbeat_manager.py | HeartbeatManager._heartbeat_loop |
| 超时管理器 | aip/resilience/timeout.py | TimeoutManager.with_timeout |
| 观察者扩展 | aip/extensions/base.py | AIPExtension |
| 日志/指标扩展 | aip/extensions/middleware.py | LoggingExtension / MetricsExtension |
| 生产服务端收发循环 | ufo/server/ws/handler.py | UFOWebSocketHandler(dispatch :527, 心跳 ack :633) |
| 生产设备端收发循环 | ufo/client/websocket.py | UFOWebSocketClient.connect_and_listen |
相邻章节: 上层"派什么任务、怎么编排"见 05 Galaxy 星座 DAG; 设备收到任务后"怎么在本机执行"见 01 主循环 到 04 混合自动化;全景导读见 index。