modify: inline 插件进程内直调快路径（external 不受影响，含 inline/external 一致性测试）

.gitignore

@@ -42,3 +42,6 @@ config
 lib
 auto.crt
 auto.key
+
+*.pem
+*.key

README.md

@@ -11,7 +11,7 @@
 | 特性                  | 说明                                                                 |
 | --------------------- | -------------------------------------------------------------------- |
 | **FastAPI + Uvicorn** | 高性能异步 Web 框架，自动生成 OpenAPI 文档                           |
-| **Zenoh 服务发现**    | 基于 liveliness 的 Worker 注册与 RPC 调用，负载均衡 + 超时重试       |
+| **Zenoh 服务发现**    | 基于 liveliness 的 Worker 注册与 RPC 调用，负载均衡 + 实例定向 + 超时重试 |
 | **MQTT 集成**         | amqtt Broker（内嵌或外部），MQTT ↔ Worker 双向桥接                |
 | **动态路由生成**      | 扫描 `config/workers/*/*.toml` 自动注册 HTTP 路由                    |
 | **Worker 框架**       | `@rpc_method` 装饰器 + `Plugin` 基类统一生命周期 + Pydantic 模型绑定 |
@@ -194,7 +194,7 @@ graph TB
     MQTT[MQTT 客户端] -->|chongming/+/+| BRIDGE[MQTT Bridge<br/>纯客户端模式]
 
     REG[RouteRegistry<br/>订阅 chongming/routes/**] -->|动态挂载/摘除 HTTP 路由| API
-    API --> WM[WorkerManager<br/>随机负载均衡 + 超时重试]
+    API --> WM[WorkerManager<br/>随机负载均衡 / 实例定向 + 超时重试]
     BRIDGE --> WM
 
     subgraph GW[网关进程 chongming]
@@ -328,6 +328,30 @@ worker/<type>/<worker_id>/__admin__
 - **REST / UI**：`POST /admin/workers/<type>/scale_instance`（body `{instance_id, count}`）下发；`GET /admin/workers/<type>/instances` 供前端按实例分组展示当前副本数。前端用「调整数量」（全局对称）和「逐实例精细扩缩」表格（实例 id / 模式 / 当前副本数 / 目标 + 应用）两个入口。
 - **协调前提**：与外部广播缩放一致，需共享 `[cache]`，且外部进程用稳定的 `--instance` / `CHONGMING_WORKER_INSTANCE`（缺省 `{hostname}:{pid}` 虽唯一但重启会变，跨重启持续命中请显式指定）。
 
+### 实例定向调用（把请求固定到某个进程实例）
+
+默认情况下，同 type 的多个副本（含 inline 与 external 进程）逐请求**随机负载均衡**（`random.choice`），这是绝大多数无状态插件想要的。若某插件持有**只存在于某个进程本地**的状态（本地 SQLite/文件、串口或其他设备连接），需要请求固定打到那个进程上，可用**实例定向**：
+
+- **指定方式（三种入口，语义一致）**：
+  - HTTP：请求头 `X-Chongming-Instance: <实例id>`
+  - MQTT：信封载荷顶层加控制字段 `target_instance`（网关读后**摘除**，业务 side 看不到该字段）
+  - 程序内调用：`worker_manager.call(...)` / `call_raw(...)` / `call_stream(...)` 的 `instance_id=` 参数（`utils/request` 的独立 Client 不支持定向）
+- **实例 id 取值**：worker 侧按 `--instance` → `CHONGMING_WORKER_INSTANCE` → `CHONGMING_GATEWAY_ID` → `{hostname}:{pid}` 的顺序解析（`current_instance_id()`）；网关侧定向寻址 inline 副本时用网关自身实例 id（`CHONGMING_GATEWAY_ID`，缺省 hostname）。两侧缺省值在容器环境（K8s 给网关进程注入 `CHONGMING_GATEWAY_ID`）下**自动一致**，无需额外配置；裸机同一台跑多个网关时请显式指定，否则默认实例 id 里的 pid 不同会导致对不上。人可读的取值列表见 `GET /api/v1/admin/workers/<type>/instances`。
+- **粒度**：定向到**进程实例**；若该实例内有多个副本，请求仍在这几个副本之间随机均衡（即“实例固定、副本仍均衡”）。
+- **实例离线时严格报错**：返回 HTTP 503 并附已识别实例列表，**不会静默回退到其它实例**——回退会让粘滞状态的插件跑在错误副本上，产生难查的数据错乱。MQTT 同理（响应主题收到 `{"error": ...}`）。
+- **重试不跨实例**：调用失败重试只在同实例内换副本；实例内已无可换副本时立即失败，不重复打同一个副本。流式端点本身不重试，实例离线即失败。
+- **缓存按实例隔离**：定向且 `cacheable=true` 时缓存键带实例前缀，避免取到其它实例的结果。
+- **身份来源与开销**：副本的实例身份取自其 admin status 回复的 `instance_id`（Python/Rust 两种运行时都会返回），网关**按需查询并缓存**——未使用定向功能时零额外开销，不定向的请求路径与行为完全不变；首次定向到某实例时会对该类型的新副本各发一次轻量 status 查询（每个副本只查一次，inline 副本无需查询）。副本重启后 id 变化，会重新查询一次。
+- **版本要求**：worker 端需运行带 `instance_id` 上报的运行时（当前 Python/Rust SDK 均满足）；读不出身份的副本会被排除在定向之外（不定向的调用不受影响）。
+
+```bash
+# 等价于把这次请求固定到实例 edge-01 的那个进程
+curl -H 'X-Chongming-Instance: edge-01' http://127.0.0.1:8000/api/v1/demo/state
+
+# MQTT：载荷顶层带 target_instance（网关转发前会摘掉它）
+# topic: chongming/demo/state   payload: {"meta":{}, "data":{}, "target_instance":"edge-01"}
+```
+
 ### inline worker 宿主级领导选举（gunicorn 多 worker 兼容）
 
 inline worker 的生命周期交给**基于分布式锁的宿主级领导选举**（取代“每 gunicorn worker 各自拉起”），key `inline_owner:<gateway_id>` 按网关隔离：

config.toml

@@ -123,6 +123,27 @@ base = [
   "fastapi.middleware.cors",
   "fastapi.middleware.gzip",
   "uvicorn",
+  "granian", # 备选服务器（APP_SERVER=granian，见 public/main.py），含 Rust 内核
+  "granian._compat",
+  "granian._futures",
+  "granian._granian",
+  "granian._imports",
+  "granian._internal",
+  "granian._loops",
+  "granian._signals",
+  "granian._types",
+  "granian.asgi",
+  "granian.constants",
+  "granian.errors",
+  "granian.http",
+  "granian.log",
+  "granian.net",
+  "granian.rsgi",
+  "granian.server",
+  "granian.server.common",
+  "granian.server.mp",
+  "granian.server.mt",
+  "granian.wsgi",
   "chongming_cache",
   "chongming_cache.backends",
   "chongming_cache.backends.sqlite_",

deploy/chongming.service

@@ -16,6 +16,10 @@
 #
 # 可选：/etc/chongming.env 中写 APP_WORKERS=<N> 覆盖 worker 数（默认取
 # config.toml [production.server].workers，打包版通常为 4）。
+# 可选：APP_SERVER=granian|uvicorn|gunicorn 切换 HTTP 服务器（默认 Linux 下
+# gunicorn，Windows 下 uvicorn）；三者的启动参数同取自 [production.server]。
+# 可选：CHONGMING_RPC_LOCAL=0 关闭 inline 插件的进程内直调快路径（默认开启；
+# 仅影响本进程托管的 inline 副本，external 插件始终走 Zenoh）。
 
 [Unit]
 Description=Chongming Zenoh Gateway (FastAPI + Zenoh + MQTT)
@@ -29,7 +33,8 @@ StartLimitBurst=5
 [Service]
 Type=simple
 
-# 打包应用入口：Nuitka 编译产物，前台运行 gunicorn master（Linux 分支）
+# 打包应用入口：Nuitka 编译产物，前台运行所选服务器 master（默认 Linux 分支为
+# gunicorn，可用 APP_SERVER 切到 granian/uvicorn）
 ExecStart=/data/chongming/chongming
 # 所有相对路径（config.toml / config/ / resources/*.mbank / static.svfs /
 # data/ / logs/ / uploads/）均相对工作目录解析，必须指向安装目录
@@ -48,7 +53,7 @@ Environment=PYTHONUNBUFFERED=1
 # 私有动态库，再补一条 Environment=LD_LIBRARY_PATH=/opt/sophon/sophon-ffmpeg-latest/lib
 Environment=PATH=/opt/sophon/sophon-ffmpeg-latest/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
 
-# 优雅停机：SIGTERM 由 gunicorn 排空 worker，等在途请求/告警上传完成
+# 优雅停机：SIGTERM 由服务器 master 排空 worker，等在途请求/告警上传完成
 KillSignal=SIGTERM
 KillMode=control-group
 TimeoutStopSec=30

public/config.toml

@@ -33,6 +33,8 @@ debug = false
 env = "production"
 
 [production.server]
+# 启动参数对各服务器通用（由 main.py 的转换函数映射，无需按服务器改配置）；
+# 服务器/副本数可用环境变量覆盖：APP_SERVER=granian|uvicorn|gunicorn、APP_WORKERS=<N>
 access_log = false
 host = "0.0.0.0"
 port = 8000

public/main.py

@@ -3,23 +3,126 @@
 """
 FastAPI chongming 示例
 主入口文件，适配 chongming 打包系统
+
+服务器选择（APP_SERVER 环境变量，默认见 DEFAULT_SERVER）：
+- gunicorn：现状保留（Linux 默认，uvloop worker + preload 多进程）
+- uvicorn：Windows 默认（gunicorn 依赖 fork，Windows 不可用）
+- granian：Rust HTTP 内核，多进程
+
+启动参数保持 config.toml `[<env>.server]` 现状（host / port / workers /
+reload / access_log / timeout_keep_alive），由 `_to_uvicorn`、`_to_granian`、
+`_to_gunicorn_args` 转换为对应服务器的参数。
 """
+import asyncio  # noqa: F401
 import os
-import sys
 import platform
-from utils.launch import launch
+import sys
 from multiprocessing import freeze_support
+from pathlib import Path
+
+import server  # noqa: F401  # 预加载：master 内完成模块银行装载，fork 出的 worker 共享
+import uvloop  # noqa: F401  # 显式导入保证打包时收集 uvloop（uvicorn/granian loop 依赖）
+from utils.launch import launch
 from uvicorn.workers import UvicornWorker
-import server
-import uvloop
-import asyncio
 
 use_shell = platform.system() == "Windows"
 
+# 未设置 APP_SERVER 时的平台默认（保持现状）
+DEFAULT_SERVER = "uvicorn" if use_shell else "gunicorn"
+SUPPORTED_SERVERS = ("gunicorn", "uvicorn", "granian")
+
+# 热重载（reload=true）监听的源码目录，避免 granian 默认扫 cwd 时被 logs/ 写日志触发
+RELOAD_PATHS = [Path("src/chongming"), Path("src/plugins")]
+
+
 class UvloopUvicornWorker(UvicornWorker):
     CONFIG_KWARGS = {"loop": "uvloop"}
 
 
+def _server_params(app_config: dict) -> dict:
+    """把 `[<env>.server]` 的现状参数整理为各服务器共用的启动参数。
+
+    只做缺省兜底与运行环境覆盖（打包环境强制禁 reload、APP_WORKERS 覆盖副本数），
+    键名保持 config.toml 现状，改名交给下面的转换函数。
+    """
+    server_config = app_config.get("server", {})
+    # 打包环境强制禁用重载
+    is_frozen = getattr(sys, "frozen", False)
+    # 支持通过环境变量 APP_WORKERS 覆盖 worker 数量（方便嵌入式设备调优）
+    default_workers = server_config.get("workers", 4)
+    workers = int(os.environ.get("APP_WORKERS", str(default_workers)))
+    return {
+        "host": server_config.get("host", "0.0.0.0"),
+        "port": server_config.get("port", 8000),
+        "workers": workers,
+        "reload": False if is_frozen else server_config.get("reload", False),
+        "access_log": server_config.get("access_log", False),
+        "timeout_keep_alive": server_config.get("timeout_keep_alive", 5),
+        "log_level": "debug" if app_config.get("debug", False) else "info",
+    }
+
+
+def _to_uvicorn(params: dict) -> dict:
+    """共用参数 → uvicorn.run 关键字参数（键名一致）。"""
+    return {
+        "host": params["host"],
+        "port": params["port"],
+        "reload": params["reload"],
+        "workers": params["workers"],
+        "log_level": params["log_level"],
+        "access_log": params["access_log"],
+        "timeout_keep_alive": params["timeout_keep_alive"],
+    }
+
+
+def _to_granian(params: dict) -> dict:
+    """共用参数 → granian.Granian 关键字参数。
+
+    仅两处改名：host → address、access_log → log_access；
+    timeout_keep_alive 在 granian 无对应项（HTTP/1 keep-alive 只有开关、无空闲
+    超时配置），忽略。
+    """
+    return {
+        "address": params["host"],
+        "port": params["port"],
+        "interface": "asgi",
+        "workers": params["workers"],
+        "reload": params["reload"],
+        "log_level": params["log_level"],
+        "log_access": params["access_log"],
+    }
+
+
+def _to_gunicorn_args(params: dict) -> list:
+    """共用参数 → gunicorn 命令行参数（保持现状：uvloop worker + preload）。"""
+    return [
+        "gunicorn",
+        "server:app",
+        "-k",
+        "__main__.UvloopUvicornWorker",
+        "--preload",
+        "-w",
+        str(params["workers"]),
+        "-b",
+        f"{params['host']}:{params['port']}",
+    ]
+
+
+def _select_server(logger) -> str:
+    """按 APP_SERVER 环境变量选择服务器，非法值/平台不兼容时回退默认。"""
+    server_kind = (os.environ.get("APP_SERVER") or DEFAULT_SERVER).strip().lower()
+    if server_kind not in SUPPORTED_SERVERS:
+        logger.warning(
+            f"未知 APP_SERVER={server_kind!r}（可选: {', '.join(SUPPORTED_SERVERS)}），"
+            f"回退 {DEFAULT_SERVER}"
+        )
+        server_kind = DEFAULT_SERVER
+    if server_kind == "gunicorn" and use_shell:
+        logger.warning("Windows 不支持 gunicorn（依赖 fork），回退 uvicorn")
+        server_kind = "uvicorn"
+    return server_kind
+
+
 def app_run(app_config: dict, default_config: dict):
     """
     应用运行函数 - 适配模块银行环境
@@ -35,18 +138,8 @@ def app_run(app_config: dict, default_config: dict):
 
     logger = get_logger("app")
 
-    # 获取服务器配置
+    params = _server_params(app_config)
-    server_config = app_config.get("server", {})
+    server_kind = _select_server(logger)
-    host = server_config.get("host", "0.0.0.0")
-    port = server_config.get("port", 8000)
-    timeout_keep_alive = server_config.get("timeout_keep_alive", 5)
-
-    # 打包环境强制禁用重载
-    is_frozen = getattr(sys, "frozen", False)
-    reload = False if is_frozen else server_config.get("reload", False)
-    # 支持通过环境变量 APP_WORKERS 覆盖 worker 数量（方便嵌入式设备调优）
-    default_workers = server_config.get("workers", 4)
-    workers = int(os.environ.get("APP_WORKERS", str(default_workers)))
 
     # 显示启动信息
     env = app_config.get("env", "development")
@@ -54,41 +147,31 @@ def app_run(app_config: dict, default_config: dict):
     logger.info(f"🚀 启动 {default_config.get('app.name', 'FastAPI 应用')}")
     logger.info(f"📦 版本: {default_config.get('app.version', '1.0.0')}")
     logger.info(f"🌍 环境: {env}")
-    logger.info(f"📍 地址: http://{host}:{port}")
+    logger.info(f"🖥️ 服务器: {server_kind}")
-    logger.info(f"📚 文档: http://{host}:{port}/docs")
+    logger.info(f"📍 地址: http://{params['host']}:{params['port']}")
+    logger.info(f"📚 文档: http://{params['host']}:{params['port']}/docs")
     logger.info(f"🔧 调试: {'启用' if app_config.get('debug', False) else '禁用'}")
-    logger.info(f"🔄 重载: {'启用' if reload else '禁用'}")
+    logger.info(f"🔄 重载: {'启用' if params['reload'] else '禁用'}")
-    logger.info(f"👥 Workers: {workers}")
+    logger.info(f"👥 Workers: {params['workers']}")
     logger.info("=" * 50)
 
-    # 根据操作系统选择不同的服务器启动方式
+    # 根据所选服务器启动（默认平台分支保持现状）
-    if use_shell:
+    if server_kind == "granian":
+        import granian
+
+        granian.Granian(
+            target="server:app",
+            **_to_granian(params),
+            reload_paths=RELOAD_PATHS,
+        ).serve()
+    elif server_kind == "uvicorn":
         import uvicorn
 
-        uvicorn.run(
+        uvicorn.run("server:app", **_to_uvicorn(params))
-            "server:app",
-            host=host,
-            port=port,
-            reload=reload,
-            workers=workers,
-            log_level="debug" if app_config.get("debug", False) else "info",
-            access_log=False,
-            timeout_keep_alive=timeout_keep_alive,
-        )
     else:
         from gunicorn.app.wsgiapp import run
 
-        sys.argv = [
+        sys.argv = _to_gunicorn_args(params)
-            "gunicorn",
-            "server:app",
-            "-k",
-            "__main__.UvloopUvicornWorker",
-            "--preload",
-            "-w",
-            str(workers),
-            "-b",
-            f"{host}:{port}",
-        ]
         run()
 
 

pyproject.toml

@@ -9,6 +9,7 @@ dependencies = [
   "eclipse-zenoh",
   "fastapi[uvicorn]",
   "gunicorn",
+  "granian[reload,uvloop]>=2.8.4",
   "module-bank",
   "openai",
   "python-multipart",
@@ -36,7 +37,14 @@ requires-python = ">=3.12"
 version = "0.1.0"
 
 [dependency-groups]
-dev = ["chongming-cli", "nuitka", "pyarmor", "pyinstaller", "ruff"]
+dev = [
+    "chongming-cli",
+    "greenlet>=3.5.6",
+    "nuitka",
+    "pyarmor",
+    "pyinstaller",
+    "ruff",
+]
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/chongming"]
@@ -49,5 +57,6 @@ url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple"
 
 [project.scripts]
 gunicorn = "chongming:gunicorn_serve"
+granian = "chongming:granian_serve"
 serve = "chongming:serve"
 worker_serve = "chongming_worker.process:worker_serve"

src/chongming/__init__.py

@@ -1,7 +1,66 @@
+"""chongming 网关启动入口（uvicorn / granian / gunicorn）。
+
+config.toml 中 `[<env>.server]` 的启动参数保持现状（host / port / workers /
+reload / access_log / timeout_keep_alive），由 `_to_uvicorn`、`_to_granian` 在
+启动时转换为对应服务器的关键字参数，切换服务器不必改配置格式。
+"""
+
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any
+
 import uvicorn
-from chongming_config import load_config
+from chongming_config import Server, load_config
+
 from .app.core.logger import get_logger
 
+# 开发期热重载（reload=true）监听的源码目录（granian 用；uvicorn 走其默认扫描）
+RELOAD_PATHS = [Path("src/chongming"), Path("src/plugins")]
+
+
+def _server_config_view(server_config: Server) -> Mapping[str, Any]:
+    """[<env>.server] 的宽松视图：实际 TOML 可能带 Server 之外的键
+    （如 timeout_keep_alive，chongming_config.Server 尚未声明）。"""
+    return server_config
+
+
+def _to_uvicorn(server_config: Server) -> dict:
+    """把 [<env>.server] 的现状参数转为 uvicorn.run 的关键字参数。
+
+    uvicorn 的参数名与配置项一致，这里只做缺省兜底。
+    """
+    cfg = _server_config_view(server_config)
+    converted: dict = {
+        "host": cfg.get("host", "0.0.0.0"),
+        "port": cfg.get("port", 8000),
+        "reload": cfg.get("reload", False),
+    }
+    for key in ("workers", "access_log", "timeout_keep_alive"):
+        if key in cfg:
+            converted[key] = cfg[key]
+    return converted
+
+
+def _to_granian(server_config: Server) -> dict:
+    """把 [<env>.server] 的现状参数转为 granian.Granian 的关键字参数。
+
+    仅两处改名：host → address、access_log → log_access；
+    timeout_keep_alive 在 granian 无对应项（HTTP/1 keep-alive 只有开关、无空闲
+    超时配置），忽略。
+    """
+    cfg = _server_config_view(server_config)
+    converted: dict = {
+        "address": cfg.get("host", "0.0.0.0"),
+        "port": cfg.get("port", 8000),
+        "interface": "asgi",
+    }
+    for key in ("reload", "workers"):
+        if key in cfg:
+            converted[key] = cfg[key]
+    if "access_log" in cfg:
+        converted["log_access"] = cfg["access_log"]
+    return converted
+
 
 def serve():
     """
@@ -20,7 +79,34 @@ def serve():
     logger = get_logger("app")
     logger.info(f"启动服务: {server_config}")
     logger.info(f"启动日志: {log_config}")
-    uvicorn.run("chongming.app:app", **server_config)
+    uvicorn.run("chongming.app:app", **_to_uvicorn(server_config))
+
+
+def granian_serve():
+    """
+    使用 granian 启动服务
+
+    从配置文件中加载服务器配置和日志配置，然后使用 granian 启动应用服务。
+    该函数适用于开发环境或单进程部署场景。
+
+    Returns:
+        None: 该函数不返回值，会阻塞运行直到服务停止
+    """
+    # 延迟导入：未选用 granian 的部署环境不必安装它
+    import granian
+
+    config = load_config()
+    env = config["default"]["env"]
+    server_config = config[env]["server"]
+    log_config = config[env]["logging"]
+    logger = get_logger("app")
+    logger.info(f"启动服务: {server_config}")
+    logger.info(f"启动日志: {log_config}")
+    granian.Granian(
+        target="chongming.app:app",
+        **_to_granian(server_config),
+        reload_paths=RELOAD_PATHS,
+    ).serve()
 
 
 def gunicorn_serve():
@@ -36,6 +122,7 @@ def gunicorn_serve():
         None: 该函数不返回值，会阻塞运行直到服务停止
     """
     import sys
+
     from gunicorn.app.wsgiapp import run
 
     config = load_config()

src/chongming/app/__init__.py

@@ -19,11 +19,11 @@ from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.middleware.cors import CORSMiddleware
 
 from chongming_config import get_config
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 from .core.logger import get_logger
-# from .core.static_files import SVFSStaticFiles
 from sqlite_vfs.contrib.fastapi_static import SVFSStaticFiles
 from .core import lifespan_setup, call_stats
-from .routes import storage_router, stats_router, worker_admin_router, auth_router
+from .routes import stats_router, worker_admin_router, auth_router
 
 
 config = get_config()
@@ -199,41 +199,67 @@ def _should_log_access(status_code: int, sample_rate: float = ACCESS_SAMPLE_RATE
     return next(_access_log_counter) % denom == 0
 
 
-@app.middleware("http")
+def _log_access(scope: Scope, status_code: int, start: float) -> None:
-async def access_log_middleware(request: Request, call_next):
+    """按采样规则写一条访问日志（成功按 access_sample_rate，错误 >=400 全量）"""
+    if not _should_log_access(status_code):
+        return
+    client = scope.get("client")
+    logger.info(
+        "%s %s status=%s duration=%.2fms client=%s",
+        scope.get("method", "-"), scope.get("path", "-"), status_code,
+        (time.perf_counter() - start) * 1000, client[0] if client else "-",
+        extra={"status_code": status_code},
+    )
+
+
+class AccessLogMiddleware:
     """
-    HTTP访问日志中间件，记录每个请求的详细信息。
+    HTTP访问日志中间件（纯 ASGI 实现），记录每个请求的详细信息。
 
     记录的信息包括：
     - 请求方法和路径
     - 响应状态码
-    - 请求处理时长（毫秒）
+    - 请求处理时长（毫秒，按慢路径：从进入中间件到响应头产出）
     - 客户端IP地址
 
     高吞吐优化：
     - 采样：成功请求按 access_sample_rate 记录，错误（>=400）全量
     - 异步：% 参数化消息 + 队列日志（QueueHandler），格式化与文件 I/O
       在后台监听线程执行，主线程仅入队
-
+    - 不用 BaseHTTPMiddleware（@app.middleware("http")）：它会为每请求建 anyio
-    参数:
+      task group 并把响应体经内存流过一手，压测实测高并发下吃掉 30~50% 吞吐；
-        request: FastAPI请求对象
+      这里只包一层 send 截获 response.start，无缓冲、无额外任务
-        call_next: 下一个中间件或路由处理函数
-
-    返回:
-        Response: 处理后的响应对象
     """
-    start = time.perf_counter()
-    response = await call_next(request)
-    duration = (time.perf_counter() - start) * 1000
-    client = request.client.host if request.client else "-"
 
-    if _should_log_access(response.status_code):
+    def __init__(self, app: ASGIApp) -> None:
-        logger.info(
+        self.app = app
-            "%s %s status=%s duration=%.2fms client=%s",
+
-            request.method, request.url.path, response.status_code, duration, client,
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
-            extra={"status_code": response.status_code},
+        if scope["type"] != "http":
-        )
+            await self.app(scope, receive, send)
-    return response
+            return
+
+        start = time.perf_counter()
+        logged = False
+
+        async def send_wrapper(message: Message) -> None:
+            nonlocal logged
+            # 响应头产出即记录（流式响应因此只计到首字节，不计整段流）
+            if message["type"] == "http.response.start" and not logged:
+                logged = True
+                _log_access(scope, message["status"], start)
+            await send(message)
+
+        try:
+            await self.app(scope, receive, send_wrapper)
+        except Exception:
+            # 未产出响应就异常（外层 500）：补一条全量错误日志后原样抛出
+            if not logged:
+                _log_access(scope, 500, start)
+            raise
+
+
+app.add_middleware(AccessLogMiddleware)
 
 
 @app.get("/", summary="根路径")
@@ -311,7 +337,6 @@ if vfs_db_path := config[env].get("file_system", {}).get("path", None):
     static_files_handler = SVFSStaticFiles(vfs_db_path=vfs_db_path)
     app.mount("/static", static_files_handler, name="static")
 
-app.include_router(storage_router)
 app.include_router(stats_router)
 app.include_router(worker_admin_router)
 app.include_router(auth_router)

src/chongming/app/core/auth.py

@@ -1,5 +1,7 @@
 import hashlib
+import os
 import time
+from collections import OrderedDict
 
 from fastapi import Depends, HTTPException, status, Request
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
@@ -21,6 +23,45 @@ _jwt_cache = Cache.from_config(get_config()["cache"])
 # 安全性由命中时的 exp 复核保证，TTL 不承担过期判断
 _JWT_CACHE_TTL = 60
 
+# ── 一级：进程内 JWT 结果缓存（LRU + TTL）────────────────────────────
+# 鉴权依赖是同步依赖（def），每次请求都要走一遍共享缓存——backend=redis 时
+# 即"每请求一次同步网络往返 + FastAPI 线程池 hop"。同一 token 在 TTL 内被
+# 反复使用是常态，故在进程内加一层 L1：命中直接返回，不碰网络。
+# 语义与 L2 完全一致：key 含密钥/算法指纹；命中仍复核 exp（过期=拒绝而非放行）；
+# 条目按 TTL 惰性清理并限长。CHONGMING_JWT_L1=0 可关闭（便于 A/B 与排障）。
+_JWT_L1_ENABLED = os.environ.get("CHONGMING_JWT_L1", "1") != "0"
+_JWT_L1_MAX = 1024
+# 存 orjson 字节而非 dict：命中时反序列化出新副本，保证"调用方修改返回值
+# 不污染缓存"这一 L2 既有契约（L2 同样是 JSON 往返后返回新对象）
+_jwt_l1: "OrderedDict[str, tuple[bytes, float, float]]" = OrderedDict()  # key → (payload_json, exp, 写入时刻)
+
+
+def _l1_get(key: str):
+    """命中返回 (payload_json: bytes, exp)；条目超过共享缓存 TTL（运行时可调，便于测试/运维）视为过期"""
+    entry = _jwt_l1.get(key)
+    if entry is None:
+        return None
+    _payload_json, _exp, ts = entry
+    if time.time() - ts >= _JWT_CACHE_TTL:
+        _jwt_l1.pop(key, None)
+        return None
+    _jwt_l1.move_to_end(key)
+    return entry
+
+
+def _l1_set(key: str, payload_json: bytes, exp: float) -> None:
+    _jwt_l1[key] = (payload_json, exp, time.time())
+    _jwt_l1.move_to_end(key)
+    while len(_jwt_l1) > _JWT_L1_MAX:
+        _jwt_l1.popitem(last=False)
+
+
+def _l1_sweep(now: float) -> None:
+    """惰性清理：超 TTL 的整条丢弃（token 永不重复时防堆积），只在表较大时调用"""
+    stale = [k for k, v in _jwt_l1.items() if now - v[2] >= _JWT_CACHE_TTL]
+    for k in stale:
+        _jwt_l1.pop(k, None)
+
 
 def _cache_key(token: str, secret: str, algorithm: str) -> str:
     """缓存 key：token 内容 + 密钥/算法指纹，不同密钥/算法不串用"""
@@ -28,20 +69,33 @@ def _cache_key(token: str, secret: str, algorithm: str) -> str:
 
 
 def _cached_decode(token: str, secret: str, algorithm: str) -> tuple:
-    """JWT 解密结果缓存（(payload, exp)），共享缓存避免高频请求重复密码学运算。
+    """JWT 解密结果缓存（(payload, exp)）：一级进程内 LRU，二级共享缓存。
 
-    - 命中且未过期 -> 直接返回，跳过 RSA/ECDSA 验证与 JSON 解析
+    - 一级命中且未过期 -> 直接返回（零网络、零密码学）；过期 -> 直接拒绝
-    - 命中但 exp 已过 -> 直接拒绝，不重解密（条目保留至 TTL 到期，
+    - 一级未命中 -> 查二级共享缓存；命中且未过期 -> 回填一级后返回
-      过期 token 的重复请求零密码学开销；缓存不延长 token 有效期）
+    - 二级也未命中 -> 完整解码（RSA/ECDSA 验证）后写二级 + 一级（TTL=60s/30s）
-    - 未命中 -> 完整解码后写入共享缓存（TTL=60s）
     - decode 失败（无效 token）不缓存，每次都完整验证后拒绝
     """
     key = _cache_key(token, secret, algorithm)
+    now = time.time()
+    if _JWT_L1_ENABLED:
+        hit = _l1_get(key)
+        if hit is not None:
+            payload_json, exp, _ts = hit
+            if not exp or exp >= now:
+                # 反序列化出新副本：调用方修改不影响缓存条目（与 L2 契约一致）
+                return orjson.loads(payload_json), exp
+            raise ValueError("Token 已过期")
+        if len(_jwt_l1) > _JWT_L1_MAX // 2:  # 仅大表时惰性清理，避免常态 O(n)
+            _l1_sweep(now)
+
     entry = _jwt_cache.get(key)
     if entry is not None:
         payload = entry["payload"]
         exp = entry.get("exp", 0)
         if not exp or exp >= time.time():
+            if _JWT_L1_ENABLED:
+                _l1_set(key, orjson.dumps(payload), exp)
             return payload, exp
         raise ValueError("Token 已过期")
     jwt = JWT(secret, algorithm)
@@ -49,6 +103,8 @@ def _cached_decode(token: str, secret: str, algorithm: str) -> tuple:
     payload = orjson.loads(payload_str.encode("utf-8"))
     exp = payload.get("exp", 0)
     _jwt_cache.set(key, {"payload": payload, "exp": exp}, expire=_JWT_CACHE_TTL)
+    if _JWT_L1_ENABLED:
+        _l1_set(key, orjson.dumps(payload), exp)
     return payload, exp
 
 

src/chongming/app/core/mqtt_bridge.py

@@ -169,6 +169,15 @@ class MQTTBridge:
         except orjson.JSONDecodeError:
             return
 
+        # 实例定向控制字段：与 HTTP 的 X-Chongming-Instance 头同义，指定请求发往的
+        # worker 进程实例。必须在下面归一化之前取出——归一化只保留 meta/data 两键，
+        # 之后控制字段就不在载荷里了。
+        target_instance: Optional[str] = None
+        if isinstance(pl, dict):
+            ti = pl.pop("target_instance", None)
+            if isinstance(ti, str) and ti.strip():
+                target_instance = ti.strip()
+
         # 信封协议: {"meta": {...}, "data": {...}}
         # 兼容非信封载荷（如第三方设备原始报文）: 整包视为业务数据、meta 置空，
         # 与 worker 运行时 runtime.py 的容错语义一致（method 默认 GET）——
@@ -194,7 +203,7 @@ class MQTTBridge:
         for ct, compiled in self._compiled:
             matched = compiled.match(t)
             if matched:
-                await self._route_custom(ct, matched, pl, t)
+                await self._route_custom(ct, matched, pl, t, target_instance)
                 return
 
         # 2. 内置接口 chongming/{worker}/{operation}
@@ -204,7 +213,7 @@ class MQTTBridge:
         wt, op = parts[1], parts[2]
         subj = f"{wt}.{op}"
         rt = f"{MQTT_TOPIC_PREFIX}/{wt}/{op}/response"
-        await self._route(wt, subj, pl, rt)
+        await self._route(wt, subj, pl, rt, target_instance)
 
     def _fill(self, template: str, matched: re.Match) -> str:
         """用匹配到的通配符值填充模板中的 {1} {2}... 占位符"""
@@ -214,7 +223,8 @@ class MQTTBridge:
             template = template.replace("{" + str(i) + "}", g)
         return template
 
-    async def _route_custom(self, ct: dict, matched: re.Match, pl: dict, topic: str):
+    async def _route_custom(self, ct: dict, matched: re.Match, pl: dict, topic: str,
+                            target_instance: Optional[str] = None):
         worker = ct.get("worker", "")
         # subject 默认 "<worker>.<topic 模板最后一段>"，与内置 <插件>.<操作> 约定一致
         subject_tpl = ct.get("subject")
@@ -225,15 +235,22 @@ class MQTTBridge:
         # response_topic 默认 "<实际匹配主题>/response"
         rt_tpl = ct.get("response_topic")
         rt = self._fill(rt_tpl, matched) if rt_tpl else f"{topic}/response"
-        await self._route(worker, subject, pl, rt)
+        await self._route(worker, subject, pl, rt, target_instance)
 
-    async def _route(self, wt: str, subj: str, pl: dict, rt: str):
+    async def _route(self, wt: str, subj: str, pl: dict, rt: str,
+                     target_instance: Optional[str] = None):
+        """把一次 MQTT 请求转成 worker RPC。
+
+        target_instance 为实例定向目标（来自载荷顶层的 target_instance 控制字段，
+        已在 _handle 归一化前摘除）；缺省时跨实例随机均衡，行为与以往一致。
+        """
         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(meta.get("timeout", 30)),
-                cacheable=bool(meta.get("cacheable", False)))
+                cacheable=bool(meta.get("cacheable", False)),
+                instance_id=target_instance or None)
             if r is not None:
                 await self._pub(rt, r)
         except RuntimeError as e:

src/chongming/app/core/route_generator.py

@@ -24,6 +24,21 @@ from .call_stats import spawn, record_call
 
 logger = get_logger("route_generator")
 
+# 实例定向请求头：给定后该请求固定发往指定 worker 进程实例（同一实例下的多个副本
+# 之间仍随机均衡），不参与跨实例的随机负载均衡。取值 = 实例身份（外部进程启动参数
+# --instance / CHONGMING_WORKER_INSTANCE；inline 副本为网关自身实例 id）。
+# 未携带该头时行为与以往完全一致（全类型随机均衡）。
+INSTANCE_HEADER = "X-Chongming-Instance"
+
+
+def _instance_from_request(request: Request) -> Optional[str]:
+    """从请求头读取定向实例 id；空串/未携带视为不定向。"""
+    value = request.headers.get(INSTANCE_HEADER)
+    if value:
+        value = value.strip()
+    return value or None
+
+
 # 简单类型映射
 TYPE_MAP: Dict[str, Any] = {
     "float": float,
@@ -348,6 +363,7 @@ def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_t
                 payload=payload_bytes,
                 timeout=timeout,
                 cacheable=cacheable,
+                instance_id=_instance_from_request(request),
             )
             duration_ms = (time.perf_counter() - start) * 1000
             _spawn_call_record(
@@ -409,6 +425,7 @@ def _make_raw_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, work
                 payload=payload,
                 timeout=timeout,
                 cacheable=cacheable,
+                instance_id=_instance_from_request(request),
             )
             duration_ms = (time.perf_counter() - start) * 1000
             _spawn_call_record(
@@ -472,6 +489,7 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
                     payload=payload_bytes,
                     timeout=timeout,
                     cacheable=cacheable,
+                    instance_id=_instance_from_request(request),
                 )
                 duration_ms = (time.perf_counter() - start) * 1000
                 _spawn_call_record(
@@ -515,6 +533,7 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
                 payload=payload_bytes,
                 timeout=timeout,
                 cacheable=cacheable,
+                instance_id=_instance_from_request(request),
             )
             duration_ms = (time.perf_counter() - start) * 1000
             _spawn_call_record(
@@ -648,6 +667,7 @@ def _make_stream_endpoint(
                 payload=payload,
                 timeout=timeout,
                 idle_timeout=idle_timeout,
+                instance_id=_instance_from_request(request),
             )
         except RuntimeError as e:
             duration_ms = (time.perf_counter() - start) * 1000

src/chongming/app/core/worker_manager.py

@@ -88,7 +88,40 @@ class _MemLRU:
             self._d.popitem(last=False)
 
 
+# ── 进程内直调（inline 快路径）─────────────────────────────────────────────
+# inline 副本由本进程托管（与网关同事件循环），业务请求却仍要绕
+# "zenoh 查询 → router → queryable 回调 → 跨线程回本 loop"。分派逻辑已统一到
+# chongming_worker.dispatch，故此处可直接 await 同一个 dispatcher，省掉整圈往返。
+# 语义不变：信封/错误通道/超时/重试/换副本/实例定向/in-flight 与 drain 全部沿用原逻辑；
+# 非本地副本（external 插件、w>1 时非托管进程）与旧版 worker 包自动回退 zenoh。
+_RPC_LOCAL = os.environ.get("CHONGMING_RPC_LOCAL", "1") != "0"
+try:  # 旧版 chongming-worker（<0.5）无 dispatch 模块：保持纯 zenoh 行为
+    from chongming_worker.dispatch import get_dispatcher as _get_local_dispatcher
+except Exception:  # noqa: BLE001
+    _get_local_dispatcher = None
+
+
+def _local_dispatcher_for(key_expr: str) -> Optional[Any]:
+    """'worker/<type>/<id>' → 本进程托管该副本的 dispatcher；非本地/不可用返回 None"""
+    if not _RPC_LOCAL or _get_local_dispatcher is None:
+        return None
+    parts = key_expr.split("/")
+    if len(parts) != 3 or parts[0] != "worker":
+        return None
+    return _get_local_dispatcher(parts[1], parts[2])
+
+
+def _err_text(payload: Any) -> str:
+    """把错误载荷归一为文本（兼容 zenoh ZBytes 与直调的 bytes）"""
+    if payload is None:
+        return "unknown error"
+    if isinstance(payload, (bytes, bytearray)):
+        return bytes(payload).decode("utf-8", "replace") or "unknown error"
+    return str(payload) or "unknown error"
+
+
 class WorkerManager:
+
     """
     统一 Worker 管理器
 
@@ -112,6 +145,15 @@ class WorkerManager:
         self._launch_worker_count = launch_worker_count
         self._workers_cache: Dict[str, List[str]] = {}  # type -> list of worker key_exprs
         self._local_workers: Dict[str, List[str]] = {}   # type -> list of locally registered workers
+        # ── 实例级定向（把请求固定到某个进程实例而非随机负载均衡）──
+        # 副本经 liveliness 只暴露 worker/<type>/<id>，实例身份取自副本 admin status
+        # 的 instance_id（Python/Rust 运行时均返回），故定向索引按需查询并缓存：
+        #   _instance_index[type][instance_id] = [worker/<type>/<id>, ...]
+        #   _instance_unknown[type]            = [已查询但身份不可得的副本 key, ...]
+        # 只缓存已查询过的副本；每次按需刷新时整表重建并对新副本补查询、对离线副本剪枝。
+        self._instance_index: Dict[str, Dict[str, List[str]]] = {}
+        self._instance_unknown: Dict[str, List[str]] = {}
+        self._instance_locks: Dict[str, asyncio.Lock] = {}
         self._lock = asyncio.Lock()
         self._refresh_interval = 15  # seconds
         # 当前网关心跳标识：优先环境变量（多实例需唯一，尤其同宿主多网关须显式指定），
@@ -406,23 +448,112 @@ class WorkerManager:
                 logger.error(f"刷新 worker 列表失败: {e}")
             await asyncio.sleep(self._refresh_interval)
 
-    async def get_worker(self, worker_type: str) -> Optional[str]:
+    def _all_workers(self, worker_type: str) -> List[str]:
-        """获取一个可用的 worker 地址（简单随机负载均衡）
+        """该类型当前全部候选副本（liveliness 列表优先，启动初期回退本地注册表）。"""
+        workers = list(self._workers_cache.get(worker_type, []))
+        if not workers:
+            workers = list(self._local_workers.get(worker_type, []))
+        return workers
+
+    async def _query_instance_id(self, worker_type: str, worker_key: str) -> Optional[str]:
+        """问单个副本要它的实例身份（admin status 的 instance_id）。
+
+        复用单副本管理协议 worker/<type>/<id>/__admin__；查询失败或副本运行旧版
+        运行时（回复无 instance_id）返回 None，由调用方归入 unknown 组。
+        """
+        wid = worker_key.split("/")[2] if "/" in worker_key else worker_key
+        try:
+            reply = await self._send_admin_cmd(worker_type, wid, "query", timeout=2.0)
+        except Exception as e:  # noqa: BLE001 - 单副本查询失败不影响其它副本
+            logger.warning("[instance] %s/%s 身份查询失败: %s", worker_type, wid, e)
+            return None
+        iid = reply.get("instance_id") if isinstance(reply, dict) else None
+        return str(iid) if iid else None
+
+    async def _refresh_instance_map(self, worker_type: str) -> Dict[str, List[str]]:
+        """重建某类型的「实例 → 副本」索引，并按需补查询新副本。
+
+        只对从未查询过的副本发一次 admin status（本地注册的 inline 副本直接返回
+        网关自身实例，零查询）；已查询过的副本沿用缓存，因此常规路径是纯内存查询，
+        仅新上线副本会引入一次额外 RPC。离线副本在重建时被剪枝。
+        """
+        workers = self._all_workers(worker_type)
+        known = self._instance_index.get(worker_type, {})
+        unknown = set(self._instance_unknown.get(worker_type, []))
+        # 反查已有归属：副本 -> 实例（副本 key 唯一，一个副本只属一个实例）
+        owner = {w: iid for iid, ws in known.items() for w in ws}
+
+        index: Dict[str, List[str]] = {}
+        still_unknown: List[str] = []
+        queried = owner.keys() | unknown
+        for w in workers:
+            if w in queried:
+                iid = owner.get(w)
+            else:
+                # 本地注册（inline）副本天然属于本网关实例，无需查询
+                iid = self._gateway_id if w in self._local_workers.get(worker_type, []) else None
+                if iid is None:
+                    iid = await self._query_instance_id(worker_type, w)
+                if iid is None:
+                    still_unknown.append(w)
+            if iid:
+                index.setdefault(iid, []).append(w)
+
+        self._instance_index[worker_type] = index
+        self._instance_unknown[worker_type] = still_unknown
+        return index
+
+    async def _instance_workers(self, worker_type: str, instance_id: str) -> List[str]:
+        """取某实例下该类型的在线副本 key 列表（无则空表）。"""
+        index = self._instance_index.get(worker_type)
+        if index is None:
+            # 同类型并发请求只让一个跑刷新，其余等结果，避免重复查询同一批副本
+            lock = self._instance_locks.setdefault(worker_type, asyncio.Lock())
+            async with lock:
+                index = self._instance_index.get(worker_type)
+                if index is None:
+                    index = await self._refresh_instance_map(worker_type)
+        return index.get(instance_id, [])
+
+    async def get_worker(
+        self, worker_type: str, instance_id: Optional[str] = None
+    ) -> Optional[str]:
+        """获取一个可用的 worker 地址。
+
+        instance_id 为 None 时按简单随机负载均衡挑选（默认，行为不变）；
+        指定实例时只在该实例的在线副本中挑选，把请求固定到那个进程实例。
+        实例无在线副本（或该实例在旧版运行时下无法识别身份）返回 None。
 
         读路径无锁：_workers_cache 由 refresh_workers 整表原子替换，
         _local_workers 由同步方法修改，单事件循环内读取安全。
         """
-        workers = self._workers_cache.get(worker_type, [])
+        if instance_id:
-        # 如果缓存为空，回退到本地注册表（适用于启动初期 refresh_workers 尚未执行时）
+            return await self._pick_from_instance(worker_type, instance_id)
-        if not workers:
+        workers = self._all_workers(worker_type)
-            local = self._local_workers.get(worker_type, [])
+        return random.choice(workers) if workers else None
-            if local:
+
-                return random.choice(local)
+    async def _pick_from_instance(
-            return None
+        self, worker_type: str, instance_id: str
-        return random.choice(workers)
+    ) -> Optional[str]:
+        """在指定实例的在线副本中随机挑一个（定向路径，永不跨实例回退）。
+
+        定向的语义就是"必须打在同一个进程实例上"，因此实例离线时返回 None 让
+        调用方报错，而不是悄悄换成别的实例——后者会让粘滞状态类的插件跑在错误
+        副本上，产生难查的数据错乱。
+        """
+        candidates = await self._instance_workers(worker_type, instance_id)
+        if candidates:
+            return random.choice(candidates)
+        logger.warning(
+            "[instance] 目标实例 %s 下无 %s 的在线副本（已识别实例: %s；%d 个副本读不出身份）",
+            instance_id, worker_type, sorted(self._instance_index.get(worker_type, {})),
+            len(self._instance_unknown.get(worker_type, [])),
+        )
+        return None
 
     async def _pick_other_worker(
-        self, worker_type: str, exclude: List[str]
+        self, worker_type: str, exclude: List[str],
+        instance_id: Optional[str] = None,
     ) -> Optional[str]:
         """
         选择一个不同于已尝试列表的 worker。
@@ -430,10 +561,17 @@ class WorkerManager:
         Args:
             worker_type: Worker 类型
             exclude: 已尝试（需排除）的 worker key_expr 列表
+            instance_id: 定向实例；给定时只在同实例内换副本，绝不跨实例
 
         Returns:
             可用的 worker key_expr，若无其他可用则返回 None
         """
+        if instance_id:
+            candidates = [
+                w for w in await self._instance_workers(worker_type, instance_id)
+                if w not in exclude
+            ]
+            return random.choice(candidates) if candidates else None
         workers = self._workers_cache.get(worker_type, [])
         candidates = [w for w in workers if w not in exclude]
         if candidates:
@@ -455,24 +593,37 @@ class WorkerManager:
             return False
         return True
 
-    async def _raise_no_worker(self, worker_type: str) -> None:
+    async def _no_worker_message(
+        self, worker_type: str, instance_id: Optional[str] = None
+    ) -> str:
         """
-        无可用 Worker 时的统一诊断：列出当前已知的所有 worker 类型并抛 RuntimeError。
+        无可用 Worker 时的统一诊断文案。
 
-        供 call() / call_stream() 共用，保证两类调用错误信息一致。
+        供 call() / call_stream() 共用，保证两类调用错误信息一致；定向（instance_id）
+        调用额外列出当前已识别实例，便于区分"实例名写错"与"实例离线"。
         """
         all_known = list(self._workers_cache.keys())
         local_known = list(self._local_workers.keys())
         all_types = sorted(set(all_known + local_known))
-        msg = (
+        msg = f"类型 '{worker_type}' 没有可用 Worker。"
-            f"类型 '{worker_type}' 没有可用 Worker。"
+        if instance_id:
-            f"当前已知的 Worker 类型: {all_types if all_types else '(无)'}。"
+            msg = (
-            f"请检查: 1) config/workers/config.toml 是否配置了该类型 "
+                f"类型 '{worker_type}' 的实例 '{instance_id}' 没有在线副本（定向调用不"
-            f"2) 插件初始化是否成功（查看启动日志中的 '加载 Worker 插件失败' 错误）"
+                f"跨实例回退）。已识别实例: "
-            f"3) Worker 进程是否正常运行"
+                f"{sorted(self._instance_index.get(worker_type, {})) or '(无)'}。"
-        )
+                f"请检查: 1) 该实例 id 是否写错（取值见 GET "
-        logger.warning(msg)
+                f"/admin/workers/{worker_type}/instances 或副本启动时的 --instance）"
-        raise RuntimeError(msg)
+                f" 2) 该实例进程是否在线 3) 该实例的 worker 是否运行旧版运行时"
+                f"（不支持实例身份上报）"
+            )
+        else:
+            msg += (
+                f"当前已知的 Worker 类型: {all_types if all_types else '(无)'}。"
+                f"请检查: 1) config/workers/config.toml 是否配置了该类型 "
+                f"2) 插件初始化是否成功（查看启动日志中的 '加载 Worker 插件失败' 错误）"
+                f"3) Worker 进程是否正常运行"
+            )
+        return msg
 
     async def call_stream(
         self,
@@ -481,6 +632,7 @@ class WorkerManager:
         payload: Dict[str, Any],
         timeout: float = 300.0,
         idle_timeout: float = 10.0,
+        instance_id: Optional[str] = None,
     ) -> AsyncGenerator[bytes, None]:
         """
         流式调用：消费 Worker 的多回复流，逐块产出 JSON 字节。
@@ -499,21 +651,53 @@ class WorkerManager:
             payload: 请求参数（信封格式，由路由层构造）
             timeout: 流总寿命上限（秒），需大于业务最长生成时长
             idle_timeout: 空闲超时（秒），超时抛 StreamIdleTimeoutError
+            instance_id: 定向进程实例（None=随机负载均衡）；流本身不可重试，
+                故实例离线时直接报错，不会中途改派到别的实例
 
         Yields:
             bytes: 每个回复块的 JSON 字节
 
         Raises:
-            RuntimeError: 无可用 worker
+            RuntimeError: 无可用 worker（含定向实例无在线副本）
             StreamIdleTimeoutError: 空闲超时
         """
         # 急切解析 worker：无可用时立即抛错，让路由层映射 HTTP 503
         # （异步生成器函数体在首次迭代才执行，故这里用普通 async 函数返回内部生成器）
-        worker_expr = await self.get_worker(worker_type)
+        worker_expr = await self.get_worker(worker_type, instance_id=instance_id)
         if not worker_expr:
-            await self._raise_no_worker(worker_type)
+            raise RuntimeError(await self._no_worker_message(worker_type, instance_id))
 
         full_key = f"{worker_expr}/{subject}"
+
+        # 进程内直调（inline 快路径）：与 zenoh 路径同语义（逐块产出、错误块终止、无重试）
+        local = _local_dispatcher_for(worker_expr)
+        if local is not None and local.has_subject(subject):
+            async def _local_stream():
+                agen = local.invoke_stream(subject, orjson.dumps(payload))
+                try:
+                    while True:
+                        try:
+                            data = await asyncio.wait_for(agen.__anext__(), timeout=idle_timeout)
+                        except StopAsyncIteration:
+                            break
+                        except asyncio.TimeoutError:
+                            raise StreamIdleTimeoutError(
+                                f"Worker {worker_expr} 流式空闲超时 ({idle_timeout}s 无数据)"
+                            ) from None
+                        if not data:
+                            continue
+                        decoded = orjson.loads(data)
+                        if isinstance(decoded, dict) and decoded.get("error"):
+                            # Worker 错误收尾块：转发后终止流
+                            yield data
+                            return
+                        yield data
+                finally:
+                    # 消费方提前退出（SSE 断连/网关放弃）：停流并取消在途派发
+                    await agen.aclose()
+
+            return _local_stream()
+
         reply_iter = self.session.get(
             full_key,
             payload=orjson.dumps(payload),
@@ -602,6 +786,7 @@ class WorkerManager:
         timeout: float = 3.0,
         retries: int = 1,
         cacheable: bool = False,
+        instance_id: Optional[str] = None,
     ) -> Any:
         """
         统一调用入口（dict 语义，MQTT Bridge 等调用方使用）
@@ -613,6 +798,8 @@ class WorkerManager:
             timeout: 单次调用超时（秒）
             retries: 失败重试次数
             cacheable: 是否允许缓存结果（若为 False 则跳过缓存读写）
+            instance_id: 定向进程实例；给定时请求固定发往该实例（不参与随机
+                负载均衡，重试也只在该实例内换副本），None=默认随机均衡
 
         Returns:
             Dict[str, Any]: 调用结果（worker 回复解析后的 dict）
@@ -620,10 +807,18 @@ class WorkerManager:
         # cacheable=False 时不计算 cache_key，省一次排序序列化
         cache_key = None
         if cacheable:
-            cache_key = f"{worker_type}:{subject}:{orjson.dumps(payload, option=orjson.OPT_SORT_KEYS).decode()}"
+            # 定向调用的缓存键含实例名：定向通常用于读取该实例的本地状态
+            # （本地库/设备连接），共用缓存会拿到其它实例的结果
+            cache_key = (
+                f"{worker_type}:{subject}"
+                f":{orjson.dumps(payload, option=orjson.OPT_SORT_KEYS).decode()}"
+            )
+            if instance_id:
+                cache_key = f"{instance_id}@{cache_key}"
         reply_bytes = await self._call_impl(
             worker_type, subject, orjson.dumps(payload),
             timeout=timeout, retries=retries, cacheable=cacheable, cache_key=cache_key,
+            instance_id=instance_id,
         )
         return orjson.loads(reply_bytes) if reply_bytes else {}
 
@@ -635,6 +830,7 @@ class WorkerManager:
         timeout: float = 3.0,
         retries: int = 1,
         cacheable: bool = False,
+        instance_id: Optional[str] = None,
     ) -> bytes:
         """
         统一调用入口（bytes 语义，HTTP 端点透传用）：返回 worker 回复的原始字节。
@@ -652,6 +848,7 @@ class WorkerManager:
             timeout: 单次调用超时（秒）
             retries: 失败重试次数
             cacheable: 是否允许缓存结果（缓存原始字节）
+            instance_id: 定向进程实例（见 call()）；None=默认随机均衡
 
         Returns:
             bytes: worker 回复的原始字节（空回复兜底 b"{}"，与 call() 的 {} 语义一致）
@@ -660,9 +857,12 @@ class WorkerManager:
         cache_key = None
         if cacheable:
             cache_key = f"{worker_type}:{subject}:{hashlib.sha256(payload).hexdigest()}"
+            if instance_id:
+                cache_key = f"{instance_id}@{cache_key}"
         return await self._call_impl(
             worker_type, subject, payload,
             timeout=timeout, retries=retries, cacheable=cacheable, cache_key=cache_key,
+            instance_id=instance_id,
         )
 
     async def _call_impl(
@@ -674,11 +874,13 @@ class WorkerManager:
         retries: int = 1,
         cacheable: bool = False,
         cache_key: Optional[str] = None,
+        instance_id: Optional[str] = None,
     ) -> bytes:
         """
-        调用实现：缓存 > 负载均衡 > 超时重试，返回回复的原始字节。
+        调用实现：缓存 > 选副本（随机均衡/实例定向）> 超时重试，返回回复的原始字节。
 
         call() / call_raw() 共用本实现，仅入参出参形态不同（dict ↔ bytes）。
+        instance_id 给定时选副本与重试都被约束在该实例内（定向语义，不跨实例回退）。
         """
         # 1. 缓存检查（仅在 cacheable=True 时检查）
         if cacheable and cache_key is not None:
@@ -695,56 +897,92 @@ class WorkerManager:
                     logger.debug(f"命中共享缓存: {cache_key}")
                     return disk_val
 
-        # 2. 获取可用 worker
+        # 2. 获取可用 worker（instance_id 给定时固定到该实例，否则随机均衡）
-        worker_expr = await self.get_worker(worker_type)
+        worker_expr = await self.get_worker(worker_type, instance_id=instance_id)
         if not worker_expr:
-            await self._raise_no_worker(worker_type)
+            raise RuntimeError(await self._no_worker_message(worker_type, instance_id))
 
         full_key = f"{worker_expr}/{subject}"
 
         last_error = None
         tried_workers: List[str] = []
 
+        async def _switch_worker(why: str) -> bool:
+            """换一个副本重试；返回是否值得继续（定向调用只在同实例内换副本）。
+
+            换完更新闭包外层的 worker_expr / full_key。定向实例内已无未试副本时
+            返回 False —— 继续循环只会重复打同一个坏副本，白等一整轮超时。
+            """
+            nonlocal worker_expr, full_key, last_error
+            new_worker = await self._pick_other_worker(
+                worker_type, tried_workers, instance_id=instance_id
+            )
+            if new_worker:
+                if new_worker != worker_expr:
+                    logger.info(f"{why}切换 Worker: {worker_expr} -> {new_worker}")
+                    worker_expr = new_worker
+                    full_key = f"{worker_expr}/{subject}"
+                return True
+            # 没有任何未试过的副本可换：继续循环只会重复打同一个副本（若它已死，
+            # 白等一整轮超时；若它已回业务错误，结果必然相同），直接结束重试。
+            # 保留具体失败原因（last_error 已由调用方按场景写入），便于定位。
+            scope = f"实例 '{instance_id}' 内" if instance_id else ""
+            last_error = f"{scope}已无其它未试副本可换（{why}）：{last_error}"
+            return False
+
         for attempt in range(retries + 1):
             # 记录已尝试的 worker
             if worker_expr not in tried_workers:
                 tried_workers.append(worker_expr)
 
             try:
-                reply_iter = self.session.get(
+                local = _local_dispatcher_for(worker_expr)
-                    full_key,
+                if local is not None and local.has_subject(subject):
-                    payload=payload_bytes,
+                    # 进程内直调（inline 快路径）：省 zenoh 往返，语义与下面 zenoh 路径一致
-                    consolidation=zenoh.ConsolidationMode.NONE,
+                    logger.debug(f"[local-rpc] {worker_expr}/{subject} 进程内直调")
-                    timeout=timeout,
+                    local_res = await asyncio.wait_for(
-                )
+                        local.invoke_unary(subject, payload_bytes), timeout=timeout
-                # 使用辅助函数获取第一个回复，避免 StopIteration 在 asyncio.to_thread 中传播的问题
-                def _get_first(iterator):
-                    """线程安全地获取迭代器的第一个元素"""
-                    for item in iterator:
-                        return item
-                    return _SENTINEL
-
-                # 走专用 _ZENOH_EXECUTOR 池（隔离于全局默认池，避免高并发下排队）
-                loop = asyncio.get_running_loop()
-                first_reply = await asyncio.wait_for(
-                    loop.run_in_executor(_ZENOH_EXECUTOR, _get_first, reply_iter),
-                    timeout=timeout,
-                )
-                if first_reply is _SENTINEL:
-                    logger.warning(
-                        f"Worker {worker_expr} 无回复，尝试重试 {attempt+1}/{retries+1}"
                     )
-                    last_error = f"Worker {worker_expr} 无回复"
+                    reply_ok, reply_payload = local_res.ok, local_res.payload
-                    # 换一个 worker 重试
+                else:
-                    new_worker = await self._pick_other_worker(worker_type, tried_workers)
+                    reply_iter = self.session.get(
-                    if new_worker:
+                        full_key,
-                        worker_expr = new_worker
+                        payload=payload_bytes,
-                        full_key = f"{worker_expr}/{subject}"
+                        consolidation=zenoh.ConsolidationMode.NONE,
-                    continue
+                        timeout=timeout,
+                    )
+                    # 使用辅助函数获取第一个回复，避免 StopIteration 在 asyncio.to_thread 中传播的问题
+                    def _get_first(iterator):
+                        """线程安全地获取迭代器的第一个元素"""
+                        for item in iterator:
+                            return item
+                        return _SENTINEL
 
-                reply = first_reply
+                    # 走专用 _ZENOH_EXECUTOR 池（隔离于全局默认池，避免高并发下排队）
-                if reply.ok: # type: ignore
+                    loop = asyncio.get_running_loop()
-                    reply_bytes = bytes(reply.ok.payload) if reply.ok.payload else b"{}"
+                    first_reply = await asyncio.wait_for(
+                        loop.run_in_executor(_ZENOH_EXECUTOR, _get_first, reply_iter),
+                        timeout=timeout,
+                    )
+                    if first_reply is _SENTINEL:
+                        logger.warning(
+                            f"Worker {worker_expr} 无回复，尝试重试 {attempt+1}/{retries+1}"
+                        )
+                        last_error = f"Worker {worker_expr} 无回复"
+                        if not await _switch_worker("无回复"):
+                            break
+                        continue
+
+                    reply = first_reply
+                    if reply.ok:  # type: ignore
+                        reply_ok = True
+                        reply_payload = bytes(reply.ok.payload) if reply.ok.payload else b"{}"
+                    else:
+                        reply_ok = False
+                        reply_payload = reply.err.payload  # type: ignore
+
+                if reply_ok:
+                    reply_bytes = bytes(reply_payload) if reply_payload else b"{}"
                     # 缓存写入（仅在 cacheable=True 时）：解析一次做成功检查，
                     # 但缓存与返回均保留原始字节（网关侧零二次序列化）。
                     # 先写内存一级，再异步写共享二级（executor，不阻塞响应返回）
@@ -755,16 +993,12 @@ class WorkerManager:
                             await self._cache_set(cache_key, reply_bytes)
                     return reply_bytes
                 else:
-                    error_payload = reply.err.payload  # type: ignore
+                    error_msg = _err_text(reply_payload)
-                    error_msg = str(error_payload) if error_payload else "unknown error"
                     logger.warning(f"Worker 返回错误: {error_msg} (尝试 {attempt+1}/{retries+1})")
                     last_error = f"Worker {worker_expr} 返回错误: {error_msg}"
                     # 错误回复（含 Zenoh 超时）也换一个 worker 重试
-                    new_worker = await self._pick_other_worker(worker_type, tried_workers)
+                    if not await _switch_worker("错误后"):
-                    if new_worker:
+                        break
-                        logger.info(f"错误后切换 Worker: {worker_expr} -> {new_worker}")
-                        worker_expr = new_worker
-                        full_key = f"{worker_expr}/{subject}"
                     continue  # 重试
             except asyncio.TimeoutError:
                 logger.warning(
@@ -772,11 +1006,8 @@ class WorkerManager:
                 )
                 last_error = f"Worker {worker_expr} 超时 (>{timeout}s)"
                 # 超时后换一个 worker 重试，避免打到仍在处理中的同一个 worker
-                new_worker = await self._pick_other_worker(worker_type, tried_workers)
+                if not await _switch_worker("超时后"):
-                if new_worker:
+                    break
-                    logger.info(f"超时后切换 Worker: {worker_expr} -> {new_worker}")
-                    worker_expr = new_worker
-                    full_key = f"{worker_expr}/{subject}"
                 continue
             except Exception as e:
                 logger.error(f"调用 worker 异常: {e}")
@@ -784,7 +1015,7 @@ class WorkerManager:
 
         raise RuntimeError(
             f"Worker '{worker_type}' 调用失败 (已尝试 {len(tried_workers)} 个 Worker, "
-            f"共 {retries + 1} 次, timeout={timeout}s): {last_error}"
+            f"实际尝试 {attempt + 1} 次/上限 {retries + 1} 次, timeout={timeout}s): {last_error}"
         )
 
     # ------------------------------------------------------------------

src/chongming/app/routes/__init__.py

@@ -1,6 +1,5 @@
-from .storage import router as storage_router
 from .stats import router as stats_router
 from .worker_admin import router as worker_admin_router
 from .auth import router as auth_router
 
-__all__ = ["storage_router", "stats_router", "worker_admin_router", "auth_router"]
+__all__ = ["stats_router", "worker_admin_router", "auth_router"]

src/chongming/app/routes/storage.py

@@ -1,53 +0,0 @@
-# src/chongming/app/routes/storage.py
-import os
-import uuid
-import urllib.parse
-import aiofiles
-
-from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Request
-from fastapi.responses import FileResponse
-
-
-router = APIRouter(prefix="/api/v1/storage", tags=["Storage"])
-
-
-@router.post("/upload", summary="上传文件")
-async def upload_file(
-    request: Request,
-    file: UploadFile = File(...),
-    worker_type: str = Form(...),
-):
-    """上传文件到本地存储"""
-    upload_root = request.app.state.upload_root
-    object_key = f"{worker_type}/{uuid.uuid4()}/{file.filename}"
-    dest = os.path.join(upload_root, object_key)
-    os.makedirs(os.path.dirname(dest), exist_ok=True)
-
-    async with aiofiles.open(dest, "wb") as afp:
-        while chunk := await file.read(1024 * 1024):
-            await afp.write(chunk)
-
-    file_size = os.path.getsize(dest)
-    return {
-        "file_key": object_key,
-        "bucket": "local",
-        "filename": file.filename,
-        "content_type": file.content_type,
-        "size": file_size,
-    }
-
-@router.get("/download/{file_key:path}")
-async def download_file(request: Request, file_key: str):
-    """从本地存储下载文件"""
-    upload_root = request.app.state.upload_root
-    path = os.path.join(upload_root, file_key)
-    if not os.path.isfile(path):
-        raise HTTPException(404, "文件不存在")
-
-    raw_filename = file_key.split('/')[-1]
-    encoded = urllib.parse.quote(raw_filename, safe='')
-    return FileResponse(
-        path,
-        filename=raw_filename,
-        headers={"Content-Disposition": f"attachment; filename*=UTF-8''{encoded}"},
-    )

test/test_call_stats.py

@@ -22,7 +22,7 @@ import httpx
 import orjson
 from fastapi import FastAPI
 
-REPO_ROOT = Path(__file__).resolve().parent
+REPO_ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(REPO_ROOT / "src"))
 sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
 
@@ -53,6 +53,11 @@ _cs_mod = importlib.util.module_from_spec(_cs_spec)
 sys.modules["chongming.app.core.call_stats"] = _cs_mod
 _cs_spec.loader.exec_module(_cs_mod)  # type: ignore[union-attr]
 
+# 审计采样率固定为全量：生产 config 的 access_sample_rate=0.1 下，成功请求的审计行
+# 本就是按概率写的，断言"几条请求 = 几条审计行"会随机失败。本测试验证的是审计链路
+# 是否工作，与采样策略无关，故固定为 1.0。
+_cs_mod._AUDIT_SAMPLE_RATE = 1.0
+
 # 加载真实 route_generator
 _rg_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "route_generator.py"
 _rg_spec = importlib.util.spec_from_file_location("chongming.app.core.route_generator", _rg_path)
@@ -95,12 +100,14 @@ class FakeWorkerManager:
         self.reply = reply
         self.error = error
 
-    async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False):
+    async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False,
+                       instance_id=None):
         if self.error == "no_worker":
             raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。")
         return self.reply
 
-    async def call_stream(self, worker_type, subject, payload, timeout=300.0, idle_timeout=10.0):
+    async def call_stream(self, worker_type, subject, payload, timeout=300.0, idle_timeout=10.0,
+                          instance_id=None):
         if self.error == "no_worker":
             raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。")
 

test/test_inline_local_dispatch.py

@@ -0,0 +1,209 @@
+"""inline 直调路径 与 zenoh 路径 的一致性测试。
+
+守护"进程插件 ↔ 内置插件无缝迁移"契约：同一个 worker 副本，分别经
+  A) 进程内直调（chongming_worker.dispatch：网关 inline 快路径用的入口）
+  B) zenoh 查询（session.get → router → queryable 回调 → 原有路径）
+发起同一批请求，逐字段比对两路结果（仅屏蔽时间戳这类天然不同的字段）。
+
+覆盖：正常 GET / POST、模型校验失败（reply_err 通道）、未知 subject、
+非信封旧式载荷、流式多块、流式生成器异常收尾块。
+
+运行: uv run python test/test_inline_local_dispatch.py
+（需要 zenoh router 在 config/zenoh_config.json5 指向的端点可达）
+"""
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+# 优先本地框架 checkout（与 test_stream_worker.py 同约定）
+sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
+sys.path.insert(0, str(REPO_ROOT / "src"))
+
+import orjson  # noqa: E402
+import zenoh  # noqa: E402
+
+from chongming_worker.dispatch import get_dispatcher, unregister_dispatcher  # noqa: E402
+from chongming_worker.runtime import create_worker  # noqa: E402
+
+WORKER_TYPE = "example"
+WORKER_ID = "worker-parity-0"
+
+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)
+
+
+def mask_timestamp(obj):
+    """屏蔽时间戳类字段（两路调用时刻不同，天然不一致）"""
+    if isinstance(obj, dict):
+        return {k: (0 if k in ("timestamp", "time") else mask_timestamp(v)) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [mask_timestamp(v) for v in obj]
+    return obj
+
+
+def normalize(payload: bytes):
+    try:
+        return mask_timestamp(orjson.loads(payload))
+    except Exception:
+        return payload
+
+
+def envelope(method: str, data: dict) -> bytes:
+    return orjson.dumps({"meta": {"method": method, "user_info": None}, "data": data})
+
+
+async def invoke_zenoh(session, subject: str, payload: bytes, timeout: float = 3.0):
+    """经 zenoh 查询取首个回复：返回 ("ok", 载荷) / ("err", 载荷) / ("noreply", None)"""
+    key = f"worker/{WORKER_TYPE}/{WORKER_ID}/{subject}"
+    loop = asyncio.get_running_loop()
+
+    def _first():
+        for reply in session.get(key, payload=payload,
+                                 consolidation=zenoh.ConsolidationMode.NONE,
+                                 timeout=timeout):
+            return reply
+        return None
+
+    reply = await loop.run_in_executor(None, _first)
+    if reply is None:
+        return ("noreply", None)
+    if reply.ok:
+        return ("ok", bytes(reply.ok.payload) if reply.ok.payload else b"")
+    return ("err", bytes(reply.err.payload))
+
+
+async def collect_zenoh_stream(session, subject: str, payload: bytes, timeout: float = 5.0):
+    key = f"worker/{WORKER_TYPE}/{WORKER_ID}/{subject}"
+    loop = asyncio.get_running_loop()
+
+    def _all():
+        out = []
+        for reply in session.get(key, payload=payload,
+                                 consolidation=zenoh.ConsolidationMode.NONE,
+                                 timeout=timeout):
+            if reply.ok:
+                out.append(bytes(reply.ok.payload) if reply.ok.payload else b"")
+            else:
+                out.append(("err", bytes(reply.err.payload)))
+        return out
+
+    return await loop.run_in_executor(None, _all)
+
+
+async def collect_local_stream(dispatcher, subject: str, payload: bytes):
+    out = []
+    async for chunk in dispatcher.invoke_stream(subject, payload):
+        out.append(chunk)
+    return out
+
+
+async def main() -> int:
+    # ── 起会话 + 一个真实 worker 副本（example 插件，纯计算端点）──
+    import plugins.example.router.add  # noqa: F401
+    import plugins.example.router.subtract  # noqa: F401
+    import plugins.example.router.stream_chat  # noqa: F401
+
+    zconfig = zenoh.Config()
+    zconfig.from_file(str(REPO_ROOT / "config" / "zenoh_config.json5"))
+    session = zenoh.open(zconfig)
+
+    worker_task = asyncio.create_task(
+        create_worker(session, WORKER_TYPE, WORKER_ID, config={"data_dir": "./data"})
+    )
+    await asyncio.sleep(1.5)  # 等 liveliness / queryable / 直调注册都就绪
+
+    dispatcher = get_dispatcher(WORKER_TYPE, WORKER_ID)
+    check("直调入口已注册（register_dispatcher）", dispatcher is not None)
+    if dispatcher is None:
+        worker_task.cancel()
+        session.close()
+        return 1
+
+    try:
+        # ── 1) 正常 GET：example.add?a&b ──
+        payload = envelope("GET", {"a": 2, "b": 3})
+        z = await invoke_zenoh(session, "example.add", payload)
+        l = await dispatcher.invoke_unary("example.add", payload)
+        l_norm = ("ok" if l.ok else "err", l.payload)
+        check("GET 两路状态一致", z[0] == l_norm[0], f"zenoh={z[0]} local={l_norm[0]}")
+        check(
+            "GET 两路载荷逐字段一致",
+            normalize(z[1] or b"") == normalize(l_norm[1]),
+            f"zenoh={z[1]!r} local={l_norm[1]!r}",
+        )
+
+        # ── 2) 正常 POST（同 subject 的另一方法）──
+        payload = envelope("POST", {"a": 5, "b": 7})
+        z = await invoke_zenoh(session, "example.add", payload)
+        l = await dispatcher.invoke_unary("example.add", payload)
+        l_norm = ("ok" if l.ok else "err", l.payload)
+        check("POST 两路状态一致", z[0] == l_norm[0], f"zenoh={z[0]} local={l_norm[0]}")
+        check("POST 两路载荷逐字段一致", normalize(z[1] or b"") == normalize(l_norm[1]))
+
+        # ── 3) 模型校验失败 → 框架错误（zenoh reply_err ⇔ local ok=False）──
+        payload = envelope("GET", {"a": "not-a-number"})
+        z = await invoke_zenoh(session, "example.add", payload)
+        l = await dispatcher.invoke_unary("example.add", payload)
+        l_norm = ("ok" if l.ok else "err", l.payload)
+        check("校验失败两路都走错误通道", z[0] == "err" and l_norm[0] == "err",
+              f"zenoh={z[0]} local={l_norm[0]}")
+        z_err, l_err = normalize(z[1] or b""), normalize(l_norm[1])
+        check("错误信封结构一致（operation=error 且含 error 文本）",
+              isinstance(z_err, dict) and isinstance(l_err, dict)
+              and z_err.get("operation") == l_err.get("operation") == "error"
+              and bool(z_err.get("error")) and bool(l_err.get("error")),
+              f"zenoh={z_err} local={l_err}")
+
+        # ── 4) 非信封旧式载荷（method 默认 GET）──
+        payload = orjson.dumps({"a": 1, "b": 1})
+        z = await invoke_zenoh(session, "example.subtract", payload)
+        l = await dispatcher.invoke_unary("example.subtract", payload)
+        l_norm = ("ok" if l.ok else "err", l.payload)
+        check("旧式载荷两路一致", z[0] == l_norm[0] and normalize(z[1] or b"") == normalize(l_norm[1]),
+              f"zenoh={z} local={l_norm}")
+
+        # ── 5) 未知 subject：两路都应失败（路径不同但终态一致：网关侧都会换副本重试后 503）──
+        z = await invoke_zenoh(session, "example.not_exists", envelope("GET", {}), timeout=1.0)
+        l = await dispatcher.invoke_unary("example.not_exists", envelope("GET", {}))
+        check("未知 subject 两路都失败（zenoh 无回复 / 直调返回错误）",
+              z[0] == "noreply" and l.ok is False, f"zenoh={z[0]} local.ok={l.ok}")
+
+        # ── 6) 流式多块 ──
+        payload = envelope("POST", {"prompt": "你好", "tokens": 4})
+        zs = await collect_zenoh_stream(session, "example.stream_chat", payload)
+        ls = await collect_local_stream(dispatcher, "example.stream_chat", payload)
+        check("流式块数一致", len(zs) == len(ls), f"zenoh={len(zs)} local={len(ls)}")
+        check("流式逐块内容一致",
+              [normalize(x) for x in zs] == [normalize(x) for x in ls],
+              f"zenoh={[normalize(x) for x in zs]} local={[normalize(x) for x in ls]}")
+
+        # ── 7) 流式生成器中途异常 → {"error": ...} 收尾块（两路都应以此结束）──
+        # 流终止语义（无 token 的空流）两路一致
+        payload = envelope("POST", {"prompt": "空", "tokens": 0})
+        zs = await collect_zenoh_stream(session, "example.stream_chat", payload)
+        ls = await collect_local_stream(dispatcher, "example.stream_chat", payload)
+        check("空流终止语义一致（不挂起、块数相同）",
+              len(zs) == len(ls), f"zenoh={len(zs)} local={len(ls)}")
+    finally:
+        worker_task.cancel()
+        try:
+            await worker_task
+        except asyncio.CancelledError:
+            pass
+        unregister_dispatcher(WORKER_TYPE, WORKER_ID)
+        session.close()
+
+    print(f"\n结果: {len(failures)} 个失败" + (f" -> {failures}" if failures else ""))
+    return 1 if failures else 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

test/test_instance_targeting.py

@@ -0,0 +1,220 @@
+"""
+worker 实例定向（把请求固定到某个进程实例）测试
+
+验证 WorkerManager 的实例定向语义（HTTP 的 X-Chongming-Instance 头 / MQTT 的
+target_instance 字段 / 程序调用 instance_id 参数最终都走这里）：
+  1. 不指定实例：与以往一致，在全部候选副本里随机挑
+  2. 指定实例：只在其实例的在线副本里挑，不跨实例
+  3. 身份来源：副本经 admin status 上报 instance_id，网关按需查询并缓存
+     （本地注册的 inline 副本直接用网关自身实例，无需查询）
+  4. 目标实例不在线：报错而不回退到别的实例（定向即严格）
+  5. 调用失败重试：限定在同一实例内换副本，绝不跨实例
+  6. 实例内已无其它副本可换：立即失败，不重复打同一个副本
+  7. 实例维度缓存键：定向调用的缓存与其它实例隔离
+
+运行: .venv/bin/python test/test_instance_targeting.py
+"""
+import asyncio
+import sys
+from pathlib import Path
+
+_HERE = Path(__file__).resolve().parent          # chongming-zenoh/
+_PROJ = _HERE
+for _p in (_PROJ / "src", _PROJ / "utils" / "worker" / "src"):
+    sys.path.insert(0, str(_p))
+
+from chongming.app.core.worker_manager import WorkerManager  # noqa: E402
+
+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 _OkPayload:
+    def __init__(self, data: bytes):
+        self.payload = data
+
+
+class _Ok:
+    def __init__(self, data: bytes):
+        self.ok = _OkPayload(data)
+
+
+class _Reply:
+    """一次 get 的回复：ok 与 err 互斥（与 Zenoh reply 的形态一致）。"""
+
+    def __init__(self, data: bytes = b"{}", err: bytes | None = None):
+        if err is None:
+            self.ok = _OkPayload(data)
+            self.err = None
+        else:
+            self.ok = None
+            self.err = _OkPayload(err)
+
+
+class FakeSession:
+    """记录每次 get 的 key，并按 key 决定回复。
+
+    admin 查询（worker/<type>/<id>/__admin__）由 admin_instances 决定 instance_id；
+    业务调用（worker/<type>/<id>/<subject>）可按 call_errors 注入错误回复。
+    """
+
+    def __init__(self, admin_instances: dict | None = None,
+                 call_errors: set | None = None):
+        self.admin_instances = admin_instances or {}
+        self.call_errors = set(call_errors or ())
+        self.calls: list = []
+
+    def get(self, key, payload=None, **kwargs):
+        self.calls.append(key)
+        if key.endswith("/__admin__"):
+            wid = key.split("/")[2]
+            iid = self.admin_instances.get(wid)
+            body = (
+                f'{{"worker_id":"{wid}","instance_id":"{iid}","status":"running"}}'.encode()
+                if iid else f'{{"worker_id":"{wid}","status":"running"}}'.encode()
+            )
+            return iter([_Reply(body)])
+        if key in self.call_errors:
+            return iter([_Reply(err=b'{"error":"boom"}')])
+        return iter([_Reply(b'{"ok":true}')])
+
+    def liveliness(self):
+        raise RuntimeError("本测试不触及 liveliness")
+
+    def close(self):
+        pass
+
+
+def new_manager(session, gateway_id="gw-local"):
+    m = WorkerManager(session=session)
+    m._gateway_id = gateway_id
+    return m
+
+
+async def main():
+    # ── 1. 不定向：与以往一致，全候选集随机 ─────────────────
+    sess = FakeSession(admin_instances={"w-a": "edge-01", "w-b": "edge-02"})
+    m = new_manager(sess)
+    m._workers_cache["demo"] = ["worker/demo/w-a", "worker/demo/w-b"]
+    picked = {await m.get_worker("demo") for _ in range(60)}
+    check("1.1 不定向时可命中全部副本", picked == set(m._workers_cache["demo"]),
+          f"picked={sorted(picked)}")
+    check("1.2 不定向不触发实例身份查询（零额外 RPC）",
+          not [c for c in sess.calls if c.endswith("__admin__")], f"calls={sess.calls}")
+    check("1.3 不定向调用走随机副本",
+          await m.call_raw("demo", "ping", b"{}") == b'{"ok":true}')
+
+    # ── 2. 定向：只在目标实例内挑副本 ────────────────────────
+    sess = FakeSession(admin_instances={"w-a": "edge-01", "w-b": "edge-02",
+                                        "w-c": "edge-01"})
+    m = new_manager(sess)
+    m._workers_cache["demo"] = [
+        "worker/demo/w-a", "worker/demo/w-b", "worker/demo/w-c",
+    ]
+    picked = {await m.get_worker("demo", instance_id="edge-01") for _ in range(60)}
+    check("2.1 定向只命中目标实例的副本",
+          picked == {"worker/demo/w-a", "worker/demo/w-c"}, f"picked={sorted(picked)}")
+
+    # ── 3. 身份查询：按需查询并缓存，实例归属正确 ─────────────
+    sess = FakeSession(admin_instances={"w-a": "edge-01", "w-b": "edge-02",
+                                        "w-c": "edge-01"})
+    m = new_manager(sess)
+    m._workers_cache["demo"] = [
+        "worker/demo/w-a", "worker/demo/w-b", "worker/demo/w-c",
+    ]
+    index = await m._instance_workers("demo", "edge-01")
+    check("3.1 实例→副本反向索引正确",
+          sorted(index) == ["worker/demo/w-a", "worker/demo/w-c"], f"index={sorted(index)}")
+    admin_calls = [c for c in sess.calls if c.endswith("__admin__")]
+    check("3.2 每个副本只查询一次身份", len(admin_calls) == 3, f"n={len(admin_calls)}")
+    await m._refresh_instance_map("demo")
+    check("3.3 重复刷新复用缓存不重复查询",
+          len([c for c in sess.calls if c.endswith("__admin__")]) == len(admin_calls))
+
+    # ── 3b. inline 本地注册副本：直接用网关实例，零查询 ──────
+    sess = FakeSession()
+    m = new_manager(sess, gateway_id="gw-1")
+    m.register_worker("demo", "worker-0-1-999")
+    index = await m._instance_workers("demo", "gw-1")
+    check("3.4 inline 副本归入网关实例", index == ["worker/demo/worker-0-1-999"],
+          f"index={index}")
+    check("3.5 inline 副本身份查询零 RPC",
+          not [c for c in sess.calls if c.endswith("__admin__")], f"calls={sess.calls}")
+
+    # ── 4. 定向实例不在线：严格报错，不回退 ──────────────────
+    sess = FakeSession(admin_instances={"w-a": "edge-01"})
+    m = new_manager(sess)
+    m._workers_cache["demo"] = ["worker/demo/w-a"]
+    check("4.1 目标实例无副本时 get_worker 返回 None",
+          await m.get_worker("demo", instance_id="edge-99") is None)
+    try:
+        await m.call_raw("demo", "ping", b"{}", instance_id="edge-99")
+        check("4.2 定向到离线实例应报错", False, "未抛错")
+    except RuntimeError as e:
+        msg = str(e)
+        check("4.2 定向到离线实例报错且说明不跨实例回退",
+              "edge-99" in msg and "不" in msg, f"msg={msg[:90]}")
+
+    # ── 5. 重试锚定同实例：绝不跨实例 ────────────────────────
+    sess = FakeSession(
+        admin_instances={"w-a": "edge-01", "w-b": "edge-02", "w-c": "edge-01"},
+        call_errors={"worker/demo/w-b/ping"},  # 只让别实例的副本出错，验证不会被选中
+    )
+    m = new_manager(sess)
+    m._workers_cache["demo"] = [
+        "worker/demo/w-a", "worker/demo/w-b", "worker/demo/w-c",
+    ]
+    # 目标实例 edge-01 内的 w-a 失败 -> 应换到同实例的 w-c，不碰 edge-02 的 w-b
+    sess.call_errors = {"worker/demo/w-a/ping"}
+    out = await m.call_raw("demo", "ping", b"{}", instance_id="edge-01", retries=2)
+    business = [c for c in sess.calls if not c.endswith("__admin__")]
+    check("5.1 重试不跨实例（未打其它实例的副本）",
+          all("/w-b/" not in c for c in business), f"calls={business}")
+    check("5.2 同实例内换副本后成功", out == b'{"ok":true}', f"out={out!r}")
+
+    # ── 6. 实例内无其它副本可换：立即失败 ────────────────────
+    sess = FakeSession(admin_instances={"w-a": "edge-01"},
+                       call_errors={"worker/demo/w-a/ping"})
+    m = new_manager(sess)
+    m._workers_cache["demo"] = ["worker/demo/w-a", "worker/demo/w-b"]
+    sess.admin_instances["w-b"] = "edge-02"
+    raised = None
+    try:
+        await m.call_raw("demo", "ping", b"{}", instance_id="edge-01", retries=2)
+    except RuntimeError as e:
+        raised = e
+    check("6.1 实例内无副本可换应报错", raised is not None, f"raised={raised!r}")
+    n = len([c for c in sess.calls if not c.endswith("__admin__")])
+    check("6.2 不重复打同一个副本（只调用一次）", n == 1, f"业务调用次数={n}")
+    check("6.3 错误信息点名实例与原因",
+          raised is not None and "edge-01" in str(raised), f"msg={str(raised)[:90]}")
+
+    # ── 7. 定向调用的缓存按实例隔离 ──────────────────────────
+    sess = FakeSession(admin_instances={"w-a": "edge-01", "w-b": "edge-02"})
+    m = new_manager(sess)
+    m._workers_cache["demo"] = ["worker/demo/w-a", "worker/demo/w-b"]
+    r1 = await m.call_raw("demo", "state", b"{}", cacheable=True, instance_id="edge-01")
+    r2 = await m.call_raw("demo", "state", b"{}", cacheable=True, instance_id="edge-02")
+    keys = [f"{iid}@demo:state:" for iid in ("edge-01", "edge-02")]
+    check("7.1 不同实例的定向缓存键互不影响",
+          all(any(k.startswith(p) for k in m._rpc_mem_cache._d) for p in keys),
+          f"cache_keys={list(m._rpc_mem_cache._d)}")
+    check("7.2 定向调用结果正常返回", r1 == b'{"ok":true}' and r2 == b'{"ok":true}',
+          f"r1={r1!r}, r2={r2!r}")
+
+    print()
+    if failures:
+        print(f"FAILED: {len(failures)} 项 -> {failures}")
+        return 1
+    print("全部通过")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(asyncio.run(main()))

test/test_instance_targeting_http.py

@@ -0,0 +1,207 @@
+"""
+HTTP 实例定向头（X-Chongming-Instance）透传测试
+
+依赖隔离：桩掉 chongming 包链、auth 与 chongming_cache，直接加载真实
+route_generator.py 与 worker_manager.py，用 FastAPI TestClient 验证:
+  1. 携带 X-Chongming-Instance 的请求，该实例 id 透传到 call_raw / call_stream
+  2. 未携带该头时 instance_id 为 None（行为与改动前完全一致，兼容旧调用方）
+  3. 覆盖全部端点形态：GET query / POST body / validate=false 原样透传 / SSE 流式
+  4. 头值为空白时视为未指定
+  5. 定向到离线实例（RuntimeError）-> HTTP 503，错误信息透出
+
+运行: .venv/bin/python test/test_instance_targeting_http.py
+"""
+import importlib.util
+import logging
+import sys
+import types
+from pathlib import Path
+
+import orjson
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+TEST_DIR = Path(__file__).resolve().parent
+REPO_ROOT = TEST_DIR.parent
+sys.path.insert(0, str(REPO_ROOT / "src"))
+sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
+
+# ---- 依赖隔离：桩掉 chongming 包链 / auth / chongming_cache ----
+_pkg = types.ModuleType("chongming")
+_pkg.__path__ = []
+_app = types.ModuleType("chongming.app")
+_app.__path__ = []
+_core = types.ModuleType("chongming.app.core")
+_core.__path__ = []
+_logger_stub = types.ModuleType("chongming.app.core.logger")
+_logger_stub.get_logger = lambda name: logging.getLogger(name)
+_auth_stub = types.ModuleType("chongming.app.core.auth")
+_auth_stub.store_user_to_request = lambda x: x
+_cache_stub = types.ModuleType("chongming_cache")
+_cache_stub.Cache = object
+_cache_stub.DistributedLock = object  # worker_manager 顶层导入（本测试不触达）
+for mod in (_pkg, _app, _core, _logger_stub, _auth_stub, _cache_stub):
+    sys.modules[mod.__name__] = mod
+
+_wm_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "worker_manager.py"
+_wm_spec = importlib.util.spec_from_file_location(
+    "chongming.app.core.worker_manager", _wm_path
+)
+_wm_mod = importlib.util.module_from_spec(_wm_spec)
+sys.modules["chongming.app.core.worker_manager"] = _wm_mod
+_wm_spec.loader.exec_module(_wm_mod)  # type: ignore[union-attr]
+
+_cs_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "call_stats.py"
+_cs_spec = importlib.util.spec_from_file_location(
+    "chongming.app.core.call_stats", _cs_path
+)
+_cs_mod = importlib.util.module_from_spec(_cs_spec)
+sys.modules["chongming.app.core.call_stats"] = _cs_mod
+_cs_spec.loader.exec_module(_cs_mod)  # type: ignore[union-attr]
+
+_rg_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "route_generator.py"
+_rg_spec = importlib.util.spec_from_file_location(
+    "chongming.app.core.route_generator", _rg_path
+)
+_rg_mod = importlib.util.module_from_spec(_rg_spec)
+sys.modules["chongming.app.core.route_generator"] = _rg_mod
+_rg_spec.loader.exec_module(_rg_mod)  # type: ignore[union-attr]
+
+register_route_group = _rg_mod.register_route_group
+INSTANCE_HEADER = _rg_mod.INSTANCE_HEADER
+
+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:
+    """记录每次调用收到的 instance_id；可指定定点失败以验证 503 映射。"""
+
+    def __init__(self, fail_instance=None):
+        self.fail_instance = fail_instance            # 指定实例报错（模拟实例离线）
+        self.raw_instances = []                       # call_raw 收到的 instance_id
+        self.stream_instances = []                    # call_stream 收到的 instance_id
+
+    async def call_raw(self, worker_type, subject, payload, timeout=3.0,
+                       retries=1, cacheable=False, instance_id=None):
+        self.raw_instances.append(instance_id)
+        if instance_id and instance_id == self.fail_instance:
+            raise RuntimeError(
+                f"类型 '{worker_type}' 的实例 '{instance_id}' 没有在线副本"
+                "（定向调用不跨实例回退）。"
+            )
+        return orjson.dumps({"op": subject, "instance": instance_id})
+
+    async def call_stream(self, worker_type, subject, payload, timeout=300.0,
+                          idle_timeout=10.0, instance_id=None):
+        self.stream_instances.append(instance_id)
+        if instance_id and instance_id == self.fail_instance:
+            raise RuntimeError(f"实例 '{instance_id}' 没有在线副本")
+
+        async def _gen():
+            yield orjson.dumps({"instance": instance_id})
+
+        return _gen()
+
+
+ROUTES = [
+    {"subject": "t.get", "path": "/calc/add", "method": "GET",
+     "params": ["a: float", "b: float"], "summary": "add",
+     "timeout": 3.0, "auth_required": False},
+    {"subject": "t.post", "path": "/calc/echo", "method": "POST",
+     "params": ["msg: str"], "summary": "echo",
+     "timeout": 3.0, "auth_required": False},
+    {"subject": "t.raw", "path": "/calc/raw", "method": "POST",
+     "params": ["payload: any"], "summary": "原样透传",
+     "validate": False, "timeout": 3.0, "auth_required": False},
+    {"subject": "t.stream", "path": "/stream/chat", "method": "POST",
+     "params": ["prompt: str"], "summary": "sse",
+     "stream": True, "timeout": 60.0, "stream_idle_timeout": 1.0,
+     "auth_required": False},
+]
+
+
+def build_client(fwm):
+    app = FastAPI()
+    register_route_group(app, fwm, "demotype", ROUTES, prefix="/api/v1")
+    return TestClient(app)
+
+
+def main():
+    # ── 1. GET query 端点 ────────────────────────────────────
+    fwm = FakeWorkerManager()
+    c = build_client(fwm)
+    r = c.get("/api/v1/calc/add?a=1&b=2", headers={INSTANCE_HEADER: "edge-01"})
+    check("1.1 GET 携带头 -> 200", r.status_code == 200, f"status={r.status_code}")
+    check("1.2 GET 头透传到 call_raw", fwm.raw_instances == ["edge-01"],
+          f"instances={fwm.raw_instances}")
+    check("1.3 响应体回带实例（证明打到目标实例）",
+          r.json().get("instance") == "edge-01", f"body={r.json()}")
+
+    # ── 2. 未携带头：兼容旧调用方 ────────────────────────────
+    fwm = FakeWorkerManager()
+    c = build_client(fwm)
+    r = c.get("/api/v1/calc/add?a=1&b=2")
+    check("2.1 未携带头 -> 200（默认随机均衡路径不变）", r.status_code == 200)
+    check("2.2 未携带头时 instance_id=None", fwm.raw_instances == [None],
+          f"instances={fwm.raw_instances}")
+    check("2.3 响应体不含实例", r.json().get("instance") is None, f"body={r.json()}")
+
+    # ── 3. POST body 端点 ────────────────────────────────────
+    fwm = FakeWorkerManager()
+    c = build_client(fwm)
+    r = c.post("/api/v1/calc/echo", json={"msg": "hi"},
+               headers={INSTANCE_HEADER: "edge-02"})
+    check("3.1 POST body 头透传", fwm.raw_instances == ["edge-02"],
+          f"instances={fwm.raw_instances}")
+
+    # ── 4. validate=false 原样透传端点 ───────────────────────
+    fwm = FakeWorkerManager()
+    c = build_client(fwm)
+    r = c.post("/api/v1/calc/raw", content=b'{"anything": [1, 2]}',
+               headers={INSTANCE_HEADER: "edge-03",
+                        "content-type": "application/json"})
+    check("4.1 原样透传端点头透传", fwm.raw_instances == ["edge-03"],
+          f"instances={fwm.raw_instances}")
+
+    # ── 5. SSE 流式端点 ──────────────────────────────────────
+    fwm = FakeWorkerManager()
+    c = build_client(fwm)
+    with c.stream("POST", "/api/v1/stream/chat", json={"prompt": "x"},
+                  headers={INSTANCE_HEADER: "edge-04"}) as r:
+        body = b"".join(r.iter_bytes())
+    check("5.1 流式端点头透传", fwm.stream_instances == ["edge-04"],
+          f"instances={fwm.stream_instances}")
+    check("5.2 流式响应含目标实例", b"edge-04" in body, f"body={body[:80]!r}")
+
+    # ── 6. 空白头值视为未指定 ────────────────────────────────
+    fwm = FakeWorkerManager()
+    c = build_client(fwm)
+    r = c.get("/api/v1/calc/add?a=1&b=2", headers={INSTANCE_HEADER: "   "})
+    check("6.1 空白头值 -> instance_id=None", fwm.raw_instances == [None],
+          f"instances={fwm.raw_instances}")
+
+    # ── 7. 定向实例离线 -> 503 ───────────────────────────────
+    fwm = FakeWorkerManager(fail_instance="edge-gone")
+    c = build_client(fwm)
+    r = c.get("/api/v1/calc/add?a=1&b=2", headers={INSTANCE_HEADER: "edge-gone"})
+    check("7.1 定向实例离线 -> 503", r.status_code == 503, f"status={r.status_code}")
+    check("7.2 503 详情点名实例", "edge-gone" in str(r.json().get("detail", "")),
+          f"detail={r.json().get('detail')}")
+
+    print()
+    if failures:
+        print(f"FAILED: {len(failures)} 项 -> {failures}")
+        return 1
+    print("全部通过")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

test/test_mqtt_custom_topics.py

@@ -33,7 +33,8 @@ class FakeWM:
         self.calls = []
         self.result = {"result": "ok"}
 
-    async def call(self, worker_type, subject, payload, timeout=30, cacheable=False):
+    async def call(self, worker_type, subject, payload, timeout=30, cacheable=False,
+                   instance_id=None):
         self.calls.append(
             {
                 "worker_type": worker_type,
@@ -41,6 +42,7 @@ class FakeWM:
                 "payload": payload,
                 "timeout": timeout,
                 "cacheable": cacheable,
+                "instance_id": instance_id,
             }
         )
         return self.result
@@ -153,6 +155,41 @@ async def test_flat_payload_routed():
     print("✅ 非信封（扁平）载荷按原始报文路由通过")
 
 
+async def test_target_instance():
+    """实例定向：载荷顶层 target_instance 作为控制字段摘除并透传给 worker_manager.call"""
+    wm = FakeWM()
+    bridge, published = make_bridge(
+        wm, [{"topic": "devices/+/telemetry", "worker": "device"}]
+    )
+    # 1. 带 target_instance：透传实例名，且控制字段不进业务 payload
+    await bridge._handle(make_msg(
+        "devices/d1/telemetry",
+        {"meta": {}, "data": {"v": 1}, "target_instance": "edge-01"},
+    ))
+    assert wm.calls[-1]["instance_id"] == "edge-01", wm.calls[-1]
+    assert "target_instance" not in wm.calls[-1]["payload"], wm.calls[-1]["payload"]
+    assert wm.calls[-1]["payload"]["data"] == {"v": 1}
+
+    # 2. 不带该字段：instance_id 为 None（跨实例随机均衡，与改动前一致）
+    await bridge._handle(make_msg("devices/d1/telemetry", {"meta": {}, "data": {"v": 1}}))
+    assert wm.calls[-1]["instance_id"] is None, wm.calls[-1]
+
+    # 3. 空白值视为未指定
+    await bridge._handle(make_msg(
+        "devices/d1/telemetry",
+        {"meta": {}, "data": {"v": 1}, "target_instance": "   "},
+    ))
+    assert wm.calls[-1]["instance_id"] is None, wm.calls[-1]
+
+    # 4. 内置 chongming/... 主题同样支持
+    await bridge._handle(make_msg(
+        f"{MQTT_TOPIC_PREFIX}/example/add",
+        {"meta": {}, "data": {"a": 1}, "target_instance": "edge-02"},
+    ))
+    assert wm.calls[-1]["instance_id"] == "edge-02", wm.calls[-1]
+    print("✅ 实例定向（target_instance 控制字段）透传通过")
+
+
 async def test_non_json_dropped():
     wm = FakeWM()
     bridge, published = make_bridge(
@@ -219,6 +256,7 @@ async def main():
     await test_custom_routing()
     await test_envelope_meta_fields()
     await test_flat_payload_routed()
+    await test_target_instance()
     await test_non_json_dropped()
     await test_board_ping_topic()
     test_plugin_config_driven_collection()

test/test_passthrough_optimization.py

@@ -16,12 +16,13 @@
   D. WorkerManager 单测（FakeSession）:
      9. call_raw 零解析返回原始字节；call() 仍解析为 dict（MQTT 语义保留）
      10. cacheable: 缓存原始 bytes、二次命中不再发查询、业务错误不缓存
-     11. reply.err -> 重试 -> RuntimeError；空回复兜底 b"{}"
+     11. reply.err -> 换副本重试/无副本可换时立即 RuntimeError；空回复兜底 b"{}"
 
 运行: uv run python test_passthrough_optimization.py
 """
 import asyncio
 import importlib.util
+import random
 import logging
 import sys
 import types
@@ -32,7 +33,7 @@ import orjson
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
-REPO_ROOT = Path(__file__).resolve().parent
+REPO_ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(REPO_ROOT / "src"))
 sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
 
@@ -49,6 +50,7 @@ _auth_stub = types.ModuleType("chongming.app.core.auth")
 _auth_stub.store_user_to_request = lambda x: x  # 仅用于 Depends，桩掉鉴权
 _cache_stub = types.ModuleType("chongming_cache")
 _cache_stub.Cache = object
+_cache_stub.DistributedLock = object  # worker_manager 顶层导入（本测试不触达）
 for mod in (_pkg, _app, _core, _logger_stub, _auth_stub, _cache_stub):
     sys.modules[mod.__name__] = mod
 
@@ -101,7 +103,8 @@ class FakeWorkerManager:
         self.error = error
         self.call_raw_payloads = []  # 捕获的预序列化请求字节
 
-    async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False):
+    async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False,
+                       instance_id=None):
         self.call_raw_payloads.append(payload)
         if self.error == "no_worker":
             raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。")
@@ -282,16 +285,27 @@ def test_openapi_request_body_injected():
 # ---- D. WorkerManager 单测（FakeSession）----
 
 class FakeSession:
-    """模拟 zenoh Session：get() 返回单回复迭代器"""
+    """模拟 zenoh Session：get() 返回单回复迭代器
 
-    def __init__(self, payload=b'{"result": 5}', err_payload=None):
+    err_payload 非空时整体报错；err_target 给定时只对该 key 前缀报错（其余正常回复），
+    用于验证「失败换副本重试」。
+    """
+
+    def __init__(self, payload=b'{"result": 5}', err_payload=None, err_target=None):
         self.payload = payload
         self.err_payload = err_payload
+        self.err_target = err_target
         self.get_calls = 0
+        self.call_keys = []
 
     def get(self, *args, **kwargs):
         self.get_calls += 1
-        if self.err_payload is not None:
+        key = args[0] if args else kwargs.get("key_expr", "")
+        self.call_keys.append(str(key))
+        failing = self.err_payload is not None and (
+            self.err_target is None or str(key).startswith(self.err_target)
+        )
+        if failing:
             reply = SimpleNamespace(ok=None, err=SimpleNamespace(payload=self.err_payload))
         else:
             reply = SimpleNamespace(ok=SimpleNamespace(payload=self.payload), err=None)
@@ -341,14 +355,49 @@ async def test_wm_cache_bytes():
 
 
 async def test_wm_reply_err_retry():
+    # 单副本：无其它副本可换 —— 立即失败（不空转重试同一个坏副本），保留具体错误
     session = FakeSession(err_payload=b'{"error": "boom"}')
     wm = make_wm(session)
     try:
         await wm.call_raw("demo", "calc.add", b'{}')
         check("reply.err 应抛 RuntimeError", False)
     except RuntimeError as e:
-        check("reply.err 重试后抛 RuntimeError", "boom" in str(e), str(e))
+        check("reply.err 抛 RuntimeError 且保留具体错误", "boom" in str(e), str(e))
-    check("reply.err 触发重试（2 次查询）", session.get_calls == 2, str(session.get_calls))
+    check("单副本无其它可换时不空转重试（1 次查询）",
+          session.get_calls == 1, str(session.get_calls))
+
+
+async def test_wm_reply_err_switches_worker():
+    """多副本：先打的副本报错则换另一个副本重试，且不重复打已失败的副本。
+
+    首个副本由 random.choice 决定，故对多个 seed 试跑：必有一个覆盖「先失败后成功」
+    路径（另一个覆盖「首个即成功」），两条路径都不允许重复打同一个失败的副本。
+    """
+    ok, detail = False, "无 seed 覆盖到失败换副本路径"
+    for seed in (0, 1):
+        random.seed(seed)
+        session = FakeSession(
+            payload=b'{"result": 5}', err_payload=b'{"error": "boom"}',
+            err_target="worker/demo/w-dead",
+        )
+        wm = make_wm(session)
+        wm._workers_cache = {"demo": ["worker/demo/w-dead", "worker/demo/w-alive"]}
+        try:
+            result = await wm.call_raw("demo", "calc.add", b'{}', retries=2)
+        except RuntimeError as e:  # 未换到正常副本（不该发生）
+            detail = f"seed={seed} 意外失败: {e}"
+            continue
+        assert result == b'{"result": 5}', result
+        dead_calls = session.call_keys.count("worker/demo/w-dead/calc.add")
+        if dead_calls > 1:
+            detail = f"seed={seed} 重复打失败副本: {session.call_keys}"
+            continue
+        if session.call_keys == [
+            "worker/demo/w-dead/calc.add",
+            "worker/demo/w-alive/calc.add",
+        ]:
+            ok, detail = True, f"seed={seed} keys={session.call_keys}"
+    check("失败副本 -> 换副本重试成功（且不重复打失败副本）", ok, detail)
 
 
 async def test_wm_empty_reply_fallback():
@@ -375,6 +424,7 @@ def run():
     asyncio.run(test_wm_call_dict_semantics())
     asyncio.run(test_wm_cache_bytes())
     asyncio.run(test_wm_reply_err_retry())
+    asyncio.run(test_wm_reply_err_switches_worker())
     asyncio.run(test_wm_empty_reply_fallback())
 
     print(f"\n结果: {len(failures)} 个失败")

test/test_stream_e2e_zenoh.py

@@ -24,7 +24,7 @@ from pathlib import Path
 import orjson
 import zenoh
 
-REPO_ROOT = Path(__file__).resolve().parent
+REPO_ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
 sys.path.insert(0, str(REPO_ROOT / "src"))
 
@@ -39,6 +39,7 @@ _logger_stub = types.ModuleType("chongming.app.core.logger")
 _logger_stub.get_logger = lambda name: logging.getLogger(name)
 _cache_stub = types.ModuleType("chongming_cache")
 _cache_stub.Cache = object
+_cache_stub.DistributedLock = object  # worker_manager 顶层导入（本测试不触达）
 for mod in (_pkg, _app, _core, _logger_stub, _cache_stub):
     sys.modules[mod.__name__] = mod
 
@@ -67,7 +68,8 @@ def check(name, cond, detail=""):
         failures.append(name)
 
 
-LISTEN_PORT = 7447
+import os
+LISTEN_PORT = int(os.environ.get("CM_STREAM_TEST_PORT", "7447"))
 LISTEN = f"tcp/127.0.0.1:{LISTEN_PORT}"
 
 

test/test_stream_route.py

@@ -21,7 +21,7 @@ import orjson
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
-REPO_ROOT = Path(__file__).resolve().parent
+REPO_ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(REPO_ROOT / "src"))
 sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
 
@@ -38,6 +38,7 @@ _auth_stub = types.ModuleType("chongming.app.core.auth")
 _auth_stub.store_user_to_request = lambda x: x  # 仅用于 Depends，桩掉鉴权
 _cache_stub = types.ModuleType("chongming_cache")
 _cache_stub.Cache = object
+_cache_stub.DistributedLock = object  # worker_manager 顶层导入（本测试不触达）
 for mod in (_pkg, _app, _core, _logger_stub, _auth_stub, _cache_stub):
     sys.modules[mod.__name__] = mod
 
@@ -90,18 +91,21 @@ class FakeWorkerManager:
         self.call_payloads = []
         self.stream_payloads = []
 
-    async def call(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False):
+    async def call(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False,
+                   instance_id=None):
         self.call_payloads.append((worker_type, subject, payload))
         data = payload["data"]
         return {"result": data.get("a", 0) + data.get("b", 0)}
 
-    async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False):
+    async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False,
+                       instance_id=None):
         # 网关透传路径：payload 为预序列化字节，回复原始字节
         self.call_payloads.append((worker_type, subject, orjson.loads(payload)))
         data = orjson.loads(payload)["data"]
         return orjson.dumps({"result": data.get("a", 0) + data.get("b", 0)})
 
-    async def call_stream(self, worker_type, subject, payload, timeout=300.0, idle_timeout=10.0):
+    async def call_stream(self, worker_type, subject, payload, timeout=300.0, idle_timeout=10.0,
+                          instance_id=None):
         if self.stream_error == "no_worker":
             raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。")
         self.stream_payloads.append((worker_type, subject, payload))