modify: 混合模式——按插件 load_mode 共存（inline/external 由 Docker/K8s 编排）

README.md

@@ -340,7 +340,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=process）
+uv run worker_serve --type <name>               # 以独立进程运行 Worker（load_mode=external）
 uv pip install -e utils/<name>                  # 安装/更新工具包
 ```
 
@@ -375,10 +375,12 @@ uv pip install -e utils/<name>                  # 安装/更新工具包
 ## 配置说明
 
 - **应用配置**：根目录 `config.toml`，包含环境、服务器、安全、MQTT 等全局设置
-- **插件加载模式**：`config.toml` 的 `[<env>.workers].load_mode`
+- **插件加载模式**：按插件粒度声明在 `config/workers/<name>/config.toml` 的 `[workers]` 节
-  - `"inline"`（默认）：插件在网关进程内加载
+  - `load_mode = "inline"`（默认）：插件在网关进程内加载
-  - `"process"`：插件以独立进程运行（`uv run worker_serve --type <name>`），
+  - `load_mode = "external"`：插件以独立进程运行（进程编排交给 Docker/K8s，
-    经 Zenoh 在 `chongming/routes/**` 注册路由，网关动态挂载 HTTP 端点
+    进程入口 `uv run worker_serve --type <name>`），经 Zenoh 在 `chongming/routes/**`
+    注册路由，网关动态挂载 HTTP 端点
+  - 两种模式可同时存在（混合部署）
 - **Zenoh 配置**：`config/zenoh_config.json5`
 - **插件注册**：`config/workers/<name>/config.toml` 的 `[workers]` 节，定义加载方式、副本数与运行时配置（旧全局 `config/workers/config.toml` 仅作兜底兼容）
 - **路由定义**：`config/workers/<name>/*.toml`，每个端点一个文件

config.toml

@@ -24,11 +24,6 @@ secret_key = "your-secret-key-change-in-production"
 debug = true
 env = "development"
 
-[development.workers]
-# load_mode: "inline" = 网关进程内加载插件（现状）；
-#            "process" = 独立 worker 进程经 Zenoh 注册路由（uv run worker_serve --type <name>）
-load_mode = "inline"
-
 [development.server]
 access_log = false
 host = "0.0.0.0"

src/chongming/__init__.py

@@ -62,6 +62,13 @@ def worker_serve():
     if wc is None:
         logger.error("未找到插件配置: %s（请检查 config/workers/%s/config.toml）", args.type, args.type)
         sys.exit(1)
+    if wc.get("load_mode", "inline") != "external":
+        logger.warning(
+            "[%s] 插件配置 load_mode 不是 \"external\"：网关也会在进程内加载该插件，"
+            "外部进程将与网关内副本重复服务。若以独立进程运行，请在插件 "
+            "config.toml 的 [workers] 节设置 load_mode = \"external\"。",
+            args.type,
+        )
 
     async def _run():
         # 1. Zenoh 会话

src/chongming/app/__init__.py

@@ -47,23 +47,24 @@ async def lifespan(app: FastAPI):
     # 4. WorkerManager
     manager = lifespan_setup.setup_worker_manager(app, session)
 
-    # 5. 加载 workers（inline 进程内加载 / process 独立进程 Zenoh 注册路由）
+    # 5. 插件加载（混合模式：inline 进程内加载；external 由独立进程运行，
-    load_mode = lifespan_setup.get_load_mode()
+    #    其路由经 Zenoh 动态注册——进程编排交给 Docker/K8s，网关不拉起进程）
-    worker_tasks = []
+    worker_tasks, custom_routers, inline_types, external_types = (
-    custom_routers = []
+        await lifespan_setup.load_workers(app, session, manager)
-    if load_mode == "process":
+    )
-        # process 模式：网关不 import 插件、不扫盘注册路由，
+    if external_types:
-        # 路由由独立 worker 进程经 chongming/routes/** 动态注册
+        # 注册表为 external 插件服务；inline 类型集合防止误报重复挂载
-        await lifespan_setup.setup_route_registry(app, session, manager)
+        await lifespan_setup.setup_route_registry(
-    else:
+            app, session, manager, inline_types=inline_types
-        # inline 模式（默认）：插件在网关进程内加载，自动发现自定义路由
+        )
-        worker_tasks, custom_routers = await lifespan_setup.load_workers(app, session, manager)
 
-        # 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)
@@ -87,13 +88,12 @@ async def lifespan(app: FastAPI):
     # 1. 停止 MQTT Bridge
     await lifespan_setup.shutdown_mqtt(app)
 
-    # 1.5 process 模式：停止路由注册表订阅
+    # 1.5 停止路由注册表订阅（存在 external 插件时）
     registry = getattr(app.state, "route_registry", None)
     if registry is not None:
         registry.stop()
 
-    # 2. 清理 workers（仅 inline 模式；process 模式下网关未加载插件）
+    # 2. 清理 inline workers（cleanup_workers 内部按 load_mode 过滤）
-    if load_mode != "process":
     await lifespan_setup.cleanup_workers(app)
 
     # 3. 取消刷新任务

src/chongming/app/core/lifespan_setup.py

@@ -222,31 +222,36 @@ def setup_worker_manager(app: FastAPI, session: zenoh.Session) -> WorkerManager:
     app.state.refresh_task = refresh_task
     return manager
 
-def get_load_mode() -> str:
+async def setup_route_registry(
+    app: FastAPI,
+    session: zenoh.Session,
+    manager: WorkerManager,
+    inline_types: list[str] | None = None,
+):
     """
-    读取插件加载模式（config.toml [<env>.workers].load_mode）。
+    启动动态路由注册表（订阅 chongming/routes/**）。
 
-    - "inline"（默认）: 网关进程内加载插件（现状行为）
+    存在 load_mode="external" 的插件时启动：external 插件以独立进程运行
-    - "process": 独立 worker 进程经 Zenoh 注册路由，网关不再 import 插件
+    （由 Docker/K8s 等编排拉起，网关不负责进程管理），其路由描述符经 Zenoh
-    """
+    动态注册；inline 插件仍在网关进程内加载，两者同时存在（混合模式）。
-    env_config = config.get(env, {})
-    return env_config.get("workers", {}).get("load_mode", "inline")
 
+    inline_types: 网关进程内加载的插件类型，注册表忽略其描述符以防重复挂载。
 
-async def setup_route_registry(app: FastAPI, session: zenoh.Session, manager: WorkerManager):
+    注意：external 插件的自定义 APIRouter（如 multipart 上传端点）无法由
-    """
+    网关发现（网关不 import 该插件），此类端点需保持 inline 或等 P2 专门通道。
-    process 模式：启动动态路由注册表（订阅 chongming/routes/**）。
-
-    注意：process 模式下插件自定义 APIRouter（如 multipart 上传端点）无法由
-    网关发现（网关不再 import 插件），此类端点需在 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"路由注册表已启动（load_mode=process, prefix={prefix}）")
+    logger.info(
+        f"路由注册表已启动（prefix={prefix}, 外部插件路由动态注册, "
+        f"inline 插件: {sorted(inline_types or []) or '无'}）"
+    )
     return registry
 
 
@@ -278,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 = []
@@ -294,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,
@@ -307,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,
         })
@@ -334,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"]
@@ -397,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:
@@ -413,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 仅启动一个）"""
@@ -534,10 +573,12 @@ async def shutdown_mqtt(app: FastAPI):
 
 
 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

@@ -45,12 +45,22 @@ class MQTTBridge:
         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")
         ]
-        self._subscribed_custom: set = set()
 
     async def start(self):
         if self._running:
@@ -80,18 +90,15 @@ class MQTTBridge:
                    + (f" 及 {len(self._compiled)} 个自定义主题" if self._compiled else ""))
         self._lt = asyncio.create_task(self._loop())
 
-    async def update_custom_topics(self, custom_topics: list):
+    async def update_dynamic_topics(self, custom_topics: list):
         """
-        动态更新自定义订阅主题（process 模式：随 worker 路由表变化推送）。
+        动态更新 external 插件的自定义订阅主题（RouteRegistry 推送）。
 
-        重建编译表并增量重订阅（退订已移除主题、订阅新增主题），
+        与构造时的 inline 静态主题合并后增量重订阅（退订已移除主题、订阅新增主题），
         运行中调用即生效。
         """
-        self._custom_topics = list(custom_topics or [])
+        self._dynamic_topics = list(custom_topics or [])
-        self._compiled = [
+        self._rebuild_compiled()
-            (ct, _compile_topic(ct.get("topic", "")))
-            for ct in self._custom_topics if ct.get("topic")
-        ]
         new_topics = {ct["topic"] for ct, _ in self._compiled}
         if not (self._running and self._client):
             self._subscribed_custom = new_topics

src/chongming/app/core/route_generator.py

@@ -238,8 +238,18 @@ def register_route_group(
     return added
 
 
-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)
@@ -254,6 +264,8 @@ 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

src/chongming/app/core/route_registry.py

@@ -69,10 +69,16 @@ class RouteRegistry:
 
         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()
@@ -90,12 +96,12 @@ class RouteRegistry:
             task.cancel()
 
     def attach_bridge(self, mqtt_bridge):
-        """挂接 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_custom_topics(topics))
+            self._loop.create_task(mqtt_bridge.update_dynamic_topics(topics))
 
     @property
     def route_types(self) -> List[str]:
@@ -133,6 +139,10 @@ class RouteRegistry:
             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
@@ -235,9 +245,9 @@ class RouteRegistry:
         # 路由表变化后 OpenAPI 缓存失效，下一次访问时重建
         self._app.openapi_schema = None
 
-        # 3. 同步 MQTT 自定义订阅主题
+        # 3. 同步 external 插件的 MQTT 动态订阅主题
         if self._mqtt_bridge is not None:
-            await self._mqtt_bridge.update_custom_topics(self._collect_mqtt_topics())
+            await self._mqtt_bridge.update_dynamic_topics(self._collect_mqtt_topics())
 
     def _collect_mqtt_topics(self) -> List[Dict[str, Any]]:
         """汇总所有插件 __meta__ 声明的自定义订阅主题。"""

test_process_mode_e2e.py

@@ -1,9 +1,9 @@
 """
-process 模式端到端测试
+混合模式（external 插件）端到端测试
 
 前置条件:
-  1. 网关已以 load_mode = "process" 启动（config.toml [development.workers]）
+  1. 网关正常运行（uv run serve）；example 插件配置为 external：
-     uv run serve
+     config/workers/example/config.toml [workers] 节 load_mode = "external"
   2. 本脚本自己拉起/停止独立 worker 进程（uv run worker_serve --type example）
 
 验证:

test_route_registry.py

@@ -1,10 +1,11 @@
 """
 RouteRegistry 单元测试（loopback Zenoh 会话 + httpx ASGI 功能断言）
 
-验证 process 模式核心链路：
+验证 external 插件路由动态注册核心链路（含混合模式防双注册）：
   RoutePublisher.put 描述符 → 网关 RouteRegistry 动态挂载 FastAPI 路由（命中返回 503，worker 桩无实例）
   RoutePublisher.delete      → 动态摘除路由（404）
-  __meta__ 中的 mqtt_topics → 推送给 MQTT Bridge（update_custom_topics）
+  __meta__ 中的 mqtt_topics → 推送给 MQTT Bridge（update_dynamic_topics）
+  set_inline_types           → 忽略 inline 插件的描述符（防双注册）
 
 注：FastAPI 0.139 的 include_router 将路由包装为 _IncludedRouter（惰性挂载），
     不能靠遍历 app.router.routes 的 path 判断，故用 ASGI 请求状态码做功能断言。
@@ -37,12 +38,12 @@ class FakeWorkerManager:
 
 
 class FakeBridge:
-    """记录 update_custom_topics 的调用结果。"""
+    """记录 update_dynamic_topics 的调用结果。"""
 
     def __init__(self):
         self.topics = None
 
-    async def update_custom_topics(self, topics):
+    async def update_dynamic_topics(self, topics):
         self.topics = list(topics)
 
 
@@ -145,6 +146,26 @@ async def main():
             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()