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

config.toml

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

deploy/chongming.service

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

public/config.toml

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

public/main.py

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

pyproject.toml

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

src/chongming/__init__.py

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

src/chongming/app/__init__.py

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

src/chongming/app/core/auth.py

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

src/chongming/app/core/worker_manager.py

@@ -88,7 +88,40 @@ class _MemLRU:
             self._d.popitem(last=False)
 
 
+# ── 进程内直调（inline 快路径）─────────────────────────────────────────────
+# inline 副本由本进程托管（与网关同事件循环），业务请求却仍要绕
+# "zenoh 查询 → router → queryable 回调 → 跨线程回本 loop"。分派逻辑已统一到
+# chongming_worker.dispatch，故此处可直接 await 同一个 dispatcher，省掉整圈往返。
+# 语义不变：信封/错误通道/超时/重试/换副本/实例定向/in-flight 与 drain 全部沿用原逻辑；
+# 非本地副本（external 插件、w>1 时非托管进程）与旧版 worker 包自动回退 zenoh。
+_RPC_LOCAL = os.environ.get("CHONGMING_RPC_LOCAL", "1") != "0"
+try:  # 旧版 chongming-worker（<0.5）无 dispatch 模块：保持纯 zenoh 行为
+    from chongming_worker.dispatch import get_dispatcher as _get_local_dispatcher
+except Exception:  # noqa: BLE001
+    _get_local_dispatcher = None
+
+
+def _local_dispatcher_for(key_expr: str) -> Optional[Any]:
+    """'worker/<type>/<id>' → 本进程托管该副本的 dispatcher；非本地/不可用返回 None"""
+    if not _RPC_LOCAL or _get_local_dispatcher is None:
+        return None
+    parts = key_expr.split("/")
+    if len(parts) != 3 or parts[0] != "worker":
+        return None
+    return _get_local_dispatcher(parts[1], parts[2])
+
+
+def _err_text(payload: Any) -> str:
+    """把错误载荷归一为文本（兼容 zenoh ZBytes 与直调的 bytes）"""
+    if payload is None:
+        return "unknown error"
+    if isinstance(payload, (bytes, bytearray)):
+        return bytes(payload).decode("utf-8", "replace") or "unknown error"
+    return str(payload) or "unknown error"
+
+
 class WorkerManager:
+
     """
     统一 Worker 管理器
 
@@ -635,6 +668,36 @@ class WorkerManager:
             raise RuntimeError(await self._no_worker_message(worker_type, instance_id))
 
         full_key = f"{worker_expr}/{subject}"
+
+        # 进程内直调（inline 快路径）：与 zenoh 路径同语义（逐块产出、错误块终止、无重试）
+        local = _local_dispatcher_for(worker_expr)
+        if local is not None and local.has_subject(subject):
+            async def _local_stream():
+                agen = local.invoke_stream(subject, orjson.dumps(payload))
+                try:
+                    while True:
+                        try:
+                            data = await asyncio.wait_for(agen.__anext__(), timeout=idle_timeout)
+                        except StopAsyncIteration:
+                            break
+                        except asyncio.TimeoutError:
+                            raise StreamIdleTimeoutError(
+                                f"Worker {worker_expr} 流式空闲超时 ({idle_timeout}s 无数据)"
+                            ) from None
+                        if not data:
+                            continue
+                        decoded = orjson.loads(data)
+                        if isinstance(decoded, dict) and decoded.get("error"):
+                            # Worker 错误收尾块：转发后终止流
+                            yield data
+                            return
+                        yield data
+                finally:
+                    # 消费方提前退出（SSE 断连/网关放弃）：停流并取消在途派发
+                    await agen.aclose()
+
+            return _local_stream()
+
         reply_iter = self.session.get(
             full_key,
             payload=orjson.dumps(payload),
@@ -873,37 +936,53 @@ class WorkerManager:
                 tried_workers.append(worker_expr)
 
             try:
-                reply_iter = self.session.get(
+                local = _local_dispatcher_for(worker_expr)
-                    full_key,
+                if local is not None and local.has_subject(subject):
-                    payload=payload_bytes,
+                    # 进程内直调（inline 快路径）：省 zenoh 往返，语义与下面 zenoh 路径一致
-                    consolidation=zenoh.ConsolidationMode.NONE,
+                    logger.debug(f"[local-rpc] {worker_expr}/{subject} 进程内直调")
-                    timeout=timeout,
+                    local_res = await asyncio.wait_for(
-                )
+                        local.invoke_unary(subject, payload_bytes), timeout=timeout
-                # 使用辅助函数获取第一个回复，避免 StopIteration 在 asyncio.to_thread 中传播的问题
-                def _get_first(iterator):
-                    """线程安全地获取迭代器的第一个元素"""
-                    for item in iterator:
-                        return item
-                    return _SENTINEL
-
-                # 走专用 _ZENOH_EXECUTOR 池（隔离于全局默认池，避免高并发下排队）
-                loop = asyncio.get_running_loop()
-                first_reply = await asyncio.wait_for(
-                    loop.run_in_executor(_ZENOH_EXECUTOR, _get_first, reply_iter),
-                    timeout=timeout,
-                )
-                if first_reply is _SENTINEL:
-                    logger.warning(
-                        f"Worker {worker_expr} 无回复，尝试重试 {attempt+1}/{retries+1}"
                     )
-                    last_error = f"Worker {worker_expr} 无回复"
+                    reply_ok, reply_payload = local_res.ok, local_res.payload
-                    if not await _switch_worker("无回复"):
+                else:
-                        break
+                    reply_iter = self.session.get(
-                    continue
+                        full_key,
+                        payload=payload_bytes,
+                        consolidation=zenoh.ConsolidationMode.NONE,
+                        timeout=timeout,
+                    )
+                    # 使用辅助函数获取第一个回复，避免 StopIteration 在 asyncio.to_thread 中传播的问题
+                    def _get_first(iterator):
+                        """线程安全地获取迭代器的第一个元素"""
+                        for item in iterator:
+                            return item
+                        return _SENTINEL
 
-                reply = first_reply
+                    # 走专用 _ZENOH_EXECUTOR 池（隔离于全局默认池，避免高并发下排队）
-                if reply.ok: # type: ignore
+                    loop = asyncio.get_running_loop()
-                    reply_bytes = bytes(reply.ok.payload) if reply.ok.payload else b"{}"
+                    first_reply = await asyncio.wait_for(
+                        loop.run_in_executor(_ZENOH_EXECUTOR, _get_first, reply_iter),
+                        timeout=timeout,
+                    )
+                    if first_reply is _SENTINEL:
+                        logger.warning(
+                            f"Worker {worker_expr} 无回复，尝试重试 {attempt+1}/{retries+1}"
+                        )
+                        last_error = f"Worker {worker_expr} 无回复"
+                        if not await _switch_worker("无回复"):
+                            break
+                        continue
+
+                    reply = first_reply
+                    if reply.ok:  # type: ignore
+                        reply_ok = True
+                        reply_payload = bytes(reply.ok.payload) if reply.ok.payload else b"{}"
+                    else:
+                        reply_ok = False
+                        reply_payload = reply.err.payload  # type: ignore
+
+                if reply_ok:
+                    reply_bytes = bytes(reply_payload) if reply_payload else b"{}"
                     # 缓存写入（仅在 cacheable=True 时）：解析一次做成功检查，
                     # 但缓存与返回均保留原始字节（网关侧零二次序列化）。
                     # 先写内存一级，再异步写共享二级（executor，不阻塞响应返回）
@@ -914,8 +993,7 @@ class WorkerManager:
                             await self._cache_set(cache_key, reply_bytes)
                     return reply_bytes
                 else:
-                    error_payload = reply.err.payload  # type: ignore
+                    error_msg = _err_text(reply_payload)
-                    error_msg = str(error_payload) if error_payload else "unknown error"
                     logger.warning(f"Worker 返回错误: {error_msg} (尝试 {attempt+1}/{retries+1})")
                     last_error = f"Worker {worker_expr} 返回错误: {error_msg}"
                     # 错误回复（含 Zenoh 超时）也换一个 worker 重试

test/test_inline_local_dispatch.py

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