modify: 微服务演进

现象：cache 模式下应用重启后插件仍按 toml 跑，而前端（读缓存）显示的是缓存值。
根因：跑起来的产物里 chongming-worker 是索引里的发布版（uv run build / uv sync 会把
uv pip install -e utils/worker 的可编辑安装替换回锁文件版本），旧版 Plugin.create_workers
在 on_init 时只拿到 config/workers/<type>/config.toml 展平出来的配置。

改动（两层对齐，规则与 SDK 一致且幂等）：
- WorkerManager.align_inline_config：创建 inline 副本之前按 [cache].launch_worker_count
  对齐 [workers.config]——cache 模式取缓存值（缺失才用 toml 引导并写回），toml 模式以
  文件为准并把该段覆盖写入缓存；未配置缓存后端时原样返回。网关侧这层让网关与 worker
  包版本解耦：worker 包仍是旧版时 cache 模式同样生效。
- _discover_plugins 透出 [workers.config] 展平前的原始段；load_workers 在注册 starter/
  创建副本之前完成对齐（插件实例 on_init 用的就是这份配置）。
- test/test_inline_config_align.py：两模式 + 无缓存/无段的边界用例，并校验缓存键与 SDK 一致。
- README：补打包陷阱（可编辑安装会被 uv sync 替换回发布版）与对齐时机说明。

.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 模型绑定 |
@@ -113,11 +113,22 @@ uv sync     # utils/* 由 uv 按 pyproject 的 git 源自动安装，无需手
 
 **修改 `utils/<name>`（工具包）**：
 
-1. 在 `utils/<name>` 内提交并 push 到对应远程
+1. 在 `utils/<name>` 内提交并 push 到对应远程（并发布新版本，见下）
 2. 回主仓库执行 `uv lock --upgrade-package <包名>` 刷新版本 pin
 3. `uv sync` 后运行相关测试
 4. 主仓库提交 `uv.lock`
 
+> ⚠️ **打包陷阱（"改了 utils/* 但产物行为没变"几乎都是它）**：`uv run build` 会先按
+> 锁文件同步环境，**`uv pip install -e utils/<name>` 的可编辑安装会被替换回锁文件里的
+> 发布版**。因此本地改了工具包后，必须走上面 1-4 步（push + 发版 → 刷新 pin → sync），
+> 否则 `build/` 产物里仍是旧版代码。仅本地联调时可临时用可编辑安装，但打包前要确认
+> venv 指向的是准备发布的版本：
+>
+> ```bash
+> uv pip install -e utils/worker --no-deps     # 仅联调：让 venv 用本地源码
+> python -c "import chongming_worker, inspect; print(inspect.getsourcefile(chongming_worker))"
+> ```
+
 **修改 `src/plugins/` / `config/` / `src/chongming-web/`**：
 
 1. 在对应子仓库内提交并 push
@@ -127,6 +138,8 @@ uv sync     # utils/* 由 uv 按 pyproject 的 git 源自动安装，无需手
 
 - [ ] 所有子仓库 `git status` 干净且已 push
 - [ ] `uv.lock` 与各 git 源最新提交一致
+- [ ] 打包前确认 venv 里 `chongming-worker` 等工具包为目标版本（改了 `utils/*` 却仍是旧版时，
+      产物行为不会变，见上一节打包陷阱）
 - [ ] 配置中无硬编码凭据——密钥一律走环境变量：
       `VOLC_ACCESSKEY`（豆包 Ark Key）、`AIBOX_MQTT_URI`（AI Box 外部 Broker URI）
 - [ ] `uv run python test_mqtt_e2e.py` 等测试通过
@@ -181,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]
@@ -297,6 +310,8 @@ worker/<type>/<worker_id>/__admin__
 
   两种模式都不回写 `config.toml`：管理接口的改动只进缓存。`cache` 模式下启动日志会打印 toml 与缓存的差异键（以缓存为准）；如需让 toml 的改动生效，先删除 `worker:cm:configs:<type>`（运行时配置）或 `worker:cm:counts:<type>`（副本数）缓存键再重启，或改用 `"toml"` 模式。`GET /admin/workers/<type>/config` 的响应带 `source`（cache/toml）与 `mode`，前端据此标注展示值的来路。
 
+- **对齐时机（两层，规则一致且幂等）**：inline 插件由**网关**在创建副本之前对齐——`WorkerManager.align_inline_config()` 在收到展平配置与 `[workers.config]` 原始段后按模式取缓存/写缓存，再交给插件的 `create_workers`；worker SDK 的 `Plugin.create_workers` 在 `_bootstrap()`/`on_init()` 之前用 `resolve_startup_config()` 再解析一次。网关侧这层是为了让网关与 worker 包版本解耦：worker 包仍是旧版时 cache 模式同样生效（否则旧版 `on_init` 只拿到 toml 展平配置，表现为"前端显示缓存值、实际跑 toml"）。**external 插件**没有网关这层兜底，由插件进程内的 SDK 解析，需要 worker 包为含该改动的版本。
+
 - **进程插件（external）实例粒度参与分摊**：网关只对 **inline** 插件用 cache 解析启动副本数（见「启动流程」第 5 步 `load_workers`），external 插件的启动副本数由外部编排按自身配置决定；但**运行时的缩放**上，external 并不退化为“每进程全额”，而是同样在**共享 cache 上登记实例参与方**来分摊，避免多实例把副本翻倍：
   - **实例登记 + 心跳租约**：每个 external 进程启动时向 `worker:cm:instances:<type>` 登记自身参与方（`[instance_id, ts]`），并每 15s 心跳续约；超过租约（60s）未心跳的死实例会被清理。`divmod` 按存活参与方 N 均分全局 count（余量补靠前实例），`N==1` 时退化为全额（单部署零影响）。
   - **实例标识默认唯一**：`--instance` / 环境变量 `CHONGMING_WORKER_INSTANCE` 显式指定（跨重启稳定、逐实例精细扩缩可持续命中）；缺省回退 `{hostname}:{pid}`——**同一机器并行多个未指定 `--instance` 的进程也能靠 pid 区分**，不会因为 id 相同被并成一个参与方导致副本翻倍或前端只显示一个实例。注意缺省 id 随进程重启变化。
@@ -313,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>` 按网关隔离：

src/chongming/app/__init__.py

@@ -20,10 +20,9 @@ from fastapi.middleware.cors import CORSMiddleware
 
 from chongming_config import get_config
 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()
@@ -311,7 +310,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/lifespan_setup.py

@@ -399,6 +399,9 @@ def _discover_plugins() -> list[dict]:
             "load_mode": load_mode,
             "mqtt_topics": mqtt_topics,
             "config": runtime_config,
+            # [workers.config] 展平前的原始段：供 load_workers 按「配置优先来源模式」
+            # 引导/覆盖共享缓存，使 inline 插件初始化即拿到缓存值（cache 模式）
+            "config_section": dict(workers_section.get("config") or {}),
         })
 
     # 2. 兼容旧写法: config/workers/config.toml（仅加载尚未发现的插件）
@@ -475,7 +478,12 @@ async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerMana
         # 否则用配置值引导并写回 cache（多网关共享同一 cache 时自动对齐）。
         count = wc["count"]
         worker_type = wc["type"]
-        worker_config = wc["config"]
+        # [workers.config] 单一事实来源：按配置优先来源模式对齐（cache 模式取缓存值、
+        # toml 模式以文件为准并覆盖缓存），必须在注册 starter/创建副本之前完成——
+        # 插件实例（on_init 重建客户端等）用的就是这份配置。
+        worker_config = manager.align_inline_config(
+            worker_type, wc["config"], wc.get("config_section")
+        )
 
         if not module_name or not factory_name:
             logger.warning(f"Worker 配置不完整: {wc}")

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

@@ -14,6 +14,11 @@ from .logger import get_logger
 
 logger = get_logger("worker_manager")
 
+# ── 插件运行时配置（chongming-cache 单一事实来源）──
+# 键与 worker SDK（chongming_worker.runtime._configs_cache_key）一致，全局唯一一份
+# [workers.config]；读写规则见 WorkerManager.align_inline_config。
+_CONFIGS_KEY = "worker:cm:configs:{type}"
+
 # ── 全局副本数（chongming-cache 单一事实来源）──
 # 多网关共用同一 cache（单实例=本地 diskcache，多实例=共享卷/redis），
 # 「count」语义 = 该类 worker 的**全局目标总量**，各网关按在线数量分摊。
@@ -107,6 +112,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
         # 当前网关心跳标识：优先环境变量（多实例需唯一，尤其同宿主多网关须显式指定），
@@ -134,6 +148,51 @@ class WorkerManager:
         """
         return "cache" if str(self._launch_worker_count).lower() == "cache" else "toml"
 
+    def align_inline_config(
+        self,
+        worker_type: str,
+        worker_config: Dict[str, Any],
+        config_section: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """inline 插件初始化前，按配置优先来源模式对齐其 [workers.config]（网关侧兜底）。
+
+        worker SDK 的 Plugin.create_workers 会做同样的事（权威实现，见
+        chongming_worker.runtime.resolve_startup_config）；网关侧再兜一层，是为了让
+        网关（随本仓库发布）与 worker 包版本解耦：worker 包仍是旧版时，旧版 on_init
+        只拿到 config/workers/<type>/config.toml 展平出来的配置，会出现“前端显示缓存值、
+        实际跑 toml”。规则与 SDK 一致：
+
+          - "cache"：缓存里有该段就用它覆盖传入配置（toml 仅作缓存缺失时的引导并写回，
+            之后以缓存为准、跨重启保持）；
+          - "toml" ：以 config.toml 为准，把该段覆盖写入缓存（前端读缓存 → 随之显示文件值）；
+          - 未配置缓存后端（cache 为 None）时保持传入配置不动。
+
+        Args:
+            worker_type: 插件类型（如 aibox_sophon_adapter）
+            worker_config: _discover_plugins 展平出来的运行时配置
+            config_section: 该插件 config.toml 的 [workers.config] 原始段（引导/覆盖缓存用）
+        """
+        if self.cache is None:
+            return worker_config
+        key = _CONFIGS_KEY.format(type=worker_type)
+        section = config_section if isinstance(config_section, dict) else None
+        if self.config_mode == "cache":
+            cached = self.cache.get(key)
+            if isinstance(cached, dict):
+                logger.info(
+                    "%s [workers.config] 取自缓存（cache 模式，%d 键）",
+                    worker_type, len(cached),
+                )
+                return {**worker_config, **cached}
+            if section:
+                self.cache[key] = dict(section)   # 缓存缺失：用 toml 引导并写回
+                logger.info("%s [workers.config] 缓存缺失，已用 config.toml 引导", worker_type)
+            return worker_config
+        if section is not None:
+            self.cache[key] = dict(section)       # 以文件为准：覆盖缓存，前端随之显示文件值
+            logger.info("%s [workers.config] 以 config.toml 为准并覆盖缓存（toml 模式）", worker_type)
+        return worker_config
+
     # ── 全局副本数协调（chongming-cache 单一事实来源）────────────
     # count/quota 的读写与分摊已收归 worker 层（chongming_worker.runtime）：
     # 网关不再直写这些缓存键，仅维护「在线网关列表」（register/heartbeat/unregister）。
@@ -356,23 +415,112 @@ class WorkerManager:
                 logger.error(f"刷新 worker 列表失败: {e}")
             await asyncio.sleep(self._refresh_interval)
 
-    async def get_worker(self, worker_type: str) -> Optional[str]:
-        """获取一个可用的 worker 地址（简单随机负载均衡）
+    def _all_workers(self, worker_type: str) -> List[str]:
+        """该类型当前全部候选副本（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, [])
-        # 如果缓存为空，回退到本地注册表（适用于启动初期 refresh_workers 尚未执行时）
-        if not workers:
-            local = self._local_workers.get(worker_type, [])
-            if local:
-                return random.choice(local)
+        if instance_id:
+            return await self._pick_from_instance(worker_type, instance_id)
+        workers = self._all_workers(worker_type)
+        return random.choice(workers) if workers else None
+
+    async def _pick_from_instance(
+        self, worker_type: str, instance_id: str
+    ) -> 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
-        return random.choice(workers)
 
     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。
@@ -380,10 +528,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:
@@ -405,24 +560,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 = f"类型 '{worker_type}' 没有可用 Worker。"
+        if instance_id:
             msg = (
-            f"类型 '{worker_type}' 没有可用 Worker。"
+                f"类型 '{worker_type}' 的实例 '{instance_id}' 没有在线副本（定向调用不"
+                f"跨实例回退）。已识别实例: "
+                f"{sorted(self._instance_index.get(worker_type, {})) or '(无)'}。"
+                f"请检查: 1) 该实例 id 是否写错（取值见 GET "
+                f"/admin/workers/{worker_type}/instances 或副本启动时的 --instance）"
+                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 进程是否正常运行"
             )
-        logger.warning(msg)
-        raise RuntimeError(msg)
+        return msg
 
     async def call_stream(
         self,
@@ -431,6 +599,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 字节。
@@ -449,19 +618,21 @@ 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}"
         reply_iter = self.session.get(
@@ -552,6 +723,7 @@ class WorkerManager:
         timeout: float = 3.0,
         retries: int = 1,
         cacheable: bool = False,
+        instance_id: Optional[str] = None,
     ) -> Any:
         """
         统一调用入口（dict 语义，MQTT Bridge 等调用方使用）
@@ -563,6 +735,8 @@ class WorkerManager:
             timeout: 单次调用超时（秒）
             retries: 失败重试次数
             cacheable: 是否允许缓存结果（若为 False 则跳过缓存读写）
+            instance_id: 定向进程实例；给定时请求固定发往该实例（不参与随机
+                负载均衡，重试也只在该实例内换副本），None=默认随机均衡
 
         Returns:
             Dict[str, Any]: 调用结果（worker 回复解析后的 dict）
@@ -570,10 +744,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 {}
 
@@ -585,6 +767,7 @@ class WorkerManager:
         timeout: float = 3.0,
         retries: int = 1,
         cacheable: bool = False,
+        instance_id: Optional[str] = None,
     ) -> bytes:
         """
         统一调用入口（bytes 语义，HTTP 端点透传用）：返回 worker 回复的原始字节。
@@ -602,6 +785,7 @@ class WorkerManager:
             timeout: 单次调用超时（秒）
             retries: 失败重试次数
             cacheable: 是否允许缓存结果（缓存原始字节）
+            instance_id: 定向进程实例（见 call()）；None=默认随机均衡
 
         Returns:
             bytes: worker 回复的原始字节（空回复兜底 b"{}"，与 call() 的 {} 语义一致）
@@ -610,9 +794,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(
@@ -624,11 +811,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:
@@ -645,16 +834,39 @@ class WorkerManager:
                     logger.debug(f"命中共享缓存: {cache_key}")
                     return disk_val
 
-        # 2. 获取可用 worker
-        worker_expr = await self.get_worker(worker_type)
+        # 2. 获取可用 worker（instance_id 给定时固定到该实例，否则随机均衡）
+        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:
@@ -685,11 +897,8 @@ class WorkerManager:
                         f"Worker {worker_expr} 无回复，尝试重试 {attempt+1}/{retries+1}"
                     )
                     last_error = f"Worker {worker_expr} 无回复"
-                    # 换一个 worker 重试
-                    new_worker = await self._pick_other_worker(worker_type, tried_workers)
-                    if new_worker:
-                        worker_expr = new_worker
-                        full_key = f"{worker_expr}/{subject}"
+                    if not await _switch_worker("无回复"):
+                        break
                     continue
 
                 reply = first_reply
@@ -710,11 +919,8 @@ class WorkerManager:
                     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 new_worker:
-                        logger.info(f"错误后切换 Worker: {worker_expr} -> {new_worker}")
-                        worker_expr = new_worker
-                        full_key = f"{worker_expr}/{subject}"
+                    if not await _switch_worker("错误后"):
+                        break
                     continue  # 重试
             except asyncio.TimeoutError:
                 logger.warning(
@@ -722,11 +928,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 new_worker:
-                    logger.info(f"超时后切换 Worker: {worker_expr} -> {new_worker}")
-                    worker_expr = new_worker
-                    full_key = f"{worker_expr}/{subject}"
+                if not await _switch_worker("超时后"):
+                    break
                 continue
             except Exception as e:
                 logger.error(f"调用 worker 异常: {e}")
@@ -734,7 +937,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_config_align.py

@@ -0,0 +1,163 @@
+"""
+inline 插件 [workers.config] 对齐（网关侧兜底）单测 —— WorkerManager.align_inline_config。
+
+网关侧在创建 inline 副本之前，按「配置优先来源模式」（根 config.toml 的
+[cache].launch_worker_count）对齐插件配置：cache 模式取共享缓存里的段，
+toml 模式以 config/workers/<type>/config.toml 为准并覆盖缓存。目的是让网关与
+worker 包版本解耦——worker 包仍是旧版时，旧版 on_init 只拿到 toml 展平配置，
+会出现“前端显示缓存值、实际跑 toml”。
+
+用 dict 作缓存桩，验证两种模式下插件拿到的配置与缓存/文件两侧的取值一致。
+"""
+import os
+import sys
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, os.path.join(REPO, "src"))
+os.chdir(REPO)  # 网关模块导入期读根 config.toml / 日志配置
+
+from chongming.app.core.worker_manager import _CONFIGS_KEY, WorkerManager  # noqa: E402
+
+PASS = 0
+FAIL = 0
+WT = "aibox_sophon_adapter"
+
+# _discover_plugins 展平出来的形态（toml 值）
+FLAT = {"type": WT, "count": 1, "data_dir": "./data", "mediamtx_addr": "rtsp://toml:8554"}
+# 该插件 config.toml 的 [workers.config] 原始段
+SECTION = {"data_dir": "./data", "mediamtx_addr": "rtsp://toml:8554", "heartbeat": {"timeout": 15}}
+# 前端下发后缓存里的段（改过 mediamtx_addr 与 heartbeat）
+CACHED = {"data_dir": "./data", "mediamtx_addr": "rtsp://cache:9999", "heartbeat": {"timeout": 30}}
+
+
+def report(name, cond, detail=""):
+    global PASS, FAIL
+    if cond:
+        PASS += 1
+        print(f"✅ {name}" + (f" — {detail}" if detail else ""))
+    else:
+        FAIL += 1
+        print(f"❌ {name}" + (f" — {detail}" if detail else ""))
+
+
+def manager(cache, mode):
+    m = WorkerManager.__new__(WorkerManager)   # 免会话构造：仅用到 cache + config_mode
+    m.cache = cache
+    m._launch_worker_count = mode
+    return m
+
+
+def run() -> int:
+    key = _CONFIGS_KEY.format(type=WT)
+
+    # ── 1. cache 模式：缓存有段 → 插件拿到缓存值（本次修复目标）──────────
+    cache = {key: dict(CACHED)}
+    got = manager(cache, "cache").align_inline_config(WT, dict(FLAT), dict(SECTION))
+    report(
+        "cache 模式：插件配置取缓存值（覆盖 toml 同名键）",
+        got.get("mediamtx_addr") == "rtsp://cache:9999"
+        and got.get("heartbeat") == {"timeout": 30},
+        f"mediamtx_addr={got.get('mediamtx_addr')}",
+    )
+    report("cache 模式：不覆盖已存在的缓存段", cache[key] == CACHED, f"cache={cache[key]}")
+
+    # ── 2. cache 模式：缓存缺失 → 用 toml 段引导并写回 ──────────────────
+    cache2 = {}
+    got2 = manager(cache2, "cache").align_inline_config(WT, dict(FLAT), dict(SECTION))
+    report(
+        "cache 模式：缓存缺失用 config.toml 引导并写回",
+        cache2.get(key) == SECTION and got2.get("mediamtx_addr") == "rtsp://toml:8554",
+        f"cache={cache2.get(key)}",
+    )
+
+    # ── 3. toml 模式：以文件为准并覆盖缓存（前端随之显示文件值）─────────
+    cache3 = {key: dict(CACHED)}
+    got3 = manager(cache3, "toml").align_inline_config(WT, dict(FLAT), dict(SECTION))
+    report(
+        "toml 模式：插件配置为 toml 值",
+        got3.get("mediamtx_addr") == "rtsp://toml:8554",
+        f"mediamtx_addr={got3.get('mediamtx_addr')}",
+    )
+    report(
+        "toml 模式：启动时用 toml 覆盖缓存（前端与运行值一致）",
+        cache3.get(key) == SECTION,
+        f"cache={cache3.get(key)}",
+    )
+
+    # ── 4. 边界：无缓存后端 / 无段（旧 config/workers/config.toml 兼容路径）──
+    cache4 = {key: dict(CACHED)}
+    m4 = manager(cache4, "cache")
+    m4.cache = None
+    report(
+        "无缓存后端：原样返回（经典 toml 行为）",
+        m4.align_inline_config(WT, dict(FLAT), dict(SECTION)) == FLAT
+        and cache4[key] == CACHED,
+    )
+    cache5 = {}
+    got5 = manager(cache5, "toml").align_inline_config(WT, dict(FLAT), None)
+    report("无 [workers.config] 段（旧写法）：不写缓存、原样返回", got5 == FLAT and not cache5)
+
+    # ── 5. 与 SDK 侧一致：缓存键必须相同；SDK 为本地版本时才比对解析结果 ──
+    # 注意：venv 里的 chongming-worker 可能是索引里的发布版（uv sync 会把可编辑安装
+    # 换回锁文件版本），新版才有 resolve_startup_config / 正确的 toml 分支语义。
+    sys.path.insert(0, os.path.join(REPO, "utils", "worker", "src"))
+    import chongming_worker.runtime as r  # noqa: E402
+
+    report(
+        "缓存键与 SDK 一致（两端必须读写同一键）",
+        r._configs_cache_key(WT) == _CONFIGS_KEY.format(type=WT),
+        f"sdk={r._configs_cache_key(WT)}",
+    )
+    local_src = os.path.join(REPO, "utils", "worker", "src") in os.path.abspath(r.__file__)
+    if not local_src:
+        print(f"⚠️  跳过解析结果比对：当前 SDK 为 {r.__file__}（非本地源码，缺本次改动）")
+    else:
+        c = {r._configs_cache_key(WT): dict(CACHED)}
+        sdk = r.resolve_runtime_config(WT, dict(FLAT), c, "cache")
+        report(
+            "cache 模式：网关侧兜底与 SDK 解析结果一致",
+            sdk.get("mediamtx_addr") == CACHED["mediamtx_addr"],
+            f"sdk={sdk.get('mediamtx_addr')}",
+        )
+
+    # ── 6. 端到端：网关对齐 + 插件初始化（部署组合：网关新、worker 包为发布版）──
+    # 该断言即线上现象的复现与验证：插件 on_init 拿到的必须是网关对齐后的缓存值，
+    # 与 venv 里 chongming-worker 是否为含本次改动的版本无关。
+    import asyncio
+
+    from chongming_worker.plugin import Plugin  # noqa: E402
+
+    seen = {}
+
+    class DemoPlugin(Plugin):
+        type = WT
+        auto_bind = False
+
+        async def _bootstrap(self):      # 隔离数据库/对象存储引导，只看配置来源
+            return None
+
+        async def on_init(self):
+            seen["on_init"] = self.config.get("mediamtx_addr")
+
+        def _create_workers_impl(self, session, config, count=1, worker_manager=None):
+            seen["impl"] = (config or {}).get("mediamtx_addr")
+            return []
+
+    aligned = manager({key: dict(CACHED)}, "cache").align_inline_config(
+        WT, dict(FLAT), dict(SECTION)
+    )
+    asyncio.run(DemoPlugin().create_workers(None, aligned, 1))
+    report(
+        "端到端：网关对齐后插件 on_init 读到缓存值（不依赖 worker 包版本）",
+        seen.get("on_init") == CACHED["mediamtx_addr"] == seen.get("impl"),
+        f"on_init={seen.get('on_init')}, impl={seen.get('impl')}, "
+        f"sdk={os.path.abspath(__import__('chongming_worker').__file__)}",
+    )
+
+    print("\n" + "=" * 56)
+    print(f"测试结果: ✅ {PASS} 通过  ❌ {FAIL} 失败")
+    return 0 if FAIL == 0 else 1
+
+
+if __name__ == "__main__":
+    sys.exit(run())

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 触发重试（2 次查询）", session.get_calls == 2, str(session.get_calls))
+        check("reply.err 抛 RuntimeError 且保留具体错误", "boom" in str(e), str(e))
+    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))

uv.lock

@@ -560,7 +560,7 @@ wheels = [
 
 [[package]]
 name = "chongming-worker"
-version = "0.4.4"
+version = "0.4.5"
 source = { registry = "http://124.71.68.6:3000/api/packages/chongming/pypi/simple/" }
 dependencies = [
     { name = "chongming-cache" },
@@ -571,9 +571,9 @@ dependencies = [
     { name = "pydantic" },
     { name = "tomli" },
 ]
-sdist = { url = "http://124.71.68.6:3000/api/packages/chongming/pypi/files/chongming-worker/0.4.4/chongming_worker-0.4.4.tar.gz", hash = "sha256:0a129bd4034f6031911839792e3c83cfc78bbb706dccadde8f41feb754fe83c8" }
+sdist = { url = "http://124.71.68.6:3000/api/packages/chongming/pypi/files/chongming-worker/0.4.5/chongming_worker-0.4.5.tar.gz", hash = "sha256:fec2b7ed7634838f67368da56805abf6918232535e85535ddfd72cd099b15fab" }
 wheels = [
-    { url = "http://124.71.68.6:3000/api/packages/chongming/pypi/files/chongming-worker/0.4.4/chongming_worker-0.4.4-py3-none-any.whl", hash = "sha256:170c0e59b4fc3565d437db3c805eda343de5b324a0b574fdfdda65e3348145b4" },
+    { url = "http://124.71.68.6:3000/api/packages/chongming/pypi/files/chongming-worker/0.4.5/chongming_worker-0.4.5-py3-none-any.whl", hash = "sha256:88a343387bf354db1ef051384c06b268dfd996a3fb25366537fa32547e58fc93" },
 ]
 
 [[package]]
@@ -940,11 +940,11 @@ wheels = [
 
 [[package]]
 name = "idna"
-version = "3.19"
+version = "3.20"
 source = { registry = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple" }
-sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/5f/f7/abb373e5757eaec4b922b92f97ec8d6d7e057cf06778247604fbc4e7c3f3/idna-3.19.tar.gz", hash = "sha256:5e0811a4383b21dc5838069f801c4fb62113b7447663d2530d2bd6e77b49bf15", size = 215237, upload-time = "2026-08-18T05:14:24.27Z" }
+sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/f5/08/8eea9d4b8302028f3abb2c0813953f7aec26d33b7a8960ed760e65ff29fa/idna-3.20.tar.gz", hash = "sha256:a7db850025b95ded1eae8a46181a1a6c56c92c96f0e2b005d9ff8dc0210cab44", size = 216463, upload-time = "2026-09-17T14:11:04.752Z" }
 wheels = [
-    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/57/b0/0e52c878c53f245edd3a11020f20979b3f490f245af532c7cae3027754b5/idna-3.19-py3-none-any.whl", hash = "sha256:815e7be7a7806d54abb586dc943addc79e8b2ee16915059658cbeff4b1b43bf4", size = 68550, upload-time = "2026-08-18T05:14:22.343Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/58/a2/bb081bab032533a855d44de1d56f8e8426114ff1ba5d1f07a438a0a654f8/idna-3.20-py3-none-any.whl", hash = "sha256:ab7ae7122974553370f0bdb919e1a960b2cd1bc1ef0276416d896db81c14582c", size = 69583, upload-time = "2026-09-17T14:11:03.168Z" },
 ]
 
 [[package]]
@@ -1137,7 +1137,7 @@ sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/c7/ce/38
 
 [[package]]
 name = "openai"
-version = "3.14.1"
+version = "3.15.0"
 source = { registry = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple" }
 dependencies = [
     { name = "anyio" },
@@ -1147,9 +1147,9 @@ dependencies = [
     { name = "sniffio" },
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/a2/a5/bc3681f645a0d026332047cc19a703315979740a11a5763f46c163abd614/openai-3.14.1.tar.gz", hash = "sha256:2f3423a71af767d9071b059cf6df1e97a721c3a8d19417375ad7993d4df0d346", size = 1634770, upload-time = "2026-09-15T23:13:35.708Z" }
+sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/82/6b/f086332d9fdefcc3853f2457209fcf221985d01d32b5021ab8096be7d506/openai-3.15.0.tar.gz", hash = "sha256:47bc15f8b9db62ba9ff7fa32164c72f4fd6333b22d890251292aac1c6bfe998f", size = 1664554, upload-time = "2026-09-18T00:53:05.192Z" }
 wheels = [
-    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/f7/13/d1caa2b2385ed22a1e10e1fd7fc09a92e282cffadad49c569827315f468a/openai-3.14.1-py3-none-any.whl", hash = "sha256:a6758739282f3b1cafe8df1204f7a2f7b1e9edbbcfa41c0efba687448fbc51cf", size = 2003545, upload-time = "2026-09-15T23:13:33.164Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/fb/92/fc65194e6e92cf0202a2e5fff75e2fef62e7dbb7a6cf2b48c1239dfedb50/openai-3.15.0-py3-none-any.whl", hash = "sha256:95d6b2caf3823c7e0fc08911317b7f31b9f9ccf93b414e7a5c8d5b226af115f6", size = 2018693, upload-time = "2026-09-18T00:53:02.507Z" },
 ]
 
 [[package]]
@@ -1931,7 +1931,7 @@ wheels = [
 
 [[package]]
 name = "volcengine-python-sdk"
-version = "5.0.49"
+version = "5.0.50"
 source = { registry = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple" }
 dependencies = [
     { name = "certifi" },
@@ -1939,9 +1939,9 @@ dependencies = [
     { name = "six" },
     { name = "urllib3" },
 ]
-sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/24/b9/813003899553d3e68fa09841a83c9753eed82298758c7f56abf7595c7d36/volcengine_python_sdk-5.0.49.tar.gz", hash = "sha256:4eb95c77bd4af3f2bfa8b569eb3280ed9b02b40dea371ba4d7f40269fed19cd6", size = 10805148, upload-time = "2026-09-10T12:02:25.301Z" }
+sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/aa/76/4fe32f3c467a1e79263abf4516a8181810567ef0d019d6a7783a2733f4a5/volcengine_python_sdk-5.0.50.tar.gz", hash = "sha256:a9e5da2c63b5982b46c233b261a39b5f8fb456cd8d7b321b303d3ad40577d6bf", size = 10779492, upload-time = "2026-09-17T11:24:27.378Z" }
 wheels = [
-    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/14/da/ca7b207d19a9e07b11afa2f6af9c5ee624345fea6da4f81c6b8adf295b8a/volcengine_python_sdk-5.0.49-py2.py3-none-any.whl", hash = "sha256:cd4322b48fcca048eaf6417efc58a1eeef36486c7f32fee6975291aa09167cca", size = 43378306, upload-time = "2026-09-10T12:02:19.274Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/ab/85/75009d3c824f2c122e987b46491c6e033b3e88b264aa4c650d6f334a0e83/volcengine_python_sdk-5.0.50-py2.py3-none-any.whl", hash = "sha256:89858a7cbb7fd0d9e74fdcdd0f6584bbfc4e8f92fdc4ed4bee5af7e24a36eda7", size = 43243933, upload-time = "2026-09-17T11:24:17.135Z" },
 ]
 
 [package.optional-dependencies]