modify: json 解析优化

docs/multi-language-plugin-plan.md

@@ -15,7 +15,7 @@
 | - | ---- | ---- |
 | P1 | **规范先行，协议是唯一契约** | 网关与插件之间只认线协议（[plugin-wire-protocol.md](./plugin-wire-protocol.md)）。任何语言实现均以规范为准，不得依赖其他语言 SDK 的内部行为 |
 | P2 | **网关零改动起步** | 阶段 1 的 Rust 插件按协议实现即可对接，网关代码不允许为适配单语言做改动 |
-| P3 | **混合部署兼容** | 同一 `<type>` 下可混布 Python 与 Rust 副本（协议相同、负载均衡透明）；但流式语义、错误格式、超时行为必须与规范完全一致 |
+| P3 | **混合部署兼容** | 同一 `<type>` 下可混布 Python 与 Rust 副本（协议相同、负载均衡透明）；但流式语义、错误格式（非流式经 `reply_err` 通道）、超时行为必须与规范完全一致 |
 | P4 | **契约校验不旁路** | Python SDK 的进程内契约校验（TOML ↔ 注册 ↔ 模型）是 Python 特有能力；其他语言实现须提供等价的自检（描述符字段完整性），网关侧强校验在阶段 2 补齐 |
 | P5 | **修改即更新规范** | 任何实现侧改动触及协议语义时，必须同步修订规范（含[附录 A 事实来源索引](./plugin-wire-protocol.md)） |
 | P6 | **性能收益只认实测** | 语言选型以基准测试为准（见 §6），不凭直觉切换 |
@@ -75,7 +75,7 @@ chongming-worker-rs/            # 独立仓库（或 utils/worker-rs/，与 Pyth
 | §2 路由描述符 | `descriptor.rs` 构造 + `publisher.rs` PUT/DELETE；`route_id = subject.replace('.','_') + "__" + METHOD` |
 | §3 服务发现 | `session.liveliness().declare_token("worker/<type>/<id>")`，`--count` 多副本 id 互异 |
 | §4 RPC | `declare_queryable("worker/<type>/<id>/<subject>")` 异步 handler；信封解析 + 分派；流式多次 `query.reply()` |
-| §4.3 异常 | 非流式：`{"result":0.0,"operation":"error","error":...,"worker":...}`；流式：`{"error":...}` 收尾块 |
+| §4.3 异常 | 非流式：`query.reply_err(...)`，载荷 `{"result":0.0,"operation":"error","error":...,"worker":...}`；流式：`{"error":...}` 收尾块（普通 reply） |
 | §5 配置 | `config.rs`：TOML + JSON5 双重解析 |
 | 生命周期 | `publisher.rs` 10s tokio interval 重发布；`tokio::signal` 监听 SIGTERM/SIGINT → DELETE 全部 key |
 

docs/plugin-wire-protocol.md

@@ -81,6 +81,7 @@ chongming/routes/<type>/__meta__      # 插件级元信息（可选，仅在声
 | `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 逐块转发多回复 |
@@ -147,17 +148,33 @@ worker/<type>/<id>/<subject>
 
 **兼容规则**：非信封载荷（不含 `meta`/`data` 键的 JSON 对象）整包视为业务 `data`，按 `method = GET` 处理。
 
+**透传模式（`validate = false` 端点）**：`data` 为请求体的**原始 JSON 值**（网关零解析/校验，
+信封由字节拼接构造，结构不变）。插件收到的 `data` 形态与校验模式一致（由 SDK 解析为 dict）；
+参数类型转换/校验失败由插件侧暴露（框架异常经 `reply_err` → 网关 503）。空/空白请求体兜底为 `{}`。
+
 ### 4.3 响应
 
 | 形态 | 行为 |
 | ---- | ---- |
 | 普通 | 单个回复，载荷为业务结果 JSON 对象 |
-| 异常（非流式） | 单个回复：`{"result": 0.0, "operation": "error", "error": "<异常信息>", "worker": "<worker_id>"}` |
-| 流式 | 对同一 Query **多次 reply** 逐块推送；生成器正常结束 = 流结束（无结束哨兵）；中途异常以 `{"error": "<异常信息>"}` **收尾块**结束流；查询已终止（客户端断开）时停止回复 |
+| 框架异常（非流式） | 单个 **`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`：取首个回复；空回复 / zenoh 层错误 / 超时 → **切换副本重试**（默认重试 1 次）
+- 普通调用 `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
 
@@ -204,6 +221,13 @@ worker/<type>/<id>/<subject>
 - **协议变更分级**：
   - 破坏性变更（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`）触及本规范描述的语义时，必须同步修订本规范
 
 ---

docs/process-plugin.md

@@ -240,6 +240,17 @@ from . import hello
 __all__ = ["hello"]
 ```
 
+**异常语义**（插件作者只需 return / raise，通道由 SDK 处理）：
+
+| 形态 | 插件写法 | 网关行为 |
+| ---- | -------- | -------- |
+| 业务错误 | `return {"code": 1, "error": "..."}` | HTTP 200 透传（业务语义，网关不解析） |
+| 框架异常（非流式） | `raise ValueError(...)` | SDK 经 `reply_err` 通道回复 → 网关重试后 **HTTP 503** |
+| 流式异常 | 生成器内 raise | SDK 发送 `{"error": ...}` 收尾块 → SSE error 事件终止 |
+
+即「**返回即业务、抛出即框架**」。注意：框架异常从历史行为「HTTP 200 + 错误信封」
+修正为 **503**（`reply_err` 通道，见 wire 协议 §4.3 兼容矩阵）。
+
 ### 第 6 步：本地调试
 
 ```bash

src/chongming/app/core/route_generator.py

@@ -5,12 +5,12 @@
 为每个端点生成带有正确 Pydantic 类型注解的处理函数，确保 OpenAPI 文档
 正确反映参数类型、响应结构，并注册到 components/schemas。
 """
-from typing import Any, AsyncGenerator, Dict, List, Optional
+from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple
 from inspect import Parameter, Signature
 
 import orjson
 from fastapi import FastAPI, HTTPException, APIRouter, Query, Path as FastAPIPath, Request, Depends
-from fastapi.responses import StreamingResponse
+from fastapi.responses import StreamingResponse, Response
 from pydantic import create_model, Field, BaseModel
 
 from chongming_worker.route_config import WORKER_CONFIG_DIR, load_worker_routes_from_dir
@@ -94,6 +94,9 @@ class RouteConfig:
         self.cacheable: bool = data.get("cacheable", False)
         self.auth_required: bool = data.get("auth_required", True)
         self.internal: bool = data.get("internal", False)
+        # validate=false 时网关不解析/校验请求体，原样透传（校验职责下沉 worker，
+        # 失败经 reply_err 以 503 暴露）；仅对 POST/PUT/PATCH 非流式端点生效
+        self.validate: bool = data.get("validate", True)
         # 流式端点：以 SSE（text/event-stream）逐块输出
         self.stream: bool = data.get("stream", False)
         # 流式空闲超时（秒）：连续未收到数据块的最大间隔（timeout 为流总寿命上限）
@@ -125,6 +128,7 @@ def register_route_group(
     routes: List[Dict[str, Any]],
     prefix: str = "",
     extra_schemas: Optional[Dict[str, type[BaseModel]]] = None,
+    raw_body_models: Optional[Dict[Tuple[str, str], str]] = None,
 ) -> List[Any]:
     """
     为一个 worker_type 的一组路由描述符创建并挂载 APIRouter。
@@ -139,6 +143,8 @@ def register_route_group(
         routes: 路由描述符列表（TOML 兼容格式，与 load_worker_routes_from_dir 输出一致）
         prefix: 全局路由前缀（如 /api/v1）
         extra_schemas: 额外 schema 收集 dict（GET 请求模型注入 OpenAPI 用）
+        raw_body_models: (path, method) -> 请求模型名 的收集 dict
+            （validate=false 端点：请求体文档经 requestBody $ref 注入，见 _inject_schemas_to_openapi）
 
     Returns:
         List[Any]: 本次新增到 app.router.routes 的 APIRoute 对象列表
@@ -146,7 +152,14 @@ def register_route_group(
     """
     router = APIRouter(prefix=prefix)
     for route_conf in routes:
-        _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas)
+        _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas, raw_body_models)
+
+    # raw_body_models 记录的 (path, method) 为端点声明路径（不含 prefix），
+    # 而 OpenAPI paths 键为最终挂载路径（prefix + path），这里统一补前缀
+    if raw_body_models is not None and prefix:
+        for key in list(raw_body_models.keys()):
+            path, method = key
+            raw_body_models[(prefix + path, method)] = raw_body_models.pop(key)
 
     before = set(id(r) for r in app.router.routes)
     app.include_router(router)
@@ -178,6 +191,8 @@ def register_dynamic_routes(
         return
 
     extra_schemas: Dict[str, type[BaseModel]] = {}
+    # validate=false 端点的 (path, method) -> 请求模型名（OpenAPI requestBody 注入用）
+    raw_body_models: Dict[Tuple[str, str], str] = {}
 
     # 扫描所有 worker 子目录（每个子目录名 = worker_type）
     for worker_dir in sorted(workers_base.iterdir()):
@@ -190,16 +205,16 @@ def register_dynamic_routes(
         if not routes:
             continue
 
-        register_route_group(app, worker_manager, worker_type, routes, prefix, extra_schemas)
+        register_route_group(app, worker_manager, worker_type, routes, prefix, extra_schemas, raw_body_models)
 
     # 将额外 schema 注册到 OpenAPI 的 components/schemas
     if extra_schemas:
-        _inject_schemas_to_openapi(app, lambda: extra_schemas)
+        _inject_schemas_to_openapi(app, lambda: extra_schemas, lambda: raw_body_models)
 
     logger.info(f"动态路由注册完成，worker 配置目录: {WORKER_CONFIG_DIR}")
 
 
-def _inject_schemas_to_openapi(app: FastAPI, schemas_provider):
+def _inject_schemas_to_openapi(app: FastAPI, schemas_provider, raw_body_provider=None):
     """
     将自定义 Pydantic 模型注册到 FastAPI 的 OpenAPI components/schemas 中。
 
@@ -210,6 +225,9 @@ def _inject_schemas_to_openapi(app: FastAPI, schemas_provider):
         schemas_provider: 无参 callable，返回 {name: BaseModel} 字典。
             盘扫路径传静态 dict；process 模式传 RouteRegistry 的动态 dict，
             使路由表变化后重新生成的 OpenAPI 能反映最新 schema。
+        raw_body_provider: 可选无参 callable，返回 {(path, method): model_name} 字典。
+            validate=false 端点未声明 body 参数（FastAPI 不会自动生成 requestBody），
+            这里按 (path, method) 补注入 requestBody $ref，保持 OpenAPI 请求体文档完整。
     """
     original_openapi = app.openapi
 
@@ -224,6 +242,19 @@ def _inject_schemas_to_openapi(app: FastAPI, schemas_provider):
         for name, model in schemas_provider().items():
             if name not in schemas_dict:
                 schemas_dict[name] = model.model_json_schema()
+        # validate=false 端点：补注入 requestBody（请求体文档，模型已注册到 components）
+        if raw_body_provider is not None:
+            for (path, method), model_name in raw_body_provider().items():
+                op = schema.get("paths", {}).get(path, {}).get(method.lower())
+                if op is not None:
+                    op["requestBody"] = {
+                        "required": True,
+                        "content": {
+                            "application/json": {
+                                "schema": {"$ref": f"#/components/schemas/{model_name}"}
+                            }
+                        },
+                    }
         app.openapi_schema = schema
         return schema
 
@@ -249,14 +280,57 @@ def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_t
             if user_info:
                 meta["user_info"] = user_info
             payload = {"meta": meta, "data": body.model_dump()}
-            result = await worker_manager.call(
+            result = await worker_manager.call_raw(
+                worker_type=worker_type,
+                subject=subject,
+                payload=orjson.dumps(payload),
+                timeout=timeout,
+                cacheable=cacheable,
+            )
+            # 回复为 worker 原始字节，零解析直接转发（response_model 仅用于 OpenAPI 文档）
+            return Response(content=result, media_type="application/json")
+        except RuntimeError as e:
+            logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e)
+            raise HTTPException(status_code=503, detail=str(e))
+        except Exception as e:
+            logger.error(f"路由 {path} 调用异常: {e}")
+            raise HTTPException(status_code=500, detail="内部服务器错误")
+
+    return endpoint
+
+
+def _make_raw_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_type: str):
+    """
+    创建 validate=false 的 body 端点：请求体原样透传（网关零解析/校验）。
+
+    信封结构不变（{"meta": ..., "data": <原始 JSON 值>}），通过字节拼接构造，
+    网关既不解析 body 也不转义；参数校验/类型转换职责下沉到 worker
+    （失败经 reply_err 通道以 503 暴露）。空/空白 body 兜底为 {}。
+    请求模型（params TOML 生成）仅用于 OpenAPI 文档（extra_schemas + requestBody 注入）。
+    """
+    path = rc.path
+    subject = rc.subject
+    timeout = rc.timeout
+    http_method = rc.method
+    cacheable = rc.cacheable
+
+    async def endpoint(request: Request) -> Any:
+        try:
+            meta = {"method": http_method}
+            user_info = getattr(request.state, "user", None)
+            if user_info:
+                meta["user_info"] = user_info
+            raw = await request.body()
+            data_part = raw.strip() or b"{}"
+            payload = b'{"meta":' + orjson.dumps(meta) + b',"data":' + data_part + b'}'
+            result = await worker_manager.call_raw(
                 worker_type=worker_type,
                 subject=subject,
                 payload=payload,
                 timeout=timeout,
                 cacheable=cacheable,
             )
-            return result
+            return Response(content=result, media_type="application/json")
         except RuntimeError as e:
             logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e)
             raise HTTPException(status_code=503, detail=str(e))
@@ -289,21 +363,21 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
     # 无参数情况
     if not param_spec_combined:
         async def _simple_endpoint(request: Request) -> Any:
-            # 信封协议: {"meta": {...}, "data": {...}}
+            # 信封协议: {"meta": {...}, "data": {}}
             meta = {"method": http_method}
             user_info = getattr(request.state, "user", None)
             if user_info:
                 meta["user_info"] = user_info
             payload = {"meta": meta, "data": {}}
             try:
-                result = await worker_manager.call(
+                result = await worker_manager.call_raw(
                     worker_type=worker_type,
                     subject=subject,
-                    payload=payload,
+                    payload=orjson.dumps(payload),
                     timeout=timeout,
                     cacheable=cacheable,
                 )
-                return result
+                return Response(content=result, media_type="application/json")
             except RuntimeError as e:
                 logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e)
                 raise HTTPException(status_code=503, detail=str(e))
@@ -323,14 +397,14 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
             meta["user_info"] = user_info
         payload = {"meta": meta, "data": {k: kwargs[k] for k in param_names}}
         try:
-            result = await worker_manager.call(
+            result = await worker_manager.call_raw(
                 worker_type=worker_type,
                 subject=subject,
-                payload=payload,
+                payload=orjson.dumps(payload),
                 timeout=timeout,
                 cacheable=cacheable,
             )
-            return result
+            return Response(content=result, media_type="application/json")
         except RuntimeError as e:
             logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e)
             raise HTTPException(status_code=503, detail=str(e))
@@ -475,6 +549,7 @@ def _register_single_route(
     worker_type: str,
     route_conf: Dict[str, Any],
     extra_schemas: Optional[Dict[str, type[BaseModel]]] = None,
+    raw_body_models: Optional[Dict[Tuple[str, str], str]] = None,
 ):
     """注册单个路由，生成正确的函数签名以便 OpenAPI 正确生成文档"""
     rc = RouteConfig(route_conf)
@@ -537,7 +612,17 @@ def _register_single_route(
         RequestModel = generate_request_model(
             param_spec, request_model_name
         )
+        if rc.validate:
             endpoint = _make_body_endpoint(worker_manager, rc, worker_type, RequestModel)
+        else:
+            # validate=false：请求体原样透传（网关零解析/校验，校验下沉 worker）。
+            # RequestModel 仅用于 OpenAPI 请求体文档：
+            # 经 extra_schemas 注册 components/schemas + requestBody $ref 注入
+            if extra_schemas is not None and request_model_name not in extra_schemas:
+                extra_schemas[request_model_name] = RequestModel
+            if raw_body_models is not None:
+                raw_body_models[(rc.path, rc.method)] = request_model_name
+            endpoint = _make_raw_body_endpoint(worker_manager, rc, worker_type)
     else:
         ResponseModel = generate_response_model(
             rc.response_model, response_model_name

src/chongming/app/core/route_registry.py

@@ -20,7 +20,7 @@
 import asyncio
 import hashlib
 import json
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 import zenoh
 
@@ -63,6 +63,8 @@ class RouteRegistry:
         self._route_objs: Dict[str, List[Any]] = {}
         # OpenAPI 额外 schema（GET 请求模型），随路由表变化
         self._schemas: Dict[str, Any] = {}
+        # validate=false 端点的 (path, method) -> 请求模型名（OpenAPI requestBody 注入用）
+        self._raw_body_models: Dict[Tuple[str, str], str] = {}
         # worker_type -> 防抖任务
         self._debounce: Dict[str, asyncio.Task] = {}
 
@@ -72,7 +74,7 @@ class RouteRegistry:
         self._inline_types: set = set()
 
         # 将动态 schema 注入 OpenAPI（provider 实时读取本注册表）
-        _inject_schemas_to_openapi(app, lambda: self._schemas)
+        _inject_schemas_to_openapi(app, lambda: self._schemas, lambda: self._raw_body_models)
 
     def set_inline_types(self, types: set):
         """设置 inline 插件类型集合（混合模式防双注册）。"""
@@ -235,7 +237,8 @@ class RouteRegistry:
         ]
         if http_routes:
             self._route_objs[wtype] = register_route_group(
-                self._app, self._wm, wtype, http_routes, self._prefix, self._schemas
+                self._app, self._wm, wtype, http_routes, self._prefix,
+                self._schemas, self._raw_body_models,
             )
         else:
             logger.info("[%s] 无 HTTP 路由可挂载（%d 个描述符，全部为 MQTT/internal）",

src/chongming/app/core/worker_manager.py

@@ -1,4 +1,5 @@
 import asyncio
+import hashlib
 import random
 import json
 import orjson
@@ -264,25 +265,78 @@ class WorkerManager:
         cacheable: bool = False,
     ) -> Any:
         """
-        统一调用入口：缓存 > 负载均衡 > 超时重试
+        统一调用入口（dict 语义，MQTT Bridge 等调用方使用）
 
         Args:
             worker_type: Worker 类型标识（如 "example"）
             subject: 调用的主题（如 "calc.add"）
-            payload: 请求参数
+            payload: 请求参数（信封格式，由路由层构造）
             timeout: 单次调用超时（秒）
             retries: 失败重试次数
             cacheable: 是否允许缓存结果（若为 False 则跳过缓存读写）
 
         Returns:
-            Dict[str, Any]: 调用结果
+            Dict[str, Any]: 调用结果（worker 回复解析后的 dict）
+        """
+        cache_key = f"{worker_type}:{subject}:{json.dumps(payload, sort_keys=True)}"
+        reply_bytes = await self._call_impl(
+            worker_type, subject, orjson.dumps(payload),
+            timeout=timeout, retries=retries, cacheable=cacheable, cache_key=cache_key,
+        )
+        return orjson.loads(reply_bytes) if reply_bytes else {}
 
-        Raises:
-            RuntimeError: 无可用 worker 或所有重试均失败
+    async def call_raw(
+        self,
+        worker_type: str,
+        subject: str,
+        payload: bytes,
+        timeout: float = 3.0,
+        retries: int = 1,
+        cacheable: bool = False,
+    ) -> bytes:
+        """
+        统一调用入口（bytes 语义，HTTP 端点透传用）：返回 worker 回复的原始字节。
+
+        与 call() 的区别:
+          - payload 为预序列化的请求字节（普通信封或字节拼接信封），网关不再二次序列化
+          - 回复不做 JSON 解析，直接返回原始字节（配合 Response(content=bytes) 零开销转发）
+          - cacheable 时仅在缓存写入前解析一次做成功检查（业务错误不缓存），
+            缓存与返回均保留原始字节
+
+        Args:
+            worker_type: Worker 类型标识（如 "example"）
+            subject: 调用的主题（如 "calc.add"）
+            payload: 预序列化的请求字节（信封 JSON）
+            timeout: 单次调用超时（秒）
+            retries: 失败重试次数
+            cacheable: 是否允许缓存结果（缓存原始字节）
+
+        Returns:
+            bytes: worker 回复的原始字节（空回复兜底 b"{}"，与 call() 的 {} 语义一致）
+        """
+        cache_key = f"{worker_type}:{subject}:{hashlib.sha256(payload).hexdigest()}"
+        return await self._call_impl(
+            worker_type, subject, payload,
+            timeout=timeout, retries=retries, cacheable=cacheable, cache_key=cache_key,
+        )
+
+    async def _call_impl(
+        self,
+        worker_type: str,
+        subject: str,
+        payload_bytes: bytes,
+        timeout: float = 3.0,
+        retries: int = 1,
+        cacheable: bool = False,
+        cache_key: Optional[str] = None,
+    ) -> bytes:
+        """
+        调用实现：缓存 > 负载均衡 > 超时重试，返回回复的原始字节。
+
+        call() / call_raw() 共用本实现，仅入参出参形态不同（dict ↔ bytes）。
         """
         # 1. 缓存检查（仅在 cacheable=True 时检查）
-        cache_key = f"{worker_type}:{subject}:{json.dumps(payload, sort_keys=True)}"
-        if cacheable and self.cache is not None:
+        if cacheable and cache_key is not None and self.cache is not None:
             if cache_key in self.cache:
                 logger.debug(f"命中缓存: {cache_key}")
                 return self.cache[cache_key]
@@ -293,7 +347,6 @@ class WorkerManager:
             await self._raise_no_worker(worker_type)
 
         full_key = f"{worker_expr}/{subject}"
-        payload_bytes = orjson.dumps(payload)
 
         last_error = None
         tried_workers: List[str] = []
@@ -335,16 +388,14 @@ class WorkerManager:
 
                 reply = first_reply
                 if reply.ok: # type: ignore
-                    # result_str = str(reply.ok.payload)
-                    payload_bytes = bytes(reply.ok.payload) if reply.ok.payload else b"" # type: ignore
-                    if not payload_bytes:
-                        result = {}  # 或根据业务返回默认值
-                    else:
-                        result = orjson.loads(payload_bytes)
-                    # 缓存结果（仅在 cacheable=True 时写入）
-                    if cacheable and self.cache is not None and self.is_success_response(result):
-                        self.cache[cache_key] = result
-                    return result
+                    reply_bytes = bytes(reply.ok.payload) if reply.ok.payload else b"{}"
+                    # 缓存写入（仅在 cacheable=True 时）：解析一次做成功检查，
+                    # 但缓存与返回均保留原始字节（网关侧零二次序列化）
+                    if cacheable and cache_key is not None and self.cache is not None and reply_bytes:
+                        parsed = orjson.loads(reply_bytes)
+                        if self.is_success_response(parsed):
+                            self.cache[cache_key] = reply_bytes
+                    return reply_bytes
                 else:
                     error_payload = reply.err.payload  # type: ignore
                     error_msg = str(error_payload) if error_payload else "unknown error"

test_passthrough_optimization.py

@@ -0,0 +1,376 @@
+"""
+透传优化回归测试（依赖隔离：桩掉 chongming 包链、auth 与 chongming_cache）
+
+覆盖:
+  A. 回复路径 bytes 透传:
+     1. 普通端点返回 worker 原始字节（byte-for-byte，零再序列化）+ application/json
+     2. 框架错误（call_raw 抛 RuntimeError）-> HTTP 503
+     3. 业务错误（{"code":1,...} 普通回复）-> HTTP 200 透传（网关不解析）
+  B. validate=false 请求透传:
+     4. POST 信封字节拼接精确断言（网关零解析/零转义）
+     5. 空 body 兜底 {"meta":...,"data":{}}
+     6. validate=true 端点仍走 Pydantic 校验（422）；validate=false 跳过校验（透传）
+     7. GET 端点不受 validate 影响（validate 仅对 body 端点生效）
+  C. OpenAPI:
+     8. validate=false 端点 requestBody $ref 注入 + components schema 完整
+  D. WorkerManager 单测（FakeSession）:
+     9. call_raw 零解析返回原始字节；call() 仍解析为 dict（MQTT 语义保留）
+     10. cacheable: 缓存原始 bytes、二次命中不再发查询、业务错误不缓存
+     11. reply.err -> 重试 -> RuntimeError；空回复兜底 b"{}"
+
+运行: uv run python test_passthrough_optimization.py
+"""
+import asyncio
+import importlib.util
+import logging
+import sys
+import types
+from pathlib import Path
+from types import SimpleNamespace
+
+import orjson
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+REPO_ROOT = Path(__file__).resolve().parent
+sys.path.insert(0, str(REPO_ROOT / "src"))
+sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
+
+# ---- 依赖隔离：桩掉 chongming 包链 / auth / chongming_cache ----
+_pkg = types.ModuleType("chongming")
+_pkg.__path__ = []
+_app = types.ModuleType("chongming.app")
+_app.__path__ = []
+_core = types.ModuleType("chongming.app.core")
+_core.__path__ = []
+_logger_stub = types.ModuleType("chongming.app.core.logger")
+_logger_stub.get_logger = lambda name: logging.getLogger(name)
+_auth_stub = types.ModuleType("chongming.app.core.auth")
+_auth_stub.store_user_to_request = lambda x: x  # 仅用于 Depends，桩掉鉴权
+_cache_stub = types.ModuleType("chongming_cache")
+_cache_stub.Cache = object
+for mod in (_pkg, _app, _core, _logger_stub, _auth_stub, _cache_stub):
+    sys.modules[mod.__name__] = mod
+
+# 加载真实 worker_manager（route_generator 依赖其 StreamIdleTimeoutError）
+_wm_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "worker_manager.py"
+_wm_spec = importlib.util.spec_from_file_location(
+    "chongming.app.core.worker_manager", _wm_path
+)
+_wm_mod = importlib.util.module_from_spec(_wm_spec)
+sys.modules["chongming.app.core.worker_manager"] = _wm_mod
+_wm_spec.loader.exec_module(_wm_mod)  # type: ignore[union-attr]
+
+# 加载真实 route_generator
+_rg_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "route_generator.py"
+_rg_spec = importlib.util.spec_from_file_location(
+    "chongming.app.core.route_generator", _rg_path
+)
+_rg_mod = importlib.util.module_from_spec(_rg_spec)
+sys.modules["chongming.app.core.route_generator"] = _rg_mod
+_rg_spec.loader.exec_module(_rg_mod)  # type: ignore[union-attr]
+
+register_route_group = _rg_mod.register_route_group
+_inject_schemas_to_openapi = _rg_mod._inject_schemas_to_openapi
+WorkerManager = _wm_mod.WorkerManager
+
+failures = []
+
+
+def check(name, cond, detail=""):
+    status = "PASS" if cond else "FAIL"
+    print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
+    if not cond:
+        failures.append(name)
+
+
+class FakeWorkerManager:
+    """模拟 WorkerManager：call_raw 返回原始字节；error="no_worker" 时抛 RuntimeError"""
+
+    def __init__(self, reply=b'{"result": 5}', error=None):
+        self.reply = reply
+        self.error = error
+        self.call_raw_payloads = []  # 捕获的预序列化请求字节
+
+    async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False):
+        self.call_raw_payloads.append(payload)
+        if self.error == "no_worker":
+            raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。")
+        return self.reply
+
+
+ROUTES = [
+    # validate=true 对照（默认）
+    {
+        "subject": "demo.echo", "path": "/echo", "method": "POST",
+        "params": ["msg: str"], "summary": "echo",
+        "timeout": 3.0, "auth_required": False,
+    },
+    {
+        "subject": "demo.echo2", "path": "/echo2", "method": "POST",
+        "params": ["n: int"], "summary": "typed echo",
+        "timeout": 3.0, "auth_required": False,
+    },
+    {
+        "subject": "demo.add", "path": "/add", "method": "GET",
+        "params": ["a: int", "b: int"], "summary": "add",
+        "timeout": 3.0, "auth_required": False,
+    },
+    # validate=false：请求体原样透传
+    {
+        "subject": "demo.raw", "path": "/raw/echo", "method": "POST",
+        "params": ["msg: str"], "summary": "raw echo",
+        "validate": False, "timeout": 3.0, "auth_required": False,
+    },
+    # validate=false + GET 对照（validate 仅对 body 端点生效）
+    {
+        "subject": "demo.rawget", "path": "/raw/get", "method": "GET",
+        "params": ["a: int"], "summary": "raw get",
+        "validate": False, "timeout": 3.0, "auth_required": False,
+    },
+]
+
+
+def build_app(fwm):
+    app = FastAPI()
+    extra_schemas = {}
+    raw_body_models = {}
+    register_route_group(
+        app, fwm, "demo", ROUTES, prefix="/api/v1",
+        extra_schemas=extra_schemas, raw_body_models=raw_body_models,
+    )
+    if extra_schemas:
+        _inject_schemas_to_openapi(app, lambda: extra_schemas, lambda: raw_body_models)
+    return TestClient(app)
+
+
+# ---- A. 回复路径 bytes 透传 ----
+
+def test_reply_bytes_passthrough():
+    fwm = FakeWorkerManager(reply=b'{"result": 5}')
+    client = build_app(fwm)
+    r = client.get("/api/v1/add?a=1&b=2")
+    check("GET 状态码 200", r.status_code == 200, str(r.status_code))
+    check(
+        "回复为 worker 原始字节（byte-for-byte 零再序列化）",
+        r.content == b'{"result": 5}',
+        repr(r.content),
+    )
+    check(
+        "Content-Type application/json",
+        r.headers["content-type"].startswith("application/json"),
+        r.headers.get("content-type", ""),
+    )
+    data = orjson.loads(fwm.call_raw_payloads[0])["data"]
+    check("query 参数透传到信封 data", data == {"a": 1, "b": 2}, str(data))
+
+
+def test_framework_error_503():
+    fwm = FakeWorkerManager(error="no_worker")
+    client = build_app(fwm)
+    r = client.get("/api/v1/add?a=1&b=2")
+    check("框架错误（无 worker）-> 503", r.status_code == 503, str(r.status_code))
+
+
+def test_business_error_passthrough_200():
+    fwm = FakeWorkerManager(reply=b'{"code": 1, "error": "nope"}')
+    client = build_app(fwm)
+    r = client.post("/api/v1/echo", json={"msg": "hi"})
+    check("业务错误（普通回复）-> 200 透传", r.status_code == 200, str(r.status_code))
+    check("业务错误体原样透传", r.content == b'{"code": 1, "error": "nope"}', repr(r.content))
+
+
+# ---- B. validate=false 请求透传 ----
+
+def test_validate_false_envelope_splice():
+    fwm = FakeWorkerManager()
+    client = build_app(fwm)
+    r = client.post("/api/v1/raw/echo", json={"msg": "hello"})
+    check("validate=false POST 状态码 200", r.status_code == 200, str(r.status_code))
+    expected = b'{"meta":{"method":"POST"},"data":{"msg":"hello"}}'
+    check(
+        "信封字节拼接（网关零解析/零转义）",
+        fwm.call_raw_payloads[0] == expected,
+        repr(fwm.call_raw_payloads[0]),
+    )
+    check("响应透传", r.content == b'{"result": 5}', repr(r.content))
+
+
+def test_validate_false_empty_body():
+    fwm = FakeWorkerManager()
+    client = build_app(fwm)
+    r = client.post("/api/v1/raw/echo", content=b"")
+    check("空 body 状态码 200", r.status_code == 200, str(r.status_code))
+    check(
+        "空 body 兜底 data:{}",
+        fwm.call_raw_payloads[0] == b'{"meta":{"method":"POST"},"data":{}}',
+        repr(fwm.call_raw_payloads[0]),
+    )
+
+
+def test_validate_true_still_422():
+    fwm = FakeWorkerManager()
+    client = build_app(fwm)
+    r = client.post("/api/v1/echo2", json={"n": "not-a-number"})
+    check("validate=true 仍 Pydantic 校验（422）", r.status_code == 422, str(r.status_code))
+    check("校验失败不发起调用", len(fwm.call_raw_payloads) == 0)
+
+
+def test_validate_false_skips_validation():
+    fwm = FakeWorkerManager()
+    client = build_app(fwm)
+    r = client.post("/api/v1/raw/echo", json={"n": "not-a-number"})
+    check("validate=false 跳过校验（200）", r.status_code == 200, str(r.status_code))
+    check(
+        "原始 body 透传（含类型错误输入）",
+        fwm.call_raw_payloads[0] == b'{"meta":{"method":"POST"},"data":{"n":"not-a-number"}}',
+        repr(fwm.call_raw_payloads[0]),
+    )
+
+
+def test_validate_false_get_unaffected():
+    fwm = FakeWorkerManager()
+    client = build_app(fwm)
+    r = client.get("/api/v1/raw/get?a=5")
+    check("validate=false GET 不受影响（200）", r.status_code == 200, str(r.status_code))
+    data = orjson.loads(fwm.call_raw_payloads[0])["data"]
+    check("GET 参数正常透传", data == {"a": 5}, str(data))
+
+
+# ---- C. OpenAPI ----
+
+def test_openapi_request_body_injected():
+    fwm = FakeWorkerManager()
+    client = build_app(fwm)
+    schema = client.app.openapi()
+
+    # validate=false 端点：requestBody $ref 注入
+    raw_op = schema["paths"]["/api/v1/raw/echo"]["post"]
+    check(
+        "validate=false 端点 requestBody 已注入",
+        raw_op.get("requestBody", {}).get("content", {}).get("application/json", {}).get("schema", {}).get("$ref")
+        == "#/components/schemas/demo_raw_echo_POST_Request",
+        str(raw_op.get("requestBody")),
+    )
+    check(
+        "请求模型注册到 components/schemas",
+        "demo_raw_echo_POST_Request" in schema["components"]["schemas"],
+    )
+    # validate=true 端点：FastAPI 原生 requestBody（不受影响）
+    ok_op = schema["paths"]["/api/v1/echo"]["post"]
+    check(
+        "validate=true 端点原生 requestBody 保留",
+        "requestBody" in ok_op,
+        str(ok_op.get("requestBody")),
+    )
+    # GET 端点无 requestBody
+    check(
+        "GET 端点无 requestBody",
+        "requestBody" not in schema["paths"]["/api/v1/raw/get"]["get"],
+    )
+
+
+# ---- D. WorkerManager 单测（FakeSession）----
+
+class FakeSession:
+    """模拟 zenoh Session：get() 返回单回复迭代器"""
+
+    def __init__(self, payload=b'{"result": 5}', err_payload=None):
+        self.payload = payload
+        self.err_payload = err_payload
+        self.get_calls = 0
+
+    def get(self, *args, **kwargs):
+        self.get_calls += 1
+        if self.err_payload is not None:
+            reply = SimpleNamespace(ok=None, err=SimpleNamespace(payload=self.err_payload))
+        else:
+            reply = SimpleNamespace(ok=SimpleNamespace(payload=self.payload), err=None)
+        return iter([reply])
+
+
+def make_wm(session, cache=None):
+    wm = WorkerManager(session, cache=cache if cache is not None else {})
+    wm._workers_cache = {"demo": ["worker/demo/w1"]}
+    return wm
+
+
+async def test_wm_call_raw_no_parse():
+    session = FakeSession(payload=b'{"result": 5}')
+    wm = make_wm(session)
+    result = await wm.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}')
+    check(
+        "call_raw 返回原始字节（零解析）",
+        isinstance(result, bytes) and result == b'{"result": 5}',
+        repr(result),
+    )
+    check("session.get 仅调用 1 次", session.get_calls == 1, str(session.get_calls))
+
+
+async def test_wm_call_dict_semantics():
+    session = FakeSession(payload=b'{"result": 5}')
+    wm = make_wm(session)
+    result = await wm.call("demo", "calc.add", {"meta": {}, "data": {}})
+    check("call() 仍解析为 dict（MQTT 语义保留）", result == {"result": 5}, repr(result))
+
+
+async def test_wm_cache_bytes():
+    cache = {}
+    session = FakeSession(payload=b'{"result": 5}')
+    wm = make_wm(session, cache)
+    r1 = await wm.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}', cacheable=True)
+    r2 = await wm.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}', cacheable=True)
+    check("缓存命中返回原始 bytes", r1 == r2 == b'{"result": 5}', repr(r1))
+    check("二次调用命中缓存（不再发查询）", session.get_calls == 1, str(session.get_calls))
+    check("缓存值为原始 bytes", all(isinstance(v, bytes) for v in cache.values()))
+
+    # 业务错误（code!=0）不缓存
+    session2 = FakeSession(payload=b'{"code": 1, "error": "x"}')
+    wm2 = make_wm(session2, {})
+    await wm2.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}', cacheable=True)
+    check("业务错误不写入缓存", len(wm2.cache) == 0, str(wm2.cache))
+
+
+async def test_wm_reply_err_retry():
+    session = FakeSession(err_payload=b'{"error": "boom"}')
+    wm = make_wm(session)
+    try:
+        await wm.call_raw("demo", "calc.add", b'{}')
+        check("reply.err 应抛 RuntimeError", False)
+    except RuntimeError as e:
+        check("reply.err 重试后抛 RuntimeError", "boom" in str(e), str(e))
+    check("reply.err 触发重试（2 次查询）", session.get_calls == 2, str(session.get_calls))
+
+
+async def test_wm_empty_reply_fallback():
+    session = FakeSession(payload=b"")
+    wm = make_wm(session)
+    check("call_raw 空回复兜底 b\"{}\"", await wm.call_raw("demo", "calc.add", b"{}") == b"{}")
+    check("call() 空回复 → {}", await wm.call("demo", "calc.add", {"meta": {}, "data": {}}) == {})
+
+
+def run():
+    # HTTP 层（TestClient）
+    test_reply_bytes_passthrough()
+    test_framework_error_503()
+    test_business_error_passthrough_200()
+    test_validate_false_envelope_splice()
+    test_validate_false_empty_body()
+    test_validate_true_still_422()
+    test_validate_false_skips_validation()
+    test_validate_false_get_unaffected()
+    test_openapi_request_body_injected()
+
+    # WorkerManager 层（FakeSession）
+    asyncio.run(test_wm_call_raw_no_parse())
+    asyncio.run(test_wm_call_dict_semantics())
+    asyncio.run(test_wm_cache_bytes())
+    asyncio.run(test_wm_reply_err_retry())
+    asyncio.run(test_wm_empty_reply_fallback())
+
+    print(f"\n结果: {len(failures)} 个失败")
+    return 1 if failures else 0
+
+
+if __name__ == "__main__":
+    sys.exit(run())

test_route_registry.py

@@ -31,11 +31,14 @@ def check(name, cond, detail=""):
 
 
 class FakeWorkerManager:
-    """桩：无可用 worker，命中路由后 call 抛 RuntimeError（网关转 503）。"""
+    """桩：无可用 worker，命中路由后 call_raw 抛 RuntimeError（网关转 503）。"""
 
     async def call(self, **kwargs):
         raise RuntimeError("无可用 worker（测试桩）")
 
+    async def call_raw(self, **kwargs):
+        raise RuntimeError("无可用 worker（测试桩）")
+
 
 class FakeBridge:
     """记录 update_dynamic_topics 的调用结果。"""

test_stream_route.py

@@ -86,6 +86,12 @@ class FakeWorkerManager:
         data = payload["data"]
         return {"result": data.get("a", 0) + data.get("b", 0)}
 
+    async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False):
+        # 网关透传路径：payload 为预序列化字节，回复原始字节
+        self.call_payloads.append((worker_type, subject, orjson.loads(payload)))
+        data = orjson.loads(payload)["data"]
+        return orjson.dumps({"result": data.get("a", 0) + data.get("b", 0)})
+
     async def call_stream(self, worker_type, subject, payload, timeout=300.0, idle_timeout=10.0):
         if self.stream_error == "no_worker":
             raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。")
@@ -203,7 +209,7 @@ def test_non_stream_compat():
           f"status={r.status_code} body={r.text}")
 
     check("非流式调用未进入 call_stream", len(fwm.stream_payloads) == 0)
-    check("非流式调用仍走 call()", len(fwm.call_payloads) == 2)
+    check("非流式调用仍走 call_raw", len(fwm.call_payloads) == 2)
 
 
 def run():