modify: 微服务演进

.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>` 按网关隔离：

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/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

@@ -112,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
         # 当前网关心跳标识：优先环境变量（多实例需唯一，尤其同宿主多网关须显式指定），
@@ -406,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]:
+    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(
+        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。
@@ -430,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:
@@ -455,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)
+        return msg
-        raise RuntimeError(msg)
 
     async def call_stream(
         self,
@@ -481,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 字节。
@@ -499,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(
@@ -602,6 +723,7 @@ class WorkerManager:
         timeout: float = 3.0,
         retries: int = 1,
         cacheable: bool = False,
+        instance_id: Optional[str] = None,
     ) -> Any:
         """
         统一调用入口（dict 语义，MQTT Bridge 等调用方使用）
@@ -613,6 +735,8 @@ class WorkerManager:
             timeout: 单次调用超时（秒）
             retries: 失败重试次数
             cacheable: 是否允许缓存结果（若为 False 则跳过缓存读写）
+            instance_id: 定向进程实例；给定时请求固定发往该实例（不参与随机
+                负载均衡，重试也只在该实例内换副本），None=默认随机均衡
 
         Returns:
             Dict[str, Any]: 调用结果（worker 回复解析后的 dict）
@@ -620,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 {}
 
@@ -635,6 +767,7 @@ class WorkerManager:
         timeout: float = 3.0,
         retries: int = 1,
         cacheable: bool = False,
+        instance_id: Optional[str] = None,
     ) -> bytes:
         """
         统一调用入口（bytes 语义，HTTP 端点透传用）：返回 worker 回复的原始字节。
@@ -652,6 +785,7 @@ class WorkerManager:
             timeout: 单次调用超时（秒）
             retries: 失败重试次数
             cacheable: 是否允许缓存结果（缓存原始字节）
+            instance_id: 定向进程实例（见 call()）；None=默认随机均衡
 
         Returns:
             bytes: worker 回复的原始字节（空回复兜底 b"{}"，与 call() 的 {} 语义一致）
@@ -660,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(
@@ -674,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:
@@ -695,16 +834,39 @@ 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:
@@ -735,11 +897,8 @@ class WorkerManager:
                         f"Worker {worker_expr} 无回复，尝试重试 {attempt+1}/{retries+1}"
                     )
                     last_error = f"Worker {worker_expr} 无回复"
-                    # 换一个 worker 重试
+                    if not await _switch_worker("无回复"):
-                    new_worker = await self._pick_other_worker(worker_type, tried_workers)
+                        break
-                    if new_worker:
-                        worker_expr = new_worker
-                        full_key = f"{worker_expr}/{subject}"
                     continue
 
                 reply = first_reply
@@ -760,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 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 +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 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 +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_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))