diff --git a/.gitignore b/.gitignore
index 793a8e6..0404c1d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -33,6 +33,7 @@ config
 
 # .agents
 .github
+.zcode
 
 # lib
 lib
diff --git a/README.md b/README.md
index ddecdb6..cdf44a6 100644
--- a/README.md
+++ b/README.md
@@ -2,7 +2,7 @@
 
 基于 **FastAPI + Zenoh + MQTT + Vue 3** 的**配置驱动微服务网关平台**。
 
-通过 TOML 配置文件定义 RPC 端点，自动生成 Pydantic 模型与 FastAPI 路由，同时暴露 **HTTP** 和 **MQTT** 双通道。底层基于 Zenoh 分布式消息协议实现 RPC 通信，Worker 插件可独立开发、热加载、水平扩展。
+通过 TOML 配置文件定义 RPC 端点，自动生成 Pydantic 模型与 FastAPI 路由，同时暴露 **HTTP** 和 **MQTT** 双通道。底层基于 Zenoh 分布式消息协议实现 RPC 通信，Worker 插件可独立开发、按副本数水平扩展。
 
 ---
 
@@ -13,8 +13,9 @@
 | **FastAPI + Uvicorn** | 高性能异步 Web 框架，自动生成 OpenAPI 文档                           |
 | **Zenoh 服务发现**    | 基于 liveliness 的 Worker 注册与 RPC 调用，负载均衡 + 超时重试       |
 | **MQTT 集成**         | 内嵌 amqtt Broker，MQTT ↔ Worker 双向桥接                            |
-| **动态路由生成**      | 扫描 `config/workers/*.toml` 自动注册 HTTP 路由                      |
+| **动态路由生成**      | 扫描 `config/workers/*/*.toml` 自动注册 HTTP 路由                    |
 | **Worker 框架**       | `@rpc_method` 装饰器 + `Plugin` 基类统一生命周期 + Pydantic 模型绑定 |
+| **启动契约校验**      | TOML 路由 ↔ `@rpc_method` ↔ 生成模型三方一致性校验，fail fast        |
 | **自动化模型生成**    | `gen_models` 根据 TOML 配置生成请求/响应模型                         |
 | **插件脚手架**        | `gen_plugin` 一键生成风格统一的 Worker 插件骨架                      |
 | **JWT 鉴权**          | Rust 实现，路由级 `auth_required` 控制                               |
@@ -36,15 +37,15 @@ chongming-zenoh/
 ├── config/                       # 配置文件目录
 │   ├── zenoh_config.json5        # Zenoh 会话配置
 │   └── workers/                  # Worker 路由定义（TOML）
-│       ├── config.toml           # 插件注册列表
-│       ├── auth/                 # 鉴权插件（login / register / me / …）
+│       ├── config.toml           # 旧式全局插件注册（仅作兜底兼容）
+│       ├── auth/                 # 鉴权插件（config.toml = 插件注册 + 运行时配置）
 │       ├── doubao/               # 豆包视觉理解插件
 │       ├── aibox_sophon_adapter/ # AI Box 设备心跳监控插件
 │       └── example/              # 示例插件（四则运算 / MQTT 端点）
 ├── src/
 │   ├── chongming/                # 核心应用
 │   │   ├── app/
-│   │   │   ├── core/             # WorkerManager, RouteGenerator, MQTTBridge, auth
+│   │   │   ├── core/             # WorkerManager, RouteGenerator, MQTTBridge, contracts(契约校验), auth
 │   │   │   └── routes/           # 内置路由（文件上传等）
 │   │   └── scripts/              # build, gen_models, gen_plugin
 │   ├── chongming-web/            # Vue 3 前端
@@ -211,17 +212,34 @@ async def upload(file: UploadFile = File(...)):
 ```toml
 # config/workers/aibox_sophon_adapter/config.toml
 [[workers.mqtt_topics]]
-topic = "/board_ping"                       # 支持 +（单级）/ #（多级，须在末尾）通配符
+topic = "box/aibox_sophon/board_ping"         # 支持 +（单级）/ #（多级，须在末尾）通配符
 subject = "aibox_sophon_adapter.board_ping" # 可选，{1} {2}... 对应通配符捕获值
 response_topic = "/board_ping/response"     # 可选，响应主题；默认 "<实际主题>/response"
 ```
 
 - `subject` 缺省为 `<插件类型>.<topic 最后一段>`；`response_topic` 缺省为 `<实际匹配主题>/response`。
 - 网关发现插件时自动收集所有 `[[workers.mqtt_topics]]`，MQTT Bridge 一并订阅并路由。
-- 请求负载同样支持 `_method` / `_timeout` / `_cacheable` 控制字段。
 - 多条自定义主题按配置顺序匹配，先匹配者优先；内置 `chongming/...` 接口最后兜底。
 - 全局 `[<env>.mqtt].custom_topics` 仍可作为通用兜底（不依赖具体插件时使用，可选）。
 
+### 请求负载格式（信封协议）
+
+MQTT 请求负载为信封结构，控制字段与业务数据分离：
+
+```json
+{
+  "meta": {
+    "method": "GET",      // 可选，默认 GET；MQTT 专用端点需显式 "MQTT"
+    "timeout": 30,        // 可选，RPC 超时（秒），默认 30
+    "cacheable": false    // 可选，是否允许结果缓存，默认 false
+  },
+  "data": { "a": 3, "b": 5 }   // 业务参数，传给 Worker
+}
+```
+
+- 非信封格式的负载会被网关丢弃并告警（严格协议，设备侧请按此格式发布）。
+- HTTP 通道由网关自动组装信封（注入 method / user_info），无需调用方关心。
+
 ### 快速测试
 
 ```bash
@@ -243,14 +261,19 @@ python publisher.py example add '{"a":3,"b":5}'  # 调用 add
 ### 添加新 Worker 插件
 
 ```text
-1. 创建 src/plugins/<name>/ — __init__.py, global_var.py, router/, models/
-2. 注册到 config/workers/config.toml 的 [[workers]]
+1. 创建 src/plugins/<name>/ — __init__.py（Plugin 子类）, global_var.py, router/, models/
+2. 创建 config/workers/<name>/config.toml，在 [workers] 节注册插件
+   （type / module / factory / cleanup / count + [workers.config] 运行时配置）
 3. 创建 config/workers/<name>/*.toml 端点定义
 4. 运行 uv run python -m chongming.scripts.gen_models 生成模型
 5. 实现 router/<op>.py，引用 models/pydantic.py 中的模型
 6. 更新 router/__init__.py 导入
 ```
 
+> 推荐用脚手架一步生成骨架：`uv run gen_plugin <name> [operation ...]`。
+> 启动时网关会做**契约校验**：TOML 路由缺少对应 `@rpc_method` 将导致启动失败，
+> 模型缺失/孤儿等不一致以警告日志提示。
+
 ### 为现有插件添加新端点
 
 ```text
@@ -303,7 +326,7 @@ uv pip install -e utils/<name>                  # 安装/更新工具包
 
 - **应用配置**：根目录 `config.toml`，包含环境、服务器、安全、MQTT 等全局设置
 - **Zenoh 配置**：`config/zenoh_config.json5`
-- **Worker 注册**：`config/workers/config.toml`，定义加载的插件及其副本数
+- **插件注册**：`config/workers/<name>/config.toml` 的 `[workers]` 节，定义加载方式、副本数与运行时配置（旧全局 `config/workers/config.toml` 仅作兜底兼容）
 - **路由定义**：`config/workers/<name>/*.toml`，每个端点一个文件
 
 详细配置说明见 [`config/README.md`](config/README.md)。
@@ -339,7 +362,7 @@ uv pip install -e utils/<name>            # 安装/更新工具包子包
 
 | 字段                 | 位置                  | 说明                                    |
 | -------------------- | --------------------- | --------------------------------------- |
-| `count`              | `workers/config.toml` | Worker 副本数                           |
+| `count`              | `<插件>/config.toml`  | Worker 副本数（`[workers]` 节）         |
 | `router_prefix`      | `<插件>/config.toml`  | API 路由前缀                            |
 | `defaults.timeout`   | `<插件>/config.toml`  | 默认超时（秒）                          |
 | `defaults.cacheable` | `<插件>/config.toml`  | 是否缓存 RPC 结果                       |
diff --git a/publisher.py b/publisher.py
index fabf0a8..7fa251c 100644
--- a/publisher.py
+++ b/publisher.py
@@ -3,10 +3,11 @@ MQTT Publisher — 向 chongming Worker 发送 MQTT 请求
 
 用法:
   python publisher.py <worker_type> <operation> [json_payload]
-  
-示例:
-  python publisher.py example add '{"a": 3, "b": 5, "_method": "GET"}'
-  python publisher.py doubao analyze '{"text": "hello", "_method": "POST"}'
+
+示例（信封协议，meta 可选）:
+  python publisher.py example add '{"meta": {"method": "GET"}, "data": {"a": 3, "b": 5}}'
+  python publisher.py example add '{"a": 3, "b": 5}'          # 扁平载荷自动包装（method=GET）
+  python publisher.py example add '{"meta": {"method": "MQTT"}, "data": {"a": 3, "b": 5}}'
 """
 
 import asyncio
@@ -24,7 +25,7 @@ PREFIX = "chongming"
 # 默认参数
 DEFAULT_WORKER = "example"
 DEFAULT_OP = "add"
-DEFAULT_PAYLOAD = {"a": 3, "b": 5, "_method": "GET"}
+DEFAULT_PAYLOAD = {"meta": {"method": "GET"}, "data": {"a": 3, "b": 5}}
 
 async def main():
     worker_type = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_WORKER
@@ -33,6 +34,9 @@ async def main():
 
     topic = f"{PREFIX}/{worker_type}/{operation}"
     payload = json.loads(payload_str)
+    # 扁平载荷自动包装为信封（method 默认 GET），已含 meta/data 则原样使用
+    if not (isinstance(payload, dict) and ("meta" in payload or "data" in payload)):
+        payload = {"meta": {"method": "GET"}, "data": payload}
 
     client = MQTTClient()
     await client.connect("mqtt://localhost:1883/")
diff --git a/pyproject.toml b/pyproject.toml
index 235c363..863fff5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -56,11 +56,11 @@ chongming-request = { git = "http://124.71.68.6:3000/chongming/chongming-utils-r
 chongming-jwt = { git = "http://124.71.68.6:3000/chongming/chongming-utils-jwt.git" }
 chongming-config = { git = "http://124.71.68.6:3000/chongming/chongming-utils-config.git" }
 chongming-password = { git = "http://124.71.68.6:3000/chongming/chongming-utils-password.git"  }
-chongming-worker = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git" }
 bytedance = { git = "http://124.71.68.6:3000/chongming/bytedance.git" }
 chongming-stream-manager = { git = "http://124.71.68.6:3000/chongming/chongming-stream-manager.git" }
 chongming-database = { git = "http://124.71.68.6:3000/chongming/chongming-utils-database.git" }
 chongming-mqtt = { git = "http://124.71.68.6:3000/chongming/chongming-mqtt.git" }
+chongming-worker = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git"}
 
 [project.scripts]
 build = "chongming.scripts.build:build"
diff --git a/src/chongming/app/__init__.py b/src/chongming/app/__init__.py
index f0e132d..3e036d2 100644
--- a/src/chongming/app/__init__.py
+++ b/src/chongming/app/__init__.py
@@ -50,6 +50,9 @@ async def lifespan(app: FastAPI):
     # 5. 加载 workers（同时自动发现插件自定义路由）
     worker_tasks, custom_routers = await lifespan_setup.load_workers(app, session, manager)
 
+    # 5.5 契约校验（TOML 路由 ↔ @rpc_method ↔ 生成模型），失败则启动终止
+    lifespan_setup.validate_contracts(app)
+
     # 6. 注册路由（含 TOML 动态路由 + 插件自定义路由）
     await lifespan_setup.register_routes(app, manager, custom_routers)
 
diff --git a/src/chongming/app/core/contracts.py b/src/chongming/app/core/contracts.py
new file mode 100644
index 0000000..98e0797
--- /dev/null
+++ b/src/chongming/app/core/contracts.py
@@ -0,0 +1,168 @@
+"""
+启动契约校验（config/workers TOML ↔ @rpc_method 注册表 ↔ gen_models 生成模型）
+
+挂载于 lifespan 的 load_workers 与 register_routes 之间，此时插件模块已导入
+（@rpc_method 注册表就绪）且 TOML 路由尚未注册到 FastAPI。
+
+策略:
+  - 硬失败（汇总后 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, Optional, Set
+
+from chongming_worker import get_all_rpc_methods
+
+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:
+    """
+    对每个发现的插件执行三方一致性校验。
+
+    Args:
+        plugins: _discover_plugins() 返回的插件配置列表
+
+    Raises:
+        RuntimeError: 存在硬失败项（HTTP 路由缺失 @rpc_method 注册）时
+    """
+    registry = get_all_rpc_methods()
+    hard_errors: List[str] = []
+    warnings: List[str] = []
+
+    for wc in plugins:
+        worker_type = wc.get("type", "")
+        if not worker_type:
+            continue
+
+        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)
+
+        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 清理）"
+                )
+
+    # 输出结果
+    if warnings:
+        logger.warning("契约校验发现 %d 个警告:", len(warnings))
+        for w in warnings:
+            logger.warning("  ⚠ %s", w)
+
+    if hard_errors:
+        logger.error("契约校验失败，发现 %d 个硬性错误:", len(hard_errors))
+        for e in hard_errors:
+            logger.error("  ✗ %s", e)
+        raise RuntimeError(
+            "启动契约校验失败: "
+            + "; ".join(hard_errors)
+        )
+
+    logger.info("契约校验通过（TOML 路由 ↔ @rpc_method 注册表一致）")
diff --git a/src/chongming/app/core/lifespan_setup.py b/src/chongming/app/core/lifespan_setup.py
index 0e818ac..3806314 100644
--- a/src/chongming/app/core/lifespan_setup.py
+++ b/src/chongming/app/core/lifespan_setup.py
@@ -21,6 +21,11 @@ from ..core.mqtt_bridge import MQTTBridge
 config = get_config()
 logger = get_logger("app")
 
+# 挂载框架/插件父 logger 的 handler：logger 模块只为具名 logger 挂 handler，
+# 子 logger（chongming_worker.runtime、plugins.doubao 等）经 propagate 到父级才可见
+get_logger("chongming_worker")
+get_logger("plugins")
+
 try:
     env = config["default"]["env"]
 except KeyError:  # pragma: no cover
@@ -377,6 +382,19 @@ async def register_routes(app: FastAPI, manager: WorkerManager, custom_routers:
             app.include_router(r, prefix=prefix)
         logger.info(f"已注册 {len(custom_routers)} 个插件自定义路由（prefix={prefix}）")
 
+
+def validate_contracts(app: FastAPI | None = None):
+    """
+    契约校验（TOML 路由 ↔ @rpc_method 注册表 ↔ gen_models 生成模型）。
+
+    在 load_workers 之后、register_routes 之前调用：此时插件模块已导入
+    （注册表就绪）。HTTP 路由缺失对应 @rpc_method 时抛 RuntimeError 终止启动，
+    其余不一致以警告日志汇总。
+    """
+    from ..core.contracts import validate_contracts as _validate
+
+    _validate(_discover_plugins())
+
 def setup_mqtt_broker(app: FastAPI):
     """启动 MQTT Broker 子进程（原子锁确保多 worker 仅启动一个）"""
     env_config = config.get(env, {})
diff --git a/src/chongming/app/core/mqtt_bridge.py b/src/chongming/app/core/mqtt_bridge.py
index aa2a9d1..70b8ddd 100644
--- a/src/chongming/app/core/mqtt_bridge.py
+++ b/src/chongming/app/core/mqtt_bridge.py
@@ -97,7 +97,19 @@ class MQTTBridge:
         t: str = m.topic; p = m.data  # type: ignore
         try: pl = json.loads(p.decode("utf-8"))
         except json.JSONDecodeError: return
-        pl.setdefault("_method", "GET")
+
+        # 信封协议（严格）: {"meta": {...}, "data": {...}}
+        # 非信封格式载荷直接丢弃并告警，请设备侧按新协议发布
+        if not isinstance(pl, dict) or ("meta" not in pl and "data" not in pl):
+            logger.warning(
+                "丢弃非信封格式载荷 [%s]: 请使用 {'meta': {...}, 'data': {...}} 格式", t
+            )
+            return
+        meta = pl.get("meta") or {}
+        data = pl.get("data") or {}
+        if not isinstance(data, dict):
+            data = {}
+        pl = {"meta": meta, "data": data}
 
         # 1. 优先匹配显式配置的自定义订阅主题（顺序即优先级）
         for ct, compiled in self._compiled:
@@ -136,9 +148,11 @@ class MQTTBridge:
 
     async def _route(self, wt: str, subj: str, pl: dict, rt: str):
         logger.info(f"MQTT -> {subj}")
+        meta = pl.get("meta") or {}
         try:
             r = await self._wm.call(worker_type=wt, subject=subj, payload=pl,
-                timeout=float(pl.pop("_timeout",30)), cacheable=bool(pl.pop("_cacheable",False)))
+                timeout=float(meta.get("timeout", 30)),
+                cacheable=bool(meta.get("cacheable", False)))
             if r is not None: await self._pub(rt, r)
         except RuntimeError as e: await self._pub(rt, {"error":str(e)})
         except Exception as e: await self._pub(rt, {"error":str(e)})
diff --git a/src/chongming/app/core/route_generator.py b/src/chongming/app/core/route_generator.py
index 34d7467..31a89fc 100644
--- a/src/chongming/app/core/route_generator.py
+++ b/src/chongming/app/core/route_generator.py
@@ -186,6 +186,8 @@ def load_worker_routes_from_dir(worker_type: str) -> List[Dict[str, Any]]:
             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(
@@ -268,11 +270,12 @@ def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_t
         body: RequestModel,  # type: ignore
     ) -> Any:
         try:
-            payload = body.model_dump()
-            payload.setdefault("_method", http_method)
+            # 信封协议: {"meta": {...}, "data": {...}}
+            meta = {"method": http_method}
             user_info = getattr(request.state, "user", None)
             if user_info:
-                payload["user_info"] = user_info
+                meta["user_info"] = user_info
+            payload = {"meta": meta, "data": body.model_dump()}
             result = await worker_manager.call(
                 worker_type=worker_type,
                 subject=subject,
@@ -313,11 +316,12 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
     # 无参数情况
     if not param_spec_combined:
         async def _simple_endpoint(request: Request) -> Any:
-            payload = {}
-            payload.setdefault("_method", http_method)
+            # 信封协议: {"meta": {...}, "data": {...}}
+            meta = {"method": http_method}
             user_info = getattr(request.state, "user", None)
             if user_info:
-                payload["user_info"] = user_info
+                meta["user_info"] = user_info
+            payload = {"meta": meta, "data": {}}
             try:
                 result = await worker_manager.call(
                     worker_type=worker_type,
@@ -339,11 +343,12 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
     param_names = list(param_spec_combined.keys())
 
     async def endpoint(request: Request, **kwargs: Any) -> Any:
-        payload = {k: kwargs[k] for k in param_names}
-        payload.setdefault("_method", http_method)
+        # 信封协议: {"meta": {...}, "data": {...}}
+        meta = {"method": http_method}
         user_info = getattr(request.state, "user", None)
         if user_info:
-            payload["user_info"] = user_info
+            meta["user_info"] = user_info
+        payload = {"meta": meta, "data": {k: kwargs[k] for k in param_names}}
         try:
             result = await worker_manager.call(
                 worker_type=worker_type,
@@ -394,7 +399,7 @@ def _register_single_route(
     # MQTT 端点：不注册 HTTP 路由，仅通过 MQTT Bridge 调用
     if rc.method == "MQTT":
         logger.info(
-            f"MQTT 端点: subject={worker_type}.{rc.subject} (仅 MQTT，不注册 HTTP 路由)"
+            f"MQTT 端点: subject={rc.subject} (仅 MQTT，不注册 HTTP 路由)"
         )
         return
 
diff --git a/test_auth_race.py b/test_auth_race.py
index 5c514f5..e7b55f5 100644
--- a/test_auth_race.py
+++ b/test_auth_race.py
@@ -20,7 +20,8 @@ def fake_create_workers(session, config, count=1, worker_manager=None):
 
 
 async def main():
-    auth_plugin._create_workers = fake_create_workers
+    # 替换 Plugin 基类的 worker 创建实现，避免依赖 zenoh 会话
+    auth_plugin.plugin._create_workers_impl = fake_create_workers
 
     with tempfile.TemporaryDirectory() as tmp:
         config = {"type": "auth", "data_dir": tmp}
diff --git a/test_mqtt_custom_topics.py b/test_mqtt_custom_topics.py
index 72cc3a5..10ca44c 100644
--- a/test_mqtt_custom_topics.py
+++ b/test_mqtt_custom_topics.py
@@ -99,20 +99,20 @@ async def test_custom_routing():
     )
 
     # 1. 默认 subject = <worker>.<最后一段>；响应 = <实际主题>/response
-    await bridge._handle(make_msg("devices/dev-01/telemetry", {"v": 1}))
+    await bridge._handle(make_msg("devices/dev-01/telemetry", {"meta": {}, "data": {"v": 1}}))
     assert wm.calls[-1]["worker_type"] == "device"
     assert wm.calls[-1]["subject"] == "device.telemetry"
     assert published[-1][0] == "devices/dev-01/telemetry/response"
 
     # 2. 带占位符 + 自定义响应主题
-    await bridge._handle(make_msg("foo/a/b", {"v": 2}))
+    await bridge._handle(make_msg("foo/a/b", {"meta": {}, "data": {"v": 2}}))
     assert wm.calls[-1]["worker_type"] == "foo"
     assert wm.calls[-1]["subject"] == "foo.handle.a/b"
     assert published[-1][0] == "out/a/b/resp"
 
     # 3. 内置 chongming 兜底
     await bridge._handle(
-        make_msg(f"{MQTT_TOPIC_PREFIX}/example/add", {"a": 1, "b": 2})
+        make_msg(f"{MQTT_TOPIC_PREFIX}/example/add", {"meta": {}, "data": {"a": 1, "b": 2}})
     )
     assert wm.calls[-1]["worker_type"] == "example"
     assert wm.calls[-1]["subject"] == "example.add"
@@ -120,7 +120,7 @@ async def test_custom_routing():
     print("✅ 自定义主题路由 + 内置兜底通过")
 
 
-async def test_custom_routing_control_fields():
+async def test_envelope_meta_fields():
     wm = FakeWM()
     bridge, published = make_bridge(
         wm, [{"topic": "devices/+/telemetry", "worker": "device"}]
@@ -128,14 +128,27 @@ async def test_custom_routing_control_fields():
     await bridge._handle(
         make_msg(
             "devices/d1/telemetry",
-            {"_timeout": 5, "_cacheable": True, "_method": "POST"},
+            {"meta": {"timeout": 5, "cacheable": True, "method": "POST"}, "data": {}},
         )
     )
     assert wm.calls[-1]["timeout"] == 5.0
     assert wm.calls[-1]["cacheable"] is True
-    assert "_timeout" not in wm.calls[-1]["payload"]
-    assert wm.calls[-1]["payload"].get("_method") == "POST"
-    print("✅ 控制字段 (_timeout/_cacheable/_method) 透传通过")
+    payload = wm.calls[-1]["payload"]
+    assert payload["meta"]["method"] == "POST"
+    assert payload["data"] == {}
+    print("✅ 信封 meta 字段 (method/timeout/cacheable) 透传通过")
+
+
+async def test_flat_payload_dropped():
+    """严格信封协议：扁平载荷（无 meta/data）应被丢弃且不调用 worker"""
+    wm = FakeWM()
+    bridge, published = make_bridge(
+        wm, [{"topic": "devices/+/telemetry", "worker": "device"}]
+    )
+    await bridge._handle(make_msg("devices/d1/telemetry", {"_method": "POST", "v": 1}))
+    assert wm.calls == []
+    assert published == []
+    print("✅ 非信封（扁平）载荷被丢弃通过")
 
 
 async def test_non_json_dropped():
@@ -165,7 +178,7 @@ async def test_board_ping_topic():
             }
         ],
     )
-    await bridge._handle(make_msg("/board_ping", {"BoardId": "B001", "Status": 1}))
+    await bridge._handle(make_msg("/board_ping", {"meta": {}, "data": {"BoardId": "B001", "Status": 1}}))
     assert wm.calls[-1]["worker_type"] == "aibox_sophon_adapter"
     assert wm.calls[-1]["subject"] == "aibox_sophon_adapter.board_ping"
     assert published[-1][0] == "/board_ping/response"
@@ -182,7 +195,8 @@ def test_plugin_config_driven_collection():
     assert "aibox_sophon_adapter" in by_type, "未发现 aibox_sophon_adapter 插件"
     topics = by_type["aibox_sophon_adapter"]["mqtt_topics"]
     assert any(
-        t["topic"] == "/board_ping" and t["worker"] == "aibox_sophon_adapter"
+        t["topic"] == "box/aibox_sophon/board_ping"
+        and t["worker"] == "aibox_sophon_adapter"
         and t["subject"] == "aibox_sophon_adapter.board_ping"
         for t in topics
     ), topics
@@ -194,7 +208,8 @@ def test_plugin_config_driven_collection():
 async def main():
     test_compile_topic()
     await test_custom_routing()
-    await test_custom_routing_control_fields()
+    await test_envelope_meta_fields()
+    await test_flat_payload_dropped()
     await test_non_json_dropped()
     await test_board_ping_topic()
     test_plugin_config_driven_collection()
diff --git a/test_mqtt_e2e.py b/test_mqtt_e2e.py
index ff5196e..1285987 100644
--- a/test_mqtt_e2e.py
+++ b/test_mqtt_e2e.py
@@ -30,11 +30,11 @@ REQUEST_TOPIC = f"{PREFIX}/{WORKER}/{OPERATION}"
 RESPONSE_TOPIC = f"{PREFIX}/{WORKER}/{OPERATION}/response"
 BROKER_URL = "mqtt://localhost:1883/"
 
-# 测试用例
+# 测试用例（信封协议: meta.method 控制分派，data 携带业务参数）
 TEST_CASES = [
-    {"a": 3, "b": 5, "_method": "MQTT", "expected": 8},
-    {"a": -1, "b": 10, "_method": "MQTT", "expected": 9},
-    {"a": 3.14, "b": 2.86, "_method": "MQTT", "expected": 6.0},
+    {"a": 3, "b": 5, "expected": 8},
+    {"a": -1, "b": 10, "expected": 9},
+    {"a": 3.14, "b": 2.86, "expected": 6.0},
 ]
 
 PASS = 0
@@ -50,7 +50,7 @@ async def run_test():
 
     for i, case in enumerate(TEST_CASES):
         a, b, expected = case["a"], case["b"], case["expected"]
-        payload = {"a": a, "b": b, "_method": "MQTT"}
+        payload = {"meta": {"method": "MQTT"}, "data": {"a": a, "b": b}}
         payload_bytes = json.dumps(payload).encode("utf-8")
 
         await client.publish(REQUEST_TOPIC, payload_bytes, QOS_1)
diff --git a/uv.lock b/uv.lock
index 725a57a..418ae3a 100644
--- a/uv.lock
+++ b/uv.lock
@@ -319,7 +319,7 @@ 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#4b8985d5bd4fc45894bc3e27ecbd541277b83e2f" }
+source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#a58e92c21e5bf4b3ada55b2abc894e8ead7d17bc" }
 dependencies = [
     { name = "eclipse-zenoh" },
     { name = "tomli" },
diff --git a/verify_example_plugin.py b/verify_example_plugin.py
index 5c95e20..7562f67 100644
--- a/verify_example_plugin.py
+++ b/verify_example_plugin.py
@@ -44,18 +44,19 @@ check("set 后 require 返回同一对象", gv.require("db") is marker)
 check("模块属性 db 已被更新", gv.db is marker)
 
 
-# ========== 2. __init__.py 导入链（router + auto_bind_models）==========
+# ========== 2. __init__.py 导入链（router + Plugin 基类）==========
 import plugins.example as plug
 
 check("plugins.example 导入成功", True)
 check("create_workers 存在", callable(getattr(plug, "create_workers", None)))
 check("cleanup 存在", callable(getattr(plug, "cleanup", None)))
-check("_initialized 默认 False", plug._initialized is False)
+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
 
-# 用桩替代 _create_workers，避免真实 zenoh 会话
+# 用桩替代 Plugin 基类的 worker 创建实现，避免真实 zenoh 会话
 calls = []
 
 
@@ -64,7 +65,7 @@ def _stub_create_workers(session, config, count=1, worker_manager=None):
     return ["task-stub"]
 
 
-plug._create_workers = _stub_create_workers  # type: ignore[attr-defined]
+plug.plugin._create_workers_impl = _stub_create_workers  # type: ignore[method-assign]
 
 
 async def main():
@@ -73,7 +74,7 @@ async def main():
     # 第一次调用：应初始化 db
     t1 = await plug.create_workers(session="s1", config=cfg, count=2)
     check("首次 create_workers 调用桩", t1 == ["task-stub"])
-    check("_initialized 置 True", plug._initialized is True)
+    check("plugin._initialized 置 True", plug.plugin._initialized is True)
     from plugins.example.global_var import db as gv_db
     check("global_var.db 已初始化", gv_db is not None)
 
@@ -86,7 +87,7 @@ async def main():
 
     # cleanup 后复位
     await plug.cleanup("worker/example/worker-i-1")
-    check("cleanup 后 _initialized 复位 False", plug._initialized is False)
+    check("cleanup 后 plugin._initialized 复位 False", plug.plugin._initialized is False)
     from plugins.example.global_var import db as gv_db3
     check("cleanup 后 db 为 None", gv_db3 is None)