chongming-core/docs/plugin-wire-protocol.md

# 进程插件线协议规范（语言无关）

> 本规范定义网关与进程插件（`load_mode = "external"`）之间的**唯一契约**：
> Zenoh keyexpr + JSON 载荷。任何拥有 Zenoh 绑定的语言（Rust / Go / C / C++ / Java 等）
> 只要按本规范实现，即可与网关零改动对接，与 Python 插件共存于同一类型、参与同一套负载均衡。
>
> **事实来源**：本规范以 `chongming_worker` SDK 与网关代码为唯一事实来源
> （见文末[附录 A](#附录-a-事实来源索引)）。修改任何一侧实现必须同步更新本规范。

---

## 1. 总览

进程插件是一个**独立进程**，生命周期由 Docker/K8s 编排拉起（网关不负责 spawn）。
插件进程启动后：

1. 打开 Zenoh 会话（配置见 §5）
2. 发布路由描述符 → 网关 RouteRegistry 订阅后动态挂载 HTTP 端点
3. 声明 liveliness token 与 Queryable → 网关 WorkerManager 发现并按负载均衡转发 HTTP 请求
4. 退出时摘除路由描述符（DELETE），网关 2s 防抖后移除 HTTP 端点

```
HTTP 客户端 ──> 网关 FastAPI 动态路由
                  │  WorkerManager（15s liveliness 发现 + 随机负载均衡 + 超时重试）
                  ▼
        Zenoh Query: worker/<type>/<id>/<subject>
                  ▲        │
                  │        ▼
        插件进程 Queryable（解析信封 → 业务处理 → JSON 回复）

插件进程 ──PUT/DELETE──> chongming/routes/<type>/<route_id>   （路由描述符）
插件进程 ──liveliness──> worker/<type>/<id>                    （服务发现）
```

### 关键时间参数

| 参数 | 值 | 语义 |
| ---- | -- | ---- |
| 描述符重发布周期 | 10s | 插件侧周期性重发布全部描述符（自愈：网关后启动 / 样本丢失） |
| 网关路由防抖 | 2s | 网关收到描述符变更后防抖重建路由（合并重启风暴） |
| liveliness 刷新 | 15s | 网关 WorkerManager 轮询 `worker/*/*` |
| 优雅退出摘除 | 即时 | 插件收到 SIGTERM/SIGINT 后 DELETE 全部已发布 key |

---

## 2. 路由描述符协议（`chongming/routes/**`）

### 2.1 keyexpr 结构

```
chongming/routes/<type>/<route_id>    # 端点描述符（PUT 发布 / DELETE 摘除）
chongming/routes/<type>/__meta__      # 插件级元信息（可选，仅在声明 MQTT 主题时发布）
```

- `<type>`：插件类型（worker type），与 `config/workers/<type>/` 目录名一致
- `<route_id>`：`<subject 的 "." 替换为 "_">__<METHOD>`，如 `demo.hello` + `GET` → `demo_hello__GET`
- 载荷一律 UTF-8 JSON

### 2.2 描述符字段（schema_version = 1）

发布端在端点描述符上附加两个字段：

```json
{
  "schema_version": 1,
  "type": "demo",
  ...以下字段来自端点定义（TOML 兼容 dict）
}
```

网关侧消费的字段（`RouteConfig`，缺失时取默认值，未知字段忽略）：

| 字段 | 类型 | 默认 | 说明 |
| ---- | ---- | ---- | ---- |
| `subject` | string | 必填 | RPC 主题，如 `demo.hello`；与 `worker/<type>/<id>/<subject>` 对应 |
| `method` | string | `"GET"` | HTTP 方法（`GET/POST/PUT/PATCH/DELETE`）；`"MQTT"` 为纯 MQTT 端点（不注册 HTTP 路由）；网关侧统一大写处理 |
| `path` | string | 空 | HTTP 路径，如 `/demo/hello`；含 `{param}` 时构成路径参数 |
| `params` | array[string] | `[]` | 参数声明，每项 `"名称: 类型"`（类型支持 `str/int/float/bool/any`、`list[...]`、`dict[...]`）；无 `:` 的项按 `str` 处理 |
| `response_model` | object | `{}` | 响应模型声明：`{"字段": {"type": "...", "required": bool, "default": ...}}` 或 `{"字段": "type"}`（仅用于网关 OpenAPI 文档与响应模型校验） |
| `summary` | string | `""` | OpenAPI 摘要 |
| `docstring` | string | `""` | OpenAPI 描述 |
| `timeout` | number | `3.0` | 单次调用超时（秒）；流式端点为其**总寿命上限**，须配大（如 300） |
| `cacheable` | bool | `false` | 网关侧结果缓存开关（成功响应按 `(type, subject, payload)` 缓存） |
| `validate` | bool | `true` | 网关是否解析并校验请求体（false = 原样透传，校验下沉 worker；仅对 POST/PUT/PATCH 非流式端点生效，见 §4.2） |
| `auth_required` | bool | `true` | 网关是否注入鉴权依赖（认证用户经信封 `meta.user_info` 透传给插件） |
| `internal` | bool | `false` | 内部端点：不注册 HTTP 路由，仅保留 Zenoh RPC 调用能力 |
| `stream` | bool | `false` | 流式端点：网关以 SSE 逐块转发多回复 |
| `stream_idle_timeout` | number | `10.0` | 流式空闲超时（秒）：连续无数据块的最大间隔 |
| `route_name` | string | 无 | 端点名（TOML 文件名），契约校验用于推断模型名；网关运行时不消费 |
| `schema_version` | int | 1 | 协议版本（见 §7 演进规则） |
| `type` | string | 插件类型 | 插件类型，网关缺失时以 keyexpr 中的 `<type>` 兜底 |

> **描述符生成约定**（Python SDK 行为，其他语言须输出等价结构）：
> - `method` 恒取自端点 TOML 节键并大写，不被 defaults 覆盖
> - `path` 缺省为 `router_prefix + "/" + 端点名`
> - `subject` 缺省为 `<type>.<端点名>`
> - `[worker.defaults]` 中的键作为字段缺省值合并进描述符（显式字段优先）

### 2.3 发布 / 摘除语义

| 操作 | 行为 |
| ---- | ---- |
| 发布 | PUT 单个端点描述符到 `chongming/routes/<type>/<route_id>`；多端点逐个 PUT |
| 元信息 | 声明了 `[[workers.mqtt_topics]]` 时，额外 PUT `chongming/routes/<type>/__meta__`，载荷 `{"schema_version": 1, "mqtt_topics": [{"topic": "...", ...}]}`；网关据此动态订阅外部主题 |
| 重发布 | 每 10s 将**全部**描述符重新 PUT（幂等） |
| 摘除 | 优雅退出（SIGTERM/SIGINT）时 DELETE 全部已发布 key；进程被 SIGKILL 时描述符残留，重启后重发布覆盖 |

### 2.4 网关侧行为（实现方须知）

- 订阅 `chongming/routes/**`；PUT → upsert，DELETE → remove
- **幂等去重**：同一 `route_id` 描述符内容 hash 相同视为多副本重复发布，忽略
- **冲突**：同一 `route_id` 内容不同 → 告警并**保留先到者**
- **防抖**：type 级 2s 防抖，变更合并为一次路由重建
- **健壮性**：非法 JSON / 非 dict 描述符 → 丢弃并告警，不影响网关
- **路由筛选**：`method == "MQTT"` 或 `internal == true` 的端点不注册 HTTP 路由，但保留在路由表中

---

## 3. 服务发现协议（liveliness）

- 插件进程为**每个副本**声明 liveliness token：`worker/<type>/<id>`
- `<id>` 无格式约束（Python SDK 惯例 `worker-<i>-<pid>`）；网关只解析 `parts[1]` 作为类型
- 网关每 15s 轮询 `worker/*/*` 刷新存活列表，调用时随机选择副本（负载均衡）

---

## 4. RPC 调用协议（Queryable）

### 4.1 keyexpr

```
worker/<type>/<id>/<subject>
```

`<subject>` 与描述符的 `subject` 字段一致（多方法共享同一 subject 时，同一 keyexpr 由多个方法分派）。

### 4.2 请求信封

```
{"meta": {"method": "GET", "user_info": {...}}, "data": {...业务参数...}}
```

| 字段 | 说明 |
| ---- | ---- |
| `meta.method` | HTTP 方法（大写）；**缺失默认 `GET`** |
| `meta.user_info` | 网关注入的认证用户信息（`auth_required = true` 时存在，结构由网关鉴权决定） |
| `data` | 业务参数：GET/DELETE 为查询/路径参数的 `{name: value}` 字典；POST/PUT/PATCH 为请求体字段 |

**兼容规则**：非信封载荷（不含 `meta`/`data` 键的 JSON 对象）整包视为业务 `data`，按 `method = GET` 处理。

**透传模式（`validate = false` 端点）**：`data` 为请求体的**原始 JSON 值**（网关零解析/校验，
信封由字节拼接构造，结构不变）。插件收到的 `data` 形态与校验模式一致（由 SDK 解析为 dict）；
参数类型转换/校验失败由插件侧暴露（框架异常经 `reply_err` → 网关 503）。空/空白请求体兜底为 `{}`。

### 4.3 响应

| 形态 | 行为 |
| ---- | ---- |
| 普通 | 单个回复，载荷为业务结果 JSON 对象 |
| 框架异常（非流式） | 单个 **`reply_err`** 回复，载荷 `{"result": 0.0, "operation": "error", "error": "<异常信息>", "worker": "<worker_id>"}`；网关据此零解析判定失败 → 重试 → HTTP 503 |
| 业务错误 | 普通回复（如 `{"code": 1, "error": "..."}`），HTTP 200 透传，网关不解析 |
| 流式 | 对同一 Query **多次 reply** 逐块推送；生成器正常结束 = 流结束（无结束哨兵）；中途异常以 `{"error": "<异常信息>"}` **收尾块**结束流（流式错误仍为普通回复，不走 `reply_err`）；查询已终止（客户端断开）时停止回复 |

> **错误通道分层**：框架异常（handler 抛错、无匹配方法、超时）→ `reply_err`；
> 业务错误（业务逻辑判定失败）→ 普通回复。插件作者语义：**返回即业务、抛出即框架**。
>
> **兼容矩阵**（新旧网关/插件任意组合均语义完好）：
>
> | 网关 \ 插件 | 旧（普通回复错误信封） | 新（reply_err） |
> | ----------- | ---------------------- | --------------- |
> | 旧 | HTTP 200 + 错误信封（历史行为） | `call()` 已处理 reply.err → 重试 → 503 |
> | 新 | HTTP 200 + 错误信封（透传不解析，业务语义） | reply_err → 重试 → **503**（语义修正） |

网关侧对应行为：

- 普通调用 `WorkerManager.call_raw`（HTTP 端点）：取首个回复，**原始字节零解析透传**给 HTTP 响应；空回复 / reply.err / 超时 → **切换副本重试**（默认重试 1 次），重试耗尽 → HTTP 503
- `call()`（MQTT Bridge 等内部调用方）：同上述重试逻辑，回复解析为 dict 后使用
- 流式调用 `WorkerManager.call_stream`：逐块消费（`ConsolidationMode.NONE`）；收到含 `error` 键的块视为终止信号转发后结束；`reply.err`（zenoh 层错误）同样终止流
- 无可用副本 → HTTP 503

---

## 5. 配置约定

插件进程自读两份配置（其他语言实现须兼容读取语义）：

| 配置 | 位置 | 用途 |
| ---- | ---- | ---- |
| 全局配置 | `config.toml`（`--config` 可指定） | 仅 `[zenoh].json5_path`（默认 `config/zenoh_config.json5`），JSON5 格式的 Zenoh 会话配置 |
| 插件配置 | `config/workers/<type>/config.toml` | `[workers]` 节（type/count/load_mode/mqtt_topics/运行时配置）与端点 TOML 定义（描述符来源） |

进程入口参数：`--type <name> [--count N] [--config config.toml]`（Python SDK 约定；其他语言可自定入口，但 `--type` 语义应保持一致）。

> 非 Python 插件的配置读取完全由自身实现，**网关不读取插件配置**
> （描述符经 Zenoh 发布，网关与插件配置解耦）。

---

## 6. 最小实现清单（其他语言插件）

一个合规的插件进程必须实现：

1. [ ] 解析 `config.toml` 的 `[zenoh].json5_path` 并打开 Zenoh 会话
2. [ ] 为每个端点构造描述符（§2.2 字段表）并 PUT 到 `chongming/routes/<type>/<route_id>`
3. [ ] 声明 `[[workers.mqtt_topics]]` 时 PUT `__meta__` 描述符
4. [ ] 为每个副本声明 liveliness token `worker/<type>/<id>`（多副本时 id 互异）
5. [ ] 为每个 subject 声明 Queryable `worker/<type>/<id>/<subject>`
6. [ ] 解析请求信封（§4.2），按 `meta.method` 分派；回复符合 §4.3 三种形态
7. [ ] 每 10s 重发布全部描述符
8. [ ] 优雅退出时 DELETE 全部已发布 key

**自检义务**：网关侧不做描述符强校验（契约校验是 Python SDK 的进程内行为），
插件须自行保证描述符字段完整（尤其 `subject` / `method` / `path` / `params` / `response_model`），
否则会挂载出错误路由。

---

## 7. 版本与演进规则

- 描述符携带 `schema_version`（当前 `1`），网关对未知版本暂不拒绝（缺失时兜底为 1）
- **协议变更分级**：
  - 破坏性变更（keyexpr 结构、信封结构、错误格式、描述符语义）→ `schema_version` 递增，网关须先发布兼容处理，再切换
  - 新增可选字段 / 新增端点形态 → 允许同版本内演进，但必须同步更新本规范
- **本版同版本内演进说明（2026-08）**：
  - 新增描述符字段 `validate`（§2.2，网关缺失时兜底 true，纯增量）
  - 非流式框架异常新增 `reply_err` 错误通道（§4.3）：信封结构与错误载荷均不变，
    仅输出通道由普通 reply 改为 reply_err；网关 `call()/call_raw()` 历史上已处理 reply.err，
    新旧组合兼容（见 §4.3 兼容矩阵），故不递增 schema_version
  - 网关普通调用由 `call`（dict 语义）切换为 `call_raw`（bytes 透传）：入参出参形态变化
    为网关内部实现，不影响线上协议
- **修改即更新**：任何实现侧改动（`routes_publisher.py` / `runtime.py` / `route_registry.py` / `worker_manager.py` / `route_generator.py`）触及本规范描述的语义时，必须同步修订本规范

---

## 附录 A：事实来源索引

| 协议项 | 唯一实现（文件:行号） |
| ---- | ---- |
| 描述符 keyexpr / route_id 编码 / 发布摘除 | `utils/worker/src/chongming_worker/routes_publisher.py:23-89` |
| 描述符字段定义与消费 | `src/chongming/app/core/route_generator.py:82-100`（RouteConfig） |
| 描述符生成约定（TOML → dict） | `utils/worker/src/chongming_worker/route_config.py:25-103` |
| 网关订阅 / 防抖 / 幂等 / 冲突 / 筛选 | `src/chongming/app/core/route_registry.py:44-261` |
| liveliness 发现 / 负载均衡 / 超时重试 / 缓存 | `src/chongming/app/core/worker_manager.py:73-379` |
| 请求信封构造（meta.method / user_info） | `src/chongming/app/core/route_generator.py:241-405` |
| 信封解析 / 分派 / 三种响应形态 | `utils/worker/src/chongming_worker/runtime.py:27-220` |
| 进程生命周期（10s 重发布 / 优雅退出） | `utils/worker/src/chongming_worker/process.py:35-197` |
| 配置读取（json5_path / 插件配置） | `utils/worker/src/chongming_worker/process.py:39-136` |
| 契约校验语义（仅 Python SDK 进程内） | `utils/worker/src/chongming_worker/contracts.py:150-189` |