modify: worker 运行时下沉 chongming-worker SDK——进程插件与网关零导入关系

README.md

@@ -384,6 +384,13 @@ uv pip install -e utils/<name>                  # 安装/更新工具包
     进程入口 `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,4 +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_serve"
+worker_serve = "chongming_worker.process:worker_serve"

src/chongming/__init__.py

@@ -23,126 +23,6 @@ def serve():
     uvicorn.run("chongming.app:app", **server_config)
 
 
-def worker_serve():
-    """
-    独立 Worker 进程入口（process 模式）
-
-    流程: 读取全局配置 → 发现本插件的 config/workers/<type>/config.toml →
-    import 插件（触发 @rpc_method 注册）→ 契约校验（硬失败则退出）→
-    创建 Worker 副本（count 进程内保留）→ 经 Zenoh 发布路由描述符 →
-    等待 SIGTERM/SIGINT → 摘除路由 + cleanup + 关闭会话。
-
-    用法: uv run worker_serve --type example [--count 3] [--config config.toml]
-    需在项目根目录运行（配置与插件均按相对路径解析）。
-    """
-    import argparse
-    import asyncio
-    import inspect
-    import os
-    import signal
-    import sys
-
-    import zenoh
-
-    from .app.core.contracts import validate_worker_contracts
-    from .app.core.lifespan_setup import _discover_plugins
-    from .app.core.route_publisher import RoutePublisher
-
-    parser = argparse.ArgumentParser(description="chongming 独立 Worker 进程")
-    parser.add_argument("--type", required=True, help="插件类型（如 example / auth）")
-    parser.add_argument("--count", type=int, default=None, help="副本数（覆盖配置中的 count）")
-    parser.add_argument("--config", default="config.toml", help="全局配置文件路径")
-    args = parser.parse_args()
-
-    # 控制台脚本启动时 sys.path 不含项目根目录，显式加入以保证
-    # plugins.<name>（inline 插件）与 workers.<name>（进程插件）均可导入
-    project_root = os.path.abspath(os.getcwd())
-    if project_root not in sys.path:
-        sys.path.insert(0, project_root)
-
-    config = load_config()
-    logger = get_logger("app")
-
-    plugins = _discover_plugins()
-    wc = next((p for p in plugins if p.get("type") == args.type), None)
-    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 会话
-        zconfig = zenoh.Config()
-        zconfig.from_file(config.get("zenoh", {}).get("json5_path", "config/zenoh_config.json5"))
-        session = zenoh.open(zconfig)
-        logger.info("[%s] Zenoh 会话已建立", args.type)
-
-        try:
-            # 2. import 插件（触发 @rpc_method 注册），然后契约校验
-            module = __import__(wc["module"], fromlist=[wc["factory"]])
-            routes = validate_worker_contracts(args.type)
-
-            # 3. 创建 Worker 副本（worker_manager=None：跨进程经 liveliness 发现）
-            factory = getattr(module, wc["factory"])
-            count = args.count or wc.get("count", 1)
-            if inspect.iscoroutinefunction(factory):
-                tasks = await factory(session, wc["config"], count, worker_manager=None)
-            else:
-                tasks = factory(session, wc["config"], count, worker_manager=None)
-
-            # 4. 发布路由描述符（网关 RouteRegistry 订阅后动态挂载 HTTP 路由）
-            publisher = RoutePublisher(session, args.type)
-            mqtt_topics = wc.get("mqtt_topics") or []
-            publisher.publish_routes(routes, mqtt_topics)
-
-            # 5. 等待退出信号；期间周期性重发布（网关后启动/样本丢失时自愈）
-            stop_event = asyncio.Event()
-            loop = asyncio.get_running_loop()
-            for sig in (signal.SIGTERM, signal.SIGINT):
-                loop.add_signal_handler(sig, stop_event.set)
-
-            async def _announce():
-                while not stop_event.is_set():
-                    try:
-                        await asyncio.wait_for(stop_event.wait(), timeout=10.0)
-                        return
-                    except asyncio.TimeoutError:
-                        publisher.publish_routes(routes, mqtt_topics)
-
-            announce_task = asyncio.create_task(_announce())
-            logger.info("[%s] Worker 进程就绪，等待退出信号...", args.type)
-            await stop_event.wait()
-            announce_task.cancel()
-            await asyncio.gather(announce_task, return_exceptions=True)
-
-            # 6. 优雅退出：摘除路由 → cleanup → 取消副本任务 → 关闭会话
-            publisher.unpublish()
-            cleanup = getattr(module, wc.get("cleanup", ""), None)
-            if cleanup is not None:
-                worker_id = f"worker/{args.type}/worker-i-{os.getpid()}"
-                if inspect.iscoroutinefunction(cleanup):
-                    await cleanup(worker_id)
-                else:
-                    cleanup(worker_id)
-            for t in tasks:
-                t.cancel()
-            if tasks:
-                await asyncio.gather(*tasks, return_exceptions=True)
-            logger.info("[%s] Worker 进程已退出", args.type)
-        finally:
-            session.close()
-
-    try:
-        asyncio.run(_run())
-    except KeyboardInterrupt:
-        pass
-
 def gunicorn_serve():
     """
     使用 gunicorn 启动服务

src/chongming/app/core/contracts.py

@@ -1,166 +1,31 @@
 """
-启动契约校验（config/workers TOML ↔ @rpc_method 注册表 ↔ gen_models 生成模型）
+启动契约校验（网关 inline 模式）
 
-挂载于 lifespan 的 load_workers 与 register_routes 之间，此时插件模块已导入
-（@rpc_method 注册表就绪）且 TOML 路由尚未注册到 FastAPI。
+校验逻辑的唯一实现在 chongming_worker.contracts（网关与 worker 进程共享），
+本模块只做网关侧的编排：对 inline 插件批量校验、日志汇总、硬失败终止启动。
 
-策略:
-  - 硬失败（汇总后 raise，启动终止）:
-    HTTP 路由（GET/POST/PUT/DELETE/PATCH 且 internal != true）在注册表中
-    缺少对应 subject#METHOD 的 @rpc_method → 该路由运行时必然 503，提前失败。
-  - 软警告（仅日志汇总，不阻断启动）:
-    [MQTT] / internal 路由缺少注册、已注册方法无 TOML 路由、
-    TOML 声明 params/response_model 但生成模型缺失（gen_models 过期）、孤儿模型。
+external 插件在自己进程内做同样校验（validate_worker_contracts），网关侧跳过。
 """
-import importlib
-from typing import Any, Dict, List, Optional, Set
+from typing import Any, Dict, List
 
 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",
-        f"workers.{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))
-                # 用 obj.__name__ 而非 dir() 名称：gen_models 生成的
-                # <Plugin>Response 别名与真实模型同一对象，别名保留原类名，
-                # 按对象去重后自动收敛到真实模型名（避免别名抢占排序位）
-                names.add(obj.__name__)
-        return names
-    return None
-
-
-def _check_plugin(
-    worker_type: str,
-    toml_routes: List[Dict[str, Any]],
-    registered: Dict[str, Any],
-    model_names: Optional[Set[str]],
-) -> tuple[List[str], List[str]]:
-    """
-    对单个插件执行 TOML ↔ 注册表 ↔ 生成模型一致性校验。
-
-    Args:
-        worker_type: 插件类型
-        toml_routes: load_worker_routes_from_dir() 输出的路由列表
-        registered: get_all_rpc_methods().get(worker_type, {}) 注册表
-        model_names: 插件 models.pydantic 的模型名集合（None 表示无生成模型）
-
-    Returns:
-        (hard_errors, warnings): 硬错误与软警告消息列表
-    """
-    hard_errors: List[str] = []
-    warnings: List[str] = []
-    toml_keys = {f"{r['subject']}#{r['method']}" for r in toml_routes}
-
-    for r in toml_routes:
-        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 清理）"
-            )
-
-    return hard_errors, warnings
-
 
 def validate_contracts(plugins: List[Dict[str, Any]]) -> None:
     """
-    对每个发现的插件执行三方一致性校验（网关 inline 模式，启动时）。
+    对每个发现的 inline 插件执行三方一致性校验。
+
+    挂载于 lifespan 的 load_workers 与 register_routes 之间，此时插件模块已导入
+    （@rpc_method 注册表就绪）且 TOML 路由尚未注册到 FastAPI。
 
     Args:
-        plugins: _discover_plugins() 返回的插件配置列表
+        plugins: _discover_plugins() 返回的 inline 插件配置列表
 
     Raises:
         RuntimeError: 存在硬失败项（HTTP 路由缺失 @rpc_method 注册）时
@@ -176,8 +41,8 @@ def validate_contracts(plugins: List[Dict[str, Any]]) -> None:
 
         toml_routes = load_worker_routes_from_dir(worker_type)
         registered = registry.get(worker_type, {})
-        model_names = _load_plugin_models(worker_type)
-        hard, warns = _check_plugin(worker_type, toml_routes, registered, model_names)
+        model_names = load_plugin_models(worker_type)
+        hard, warns = check_plugin(worker_type, toml_routes, registered, model_names)
         hard_errors.extend(hard)
         warnings.extend(warns)
 
@@ -197,45 +62,3 @@ def validate_contracts(plugins: List[Dict[str, Any]]) -> None:
         )
 
     logger.info("契约校验通过（TOML 路由 ↔ @rpc_method 注册表一致）")
-
-
-def validate_worker_contracts(worker_type: str) -> List[Dict[str, Any]]:
-    """
-    Worker 进程侧契约校验（process 模式）：校验单个插件并返回可发布的路由描述符。
-
-    调用时机：worker 进程 import 插件模块后（@rpc_method 注册表就绪）、
-    发布路由描述符之前。校验逻辑与网关 validate_contracts 共用 _check_plugin。
-
-    Args:
-        worker_type: 插件类型（与 config/workers/<type> 目录一致）
-
-    Returns:
-        通过校验的 TOML 路由描述符列表（直接用于 RoutePublisher 发布）
-
-    Raises:
-        RuntimeError: 存在硬失败项（HTTP 路由缺失 @rpc_method 注册）时，
-                      调用方应终止 worker 进程，避免发布残缺路由。
-    """
-    registered = get_all_rpc_methods().get(worker_type, {})
-    toml_routes = load_worker_routes_from_dir(worker_type)
-    model_names = _load_plugin_models(worker_type)
-
-    hard_errors, warnings = _check_plugin(
-        worker_type, toml_routes, registered, model_names
-    )
-
-    for w in warnings:
-        logger.warning("  ⚠ %s", w)
-
-    if hard_errors:
-        logger.error("Worker 契约校验失败，发现 %d 个硬性错误:", len(hard_errors))
-        for e in hard_errors:
-            logger.error("  ✗ %s", e)
-        raise RuntimeError(
-            f"[{worker_type}] 契约校验失败: " + "; ".join(hard_errors)
-        )
-
-    logger.info(
-        "[%s] 契约校验通过，可发布 %d 个路由描述符", worker_type, len(toml_routes)
-    )
-    return toml_routes

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,94 +112,6 @@ class RouteConfig:
         ]
 
 
-# 默认的 worker 配置目录
-WORKER_CONFIG_DIR = "config/workers"
-
-
-def load_worker_routes_from_dir(worker_type: str) -> List[Dict[str, Any]]:
-    """
-    从 config/workers/{worker_type}/ 目录加载路由配置。
-
-    目录结构:
-      config/workers/{worker_type}/
-        config.toml           # [worker] 分组信息 + registration + defaults
-        add.toml              # 接口名 = 文件名（不含扩展名）
-          [GET]
-          summary = ...
-          params = [...]
-          response_model = { ... }
-          [POST]
-          ...                 # 同一个接口支持多个请求方法
-          [MQTT]
-          ...                 # MQTT 专用端点（不注册 HTTP 路由）
-
-    返回: List[Dict], 每个 dict 包含 flatten 后的路由配置,
-          每个方法节(如 GET/POST/MQTT) 生成一个独立的路由条目。
-    """
-    from pathlib import Path
-    import tomllib
-
-    base_dir = Path(WORKER_CONFIG_DIR) / worker_type
-    if not base_dir.is_dir():
-        logger.warning(f"Worker 配置目录不存在: {base_dir}")
-        return []
-
-    # 1. 读取分组配置
-    group_cfg_path = base_dir / "config.toml"
-    group_defaults: Dict[str, Any] = {}
-    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_route_group(
     app: FastAPI,
     worker_manager: WorkerManager,

src/chongming/app/core/route_publisher.py

@@ -1,90 +0,0 @@
-"""
-路由描述符发布器（worker 进程侧）
-
-独立 worker 进程在完成契约校验后，将路由描述符 put 到 Zenoh 的
-`chongming/routes/<type>/...` 命名空间；进程优雅退出时 delete 摘除。
-
-描述符 schema_version=1，字段与网关 RouteConfig 完全对位（TOML 兼容 dict），
-网关 RouteRegistry 订阅同名空间并动态挂载 HTTP 路由。
-"""
-import json
-from typing import Any, Dict, List, Optional
-
-import zenoh
-
-from .logger import get_logger
-
-logger = get_logger("route_publisher")
-
-ROUTES_PREFIX = "chongming/routes"
-META_KEY = "__meta__"
-SCHEMA_VERSION = 1
-
-
-def route_id(subject: str, method: str) -> str:
-    """route_id = subject 与 method 的编码（避免 keyexpr 特殊字符歧义）"""
-    return f"{subject.replace('.', '_')}__{method.upper()}"
-
-
-class RoutePublisher:
-    """发布/摘除某 worker_type 的路由描述符。"""
-
-    def __init__(self, session: zenoh.Session, worker_type: str):
-        self._session = session
-        self._worker_type = worker_type
-        self._keys: List[str] = []
-
-    def publish_routes(
-        self,
-        routes: List[Dict[str, Any]],
-        mqtt_topics: Optional[List[Dict[str, Any]]] = None,
-    ):
-        """
-        发布路由描述符。
-
-        Args:
-            routes: 契约校验通过的路由描述符列表（TOML 兼容格式）
-            mqtt_topics: 插件声明的 [[workers.mqtt_topics]]（可空）
-        """
-        for r in routes:
-            subject = r.get("subject", "")
-            method = r.get("method", "GET")
-            if not subject:
-                logger.warning("跳过无 subject 的路由描述符: %s", r)
-                continue
-            descriptor = dict(r)
-            descriptor["schema_version"] = SCHEMA_VERSION
-            descriptor["type"] = self._worker_type
-            key = f"{ROUTES_PREFIX}/{self._worker_type}/{route_id(subject, method)}"
-            self._session.put(key, json.dumps(descriptor, ensure_ascii=False).encode("utf-8"))
-            self._keys.append(key)
-            logger.info(
-                "已发布路由: %s %s (subject=%s)",
-                method.upper(), descriptor.get("path", "?"), subject,
-            )
-
-        if mqtt_topics:
-            meta_key = f"{ROUTES_PREFIX}/{self._worker_type}/{META_KEY}"
-            meta = {"schema_version": SCHEMA_VERSION, "mqtt_topics": mqtt_topics}
-            self._session.put(
-                meta_key, json.dumps(meta, ensure_ascii=False).encode("utf-8")
-            )
-            self._keys.append(meta_key)
-            logger.info("已发布插件元信息: %d 个自定义 MQTT 主题", len(mqtt_topics))
-
-        # 去重（周期性重发布会重复调用本方法）
-        self._keys = list(dict.fromkeys(self._keys))
-        logger.info(
-            "[%s] 路由发布完成: %d 端点 + %s", self._worker_type,
-            len(self._keys), "元信息" if mqtt_topics else "无元信息",
-        )
-
-    def unpublish(self):
-        """摘除全部已发布的路由描述符（优雅退出时调用）。"""
-        for key in self._keys:
-            try:
-                self._session.delete(key)
-            except Exception as e:  # pragma: no cover
-                logger.warning("摘除路由失败 [%s]: %s", key, e)
-        logger.info("[%s] 已摘除 %d 个路由描述符", self._worker_type, len(self._keys))
-        self._keys.clear()

src/chongming/app/core/route_registry.py

@@ -24,14 +24,13 @@ 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")
 
-ROUTES_PREFIX = "chongming/routes"
-META_KEY = "__meta__"
-SCHEMA_VERSION = 1
 # worker 重启风暴防抖窗口（秒）
 DEBOUNCE_SECONDS = 2.0
 

test_route_registry.py

@@ -17,8 +17,8 @@ import httpx
 import zenoh
 from fastapi import FastAPI
 
+from chongming_worker.routes_publisher import RoutePublisher
 from src.chongming.app.core.route_registry import RouteRegistry
-from src.chongming.app.core.route_publisher import RoutePublisher
 
 failures = []
 

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

@@ -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#4908378e597eb200e255c71df92d16e272b4b12e" }
+source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#c0a5b45b72a606c2c69850403d435c4883b4ff78" }
 dependencies = [
     { name = "eclipse-zenoh" },
+    { name = "pydantic" },
     { name = "tomli" },
 ]