modify: README 标注 workers/ 为本地开发目录

.gitignore

@@ -19,6 +19,9 @@ output.md
 # plugins
 src/plugins/
 
+# 进程插件（开发期本地目录；生产在独立容器部署，不入库）
+workers/
+
 # chongming-web
 src/chongming-web
 

README.md

@@ -53,11 +53,13 @@ chongming-zenoh/
 │   │       ├── api/generated/    # 自动生成 API 客户端
 │   │       ├── views/            # 页面组件
 │   │       └── router/           # Vue Router 配置
-│   └── plugins/                  # Worker 插件源码
+│   └── plugins/                  # Worker 插件源码（inline 模式）
 │       ├── auth/                 # JWT 鉴权
 │       ├── doubao/               # 豆包 AI（支持 AICC 安全通道）
 │       ├── aibox_sophon_adapter/ # 设备心跳 + 告警
 │       └── example/              # 演示示例
+├── workers/                      # 进程插件（external 模式，本地开发目录，不入库；
+│   └── demo/                     #   生产在独立容器部署，示例见本地目录）
 ├── utils/                        # 独立工具包
 │   ├── bytedance/   # 字节跳动安全通信 SDK
 │   ├── config/      # TOML 配置加载
@@ -79,6 +81,53 @@ chongming-zenoh/
 
 ---
 
+## 多仓库结构与开发工作流
+
+本仓库（**chongming-core**）只跟踪网关核心代码，其余资产分布在独立仓库中：
+
+| 目录 / 包        | 独立仓库                                        | 说明                       |
+| ---------------- | ----------------------------------------------- | -------------------------- |
+| `src/chongming/` | 本仓库（chongming-core）                        | 网关核心（FastAPI + Zenoh）|
+| `utils/<name>/`  | `chongming-utils-*`（9 个，见「工具包列表」）    | 工具包，由 `uv sync` 按 git 源安装 |
+| `src/plugins/`   | `chongming-plugins`                             | Worker 插件源码            |
+| `config/`        | `chongming-config`                              | 应用与 Worker 路由配置     |
+| `src/chongming-web/` | `chongming-web`                             | Vue 3 前端                 |
+
+### 首次克隆（开发环境引导）
+
+```bash
+git clone http://124.71.68.6:3000/chongming/chongming-core.git chongming-zenoh
+cd chongming-zenoh
+git clone http://124.71.68.6:3000/chongming/chongming-plugins.git src/plugins
+git clone http://124.71.68.6:3000/chongming/chongming-config.git config
+git clone http://124.71.68.6:3000/chongming/chongming-web.git src/chongming-web
+uv sync     # utils/* 由 uv 按 pyproject 的 git 源自动安装，无需手工 clone
+```
+
+### 开发与发布检查清单
+
+**修改 `utils/<name>`（工具包）**：
+
+1. 在 `utils/<name>` 内提交并 push 到对应远程
+2. 回主仓库执行 `uv lock --upgrade-package <包名>` 刷新版本 pin
+3. `uv sync` 后运行相关测试
+4. 主仓库提交 `uv.lock`
+
+**修改 `src/plugins/` / `config/` / `src/chongming-web/`**：
+
+1. 在对应子仓库内提交并 push
+2. 回主仓库启动验证（`uv run serve` + 相关测试）
+
+**发布前总检查**：
+
+- [ ] 所有子仓库 `git status` 干净且已 push
+- [ ] `uv.lock` 与各 git 源最新提交一致
+- [ ] 配置中无硬编码凭据——密钥一律走环境变量：
+      `VOLC_ACCESSKEY`（豆包 Ark Key）、`AIBOX_MQTT_URI`（AI Box 外部 Broker URI）
+- [ ] `uv run python test_mqtt_e2e.py` 等测试通过
+
+---
+
 ## 快速开始
 
 ### 环境要求
@@ -91,6 +140,8 @@ chongming-zenoh/
 
 ### 安装与启动
 
+> 首次克隆：先按「多仓库结构与开发工作流」拉取 `src/plugins` / `config` / `src/chongming-web` 子仓库，再执行以下步骤。
+
 ```bash
 # 同步依赖（自动编译 Rust 扩展 + 安装 Python 包）
 uv sync
@@ -291,6 +342,7 @@ uv run serve                                    # 启动开发服务
 uv run gen_plugin <name>                        # 生成统一风格插件骨架
 uv run python -m chongming.scripts.gen_models   # 生成 Pydantic 模型
 uv run python -m chongming.scripts.build        # 构建打包（PyInstaller/Nuitka）
+uv run worker_serve --type <name>               # 以独立进程运行 Worker（load_mode=external）
 uv pip install -e utils/<name>                  # 安装/更新工具包
 ```
 
@@ -304,6 +356,7 @@ uv pip install -e utils/<name>                  # 安装/更新工具包
 | **doubao**               | `src/plugins/doubao/`               | 豆包视觉理解，支持 AICC 安全通信通道                    |
 | **aibox_sophon_adapter** | `src/plugins/aibox_sophon_adapter/` | AI Box 设备心跳监控、离线检测、告警上报与查询、视频上传 |
 | **example**              | `src/plugins/example/`              | 四则运算 + 心跳演示（含 MQTT 端点示例）                 |
+| **demo**（进程插件示例） | `workers/demo/`（本地目录，不入库） | 独立进程运行（load_mode=external），经 Zenoh 注册路由 |
 
 ---
 
@@ -325,6 +378,19 @@ uv pip install -e utils/<name>                  # 安装/更新工具包
 ## 配置说明
 
 - **应用配置**：根目录 `config.toml`，包含环境、服务器、安全、MQTT 等全局设置
+- **插件加载模式**：按插件粒度声明在 `config/workers/<name>/config.toml` 的 `[workers]` 节
+  - `load_mode = "inline"`（默认）：插件在网关进程内加载
+  - `load_mode = "external"`：插件以独立进程运行（进程编排交给 Docker/K8s，
+    进程入口 `uv run worker_serve --type <name>`），经 Zenoh 在 `chongming/routes/**`
+    注册路由，网关动态挂载 HTTP 端点
+  - 两种模式可同时存在（混合部署）
+- **进程插件与网关完全隔离**：进程插件只依赖 `chongming-worker` SDK + zenoh，
+  与网关（`chongming` 包）零导入关系，仅通过 Zenoh（`chongming/routes/**` 注册路由、
+  `worker/<type>/<id>` 服务发现）通讯。`workers/` 目录仅为开发便利，
+  **生产部署时进程插件与网关分别打包到不同容器**：
+  - 网关容器：`chongming`（含 `src/chongming/`）+ 依赖
+  - 进程插件容器：插件代码（`workers/<name>/`）+ `chongming-worker` + 依赖，
+    入口 `chongming-worker --type <name>`（或 `uv run worker_serve --type <name>`）
 - **Zenoh 配置**：`config/zenoh_config.json5`
 - **插件注册**：`config/workers/<name>/config.toml` 的 `[workers]` 节，定义加载方式、副本数与运行时配置（旧全局 `config/workers/config.toml` 仅作兜底兼容）
 - **路由定义**：`config/workers/<name>/*.toml`，每个端点一个文件

pyproject.toml

@@ -68,3 +68,4 @@ gen_models = "chongming.scripts.gen_models:gen_models"
 gen_plugin = "chongming.scripts.gen_plugin:gen_plugin"
 gunicorn = "chongming:gunicorn_serve"
 serve = "chongming:serve"
+worker_serve = "chongming_worker.process:worker_serve"

src/chongming/__init__.py

@@ -22,6 +22,7 @@ def serve():
     logger.info(f"启动日志: {log_config}")
     uvicorn.run("chongming.app:app", **server_config)
 
+
 def gunicorn_serve():
     """
     使用 gunicorn 启动服务

src/chongming/app/__init__.py

@@ -47,14 +47,24 @@ async def lifespan(app: FastAPI):
     # 4. WorkerManager
     manager = lifespan_setup.setup_worker_manager(app, session)
 
-    # 5. 加载 workers（同时自动发现插件自定义路由）
+    # 5. 插件加载（混合模式：inline 进程内加载；external 由独立进程运行，
-    worker_tasks, custom_routers = await lifespan_setup.load_workers(app, session, manager)
+    #    其路由经 Zenoh 动态注册——进程编排交给 Docker/K8s，网关不拉起进程）
+    worker_tasks, custom_routers, inline_types, external_types = (
+        await lifespan_setup.load_workers(app, session, manager)
+    )
+    if external_types:
+        # 注册表为 external 插件服务；inline 类型集合防止误报重复挂载
+        await lifespan_setup.setup_route_registry(
+            app, session, manager, inline_types=inline_types
+        )
 
-    # 5.5 契约校验（TOML 路由 ↔ @rpc_method ↔ 生成模型），失败则启动终止
+    # 5.5 契约校验（仅 inline 插件；external 插件在自己进程内校验），失败则启动终止
     lifespan_setup.validate_contracts(app)
 
-    # 6. 注册路由（含 TOML 动态路由 + 插件自定义路由）
+    # 6. 注册路由（inline 插件的 TOML 动态路由 + 插件自定义路由）
-    await lifespan_setup.register_routes(app, manager, custom_routers)
+    await lifespan_setup.register_routes(
+        app, manager, custom_routers, worker_types=inline_types
+    )
 
     # 7. 启动 MQTT Bridge（内嵌 Broker + Worker 桥接）
     await lifespan_setup.setup_mqtt(app, manager)
@@ -78,7 +88,12 @@ async def lifespan(app: FastAPI):
     # 1. 停止 MQTT Bridge
     await lifespan_setup.shutdown_mqtt(app)
 
-    # 2. 清理 workers（调用 cleanup）
+    # 1.5 停止路由注册表订阅（存在 external 插件时）
+    registry = getattr(app.state, "route_registry", None)
+    if registry is not None:
+        registry.stop()
+
+    # 2. 清理 inline workers（cleanup_workers 内部按 load_mode 过滤）
     await lifespan_setup.cleanup_workers(app)
 
     # 3. 取消刷新任务

src/chongming/app/core/contracts.py

@@ -1,78 +1,31 @@
 """
-启动契约校验（config/workers TOML ↔ @rpc_method 注册表 ↔ gen_models 生成模型）
+启动契约校验（网关 inline 模式）
 
-挂载于 lifespan 的 load_workers 与 register_routes 之间，此时插件模块已导入
+校验逻辑的唯一实现在 chongming_worker.contracts（网关与 worker 进程共享），
-（@rpc_method 注册表就绪）且 TOML 路由尚未注册到 FastAPI。
+本模块只做网关侧的编排：对 inline 插件批量校验、日志汇总、硬失败终止启动。
 
-策略:
+external 插件在自己进程内做同样校验（validate_worker_contracts），网关侧跳过。
-  - 硬失败（汇总后 raise，启动终止）:
-    HTTP 路由（GET/POST/PUT/DELETE/PATCH 且 internal != true）在注册表中
-    缺少对应 subject#METHOD 的 @rpc_method → 该路由运行时必然 503，提前失败。
-  - 软警告（仅日志汇总，不阻断启动）:
-    [MQTT] / internal 路由缺少注册、已注册方法无 TOML 路由、
-    TOML 声明 params/response_model 但生成模型缺失（gen_models 过期）、孤儿模型。
 """
-import importlib
+from typing import Any, Dict, List
-from typing import Any, Dict, List, Optional, Set
 
 from chongming_worker import get_all_rpc_methods
+from chongming_worker.contracts import check_plugin, load_plugin_models
+from chongming_worker.route_config import load_worker_routes_from_dir
 
-from .route_generator import load_worker_routes_from_dir
 from .logger import get_logger
 
 logger = get_logger("contracts")
 
-# 参与 HTTP 路由注册的方法（与 route_generator 的注册行为一致）
-HTTP_METHODS = ("GET", "POST", "PUT", "DELETE", "PATCH")
-
-
-def _model_name(base: str, method: str, suffix: str) -> str:
-    """与 gen_models._model_name 保持一致的模型命名: <Op><Method><Suffix>"""
-    method_title = "Mqtt" if method == "MQTT" else method.title()
-    return f"{base.title().replace('-', '').replace('_', '')}{method_title}{suffix}"
-
-
-def _load_plugin_models(worker_type: str) -> Optional[Set[str]]:
-    """
-    加载插件 models.pydantic 中所有 Pydantic 模型的类名集合。
-
-    返回 None 表示模块不存在（插件无生成模型，跳过模型相关校验）。
-    注: gen_models 生成的 `<Plugin>Response` 联合类型——多模型时为 UnionType
-    （自动被 isinstance(type) 过滤），单模型时是类的别名（与真实模型同一对象，
-    按对象去重排除），两者都不会被误报为孤儿模型。
-    """
-    from pydantic import BaseModel
-
-    for mod_name in (
-        f"plugins.{worker_type}.models.pydantic",
-        f"src.plugins.{worker_type}.models.pydantic",
-    ):
-        try:
-            mod = importlib.import_module(mod_name)
-        except ImportError:
-            continue
-        names: Set[str] = set()
-        seen_objs: Set[int] = set()
-        for name in dir(mod):
-            obj = getattr(mod, name)
-            if (
-                isinstance(obj, type)
-                and issubclass(obj, BaseModel)
-                and obj is not BaseModel
-                and id(obj) not in seen_objs
-            ):
-                seen_objs.add(id(obj))
-                names.add(name)
-        return names
-    return None
-
 
 def validate_contracts(plugins: List[Dict[str, Any]]) -> None:
     """
-    对每个发现的插件执行三方一致性校验。
+    对每个发现的 inline 插件执行三方一致性校验。
+
+    挂载于 lifespan 的 load_workers 与 register_routes 之间，此时插件模块已导入
+    （@rpc_method 注册表就绪）且 TOML 路由尚未注册到 FastAPI。
 
     Args:
-        plugins: _discover_plugins() 返回的插件配置列表
+        plugins: _discover_plugins() 返回的 inline 插件配置列表
 
     Raises:
         RuntimeError: 存在硬失败项（HTTP 路由缺失 @rpc_method 注册）时
@@ -88,67 +41,10 @@ def validate_contracts(plugins: List[Dict[str, Any]]) -> None:
 
         toml_routes = load_worker_routes_from_dir(worker_type)
         registered = registry.get(worker_type, {})
-        toml_keys = {f"{r['subject']}#{r['method']}" for r in toml_routes}
+        model_names = load_plugin_models(worker_type)
-        model_names = _load_plugin_models(worker_type)
+        hard, warns = check_plugin(worker_type, toml_routes, registered, model_names)
-
+        hard_errors.extend(hard)
-        for r in toml_routes:
+        warnings.extend(warns)
-            key = f"{r['subject']}#{r['method']}"
-            route_desc = f"{r['method']} {r['path']} -> subject '{r['subject']}'"
-
-            if key in registered:
-                continue
-
-            if r["method"] in HTTP_METHODS and not r.get("internal", False):
-                # 硬失败：运行时必然 503
-                hint = ""
-                if not registered:
-                    hint = "（该插件注册表为空，请检查启动日志中的 '加载 Worker 插件失败' 错误）"
-                hard_errors.append(
-                    f"[{worker_type}] {route_desc} 缺少对应的 @rpc_method 注册{hint}"
-                )
-            else:
-                # 软警告：MQTT / internal 端点缺失注册
-                warnings.append(
-                    f"[{worker_type}] {route_desc} 无对应的 @rpc_method(method='{r['method']}') 注册"
-                )
-
-        # 已注册方法无 TOML 路由（反向漂移检测）
-        for key in registered:
-            if key not in toml_keys:
-                warnings.append(
-                    f"[{worker_type}] 已注册 '{key}' 但无对应 TOML 路由定义"
-                )
-
-        # 模型一致性（仅当插件存在生成模型时校验）
-        if model_names is not None:
-            referenced: Set[str] = set()
-            for r in toml_routes:
-                route_name = r.get("route_name", "")
-                if not route_name:
-                    continue
-                method = r["method"]
-                if r.get("params"):
-                    expected = _model_name(route_name, method, "Request")
-                    referenced.add(expected)
-                    if expected not in model_names:
-                        warnings.append(
-                            f"[{worker_type}] '{r['subject']}' [{method}] 声明了 params，"
-                            f"但模型 {expected} 不存在（可能未运行 gen_models）"
-                        )
-                if r.get("response_model"):
-                    expected = _model_name(route_name, method, "Response")
-                    referenced.add(expected)
-                    if expected not in model_names:
-                        warnings.append(
-                            f"[{worker_type}] '{r['subject']}' [{method}] 声明了 response_model，"
-                            f"但模型 {expected} 不存在（可能未运行 gen_models）"
-                        )
-
-            for name in sorted(model_names - referenced):
-                warnings.append(
-                    f"[{worker_type}] 模型 {name} 未被任何 TOML 路由引用"
-                    f"（孤儿模型，可重新运行 gen_models 清理）"
-                )
 
     # 输出结果
     if warnings:

src/chongming/app/core/lifespan_setup.py

@@ -222,6 +222,39 @@ def setup_worker_manager(app: FastAPI, session: zenoh.Session) -> WorkerManager:
     app.state.refresh_task = refresh_task
     return manager
 
+async def setup_route_registry(
+    app: FastAPI,
+    session: zenoh.Session,
+    manager: WorkerManager,
+    inline_types: list[str] | None = None,
+):
+    """
+    启动动态路由注册表（订阅 chongming/routes/**）。
+
+    存在 load_mode="external" 的插件时启动：external 插件以独立进程运行
+    （由 Docker/K8s 等编排拉起，网关不负责进程管理），其路由描述符经 Zenoh
+    动态注册；inline 插件仍在网关进程内加载，两者同时存在（混合模式）。
+
+    inline_types: 网关进程内加载的插件类型，注册表忽略其描述符以防重复挂载。
+
+    注意：external 插件的自定义 APIRouter（如 multipart 上传端点）无法由
+    网关发现（网关不 import 该插件），此类端点需保持 inline 或等 P2 专门通道。
+    """
+    from ..core.route_registry import RouteRegistry
+
+    prefix = config["default"]["prefix"]
+    registry = RouteRegistry(app, manager, prefix=prefix)
+    if inline_types:
+        registry.set_inline_types(set(inline_types))
+    await registry.start(session)
+    app.state.route_registry = registry
+    logger.info(
+        f"路由注册表已启动（prefix={prefix}, 外部插件路由动态注册, "
+        f"inline 插件: {sorted(inline_types or []) or '无'}）"
+    )
+    return registry
+
+
 def _discover_plugins() -> list[dict]:
     """
     发现所有插件配置。
@@ -250,6 +283,10 @@ def _discover_plugins() -> list[dict]:
             continue
 
         plugin_type = workers_section.get("type", plugin_type)
+        # 插件运行模式（按插件粒度，混合共存）:
+        #   "inline"（默认） = 网关进程内加载
+        #   "external"       = 独立进程运行（Docker/K8s 编排），经 Zenoh 注册路由
+        load_mode = workers_section.get("load_mode", "inline")
         # 插件声明的 MQTT Bridge 额外订阅主题（[[workers.mqtt_topics]]）
         # 网关级路由配置，不传给插件运行时；worker 目标即本插件类型。
         mqtt_topics = []
@@ -266,7 +303,7 @@ def _discover_plugins() -> list[dict]:
         #      — 其中 type 被 chongming_worker.runtime.create_workers 用于匹配 RPC 方法
         #   3. [workers.config] 展平到顶层 (api_key, data_dir, heartbeat 等)
         workers_flat = {k: v for k, v in workers_section.items()
-                        if k not in ("config", "mqtt_topics")}
+                        if k not in ("config", "mqtt_topics", "load_mode")}
         runtime_config = {
             **plugin_conf,
             **workers_flat,
@@ -279,6 +316,7 @@ def _discover_plugins() -> list[dict]:
             "factory": workers_section.get("factory", "create_workers"),
             "cleanup": workers_section.get("cleanup", "cleanup"),
             "count": workers_section.get("count", 1),
+            "load_mode": load_mode,
             "mqtt_topics": mqtt_topics,
             "config": runtime_config,
         })
@@ -306,29 +344,43 @@ def _discover_plugins() -> list[dict]:
 
 async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerManager):
     """
-    加载配置中的 worker 插件，启动对应的 worker 任务。
+    加载 inline 插件，启动对应的 worker 任务（混合模式）。
 
-    自动从 config/workers/<type>/config.toml 发现并加载插件。
+    自动从 config/workers/<type>/config.toml 发现插件；load_mode="external"
-    每个插件的 [workers] 节定义加载参数，[workers.config] 节定义运行时配置。
+    的插件不在网关内加载（由独立进程运行，其路由经 RouteRegistry 动态注册）。
 
-    同时自动发现插件模块中暴露的 APIRouter（约定：模块顶层名为 `router` 的 APIRouter 实例），
+    同时自动发现插件模块中暴露的 APIRouter（约定：模块顶层名为 `router` 的
-    收集后供 register_custom_routes() 统一注册，实现网关与插件的解耦。
+    APIRouter 实例），收集后供 register_custom_routes() 统一注册。
 
-    返回: (worker_tasks, custom_routers)
+    返回: (worker_tasks, custom_routers, inline_types, external_types)
     """
     workers_conf = _discover_plugins()
-    logger.info(f"发现 {len(workers_conf)} 个 worker 插件")
+    inline_plugins = [wc for wc in workers_conf if wc.get("load_mode", "inline") != "external"]
+    external_plugins = [wc for wc in workers_conf if wc.get("load_mode", "inline") == "external"]
+    inline_types = sorted({wc["type"] for wc in inline_plugins})
+    external_types = sorted({wc["type"] for wc in external_plugins})
+    logger.info(
+        f"发现 {len(workers_conf)} 个 worker 插件: "
+        f"inline={inline_types or '无'}, external={external_types or '无'}"
+    )
 
-    # 汇总各插件声明的 MQTT Bridge 额外订阅主题，供 setup_mqtt 使用
+    for wc in external_plugins:
+        logger.info(
+            "插件 %s 为 external 模式，跳过进程内加载（由独立进程经 Zenoh 注册路由）",
+            wc["type"],
+        )
+
+    # 汇总 inline 插件声明的 MQTT Bridge 额外订阅主题，供 setup_mqtt 使用
+    # （external 插件的主题由 RouteRegistry 经 __meta__ 动态推送）
     app.state.plugin_mqtt_topics = [
         topic
-        for wc in workers_conf
+        for wc in inline_plugins
         for topic in (wc.get("mqtt_topics") or [])
     ]
 
     worker_tasks = []
     custom_routers = []
-    for wc in workers_conf:
+    for wc in inline_plugins:
         module_name = wc["module"]
         factory_name = wc["factory"]
         count = wc["count"]
@@ -369,12 +421,23 @@ async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerMana
             )
 
     app.state.worker_tasks = worker_tasks
-    return worker_tasks, custom_routers
+    app.state.inline_types = inline_types
+    app.state.external_types = external_types
+    return worker_tasks, custom_routers, inline_types, external_types
 
-async def register_routes(app: FastAPI, manager: WorkerManager, custom_routers: list | None = None):
+async def register_routes(
-    """注册动态路由和插件自定义路由。"""
+    app: FastAPI,
+    manager: WorkerManager,
+    custom_routers: list | None = None,
+    worker_types: list[str] | None = None,
+):
+    """注册动态路由和插件自定义路由。
+
+    worker_types: 仅注册这些插件类型的 TOML 动态路由（混合模式下只注册
+    inline 插件；external 插件路由由 RouteRegistry 动态挂载）。
+    """
     prefix = config["default"]["prefix"]
-    register_dynamic_routes(app, manager, prefix=prefix)
+    register_dynamic_routes(app, manager, prefix=prefix, worker_types=worker_types)
     # 注册插件自定义路由（如 FORM-DATA 文件上传等 TOML 无法描述的端点）
     # 统一加上全局 prefix（如 /api/v1），与 TOML 动态路由保持一致
     if custom_routers:
@@ -385,15 +448,19 @@ async def register_routes(app: FastAPI, manager: WorkerManager, custom_routers:
 
 def validate_contracts(app: FastAPI | None = None):
     """
-    契约校验（TOML 路由 ↔ @rpc_method 注册表 ↔ gen_models 生成模型）。
+    契约校验（TOML 路由 ↔ @rpc_method 注册表 ↔ 生成模型），仅覆盖 inline 插件。
 
     在 load_workers 之后、register_routes 之前调用：此时插件模块已导入
     （注册表就绪）。HTTP 路由缺失对应 @rpc_method 时抛 RuntimeError 终止启动，
     其余不一致以警告日志汇总。
+
+    external 插件在自己进程内做同样校验（validate_worker_contracts），网关侧跳过。
     """
     from ..core.contracts import validate_contracts as _validate
 
-    _validate(_discover_plugins())
+    plugins = _discover_plugins()
+    inline_plugins = [p for p in plugins if p.get("load_mode", "inline") != "external"]
+    _validate(inline_plugins)
 
 def setup_mqtt_broker(app: FastAPI):
     """启动 MQTT Broker 子进程（原子锁确保多 worker 仅启动一个）"""
@@ -413,8 +480,13 @@ def setup_mqtt_broker(app: FastAPI):
             time.sleep(0.1)
         raise RuntimeError("MQTT Broker 端口未就绪（已被其他 worker 锁定）")
 
+    # 记录本进程持有锁，关闭时由持有者释放（避免误删其他 worker 的锁）
+    app.state.mqtt_broker_lock_acquired = True
+
     if is_port_in_use(port, host):
         logger.info(f"MQTT Broker 已在 {host}:{port} 运行，跳过")
+        _release_singleton_lock("mqtt_broker")
+        app.state.mqtt_broker_lock_acquired = False
         return
     proc = multiprocessing.Process(
         target=_run_mqtt_broker, args=(host, port), daemon=False
@@ -433,7 +505,7 @@ def setup_mqtt_broker(app: FastAPI):
 
 
 async def shutdown_mqtt_broker(app: FastAPI):
-    """终止 MQTT Broker 子进程"""
+    """终止 MQTT Broker 子进程（仅锁持有者释放锁）"""
     proc = getattr(app.state, "mqtt_broker_process", None)
     if proc and proc.is_alive():
         proc.terminate()
@@ -441,6 +513,8 @@ async def shutdown_mqtt_broker(app: FastAPI):
         if proc.is_alive():
             proc.kill()
         logger.info("MQTT Broker 子进程已终止")
+    if getattr(app.state, "mqtt_broker_lock_acquired", False):
+        _release_singleton_lock("mqtt_broker")
 
 
 async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge | None:
@@ -463,13 +537,14 @@ async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge
     custom_topics = list(mqtt_config.get("custom_topics", []) or [])
     custom_topics += list(getattr(app.state, "plugin_mqtt_topics", []) or [])
 
-    # 1. 启动 Broker 子进程（文件锁已确保只有一个）
+    # Broker 子进程已在 lifespan 第 3 步启动（setup_mqtt_broker + 文件锁单例）
-    # setup_mqtt_broker(app, host, port)
 
-    # 2. 原子锁确保只有一个 worker 启动 MQTT Bridge
+    # 原子锁确保只有一个 worker 启动 MQTT Bridge
     if not _acquire_singleton_lock("mqtt_bridge"):
         logger.info("MQTT Bridge 已由其他 worker 启动，跳过")
         return None
+    # 记录本进程持有锁，关闭时由持有者释放（避免误删其他 worker 的锁）
+    app.state.mqtt_bridge_lock_acquired = True
 
     bridge = MQTTBridge(
         worker_manager=worker_manager,
@@ -479,23 +554,31 @@ async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge
     )
     app.state.mqtt_bridge = bridge
     await bridge.start()
+    # process 模式：把 Bridge 挂接给路由注册表（动态主题 + 补推已到达的元信息）
+    registry = getattr(app.state, "route_registry", None)
+    if registry is not None:
+        registry.attach_bridge(bridge)
     logger.info(f"MQTT Bridge 已启动 ({host}:{port})")
     return bridge
 
 
 async def shutdown_mqtt(app: FastAPI):
-    """停止 MQTT Bridge 和 Broker 子进程"""
+    """停止 MQTT Bridge 和 Broker 子进程（仅锁持有者释放锁）"""
     bridge: MQTTBridge | None = getattr(app.state, "mqtt_bridge", None)
     if bridge and bridge.is_running:
         await bridge.stop()
+    if getattr(app.state, "mqtt_bridge_lock_acquired", False):
+        _release_singleton_lock("mqtt_bridge")
     await shutdown_mqtt_broker(app)
 
 
 async def cleanup_workers(app: FastAPI):
-    """调用每个 worker 配置中定义的 cleanup 函数。"""
+    """调用每个 inline worker 配置中定义的 cleanup 函数。"""
     workers_conf = _discover_plugins()
 
     for wc in workers_conf:
+        if wc.get("load_mode", "inline") == "external":
+            continue
         module_name = wc.get("module", "")
         cleanup_name = wc.get("cleanup", "")
         worker_type = wc.get("type", "")

src/chongming/app/core/mqtt_bridge.py

@@ -1,9 +1,13 @@
 """MQTT Bridge — 纯客户端模式，Broker 由子进程管理"""
-import asyncio, json, re
+import asyncio
-from typing import Any, Optional
+import json
+import re
+from typing import Optional
+
 from amqtt.client import MQTTClient
 from amqtt.mqtt.constants import QOS_1
 from .logger import get_logger
+
 logger = get_logger("mqtt")
 MQTT_TOPIC_PREFIX = "chongming"
 MQTT_SUBSCRIBE_TOPIC = f"{MQTT_TOPIC_PREFIX}/+/+"
@@ -34,27 +38,44 @@ def _compile_topic(topic: str) -> re.Pattern:
 
 class MQTTBridge:
     def __init__(self, worker_manager, host="localhost", port=1883, custom_topics=None):
-        self._wm = worker_manager; self._host = host; self._port = port
+        self._wm = worker_manager
+        self._host = host
+        self._port = port
         self._client: Optional[MQTTClient] = None
         self._lt: Optional[asyncio.Task] = None
         self._running = False
         # custom_topics: [{"topic","worker","subject"?,"response_topic"?}, ...]
-        self._custom_topics = custom_topics or []
+        # 混合模式双来源：
+        #   _static_topics  — inline 插件，构造时由网关从配置收集
+        #   _dynamic_topics — external 插件，运行时由 RouteRegistry 推送
+        self._static_topics = custom_topics or []
+        self._dynamic_topics: list = []
+        self._compiled: list = []
+        self._rebuild_compiled()
+        self._subscribed_custom: set = set()
+
+    def _rebuild_compiled(self):
+        """合并静态/动态主题并重编译匹配正则。"""
+        merged = self._static_topics + self._dynamic_topics
         self._compiled = [
             (ct, _compile_topic(ct.get("topic", "")))
-            for ct in self._custom_topics if ct.get("topic")
+            for ct in merged if ct.get("topic")
         ]
 
     async def start(self):
-        if self._running: return
+        if self._running:
-        try: await self._sc()
+            return
+        try:
+            await self._sc()
         except Exception as e:
-            logger.warning(f"MQTT 不可用: {e}"); return
+            logger.warning(f"MQTT 不可用: {e}")
+            return
         self._running = True
         logger.info(f"MQTT Bridge 已连接 {self._host}:{self._port}")
 
     async def stop(self):
-        if not self._running: return
+        if not self._running:
+            return
         self._running = False
         await self._sc_stop()
 
@@ -64,39 +85,74 @@ class MQTTBridge:
         topics = [(MQTT_SUBSCRIBE_TOPIC, QOS_1)]
         topics += [(ct["topic"], QOS_1) for ct, _ in self._compiled]
         await self._client.subscribe(topics)
+        self._subscribed_custom = {ct["topic"] for ct, _ in self._compiled}
         logger.info(f"已订阅 {MQTT_SUBSCRIBE_TOPIC}"
                    + (f" 及 {len(self._compiled)} 个自定义主题" if self._compiled else ""))
         self._lt = asyncio.create_task(self._loop())
 
+    async def update_dynamic_topics(self, custom_topics: list):
+        """
+        动态更新 external 插件的自定义订阅主题（RouteRegistry 推送）。
+
+        与构造时的 inline 静态主题合并后增量重订阅（退订已移除主题、订阅新增主题），
+        运行中调用即生效。
+        """
+        self._dynamic_topics = list(custom_topics or [])
+        self._rebuild_compiled()
+        new_topics = {ct["topic"] for ct, _ in self._compiled}
+        if not (self._running and self._client):
+            self._subscribed_custom = new_topics
+            return
+        removed = self._subscribed_custom - new_topics
+        added = new_topics - self._subscribed_custom
+        if removed:
+            await self._client.unsubscribe(list(removed))
+        if added:
+            await self._client.subscribe([(t, QOS_1) for t in added])
+        self._subscribed_custom = new_topics
+        if added or removed:
+            logger.info(
+                f"自定义订阅主题更新: +{len(added)} -{len(removed)}"
+                f"（当前 {len(new_topics)} 个）"
+            )
+
     async def _sc_stop(self):
         # 先断开连接，让 deliver_message() 解锁
         if self._client:
             try:
                 await self._client.disconnect()
-            except Exception: pass
+            except Exception:
+                pass
             self._client = None
         # 再取消消息循环
         if self._lt and not self._lt.done():
             self._lt.cancel()
-            try: await self._lt
+            try:
-            except asyncio.CancelledError: pass
+                await self._lt
+            except asyncio.CancelledError:
+                pass
 
     async def _loop(self):
         while self._running:
             try:
                 m = await self._client.deliver_message()  # type: ignore
-                if m is None: break
+                if m is None:
+                    break
                 await self._handle(m)
-            except asyncio.CancelledError: break
+            except asyncio.CancelledError:
+                break
             except Exception as e:
                 if self._running:
                     logger.error(f"消息异常: {e}")
                 break
 
     async def _handle(self, m):
-        t: str = m.topic; p = m.data  # type: ignore
+        t: str = m.topic
-        try: pl = json.loads(p.decode("utf-8"))
+        p = m.data  # type: ignore
-        except json.JSONDecodeError: return
+        try:
+            pl = json.loads(p.decode("utf-8"))
+        except json.JSONDecodeError:
+            return
 
         # 信封协议（严格）: {"meta": {...}, "data": {...}}
         # 非信封格式载荷直接丢弃并告警，请设备侧按新协议发布
@@ -120,7 +176,8 @@ class MQTTBridge:
 
         # 2. 内置接口 chongming/{worker}/{operation}
         parts = t.split("/")
-        if len(parts) < 3: return
+        if len(parts) < 3:
+            return
         wt, op = parts[1], parts[2]
         subj = f"{wt}.{op}"
         rt = f"{MQTT_TOPIC_PREFIX}/{wt}/{op}/response"
@@ -128,7 +185,8 @@ class MQTTBridge:
 
     def _fill(self, template: str, matched: re.Match) -> str:
         """用匹配到的通配符值填充模板中的 {1} {2}... 占位符"""
-        if "{" not in template: return template
+        if "{" not in template:
+            return template
         for i, g in enumerate(matched.groups(), start=1):
             template = template.replace("{" + str(i) + "}", g)
         return template
@@ -153,15 +211,20 @@ class MQTTBridge:
             r = await self._wm.call(worker_type=wt, subject=subj, payload=pl,
                 timeout=float(meta.get("timeout", 30)),
                 cacheable=bool(meta.get("cacheable", False)))
-            if r is not None: await self._pub(rt, r)
+            if r is not None:
-        except RuntimeError as e: await self._pub(rt, {"error":str(e)})
+                await self._pub(rt, r)
-        except Exception as e: await self._pub(rt, {"error":str(e)})
+        except RuntimeError as e:
+            await self._pub(rt, {"error": str(e)})
+        except Exception as e:
+            await self._pub(rt, {"error": str(e)})
 
     async def _pub(self, topic, data):
         try:
             pd = json.dumps(data, ensure_ascii=False, default=str)
             await self._client.publish(topic, pd.encode("utf-8"), QOS_1)  # type: ignore
-        except Exception as e: logger.error(f"发布失败 [{topic}]: {e}")
+        except Exception as e:
+            logger.error(f"发布失败 [{topic}]: {e}")
 
     @property
-    def is_running(self): return self._running
+    def is_running(self):
+        return self._running

src/chongming/app/core/route_generator.py

@@ -11,6 +11,8 @@ from inspect import Parameter, Signature
 from fastapi import FastAPI, HTTPException, APIRouter, Query, Path as FastAPIPath, Request, Depends
 from pydantic import create_model, Field, BaseModel
 
+from chongming_worker.route_config import WORKER_CONFIG_DIR, load_worker_routes_from_dir
+
 from .worker_manager import WorkerManager
 from .logger import get_logger
 from .auth import store_user_to_request
@@ -110,96 +112,58 @@ class RouteConfig:
         ]
 
 
-# 默认的 worker 配置目录
+def register_route_group(
-WORKER_CONFIG_DIR = "config/workers"
+    app: FastAPI,
-
+    worker_manager: WorkerManager,
-
+    worker_type: str,
-def load_worker_routes_from_dir(worker_type: str) -> List[Dict[str, Any]]:
+    routes: List[Dict[str, Any]],
+    prefix: str = "",
+    extra_schemas: Optional[Dict[str, type[BaseModel]]] = None,
+) -> List[Any]:
     """
-    从 config/workers/{worker_type}/ 目录加载路由配置。
+    为一个 worker_type 的一组路由描述符创建并挂载 APIRouter。
 
-    目录结构:
+    与 register_dynamic_routes（盘扫路径）共用同一注册逻辑，
-      config/workers/{worker_type}/
+    process 模式下 RouteRegistry 也调用本函数做动态挂载。
-        config.toml           # [worker] 分组信息 + registration + defaults
-        add.toml              # 接口名 = 文件名（不含扩展名）
-          [GET]
-          summary = ...
-          params = [...]
-          response_model = { ... }
-          [POST]
-          ...                 # 同一个接口支持多个请求方法
-          [MQTT]
-          ...                 # MQTT 专用端点（不注册 HTTP 路由）
 
-    返回: List[Dict], 每个 dict 包含 flatten 后的路由配置,
+    Args:
-          每个方法节(如 GET/POST/MQTT) 生成一个独立的路由条目。
+        app: FastAPI 应用
+        worker_manager: WorkerManager 实例
+        worker_type: worker 类型（OpenAPI 分组标签）
+        routes: 路由描述符列表（TOML 兼容格式，与 load_worker_routes_from_dir 输出一致）
+        prefix: 全局路由前缀（如 /api/v1）
+        extra_schemas: 额外 schema 收集 dict（GET 请求模型注入 OpenAPI 用）
+
+    Returns:
+        List[Any]: 本次新增到 app.router.routes 的 APIRoute 对象列表
+        （调用方可记录引用，用于动态移除）
     """
-    from pathlib import Path
+    router = APIRouter(prefix=prefix)
-    import tomllib
+    for route_conf in routes:
+        _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas)
 
-    base_dir = Path(WORKER_CONFIG_DIR) / worker_type
+    before = set(id(r) for r in app.router.routes)
-    if not base_dir.is_dir():
+    app.include_router(router)
-        logger.warning(f"Worker 配置目录不存在: {base_dir}")
+    added = [r for r in app.router.routes if id(r) not in before]
-        return []
+    logger.info(
-
+        f"已挂载 worker '{worker_type}' 的路由组: {len(routes)} 个描述符 -> "
-    # 1. 读取分组配置
+        f"{len(added)} 条 HTTP 路由 (prefix={prefix or '/'})"
-    group_cfg_path = base_dir / "config.toml"
+    )
-    group_defaults: Dict[str, Any] = {}
+    return added
-    registration: Dict[str, Any] = {}
-    router_prefix: str = ""
-    if group_cfg_path.exists():
-        with open(group_cfg_path, "rb") as f:
-            cfg = tomllib.load(f)
-        w = cfg.get("worker", {})
-        group_defaults = w.get("defaults", {})
-        registration = w.get("registration", {})
-        router_prefix = registration.get("router_prefix", "")
-
-    # 2. 遍历所有接口 toml 文件
-    routes: List[Dict[str, Any]] = []
-    for route_file in sorted(base_dir.glob("*.toml")):
-        if route_file.name == "config.toml":
-            continue
-
-        # 文件名（不含 .toml）作为路径名
-        route_name = route_file.stem  # 例如 "add"
-        base_path = router_prefix + "/" + route_name  # 例如 "/calc/add"
-
-        with open(route_file, "rb") as f:
-            route_data = tomllib.load(f)
-
-        # 每个方法节（GET/POST/PUT/DELETE）生成一个独立路由
-        for method, spec in route_data.items():
-            if not isinstance(spec, dict):
-                logger.warning(f"忽略非 dict 的方法节: {route_file.name}[{method}]")
-                continue
-
-            # 合并: spec 显式字段优先，然后 group_defaults 兜底
-            route = dict(spec)
-            for key, default_val in group_defaults.items():
-                if key not in route:
-                    route[key] = default_val
-
-            # 覆盖关键字段 - method 始终从 TOML 节键获取，不被 defaults 覆盖
-            route["method"] = method.upper()
-            route["path"] = route.get("path", base_path)
-            if "subject" not in route:
-                # 自动生成 subject: worker_type.route_name (如 "calc.add")
-                route["subject"] = f"{worker_type}.{route_name}"
-            # 记录来源文件名（契约校验用其推断 gen_models 模型命名）
-            route["route_name"] = route_name
-
-            routes.append(route)
-            logger.debug(
-                f"  已加载: {route['method']} {route['path']} -> {route['subject']}"
-            )
-
-    logger.info(f"从 {base_dir} 加载了 {len(routes)} 个路由")
-    return routes
 
 
-def register_dynamic_routes(app: FastAPI, worker_manager: WorkerManager, prefix: str = ""):
+def register_dynamic_routes(
-    """根据 config/workers/ 目录结构注册所有动态路由"""
+    app: FastAPI,
+    worker_manager: WorkerManager,
+    prefix: str = "",
+    worker_types: Optional[List[str]] = None,
+):
+    """
+    根据 config/workers/ 目录结构注册所有动态路由。
+
+    worker_types: 白名单（混合模式下只注册 inline 插件；
+    None = 全部注册，兼容纯 inline 部署）。
+    """
     from pathlib import Path
 
     workers_base = Path(WORKER_CONFIG_DIR)
@@ -207,7 +171,6 @@ def register_dynamic_routes(app: FastAPI, worker_manager: WorkerManager, prefix:
         logger.info(f"Worker 配置目录不存在: {WORKER_CONFIG_DIR}")
         return
 
-    router = APIRouter(prefix=prefix)
     extra_schemas: Dict[str, type[BaseModel]] = {}
 
     # 扫描所有 worker 子目录（每个子目录名 = worker_type）
@@ -215,28 +178,32 @@ def register_dynamic_routes(app: FastAPI, worker_manager: WorkerManager, prefix:
         if not worker_dir.is_dir():
             continue
         worker_type = worker_dir.name
+        if worker_types is not None and worker_type not in worker_types:
+            continue
         routes = load_worker_routes_from_dir(worker_type)
         if not routes:
             continue
 
-        for route_conf in routes:
+        register_route_group(app, worker_manager, worker_type, routes, prefix, extra_schemas)
-            _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas)
-
-    app.include_router(router)
 
     # 将额外 schema 注册到 OpenAPI 的 components/schemas
     if extra_schemas:
-        _inject_schemas_to_openapi(app, extra_schemas)
+        _inject_schemas_to_openapi(app, lambda: extra_schemas)
 
     logger.info(f"动态路由注册完成，worker 配置目录: {WORKER_CONFIG_DIR}")
 
 
-def _inject_schemas_to_openapi(app: FastAPI, schemas: Dict[str, type[BaseModel]]):
+def _inject_schemas_to_openapi(app: FastAPI, schemas_provider):
     """
     将自定义 Pydantic 模型注册到 FastAPI 的 OpenAPI components/schemas 中。
 
     通过重写 app.openapi() 方法注入，因为 FastAPI 在生命周期内只生成一次
     OpenAPI schema，之后会被缓存。
+
+    Args:
+        schemas_provider: 无参 callable，返回 {name: BaseModel} 字典。
+            盘扫路径传静态 dict；process 模式传 RouteRegistry 的动态 dict，
+            使路由表变化后重新生成的 OpenAPI 能反映最新 schema。
     """
     original_openapi = app.openapi
 
@@ -248,7 +215,7 @@ def _inject_schemas_to_openapi(app: FastAPI, schemas: Dict[str, type[BaseModel]]
             return {"openapi": "3.1.0", "info": {"title": "Chongming", "version": "1.0.0"}, "paths": {}}
         # 注入我们的 schema
         schemas_dict = schema.setdefault("components", {}).setdefault("schemas", {})
-        for name, model in schemas.items():
+        for name, model in schemas_provider().items():
             if name not in schemas_dict:
                 schemas_dict[name] = model.model_json_schema()
         app.openapi_schema = schema

src/chongming/app/core/route_registry.py

@@ -0,0 +1,261 @@
+"""
+动态路由注册表（process 模式）
+
+网关订阅 `chongming/routes/**` 命名空间，独立 worker 进程上线时 put 路由描述符、
+下线时 delete，网关据此动态挂载/摘除 FastAPI 路由并同步 MQTT 自定义订阅主题。
+
+协议（描述符 schema_version=1）:
+  chongming/routes/<type>/__meta__  插件级描述: {"schema_version":1,"mqtt_topics":[...]}
+  chongming/routes/<type>/<route_id> 端点描述（TOML 兼容 dict，与 RouteConfig 对位）:
+    {"schema_version":1, "type","subject","method","path","params",
+     "response_model","summary","timeout","cacheable","auth_required","internal"}
+
+特性:
+  - SampleKind.PUT -> upsert；DELETE -> remove
+  - 同一 route_id 内容相同（hash）视为多副本重复发布，忽略；不同 -> 告警并保留先到者
+  - type 级 2 秒防抖：worker 重启风暴合并为一次路由重建
+  - 迟到一致性：worker 侧周期性重发布描述符（RoutePublisher 由调用方周期调用），
+    网关后启动或样本丢失时，路由在一个发布周期内自动补齐
+"""
+import asyncio
+import hashlib
+import json
+from typing import Any, Dict, List, Optional
+
+import zenoh
+
+from chongming_worker.routes_publisher import META_KEY, ROUTES_PREFIX, SCHEMA_VERSION
+
+from .logger import get_logger
+from .route_generator import register_route_group, _inject_schemas_to_openapi
+
+logger = get_logger("route_registry")
+
+# worker 重启风暴防抖窗口（秒）
+DEBOUNCE_SECONDS = 2.0
+
+
+def _descriptor_hash(descriptor: Dict[str, Any]) -> str:
+    """描述符内容 hash（多副本幂等去重用）"""
+    payload = json.dumps(descriptor, sort_keys=True, ensure_ascii=False)
+    return hashlib.sha1(payload.encode("utf-8")).hexdigest()
+
+
+class RouteRegistry:
+    """process 模式下的网关侧动态路由注册表。"""
+
+    def __init__(
+        self,
+        app,
+        worker_manager,
+        prefix: str = "",
+    ):
+        self._app = app
+        self._wm = worker_manager
+        self._prefix = prefix
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+
+        # worker_type -> route_id -> descriptor
+        self._groups: Dict[str, Dict[str, Dict[str, Any]]] = {}
+        # worker_type -> 插件级描述（mqtt_topics 等）
+        self._meta: Dict[str, Dict[str, Any]] = {}
+        # worker_type -> 已挂载的 APIRoute 对象（用于动态摘除）
+        self._route_objs: Dict[str, List[Any]] = {}
+        # OpenAPI 额外 schema（GET 请求模型），随路由表变化
+        self._schemas: Dict[str, Any] = {}
+        # worker_type -> 防抖任务
+        self._debounce: Dict[str, asyncio.Task] = {}
+
+        self._subscriber: Optional[zenoh.Subscriber] = None
+        self._mqtt_bridge = None
+        # 网关进程内加载（inline）的插件类型：忽略其描述符，防止与盘扫注册重复挂载
+        self._inline_types: set = set()
+
+        # 将动态 schema 注入 OpenAPI（provider 实时读取本注册表）
+        _inject_schemas_to_openapi(app, lambda: self._schemas)
+
+    def set_inline_types(self, types: set):
+        """设置 inline 插件类型集合（混合模式防双注册）。"""
+        self._inline_types = set(types)
+
+    async def start(self, session: zenoh.Session):
+        """订阅 chongming/routes/**，开始接收路由描述符。"""
+        self._loop = asyncio.get_running_loop()
+        self._subscriber = session.declare_subscriber(
+            f"{ROUTES_PREFIX}/**", self._on_sample
+        )
+        logger.info(f"路由注册表已就绪: 订阅 {ROUTES_PREFIX}/**")
+
+    def stop(self):
+        """停止订阅（shutdown 时调用）。"""
+        if self._subscriber is not None:
+            self._subscriber.undeclare()
+            self._subscriber = None
+        for task in self._debounce.values():
+            task.cancel()
+
+    def attach_bridge(self, mqtt_bridge):
+        """挂接 MQTT Bridge：后续路由表变化会推送 external 插件的动态订阅主题。"""
+        self._mqtt_bridge = mqtt_bridge
+        # Bridge 晚于路由到达时，立即补推当前已收集的主题
+        topics = self._collect_mqtt_topics()
+        if topics:
+            self._loop.create_task(mqtt_bridge.update_dynamic_topics(topics))
+
+    @property
+    def route_types(self) -> List[str]:
+        """当前已注册路由的 worker 类型列表（观测用）。"""
+        return sorted(self._groups.keys())
+
+    def ready_summary(self) -> str:
+        """路由就绪状态摘要（日志/观测用）。"""
+        parts = []
+        for wtype in sorted(self._groups):
+            http = sum(
+                1 for d in self._groups[wtype].values()
+                if d.get("method") not in ("MQTT",) and not d.get("internal")
+            )
+            parts.append(f"{wtype}={len(self._groups[wtype])} 端点/{http} HTTP")
+        return ", ".join(parts) if parts else "(空)"
+
+    # ---- zenoh 回调线程 -> 事件循环 ----
+
+    def _on_sample(self, sample: zenoh.Sample):
+        """zenoh 后台线程回调：转交事件循环处理。"""
+        if self._loop is None:
+            return
+        key = str(sample.key_expr)
+        kind = sample.kind
+        payload = bytes(sample.payload) if sample.payload else b""
+        asyncio.run_coroutine_threadsafe(
+            self._handle_sample(key, kind, payload), self._loop
+        )
+
+    async def _handle_sample(self, key: str, kind: zenoh.SampleKind, payload: bytes):
+        """解析样本并更新内存路由表（在事件循环内执行）。"""
+        parts = key.split("/")
+        if len(parts) < 4:
+            return
+        wtype, route_id = parts[2], parts[3]
+
+        if wtype in self._inline_types:
+            # 该插件由网关进程内加载（盘扫注册路由），忽略外部描述符防双注册
+            return
+
+        if kind == zenoh.SampleKind.DELETE:
+            self._remove(wtype, route_id)
+            return
+
+        try:
+            descriptor = json.loads(payload.decode("utf-8"))
+        except (json.JSONDecodeError, UnicodeDecodeError):
+            logger.warning("丢弃非法路由描述符 [%s]: JSON 解析失败", key)
+            return
+        if not isinstance(descriptor, dict):
+            logger.warning("丢弃非法路由描述符 [%s]: 非 dict", key)
+            return
+
+        if route_id == META_KEY:
+            self._upsert_meta(wtype, descriptor)
+        else:
+            self._upsert_route(wtype, route_id, descriptor)
+
+    def _upsert_meta(self, wtype: str, descriptor: Dict[str, Any]):
+        old = self._meta.get(wtype)
+        new_hash = _descriptor_hash(descriptor)
+        if old is not None and _descriptor_hash(old) == new_hash:
+            return
+        self._meta[wtype] = descriptor
+        self._schedule_rebuild(wtype)
+
+    def _upsert_route(self, wtype: str, route_id: str, descriptor: Dict[str, Any]):
+        group = self._groups.setdefault(wtype, {})
+        existing = group.get(route_id)
+        if existing is not None:
+            if _descriptor_hash(existing) == _descriptor_hash(descriptor):
+                # 多副本重复发布同一描述符，幂等忽略
+                return
+            logger.warning(
+                "路由描述符冲突 [%s/%s]: 已存在版本保留，忽略新版本",
+                wtype, route_id,
+            )
+            return
+        # 补全描述符字段（发布端缺省时的兜底）
+        descriptor.setdefault("type", wtype)
+        descriptor.setdefault("schema_version", SCHEMA_VERSION)
+        group[route_id] = descriptor
+        logger.info(
+            "收到路由注册: %s %s %s (subject=%s)",
+            descriptor.get("method", "?"),
+            descriptor.get("path", "?"),
+            f"[{wtype}]",
+            descriptor.get("subject", "?"),
+        )
+        self._schedule_rebuild(wtype)
+
+    def _remove(self, wtype: str, route_id: str):
+        if route_id == META_KEY:
+            if self._meta.pop(wtype, None) is not None:
+                self._schedule_rebuild(wtype)
+            return
+        group = self._groups.get(wtype)
+        if group and group.pop(route_id, None) is not None:
+            logger.info("收到路由摘除: [%s] %s", wtype, route_id)
+            self._schedule_rebuild(wtype)
+
+    def _schedule_rebuild(self, wtype: str):
+        """type 级防抖：合并短时间内的多次变更。"""
+        prev = self._debounce.get(wtype)
+        if prev is not None and not prev.done():
+            prev.cancel()
+        self._debounce[wtype] = asyncio.create_task(self._debounced_rebuild(wtype))
+
+    async def _debounced_rebuild(self, wtype: str):
+        try:
+            await asyncio.sleep(DEBOUNCE_SECONDS)
+        except asyncio.CancelledError:
+            return
+        self._debounce.pop(wtype, None)
+        await self._rebuild(wtype)
+
+    async def _rebuild(self, wtype: str):
+        """整组重建该 worker_type 的 HTTP 路由 + 同步 MQTT 自定义主题。"""
+        routes = list(self._groups.get(wtype, {}).values())
+        routes.sort(key=lambda r: (r.get("path", ""), r.get("method", "")))
+
+        # 1. 摘除旧路由（按对象引用从 router.routes 过滤）
+        for r in self._route_objs.pop(wtype, []):
+            if r in self._app.router.routes:
+                self._app.router.routes.remove(r)
+
+        # 2. 挂载新路由（MQTT / internal 端点不注册 HTTP，但保留在路由表中）
+        http_routes = [
+            r for r in routes
+            if r.get("method") not in ("MQTT",) and not r.get("internal")
+        ]
+        if http_routes:
+            self._route_objs[wtype] = register_route_group(
+                self._app, self._wm, wtype, http_routes, self._prefix, self._schemas
+            )
+        else:
+            logger.info("[%s] 无 HTTP 路由可挂载（%d 个描述符，全部为 MQTT/internal）",
+                        wtype, len(routes))
+
+        # 路由表变化后 OpenAPI 缓存失效，下一次访问时重建
+        self._app.openapi_schema = None
+
+        # 3. 同步 external 插件的 MQTT 动态订阅主题
+        if self._mqtt_bridge is not None:
+            await self._mqtt_bridge.update_dynamic_topics(self._collect_mqtt_topics())
+
+    def _collect_mqtt_topics(self) -> List[Dict[str, Any]]:
+        """汇总所有插件 __meta__ 声明的自定义订阅主题。"""
+        topics: List[Dict[str, Any]] = []
+        for wtype in sorted(self._meta):
+            meta = self._meta[wtype]
+            for t in meta.get("mqtt_topics") or []:
+                if isinstance(t, dict) and t.get("topic"):
+                    topic = dict(t)
+                    topic.setdefault("worker", wtype)
+                    topics.append(topic)
+        return topics

src/chongming/app/core/worker_manager.py

@@ -2,8 +2,7 @@ import asyncio
 import random
 import json
 import orjson
-import time
+from typing import Any, Dict, List, Optional
-from typing import Any, Dict, List, Optional, Set
 import zenoh
 from diskcache import Cache
 from .logger import get_logger

src/chongming/scripts/build.py

@@ -72,13 +72,13 @@ class BuildReport:
         """打印结构化的构建总结"""
         sep = "=" * 60
         print(f"\n{sep}")
-        print(f"  🏗️  CHONGMING 构建报告")
+        print("  🏗️  CHONGMING 构建报告")
         print(f"{sep}")
         print(f"  构建模式    : {self.mode}")
         print(f"  总耗时      : {self.total_duration:.2f}s")
         print(f"  构建状态    : {'✅ 成功' if self.all_success else '❌ 失败'}")
         print(f"{sep}")
-        print(f"  步骤明细:")
+        print("  步骤明细:")
         for i, step in enumerate(self.steps, 1):
             icon = "✅" if step.success else "❌"
             msg = f" ({step.message})" if step.message else ""

src/chongming/scripts/gen_models.py

@@ -101,6 +101,23 @@ def generate_pydantic_code(
     routes: list[dict[str, Any]],
 ) -> str:
     """生成 pydantic.py 文件内容"""
+    # 收集全部字段类型，按需生成 typing 导入（避免未使用导入触发 lint）
+    type_strings: list[str] = []
+    has_optional_field = False
+    for route in routes:
+        for p in route.get("parsed_params", []):
+            type_strings.append(p.get("type", "str"))
+        for f in route.get("parsed_response", []):
+            type_strings.append(f.get("type", "str"))
+            # 非必填字段渲染为 Optional[...]
+            if not f.get("required", True):
+                has_optional_field = True
+    typing_imports = []
+    if has_optional_field or any("Optional" in t for t in type_strings):
+        typing_imports.append("Optional")
+    if any("Any" in t for t in type_strings):
+        typing_imports.append("Any")
+
     lines = []
     lines.append('"""')
     lines.append(f"自动生成的 Pydantic 模型 — {plugin_name}")
@@ -109,7 +126,8 @@ def generate_pydantic_code(
     lines.append("")
     lines.append("from __future__ import annotations")
     lines.append("")
-    lines.append("from typing import Any, Optional")
+    if typing_imports:
+        lines.append(f"from typing import {', '.join(typing_imports)}")
     lines.append("from pydantic import BaseModel, Field")
     lines.append("")
 
@@ -310,7 +328,7 @@ def main():
     init_file.write_text(init_code, encoding="utf-8")
     print(f"✅ 生成: {init_file}")
 
-    print(f"\n📋 生成的模型:")
+    print("\n📋 生成的模型:")
     for route in routes:
         method = route["method"]
         name = route["name"]

src/chongming/scripts/gen_plugin.py

@@ -299,7 +299,7 @@ def generate_plugin(name: str, operations: list[str]) -> int:
     print(f"\n📋 插件 '{name}' 已就绪，下一步:")
     print(f"   1. uv run gen_models config/workers/{name} src/plugins/{name}")
     print(f"   2. 编辑 src/plugins/{name}/router/ 实现业务逻辑")
-    print(f"   3. uv run serve")
+    print("   3. uv run serve")
     return 0
 
 

test_auth_race.py

@@ -52,7 +52,7 @@ async def main():
             res = await sess.execute(select(Users))
             rows = res.scalars().all()
 
-        print(f"✅ 4 个并发启动全部成功")
+        print("✅ 4 个并发启动全部成功")
         print(f"✅ users 表记录数: {len(rows)}（应为 1）")
         for u in rows:
             print(f"   - {u.username} <{u.email}> superuser={u.is_superuser}")

test_process_mode_e2e.py

@@ -0,0 +1,107 @@
+"""
+混合模式（external 插件）端到端测试
+
+前置条件:
+  1. 网关正常运行（uv run serve）；example 插件配置为 external：
+     config/workers/example/config.toml [workers] 节 load_mode = "external"
+  2. 本脚本自己拉起/停止独立 worker 进程（uv run worker_serve --type example）
+
+验证:
+  1. worker 启动后 → HTTP 路由动态出现且跨进程 RPC 可调
+  2. worker 优雅退出 → 路由动态摘除（404）
+  3. worker 重启 → 路由恢复
+"""
+import subprocess
+import sys
+import time
+import urllib.request
+from pathlib import Path
+
+GATEWAY_URL = "http://localhost:8000"
+TEST_URL = f"{GATEWAY_URL}/api/v1/calc/add?a=3&b=4"
+POLL_TIMEOUT = 30.0
+
+# 优先使用 venv 控制台脚本（uv sync 后生成），不存在时退回 uv run
+_WORKER_CMD = [str(Path(sys.executable).parent / "worker_serve"), "--type", "example"]
+if not Path(_WORKER_CMD[0]).exists():
+    _WORKER_CMD = ["uv", "run", "worker_serve", "--type", "example"]
+
+PASS = 0
+FAIL = 0
+
+
+def http_get(url):
+    try:
+        with urllib.request.urlopen(url, timeout=3) as resp:
+            return resp.status, resp.read().decode("utf-8")
+    except urllib.error.HTTPError as e:
+        return e.code, ""
+    except Exception as e:
+        return None, str(e)
+
+
+def report(name, cond, detail=""):
+    global PASS, FAIL
+    if cond:
+        PASS += 1
+        print(f"✅ {name}" + (f" — {detail}" if detail else ""))
+    else:
+        FAIL += 1
+        print(f"❌ {name}" + (f" — {detail}" if detail else ""))
+
+
+def wait_http(target_status=200, timeout=POLL_TIMEOUT):
+    deadline = time.monotonic() + timeout
+    last = None
+    while time.monotonic() < deadline:
+        status, body = http_get(TEST_URL)
+        last = (status, body)
+        if status == target_status:
+            return True, body
+        time.sleep(0.5)
+    return False, last
+
+
+def run():
+    # 1. 拉起独立 worker 进程
+    proc = subprocess.Popen(
+        _WORKER_CMD,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+    )
+    try:
+        ok, body = wait_http(target_status=200)
+        report("worker 启动后路由动态注册 + 跨进程 RPC", ok, f"body={body}")
+
+        # 2. 优雅退出 → 路由摘除（404）
+        proc.terminate()
+        proc.wait(timeout=15)
+        ok, _ = wait_http(target_status=404)
+        report("worker 退出后路由动态摘除 (404)", ok)
+
+        # 3. 重启 worker → 路由恢复
+        proc2 = subprocess.Popen(
+            _WORKER_CMD,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+        )
+        try:
+            ok, body = wait_http(target_status=200)
+            report("worker 重启后路由恢复", ok, f"body={body}")
+        finally:
+            proc2.terminate()
+            proc2.wait(timeout=15)
+    finally:
+        if proc.poll() is None:
+            proc.terminate()
+
+    print("\n" + "=" * 50)
+    print(f"测试结果: ✅ {PASS} 通过  ❌ {FAIL} 失败")
+    return 0 if FAIL == 0 else 1
+
+
+if __name__ == "__main__":
+    print("提示: 请先以 load_mode=process 启动网关 (uv run serve)\n")
+    sys.exit(run())

test_route_registry.py

@@ -0,0 +1,181 @@
+"""
+RouteRegistry 单元测试（loopback Zenoh 会话 + httpx ASGI 功能断言）
+
+验证 external 插件路由动态注册核心链路（含混合模式防双注册）：
+  RoutePublisher.put 描述符 → 网关 RouteRegistry 动态挂载 FastAPI 路由（命中返回 503，worker 桩无实例）
+  RoutePublisher.delete      → 动态摘除路由（404）
+  __meta__ 中的 mqtt_topics → 推送给 MQTT Bridge（update_dynamic_topics）
+  set_inline_types           → 忽略 inline 插件的描述符（防双注册）
+
+注：FastAPI 0.139 的 include_router 将路由包装为 _IncludedRouter（惰性挂载），
+    不能靠遍历 app.router.routes 的 path 判断，故用 ASGI 请求状态码做功能断言。
+"""
+import asyncio
+import time
+
+import httpx
+import zenoh
+from fastapi import FastAPI
+
+from chongming_worker.routes_publisher import RoutePublisher
+from src.chongming.app.core.route_registry import RouteRegistry
+
+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:
+    """桩：无可用 worker，命中路由后 call 抛 RuntimeError（网关转 503）。"""
+
+    async def call(self, **kwargs):
+        raise RuntimeError("无可用 worker（测试桩）")
+
+
+class FakeBridge:
+    """记录 update_dynamic_topics 的调用结果。"""
+
+    def __init__(self):
+        self.topics = None
+
+    async def update_dynamic_topics(self, topics):
+        self.topics = list(topics)
+
+
+async def wait_until(cond, timeout=10.0, interval=0.2) -> bool:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if cond():
+            return True
+        await asyncio.sleep(interval)
+    return False
+
+
+HTTP_DESCRIPTOR = {
+    "subject": "example.add",
+    "method": "GET",
+    "path": "/calc/add",
+    "params": ["a: float", "b: float"],
+    "response_model": {"result": {"type": "float", "required": True}},
+    "summary": "加法运算",
+    "timeout": 30.0,
+    "cacheable": False,
+    "auth_required": False,
+    "internal": False,
+}
+
+MQTT_DESCRIPTOR = {
+    "subject": "example.add",
+    "method": "MQTT",
+    "path": "/calc/add_mqtt",
+    "params": ["a: float", "b: float"],
+    "summary": "MQTT 加法",
+    "timeout": 30.0,
+    "cacheable": False,
+    "auth_required": False,
+    "internal": False,
+}
+
+MQTT_TOPICS = [{"topic": "box/aibox_sophon/board_ping"}]
+# registry 会为每个主题补 worker 字段
+MQTT_TOPICS_EXPECTED = [
+    {"topic": "box/aibox_sophon/board_ping", "worker": "example"}
+]
+
+
+async def main():
+    zconfig = zenoh.Config()
+    try:
+        zconfig.from_file("config/zenoh_config.json5")
+    except Exception:
+        pass  # 回退默认配置
+
+    gw_session = zenoh.open(zconfig)
+    pub_session = zenoh.open(zconfig)
+    try:
+        app = FastAPI()
+        registry = RouteRegistry(app, FakeWorkerManager(), prefix="/api/v1")
+        await registry.start(gw_session)
+        await asyncio.sleep(0.5)  # 等待订阅生效
+
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+            # 1. 发布 → HTTP 路由动态挂载（命中 503：路由存在但无 worker）
+            publisher = RoutePublisher(pub_session, "example")
+            publisher.publish_routes(
+                [HTTP_DESCRIPTOR, MQTT_DESCRIPTOR], mqtt_topics=MQTT_TOPICS
+            )
+
+            async def mounted():
+                resp = await client.get("/api/v1/calc/add?a=1&b=2")
+                return resp.status_code == 503
+
+            ok = await wait_until(mounted)
+            check("put 后 HTTP 路由动态挂载（503 命中）", ok)
+
+            resp = await client.get("/api/v1/calc/add_mqtt?a=1&b=2")
+            check("MQTT 端点不挂载 HTTP 路由（404）", resp.status_code == 404,
+                  f"status={resp.status_code}")
+
+            # 2. __meta__ 主题推送给 Bridge
+            bridge = FakeBridge()
+            registry.attach_bridge(bridge)
+            ok = await wait_until(lambda: bridge.topics == MQTT_TOPICS_EXPECTED)
+            check("mqtt_topics 推送给 Bridge", ok, f"bridge.topics={bridge.topics}")
+
+            # 3. 重复发布相同描述符 → 幂等（路由不重复、仍可命中）
+            publisher.publish_routes([HTTP_DESCRIPTOR])
+            await asyncio.sleep(3.5)  # 覆盖防抖窗口
+            resp = await client.get("/api/v1/calc/add?a=1&b=2")
+            check("重复发布幂等（仍正常命中）", resp.status_code == 503,
+                  f"status={resp.status_code}")
+
+            # 4. 摘除 → 路由动态移除（404）
+            publisher.unpublish()
+
+            async def unmounted():
+                resp = await client.get("/api/v1/calc/add?a=1&b=2")
+                return resp.status_code == 404
+
+            ok = await wait_until(unmounted)
+            check("delete 后 HTTP 路由动态摘除（404）", ok)
+            ok = await wait_until(lambda: bridge.topics == [])
+            check("元信息摘除后 Bridge 主题清空", ok, f"bridge.topics={bridge.topics}")
+
+            # 5. 混合模式：inline 类型的描述符被忽略（防与盘扫注册重复挂载）
+            registry.set_inline_types({"example"})
+            publisher.publish_routes([HTTP_DESCRIPTOR])
+            await asyncio.sleep(3.5)  # 覆盖防抖窗口
+            resp = await client.get("/api/v1/calc/add?a=1&b=2")
+            check("inline 类型描述符被忽略（防双注册，404）", resp.status_code == 404,
+                  f"status={resp.status_code}")
+
+            # 清除 inline 集合后恢复动态挂载
+            registry.set_inline_types(set())
+            publisher.publish_routes([HTTP_DESCRIPTOR])
+
+            async def remounted():
+                resp = await client.get("/api/v1/calc/add?a=1&b=2")
+                return resp.status_code == 503
+
+            ok = await wait_until(remounted)
+            check("清除 inline 集合后恢复动态挂载", ok)
+            publisher.unpublish()
+    finally:
+        registry.stop()
+        gw_session.close()
+        pub_session.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+    print("\n" + ("=" * 40))
+    if failures:
+        print(f"❌ {len(failures)} 项未通过: {failures}")
+        raise SystemExit(1)
+    print("✅ RouteRegistry 全部通过")

test_worker_isolation.py

@@ -0,0 +1,91 @@
+"""
+Worker 进程隔离性测试
+
+验证进程插件与网关完全隔离：阻断 chongming（网关包）的两种导入路径后，
+worker 侧全链路（SDK 导入 → 插件配置自读 → 插件 import → 契约校验）仍可用。
+
+任何对 chongming.* / src.chongming.* 的导入都会直接触发 ImportError。
+"""
+import sys
+from pathlib import Path
+
+SRC = Path(__file__).parent
+sys.path.insert(0, str(SRC))
+
+# 阻断网关包导入（两种访问路径）。chongming_worker / zenoh / pydantic 等不受影响。
+sys.modules["chongming"] = None  # type: ignore[assignment]
+sys.modules["src.chongming"] = None  # type: ignore[assignment]
+
+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)
+
+
+# ========== 1. worker SDK 导入（若依赖网关包将在此抛 ImportError） ==========
+import chongming_worker.contracts as con_mod  # noqa: E402
+import chongming_worker.process as proc_mod  # noqa: E402
+import chongming_worker.route_config as rc_mod  # noqa: E402
+import chongming_worker.routes_publisher as pub_mod  # noqa: E402
+
+check("worker SDK 导入（阻断 chongming 后成功）", True)
+
+# ========== 2. 插件配置自读（不依赖网关 _discover_plugins） ==========
+wc = proc_mod.load_plugin_config("demo")
+check(
+    "插件配置自读（type/load_mode/module）",
+    wc["type"] == "demo"
+    and wc["load_mode"] == "external"
+    and wc["module"] == "workers.demo",
+    f"type={wc['type']}, load_mode={wc['load_mode']}, module={wc['module']}",
+)
+
+# ========== 3. 完整 worker 启动前置链路：import 插件 + 契约校验 ==========
+import workers.demo  # noqa: E402,F401  触发 @rpc_method 注册
+
+routes = con_mod.validate_worker_contracts("demo")
+check(
+    "进程内契约校验通过（TOML ↔ 注册表 ↔ 模型）",
+    len(routes) == 1 and routes[0]["subject"] == "demo.hello",
+    f"routes={[r['subject'] for r in routes]}",
+)
+
+# ========== 4. 代码级双重保险：SDK 无对网关包的 import 语句（AST 精确检查） ==========
+import ast  # noqa: E402
+import inspect  # noqa: E402
+
+
+def _imports_gateway(mod) -> list:
+    """返回模块中引用 chongming（非 chongming_worker）的 import 节点列表"""
+    found = []
+    tree = ast.parse(inspect.getsource(mod))
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                if alias.name == "chongming" or alias.name.startswith("chongming."):
+                    found.append(alias.name)
+        elif isinstance(node, ast.ImportFrom):
+            if node.module and (
+                node.module == "chongming" or node.module.startswith("chongming.")
+            ):
+                found.append(node.module)
+    return found
+
+
+for mod in (proc_mod, con_mod, pub_mod, rc_mod):
+    bad = _imports_gateway(mod)
+    if bad:
+        check(f"{mod.__name__} import 了网关包: {bad}", False)
+        break
+else:
+    check("worker SDK 无对网关包的 import 语句", True)
+
+print("\n" + ("=" * 40))
+if failures:
+    print(f"❌ {len(failures)} 项未通过: {failures}")
+    sys.exit(1)
+print("✅ Worker 进程隔离性全部通过")

uv.lock

@@ -306,7 +306,7 @@ source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-password.git
 [[package]]
 name = "chongming-request"
 version = "0.1.0"
-source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-request.git#a3b8dd032331660af7256eef360f075b45d6222d" }
+source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-request.git#3800d5fbb61c3f87937bf2c1a7d0762278c94db9" }
 dependencies = [
     { name = "eclipse-zenoh" },
 ]
@@ -319,9 +319,10 @@ source = { git = "http://124.71.68.6:3000/chongming/chongming-stream-manager.git
 [[package]]
 name = "chongming-worker"
 version = "0.1.0"
-source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#a58e92c21e5bf4b3ada55b2abc894e8ead7d17bc" }
+source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#c0a5b45b72a606c2c69850403d435c4883b4ff78" }
 dependencies = [
     { name = "eclipse-zenoh" },
+    { name = "pydantic" },
     { name = "tomli" },
 ]
 

verify_example_plugin.py

@@ -16,7 +16,7 @@ def check(name, cond, detail=""):
 
 
 # ========== 1. global_var 白名单 / require / __all__ ==========
-from plugins.example import global_var as gv
+from plugins.example import global_var as gv  # noqa: E402
 
 check("global_var.__all__ 包含 db", "db" in gv.__all__)
 check("global_var.__all__ 包含 set_global_var", "set_global_var" in gv.__all__)
@@ -45,7 +45,7 @@ check("模块属性 db 已被更新", gv.db is marker)
 
 
 # ========== 2. __init__.py 导入链（router + Plugin 基类）==========
-import plugins.example as plug
+import plugins.example as plug  # noqa: E402
 
 check("plugins.example 导入成功", True)
 check("create_workers 存在", callable(getattr(plug, "create_workers", None)))
@@ -54,7 +54,7 @@ check("plugin 实例存在", getattr(plug, "plugin", None) is not None)
 check("plugin._initialized 默认 False", plug.plugin._initialized is False)
 
 # ========== 3. create_workers 幂等性（mock session/config）==========
-import asyncio
+import asyncio  # noqa: E402
 
 # 用桩替代 Plugin 基类的 worker 创建实现，避免真实 zenoh 会话
 calls = []