MoShiro/docs/protocol.md

# 线协议

MoShiro 的两条控制链路都建立在 Fleet 之上：

- **Agent ↔ Backend**：Fleet 连接服务端（wss），见 [../backend/fleet/README.md](../backend/fleet/README.md)；
- **Probe ↔ Agent**：Fleet 节点间点对点会话（TLS），见 [../backend/fleet/README.md](../backend/fleet/README.md) 的「点对点会话」一节。

认证、心跳、重连、能力协商、可靠发件箱、期望状态同步、JSON-RPC 信封都由 Fleet 负责。本文只定义 **MoShiro 自己的内容**：`moshiro.*` 方法、状态文档、二进制通道，以及 Fleet 之外的数据面（UDP 包和数据连接）。所有类型都放在 `MoShiro.Protocol` 中。

## 1. Fleet 能力的映射

| MoShiro 需要的能力 | 使用的 Fleet 机制 | MoShiro 提供的内容 |
| --- | --- | --- |
| 下发任务 | 期望状态 `fleet.state.sync` | 文档类型 `TaskSet`，包括 `tasks[]`、`peers`（允许连这个 Agent 的 Probe 公钥，已按线路开关过滤） |
| 下发 Agent 列表给 Probe | 期望状态 `fleet.state.sync`（Probe 直连面板） | 文档类型 `probe_agents`：`version`、`agents[]`（`agent_id`、`host`、`tcp_port`、`udp_ports`、`spki`），已按线路开关过滤；Agent 集合或线路开关变化时重新推送 |
| 更新 | 通知 `moshiro.update.offer` / `moshiro.update.apply`（B→A、B→P），心跳附加字段 `update` | `offer` 带 CI 签名的发布清单原文和签名，节点只写进数据目录；`apply` 请求更新；`update` 是宿主机更新器的状态（`state`、`current`、`available`、`error`、`at`），见 [deployment.md](deployment.md) |
| 上报原始块 | 发件箱，二进制通道 **16** | `RawChunk`（自描述，见 [measurement.md](measurement.md) 第 4 节），`msg_id = chunk_id` |
| Probe 交付列块 | 点对点会话的发件箱，二进制通道 **17** | 列块：`u32 stream_id \| u32 first_seq \| u32 count \| 列数据(Brotli)` |
| 节点自身状态 | `fleet.heartbeat` 的附加字段 | Agent：`sessions`、`probes[]`、`pps_in/out`、`dropped_invalid`、`dropped_rate_limited`、`queue_bytes`、`update`；Probe：只有 `update` |
| 版本与能力 | `fleet.hello` | capabilities 包括 `moshiro.kinds:<列表>`、`moshiro.blocks:<列表>`；Agent 附加 `platform`（系统、架构、版本、tag、提交、构建时间）、`tls_spki`、`probe_ports`、`local_limits` 摘要；Probe 附加 `platform` |

## 2. `moshiro.*` 方法

### Agent ↔ Backend

| 方法 | 方向 | 类型 | 参数与结果 |
| --- | --- | --- | --- |
| `moshiro.test.run` | B→A | 请求 | `test_id`、`probe_id`（在哪个 Probe 上跑）、`plan`、`phases[]` → `accepted`，或错误 `test_busy`、`probe_offline` |
| `moshiro.probe.disconnect` | B→A | 请求 | `probe_id` → `accepted`（Agent 是否真的有这个会话）；Agent 关闭该 Probe 的会话 |
| `moshiro.test.cancel` | B→A | 请求 | `test_id` |
| `moshiro.test.status` | A→B | 通知 | `test_id`、`status`（`running`、`phase_done`、`succeeded`、`failed`、`aborted`）、`phase?`、`error_code?` |
| `moshiro.event` | A→B | 通知 | `kind`（`probe_connected`（`detail` 是 `probe_id`，`probe` 里带版本、远端地址和连接时间）、`probe_disconnected`、`endpoint_changed`、`session_rejected`、`clock_step`、`gap`、`cpu_saturated`（节点进程 CPU 饱和，可能影响测量，由 Probe 或 Agent 上报））、`at_utc`、`detail` |

心跳的附加字段里有 `probes[]`：Agent 当前连着的每个 Probe（`id`、`version`、`remote`、`since_utc_ms`），每次都是完整快照，Backend 靠它纠正错过的连接事件。点对点握手的认证消息里可以带 Probe 的 `version`（只用于展示）。块头里的 `probe_id` 以 Agent 认证过的会话为准，Probe 自报的值会被覆盖。

### Probe ↔ Agent（点对点会话）

| 方法 | 方向 | 类型 | 参数与结果 |
| --- | --- | --- | --- |
| `fleet.state.sync`（文档类型 `line_task_set`） | A→P | 请求 | 这条线路的任务（`id`、`kind`、`params`、`stream_id`、`port`）和 Agent 的数据面端口 |
| `moshiro.clock` | P→A | 请求 | `probe_mono_us` → `agent_mono_us`；Probe 按往返中点估计两端单调时钟的偏移（取最小往返的一次），记入块头 `probe_offset_us` |
| `moshiro.test.phase` | A→P | 请求 | 测试的一个阶段：`test_id`、`phase`、`kind`、`params`、`stream_id`、`duration_s`、`port` → `accepted`；Probe 在回复之前准备好自己一侧的角色（单向数据流：先发几个探活包让 Agent 学到源地址） |
| `moshiro.test.phase_end` | A→P | 请求 | `sent_count`（Agent 发出的包数）→ `fragments`：Probe 收尾，把列块交给 Agent，回复交出的个数；Agent 等全部列块合并上传之后才进入下一阶段 |
| `moshiro.test.abort` | A→P | 通知 | `test_id` |
| `fleet.outbox.ack` | 双向 | 通知 | Probe 的列块（通道 17）是至少一次投递，Agent 每收到一个就确认，重复到达的只处理一次 |

会话密钥不需要单独的方法：点对点握手（TLS 固定证书 → `peer.challenge` → Probe 用身份私钥签 `(peer_id, nonce, client_pub, server_pub)` → 两端各自 ECDH + HKDF）直接得到 `session_id` 和 `session_key`，`peer.auth` 的应答只回显 `session_id` 用来互相确认。

### 列块（Probe → Agent，通道 17）

Probe 只记录自己一侧的列，Agent 在数据面记录另一侧，由 Agent 合并、盖章、上传（见 `measurement.md` 第 5 节）。合并**由 Kind 的流声明驱动**（`StreamSpec`：`Driver`、`Key`，每一列的归属由它的时钟决定），没有按 Kind 手写的模式：

| 流声明 | 例子 | Probe 的列块 | Agent 补充 |
| --- | --- | --- | --- |
| `Driver = Probe`，有 `Key`，Agent 一侧有列 | `udp_echo`、`tcp_echo` 的 `msg` | `seq`、`size`、`t1`、`t4` | 按 `seq` 补上 `t2`、`t3`（没有记录表示上行丢失） |
| `Driver = Agent`，有 `Key` | `udp_train` 的 `pkt` | `pkt(seq, t_recv)`，头部 `_seq_from` / `_seq_to` | 行由 Agent 发出的包决定（`size`、`t_send`），按 `seq` 取 `t_recv`（缺失表示丢失） |
| 没有 `Key`，或只有主导方一侧有列 | `connect`、`hold`、`step` | 完整的行 | —（按声明的列顺序原样透传） |

单块最多 `MaxRecordsPerChunk`（50 000）条记录：Probe 必须把大的阶段按序号范围切成多个列块（例如 100 Mbit/s、1000 B 的 10 秒阶段是 125 000 个包 → 4 个列块），`phase_end` 的应答报告总块数。块头里 Probe 必须带上 `agent_id`（它知道自己在和哪个 Agent 通信）。

新增方法只需要在 `MoShiro.Protocol/Rpc` 中定义，并在节点组装处加一行 `.On(...)`。旧节点会通过 capabilities 协商自动避开不支持的方法。

## 3. 数据面（不属于 Fleet）

### 3.1 UDP 包

```text
偏移  长度  字段
0     2     magic      0x4D4E ("MN")
2     1     version    1
3     1     type       1 = EchoRequest，2 = EchoReply，3 = Data（单向）
4     8     session_id
12    4     stream_id  标识任务或测试阶段
16    4     seq        在同一 stream 内单调递增
20    8     send_ts_us 发送方单调时钟（EchoReply 原样带回请求中的值）
28    N     padding    补足到 Payload 积木决定的包大小
28+N  16    mac        HMAC-SHA256(session_key, 前面全部字节)，截断至 16 字节
```

**接收方处理规则**（任一步失败即静默丢弃并计数，**不回复**）：

1. 长度、magic、version、type 合法；
2. `session_id` 和 `stream_id` 都处于活跃状态；
3. MAC 校验通过；
4. `seq` 不在重放窗口中；
5. 速率不超过声明速率的 1.2 倍，也不超过本地硬上限。

通过后先**记录原始列**。收到 `EchoRequest` 时，立即回复一个长度**完全相同**的包，发往请求的**实际源地址**（这样可以跟随 NAT 映射变化），并记录回复时刻。

### 3.2 TCP/TLS 数据连接

- 连接都由 Probe 发起；首帧为 `session_id | stream_id | nonce | mac`，认证失败直接关闭连接。
- 应用层消息使用 `VarInt 长度 | 正文` 分帧；回显类消息的正文为 `seq | send_ts_us | padding`。

## 4. 版本演进

- Fleet 层的兼容规则见 [../backend/fleet/README.md](../backend/fleet/README.md)。
- `moshiro.*` 方法：新增方法或可选字段不需要升级版本，接收方忽略未知字段。
- RawChunk：块头自描述，新增列或新增 Kind 不需要升级 `container_version`；只有容器格式变化时才升级。
- UDP 包和数据连接首帧的布局变化时，必须升级 `protocol_version`，并更新黄金字节测试。
- Agent 同时支持当前版本和上一个版本的 Probe；Backend 能解码保留期内所有的 `container_version`。