modify: 性能优化

config.toml

@@ -42,6 +42,9 @@ reload = true
 file = "logs"
 format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
 level = "DEBUG"
+# access 日志采样率：0.1 = 只记录 10% 的成功请求（错误 >=400 始终全量）
+# 1.0 = 全量记录；配合异步日志队列降低高吞吐下主线程开销
+access_sample_rate = 0.1
 
 [development.cors]
 allow_credentials = true

pyproject.toml

@@ -19,16 +19,17 @@ dependencies = [
   "tomli",
   "typing-extensions",
   "uvicorn",
-  "volcengine-python-sdk[ark]>=5.0.41",
+  "volcengine-python-sdk[ark]",
   "chongming-config",
   "chongming-password",
   "chongming-worker",
-  "amqtt>=0.11.3",
-  "aiofiles>=25.1.0",
+  "amqtt",
+  "aiofiles",
   "bytedance",
   "chongming_stream_manager",
-  "uvloop>=0.22.1",
+  "uvloop",
   "chongming-mqtt",
+  "httptools",
 ]
 description = "基于 modules-bank 的 fastapi 打包脚手架"
 name = "chongming-zenoh"

src/chongming/app/__init__.py

@@ -9,18 +9,20 @@ chongming FastAPI 应用入口
 
 import time
 import asyncio
+import itertools
 from contextlib import asynccontextmanager
 
 from fastapi import FastAPI, Request
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
+from fastapi.middleware.gzip import GZipMiddleware
 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 .core import lifespan_setup
-from .routes import storage_router
+from .core import lifespan_setup, call_stats
+from .routes import storage_router, stats_router
 
 
 config = get_config()
@@ -112,6 +114,9 @@ async def lifespan(app: FastAPI):
     if worker_tasks:
         await asyncio.gather(*worker_tasks, return_exceptions=True)
 
+    # 4.5 取消后台统计/审计任务
+    await call_stats.shutdown()
+
     # 5. 关闭 Zenoh 会话
     session.close()
     logger.info("Zenoh 会话已关闭")
@@ -131,6 +136,24 @@ app = FastAPI(
 )
 
 app.add_middleware(CORSMiddleware, **cors)
+# GZip 压缩响应（<500B 不压缩；starlette 默认排除 text/event-stream，
+# 保证 SSE 流式端点即时性；>=128KB 的响应在独立线程压缩，不阻塞事件循环）
+app.add_middleware(GZipMiddleware)
+
+
+# access 日志采样率（logging.access_sample_rate，0.1 = 只记录 10% 的成功请求；
+# 错误 >=400 始终全量记录，保证可追溯；1.0/缺省 = 全量）
+ACCESS_SAMPLE_RATE = float(config[env].get("logging", {}).get("access_sample_rate", 1.0))
+# 成功请求采样计数器（每进程独立；错误请求不消耗计数）
+_access_log_counter = itertools.count()
+
+
+def _should_log_access(status_code: int, sample_rate: float = ACCESS_SAMPLE_RATE) -> bool:
+    """错误全量；成功按采样率（采样为确定性计数，无随机数开销）"""
+    if status_code >= 400 or sample_rate >= 1.0:
+        return True
+    denom = max(1, int(round(1.0 / sample_rate)))
+    return next(_access_log_counter) % denom == 0
 
 
 @app.middleware("http")
@@ -144,6 +167,11 @@ async def access_log_middleware(request: Request, call_next):
     - 请求处理时长（毫秒）
     - 客户端IP地址
 
+    高吞吐优化：
+    - 采样：成功请求按 access_sample_rate 记录，错误（>=400）全量
+    - 异步：% 参数化消息 + 队列日志（QueueHandler），格式化与文件 I/O
+      在后台监听线程执行，主线程仅入队
+
     参数:
         request: FastAPI请求对象
         call_next: 下一个中间件或路由处理函数
@@ -156,9 +184,10 @@ async def access_log_middleware(request: Request, call_next):
     duration = (time.perf_counter() - start) * 1000
     client = request.client.host if request.client else "-"
 
+    if _should_log_access(response.status_code):
         logger.info(
-        f"{request.method} {request.url.path} status={response.status_code} "
-        f"duration={duration:.2f}ms client={client}",
+            "%s %s status=%s duration=%.2fms client=%s",
+            request.method, request.url.path, response.status_code, duration, client,
             extra={"status_code": response.status_code},
         )
     return response
@@ -240,3 +269,4 @@ if vfs_db_path := config[env].get("file_system", {}).get("path", None):
     app.mount("/static", static_files_handler, name="static")
 
 app.include_router(storage_router)
+app.include_router(stats_router)

src/chongming/app/core/auth.py

@@ -1,20 +1,64 @@
+import hashlib
+import time
+
 from fastapi import Depends, HTTPException, status, Request
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from chongming_jwt import JWT
+from chongming_cache import Cache
 from chongming_config import get_config
+import orjson
 
 
 security_scheme = HTTPBearer(auto_error=False)
 
+# JWT 解析共享缓存：与 WorkerManager 共用 [cache] 配置段。
+# 默认 diskcache（本地）；横向拓展时 config.toml 切 backend = "redis"，
+# 多网关实例共享同一 Redis 即可全局只解密一次，代码零改动。
+# 注意：缓存存的是"已验证结果"，信任模型与 worker 回复缓存一致
+# （缓存基础设施可信；Redis 请配置密码并置于内网）。
+_jwt_cache = Cache.from_config(get_config()["cache"])
+# 条目 TTL（秒）：控制存储占用与密钥轮换后的自然清理；
+# 安全性由命中时的 exp 复核保证，TTL 不承担过期判断
+_JWT_CACHE_TTL = 60
+
+
+def _cache_key(token: str, secret: str, algorithm: str) -> str:
+    """缓存 key：token 内容 + 密钥/算法指纹，不同密钥/算法不串用"""
+    return "jwt:" + hashlib.sha256(f"{secret}:{algorithm}:{token}".encode()).hexdigest()
+
+
+def _cached_decode(token: str, secret: str, algorithm: str) -> tuple:
+    """JWT 解密结果缓存（(payload, exp)），共享缓存避免高频请求重复密码学运算。
+
+    - 命中且未过期 -> 直接返回，跳过 RSA/ECDSA 验证与 JSON 解析
+    - 命中但 exp 已过 -> 直接拒绝，不重解密（条目保留至 TTL 到期，
+      过期 token 的重复请求零密码学开销；缓存不延长 token 有效期）
+    - 未命中 -> 完整解码后写入共享缓存（TTL=60s）
+    - decode 失败（无效 token）不缓存，每次都完整验证后拒绝
+    """
+    key = _cache_key(token, secret, algorithm)
+    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():
+            return payload, exp
+        raise ValueError("Token 已过期")
+    jwt = JWT(secret, algorithm)
+    payload_str = jwt.decode(token)
+    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)
+    return payload, exp
+
 
 def jwt_auth_dependency(
     credentials: HTTPAuthorizationCredentials = Depends(security_scheme)
 ) -> dict:
     """JWT 校验依赖，返回解码后的 payload，校验失败抛出 401"""
-    config =get_config()
+    config = get_config()
     secret = config["security"]["secret_key"]
     algorithm = config["security"]["algorithm"]
-    jwt = JWT(secret, algorithm)
 
     if not credentials:
         raise HTTPException(
@@ -24,9 +68,8 @@ def jwt_auth_dependency(
         )
     token = credentials.credentials
     try:
-        payload_str = jwt.decode(token)
-        import json
-        payload = json.loads(payload_str)
+        payload, _ = _cached_decode(token, secret, algorithm)
+        # 缓存后端为 pickle 序列化，返回的是独立对象，调用方修改不影响缓存
         return payload
     except Exception as e:
         raise HTTPException(

src/chongming/app/core/call_stats.py

@@ -0,0 +1,150 @@
+"""
+调用统计与审计日志（后台任务执行）
+
+route_generator 端点在 call_raw 返回后仅做非关键后置操作，全部经
+spawn() 以 fire-and-forget 后台任务执行，端点立即返回，不增加接口
+响应延迟、不占用请求协程资源：
+  1. 更新内存调用统计（per worker_type/subject/status 计数 + 延迟聚合 + 最近调用缓冲）
+  2. 写审计日志（logs/audit-YYYY-MM-DD.log，复用 logger 按日切分 + 大小轮转设施）
+
+record_call 内部全兜底：统计或审计任一步失败仅记 warning，绝不影响响应。
+统计为进程内视角（gunicorn 多 worker 下各进程独立聚合）。
+
+后台任务随应用生命周期管理：lifespan 关闭阶段调用 shutdown() 统一取消。
+"""
+import asyncio
+import datetime
+import orjson
+import time
+from collections import deque
+from typing import Any, Coroutine, Dict
+
+from .logger import get_logger
+
+logger = get_logger("call_stats")
+# 审计日志独立文件（audit-YYYY-MM-DD.log），与业务日志分离便于检索
+audit_logger = get_logger("audit")
+
+# 最近调用环形缓冲上限
+MAX_RECENT = 1000
+
+# 后台任务集合（fire-and-forget，shutdown 时统一取消）
+_tasks: set[asyncio.Task] = set()
+
+
+def spawn(coro: Coroutine) -> asyncio.Task:
+    """以 fire-and-forget 方式创建后台任务并纳入跟踪（shutdown 时取消）"""
+    task = asyncio.create_task(coro)
+    _tasks.add(task)
+    task.add_done_callback(_tasks.discard)
+    return task
+
+
+async def shutdown() -> None:
+    """取消所有后台统计/审计任务（应用关闭时调用）"""
+    tasks = list(_tasks)
+    if not tasks:
+        return
+    for task in tasks:
+        task.cancel()
+    await asyncio.gather(*tasks, return_exceptions=True)
+
+
+class CallStats:
+    """内存调用统计：计数 + 延迟聚合 + 最近调用环形缓冲。
+
+    asyncio 单线程事件循环内访问，无需加锁。
+    """
+
+    def __init__(self, max_recent: int = MAX_RECENT):
+        self._counts: Dict[tuple, int] = {}
+        self._latency: Dict[str, Dict[str, float]] = {}
+        self._recent: deque = deque(maxlen=max_recent)
+
+    def record(self, entry: Dict[str, Any]) -> None:
+        key = (entry["worker_type"], entry["subject"], entry["status"])
+        self._counts[key] = self._counts.get(key, 0) + 1
+
+        wt = entry["worker_type"]
+        lat = self._latency.setdefault(wt, {"total": 0.0, "count": 0, "max": 0.0})
+        dur = entry.get("duration_ms", 0.0)
+        lat["total"] += dur
+        lat["count"] += 1
+        if dur > lat["max"]:
+            lat["max"] = dur
+
+        self._recent.append(entry)
+
+    def summary(self) -> Dict[str, Any]:
+        """聚合快照：总量、per worker/subject/status 计数、延迟、最近调用"""
+        calls = [
+            {"worker_type": wt, "subject": subj, "status": status, "count": count}
+            for (wt, subj, status), count in sorted(self._counts.items())
+        ]
+        latency = [
+            {
+                "worker_type": wt,
+                "count": lat["count"],
+                "avg_ms": round(lat["total"] / lat["count"], 2) if lat["count"] else 0.0,
+                "max_ms": round(lat["max"], 2),
+            }
+            for wt, lat in sorted(self._latency.items())
+        ]
+        return {
+            "total_calls": sum(self._counts.values()),
+            "calls": calls,
+            "latency": latency,
+            "recent": list(self._recent)[-50:],
+        }
+
+
+_stats = CallStats()
+
+
+def summary() -> Dict[str, Any]:
+    """全局统计快照（供 GET /stats 端点使用）"""
+    return _stats.summary()
+
+
+def _format_audit(entry: Dict[str, Any]) -> str:
+    """审计日志单行 key=value 格式，便于 grep 检索"""
+    ts = entry.get("ts") or time.time()
+    ts_str = datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
+    user = entry.get("user")
+    if isinstance(user, dict):
+        # JWT payload（auth.py store_user_to_request 注入），orjson 默认紧凑单行输出
+        user_str = orjson.dumps(user).decode()
+    elif isinstance(user, str):
+        user_str = user
+    else:
+        user_str = "-"
+    parts = [
+        "AUDIT",
+        f"ts={ts_str}",
+        f"worker={entry.get('worker_type', '-')}",
+        f"subject={entry.get('subject', '-')}",
+        f"method={entry.get('method', '-')}",
+        f"path={entry.get('path', '-')}",
+        f"user={user_str}",
+        f"client={entry.get('client') or '-'}",
+        f"status={entry.get('status', '-')}",
+        f"duration_ms={entry.get('duration_ms', 0):.3f}",
+        f"req_bytes={entry.get('req_bytes', 0)}",
+        f"resp_bytes={entry.get('resp_bytes', 0)}",
+    ]
+    return " ".join(parts)
+
+
+async def record_call(entry: Dict[str, Any]) -> None:
+    """后台执行体：更新内存统计 + 写审计日志。
+
+    整体兜底：任一步失败仅记 warning，绝不向端点冒泡。
+    """
+    try:
+        _stats.record(entry)
+    except Exception as e:  # pragma: no cover - 防御性兜底
+        logger.warning("调用统计记录失败: %s", e)
+    try:
+        audit_logger.info(_format_audit(entry))
+    except Exception as e:  # pragma: no cover - 防御性兜底
+        logger.warning("审计日志写入失败: %s", e)

src/chongming/app/core/logger.py

@@ -1,8 +1,12 @@
 import re
+import os
+import atexit
+import queue
 import logging
 import datetime
 from pathlib import Path
-from logging.handlers import RotatingFileHandler
+from typing import List, Optional
+from logging.handlers import RotatingFileHandler, QueueHandler, QueueListener
 
 from chongming_config import get_config
 
@@ -177,6 +181,103 @@ class ExcludeWatchFilesFilter(logging.Filter):
         return not record.name.startswith(self._filter_starts_with_tulpe)
 
 
+# ---- 异步日志队列（QueueHandler + QueueListener）----
+# 主线程/事件循环只做"入队"（LogRecord 创建 + put，µs 级），文件/控制台 I/O
+# 由后台监听线程执行，日志记录不再阻塞业务协程。
+# 队列满时丢弃新日志（日志系统故障不影响业务）；错误日志（>=400）不受采样影响。
+
+_LOG_QUEUE_SIZE = 10000
+
+
+class _AsyncQueueHandler(QueueHandler):
+    """非阻塞入队 + 惰性格式化。
+
+    - 队列满时丢弃（而不是抛 queue.Full 打断业务）
+    - 不预格式化：% 样式消息的格式化推迟到监听线程（formatter 执行），
+      主线程只承担 LogRecord 创建与入队
+    - queue 经模块全局读取：gunicorn preload/fork 后自动指向子进程的新队列
+    """
+
+    @property
+    def queue(self):
+        return _log_queue
+
+    def __init__(self):
+        # queue 由 property 动态提供（fork 后自动指向新队列）。
+        # 只初始化 Handler 基类（level/filters 等），不走 QueueHandler.__init__
+        # 的 self.queue = queue 赋值（会与 property 冲突）
+        logging.Handler.__init__(self)
+
+    def enqueue(self, record):
+        try:
+            self.queue.put_nowait(record)
+        except queue.Full:
+            pass  # 队列满：丢弃该日志
+
+    def prepare(self, record):
+        return record  # 格式化留给监听线程的 formatter
+
+
+class _RegistryListener(QueueListener):
+    """handler 可动态注册的 QueueListener（每个 log_name 有独立 file handler）"""
+
+    def __init__(self, q: queue.Queue):
+        super().__init__(q)
+        self.handlers: list = []  # 可变列表，运行期可追加
+
+    def register(self, *handlers) -> None:
+        self.handlers.extend(handlers)
+
+
+_log_queue: queue.Queue = queue.Queue(maxsize=_LOG_QUEUE_SIZE)
+_listener: Optional[_RegistryListener] = None
+# 全部真实 handler（文件/控制台），供 fork 后重建监听线程
+_listener_handlers: List[logging.Handler] = []
+
+
+def _start_listener() -> None:
+    global _listener
+    _listener = _RegistryListener(_log_queue)
+    _listener.register(*_listener_handlers)
+    _listener.start()
+
+
+def _after_fork_child() -> None:
+    """gunicorn preload/fork 场景：线程不跨 fork，子进程重建队列与监听线程。
+
+    队列重建避免 fork 时锁状态不可用导致的死锁；旧队列中的记录在 fork
+    瞬间丢失（启动窗口内可接受）。
+    """
+    global _log_queue
+    _log_queue = queue.Queue(maxsize=_LOG_QUEUE_SIZE)
+    _start_listener()
+
+
+if hasattr(os, "register_at_fork"):
+    os.register_at_fork(after_in_child=_after_fork_child)
+
+
+def shutdown_logging() -> None:
+    """停止监听线程并冲刷剩余日志（进程退出 / 应用关闭时调用），幂等。
+
+    QueueListener.stop() 停止后会将 _thread 置 None，重复调用需跳过；
+    幂等保证 atexit 与显式调用共存。
+    """
+    global _listener
+    if _listener is not None:
+        if _listener._thread is not None:
+            _listener.stop()  # flush=True：处理完队列中已有记录再停止
+        _listener = None
+    for h in _listener_handlers:
+        try:
+            h.close()
+        except Exception:  # pragma: no cover - 关闭失败不影响退出
+            pass
+
+
+atexit.register(shutdown_logging)
+
+
 _configured_loggers = set()
 
 
@@ -237,9 +338,14 @@ def setup_logging(logger: logging.Logger, log_name: str = "app"):
     console_handler.setFormatter(color_formatter)
     console_handler.addFilter(ExcludeWatchFilesFilter())
 
-    # 添加处理器到根日志记录器
-    logger.addHandler(file_handler)
-    logger.addHandler(console_handler)
+    # 真实 handler（文件/控制台）挂到后台监听线程；logger 上只挂队列 handler，
+    # 日志调用在主线程仅入队，I/O 由监听线程执行
+    _listener_handlers.extend([file_handler, console_handler])
+    if _listener is None:
+        _start_listener()
+    else:
+        _listener.register(file_handler, console_handler)
+    logger.addHandler(_AsyncQueueHandler())
 
     print(
         f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']} "

src/chongming/app/core/mqtt_bridge.py

@@ -1,6 +1,6 @@
 """MQTT Bridge — 纯客户端模式，Broker 由子进程管理"""
 import asyncio
-import json
+import orjson
 import re
 from typing import Optional
 from urllib.parse import quote
@@ -165,8 +165,8 @@ class MQTTBridge:
         t: str = m.topic
         p = m.data  # type: ignore
         try:
-            pl = json.loads(p.decode("utf-8"))
-        except json.JSONDecodeError:
+            pl = orjson.loads(p)
+        except orjson.JSONDecodeError:
             return
 
         # 信封协议: {"meta": {...}, "data": {...}}
@@ -243,8 +243,9 @@ class MQTTBridge:
 
     async def _pub(self, topic, data):
         try:
-            pd = json.dumps(data, ensure_ascii=False, default=str)
-            await self._client.publish(topic, pd.encode("utf-8"), QOS_1)  # type: ignore
+            # data 均为 JSON 可序列化类型（worker 回复解析结果 / {"error": ...}），
+            # orjson.dumps 直接产出 UTF-8 字节，免去 str -> encode 两步
+            await self._client.publish(topic, orjson.dumps(data), QOS_1)  # type: ignore
         except Exception as e:
             logger.error(f"发布失败 [{topic}]: {e}")
 

src/chongming/app/core/route_generator.py

@@ -7,6 +7,8 @@
 """
 from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple
 from inspect import Parameter, Signature
+import asyncio
+import time
 
 import orjson
 from fastapi import FastAPI, HTTPException, APIRouter, Query, Path as FastAPIPath, Request, Depends
@@ -18,6 +20,7 @@ from chongming_worker.route_config import WORKER_CONFIG_DIR, load_worker_routes_
 from .worker_manager import WorkerManager, StreamIdleTimeoutError
 from .logger import get_logger
 from .auth import store_user_to_request
+from .call_stats import spawn, record_call
 
 logger = get_logger("route_generator")
 
@@ -129,6 +132,7 @@ def register_route_group(
     prefix: str = "",
     extra_schemas: Optional[Dict[str, type[BaseModel]]] = None,
     raw_body_models: Optional[Dict[Tuple[str, str], str]] = None,
+    response_models: Optional[Dict[Tuple[str, str], str]] = None,
 ) -> List[Any]:
     """
     为一个 worker_type 的一组路由描述符创建并挂载 APIRouter。
@@ -145,6 +149,9 @@ def register_route_group(
         extra_schemas: 额外 schema 收集 dict（GET 请求模型注入 OpenAPI 用）
         raw_body_models: (path, method) -> 请求模型名 的收集 dict
             （validate=false 端点：请求体文档经 requestBody $ref 注入，见 _inject_schemas_to_openapi）
+        response_models: (path, method) -> 响应模型名 的收集 dict
+            （端点注册不传 response_model 以砍掉注册期模型反射；响应文档经
+            responses[200].content $ref 手动注入，见 _inject_schemas_to_openapi）
 
     Returns:
         List[Any]: 本次新增到 app.router.routes 的 APIRoute 对象列表
@@ -152,14 +159,17 @@ def register_route_group(
     """
     router = APIRouter(prefix=prefix)
     for route_conf in routes:
-        _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas, raw_body_models)
+        _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas, raw_body_models, response_models)
 
-    # raw_body_models 记录的 (path, method) 为端点声明路径（不含 prefix），
+    # raw_body_models / response_models 记录的 (path, method) 为端点声明路径（不含 prefix），
     # 而 OpenAPI paths 键为最终挂载路径（prefix + path），这里统一补前缀
-    if raw_body_models is not None and prefix:
-        for key in list(raw_body_models.keys()):
+    if prefix:
+        for key in list((raw_body_models or {}).keys()):
             path, method = key
             raw_body_models[(prefix + path, method)] = raw_body_models.pop(key)
+        for key in list((response_models or {}).keys()):
+            path, method = key
+            response_models[(prefix + path, method)] = response_models.pop(key)
 
     before = set(id(r) for r in app.router.routes)
     app.include_router(router)
@@ -193,6 +203,8 @@ def register_dynamic_routes(
     extra_schemas: Dict[str, type[BaseModel]] = {}
     # validate=false 端点的 (path, method) -> 请求模型名（OpenAPI requestBody 注入用）
     raw_body_models: Dict[Tuple[str, str], str] = {}
+    # 非流式端点的 (path, method) -> 响应模型名（OpenAPI responses[200] $ref 注入用）
+    response_models: Dict[Tuple[str, str], str] = {}
 
     # 扫描所有 worker 子目录（每个子目录名 = worker_type）
     for worker_dir in sorted(workers_base.iterdir()):
@@ -205,16 +217,16 @@ def register_dynamic_routes(
         if not routes:
             continue
 
-        register_route_group(app, worker_manager, worker_type, routes, prefix, extra_schemas, raw_body_models)
+        register_route_group(app, worker_manager, worker_type, routes, prefix, extra_schemas, raw_body_models, response_models)
 
     # 将额外 schema 注册到 OpenAPI 的 components/schemas
     if extra_schemas:
-        _inject_schemas_to_openapi(app, lambda: extra_schemas, lambda: raw_body_models)
+        _inject_schemas_to_openapi(app, lambda: extra_schemas, lambda: raw_body_models, lambda: response_models)
 
     logger.info(f"动态路由注册完成，worker 配置目录: {WORKER_CONFIG_DIR}")
 
 
-def _inject_schemas_to_openapi(app: FastAPI, schemas_provider, raw_body_provider=None):
+def _inject_schemas_to_openapi(app: FastAPI, schemas_provider, raw_body_provider=None, response_provider=None):
     """
     将自定义 Pydantic 模型注册到 FastAPI 的 OpenAPI components/schemas 中。
 
@@ -228,6 +240,9 @@ def _inject_schemas_to_openapi(app: FastAPI, schemas_provider, raw_body_provider
         raw_body_provider: 可选无参 callable，返回 {(path, method): model_name} 字典。
             validate=false 端点未声明 body 参数（FastAPI 不会自动生成 requestBody），
             这里按 (path, method) 补注入 requestBody $ref，保持 OpenAPI 请求体文档完整。
+        response_provider: 可选无参 callable，返回 {(path, method): model_name} 字典。
+            端点注册未传 response_model（见 _register_single_route：砍掉注册期模型反射），
+            这里按 (path, method) 补注入 responses[200] 的 $ref，保持响应文档完整。
     """
     original_openapi = app.openapi
 
@@ -255,12 +270,55 @@ def _inject_schemas_to_openapi(app: FastAPI, schemas_provider, raw_body_provider
                             }
                         },
                     }
+        # 响应模型：补注入 responses[200] 的 $ref（保留 description，仅加 content）
+        if response_provider is not None:
+            for (path, method), model_name in response_provider().items():
+                op = schema.get("paths", {}).get(path, {}).get(method.lower())
+                if op is not None:
+                    op.setdefault("responses", {}).setdefault("200", {})
+                    op["responses"]["200"].setdefault(
+                        "content", {}
+                    ).setdefault("application/json", {})["schema"] = {
+                        "$ref": f"#/components/schemas/{model_name}"
+                    }
         app.openapi_schema = schema
         return schema
 
     app.openapi = patched_openapi
 
 
+def _spawn_call_record(
+    request: Request,
+    worker_type: str,
+    subject: str,
+    rc: RouteConfig,
+    status: int,
+    duration_ms: float,
+    req_bytes: int = 0,
+    resp_bytes: int = 0,
+) -> None:
+    """
+    以后台任务记录一次调用（统计 + 审计），非关键后置操作。
+
+    通过 spawn() 以 fire-and-forget 方式执行，端点立即返回，
+    不增加接口响应延迟；后台失败由 record_call 内部兜底。
+    """
+    entry = {
+        "ts": time.time(),
+        "worker_type": worker_type,
+        "subject": subject,
+        "method": rc.method,
+        "path": request.url.path,
+        "user": getattr(request.state, "user", None),
+        "client": request.client.host if request.client else None,
+        "status": status,
+        "duration_ms": round(duration_ms, 3),
+        "req_bytes": req_bytes,
+        "resp_bytes": resp_bytes,
+    }
+    spawn(record_call(entry))
+
+
 def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_type: str, RequestModel: type[BaseModel]):
     """创建用于 Body 请求（POST/PUT/PATCH）的端点函数"""
     path = rc.path
@@ -273,6 +331,8 @@ def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_t
         request: Request,
         body: RequestModel,  # type: ignore
     ) -> Any:
+        start: Optional[float] = None
+        payload_bytes = b""
         try:
             # 信封协议: {"meta": {...}, "data": {...}}
             meta = {"method": http_method}
@@ -280,19 +340,36 @@ def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_t
             if user_info:
                 meta["user_info"] = user_info
             payload = {"meta": meta, "data": body.model_dump()}
+            payload_bytes = orjson.dumps(payload)
+            start = time.perf_counter()
             result = await worker_manager.call_raw(
                 worker_type=worker_type,
                 subject=subject,
-                payload=orjson.dumps(payload),
+                payload=payload_bytes,
                 timeout=timeout,
                 cacheable=cacheable,
             )
+            duration_ms = (time.perf_counter() - start) * 1000
+            _spawn_call_record(
+                request, worker_type, subject, rc, 200, duration_ms,
+                req_bytes=len(payload_bytes), resp_bytes=len(result),
+            )
             # 回复为 worker 原始字节，零解析直接转发（response_model 仅用于 OpenAPI 文档）
             return Response(content=result, media_type="application/json")
         except RuntimeError as e:
+            duration_ms = (time.perf_counter() - start) * 1000 if start else 0.0
+            _spawn_call_record(
+                request, worker_type, subject, rc, 503, duration_ms,
+                req_bytes=len(payload_bytes),
+            )
             logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e)
             raise HTTPException(status_code=503, detail=str(e))
         except Exception as e:
+            duration_ms = (time.perf_counter() - start) * 1000 if start else 0.0
+            _spawn_call_record(
+                request, worker_type, subject, rc, 500, duration_ms,
+                req_bytes=len(payload_bytes),
+            )
             logger.error(f"路由 {path} 调用异常: {e}")
             raise HTTPException(status_code=500, detail="内部服务器错误")
 
@@ -315,6 +392,8 @@ def _make_raw_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, work
     cacheable = rc.cacheable
 
     async def endpoint(request: Request) -> Any:
+        start: Optional[float] = None
+        payload = b""
         try:
             meta = {"method": http_method}
             user_info = getattr(request.state, "user", None)
@@ -323,6 +402,7 @@ def _make_raw_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, work
             raw = await request.body()
             data_part = raw.strip() or b"{}"
             payload = b'{"meta":' + orjson.dumps(meta) + b',"data":' + data_part + b'}'
+            start = time.perf_counter()
             result = await worker_manager.call_raw(
                 worker_type=worker_type,
                 subject=subject,
@@ -330,11 +410,26 @@ def _make_raw_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, work
                 timeout=timeout,
                 cacheable=cacheable,
             )
+            duration_ms = (time.perf_counter() - start) * 1000
+            _spawn_call_record(
+                request, worker_type, subject, rc, 200, duration_ms,
+                req_bytes=len(payload), resp_bytes=len(result),
+            )
             return Response(content=result, media_type="application/json")
         except RuntimeError as e:
+            duration_ms = (time.perf_counter() - start) * 1000 if start else 0.0
+            _spawn_call_record(
+                request, worker_type, subject, rc, 503, duration_ms,
+                req_bytes=len(payload),
+            )
             logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e)
             raise HTTPException(status_code=503, detail=str(e))
         except Exception as e:
+            duration_ms = (time.perf_counter() - start) * 1000 if start else 0.0
+            _spawn_call_record(
+                request, worker_type, subject, rc, 500, duration_ms,
+                req_bytes=len(payload),
+            )
             logger.error(f"路由 {path} 调用异常: {e}")
             raise HTTPException(status_code=500, detail="内部服务器错误")
 
@@ -368,20 +463,36 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
             user_info = getattr(request.state, "user", None)
             if user_info:
                 meta["user_info"] = user_info
-            payload = {"meta": meta, "data": {}}
+            payload_bytes = orjson.dumps({"meta": meta, "data": {}})
+            start = time.perf_counter()
             try:
                 result = await worker_manager.call_raw(
                     worker_type=worker_type,
                     subject=subject,
-                    payload=orjson.dumps(payload),
+                    payload=payload_bytes,
                     timeout=timeout,
                     cacheable=cacheable,
                 )
+                duration_ms = (time.perf_counter() - start) * 1000
+                _spawn_call_record(
+                    request, worker_type, subject, rc, 200, duration_ms,
+                    req_bytes=len(payload_bytes), resp_bytes=len(result),
+                )
                 return Response(content=result, media_type="application/json")
             except RuntimeError as e:
+                duration_ms = (time.perf_counter() - start) * 1000
+                _spawn_call_record(
+                    request, worker_type, subject, rc, 503, duration_ms,
+                    req_bytes=len(payload_bytes),
+                )
                 logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e)
                 raise HTTPException(status_code=503, detail=str(e))
             except Exception as e:
+                duration_ms = (time.perf_counter() - start) * 1000
+                _spawn_call_record(
+                    request, worker_type, subject, rc, 500, duration_ms,
+                    req_bytes=len(payload_bytes),
+                )
                 logger.error(f"路由 {path} 调用异常: {e}")
                 raise HTTPException(status_code=500, detail="内部服务器错误")
         return _simple_endpoint
@@ -395,20 +506,36 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
         user_info = getattr(request.state, "user", None)
         if user_info:
             meta["user_info"] = user_info
-        payload = {"meta": meta, "data": {k: kwargs[k] for k in param_names}}
+        payload_bytes = orjson.dumps({"meta": meta, "data": {k: kwargs[k] for k in param_names}})
+        start = time.perf_counter()
         try:
             result = await worker_manager.call_raw(
                 worker_type=worker_type,
                 subject=subject,
-                payload=orjson.dumps(payload),
+                payload=payload_bytes,
                 timeout=timeout,
                 cacheable=cacheable,
             )
+            duration_ms = (time.perf_counter() - start) * 1000
+            _spawn_call_record(
+                request, worker_type, subject, rc, 200, duration_ms,
+                req_bytes=len(payload_bytes), resp_bytes=len(result),
+            )
             return Response(content=result, media_type="application/json")
         except RuntimeError as e:
+            duration_ms = (time.perf_counter() - start) * 1000
+            _spawn_call_record(
+                request, worker_type, subject, rc, 503, duration_ms,
+                req_bytes=len(payload_bytes),
+            )
             logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e)
             raise HTTPException(status_code=503, detail=str(e))
         except Exception as e:
+            duration_ms = (time.perf_counter() - start) * 1000
+            _spawn_call_record(
+                request, worker_type, subject, rc, 500, duration_ms,
+                req_bytes=len(payload_bytes),
+            )
             logger.error(f"路由 {path} 调用异常: {e}")
             raise HTTPException(status_code=500, detail="内部服务器错误")
 
@@ -480,19 +607,40 @@ def _make_stream_endpoint(
     def _sse_frame(data: Dict[str, Any]) -> str:
         return f"data: {orjson.dumps(data).decode('utf-8')}\n\n"
 
-    async def _sse_body(gen) -> AsyncGenerator[str, None]:
-        """SSE 分帧；流中异常转为 SSE error 事件（HTTP 200 已提交）"""
+    async def _sse_body(gen, start: float, request: Request, req_bytes: int) -> AsyncGenerator[str, None]:
+        """SSE 分帧；流中异常转为 SSE error 事件（HTTP 200 已提交）。
+
+        流结束（正常/空闲超时/中断/客户端断开）后在 finally 中后台记录
+        调用统计与审计，时长覆盖整个流。
+        """
+        status = 200
         try:
             async for chunk in gen:
                 yield f"data: {chunk.decode('utf-8')}\n\n"
         except StreamIdleTimeoutError as e:
+            status = 504
             logger.warning("路由 %s 流式空闲超时: %s", path, e)
             yield _sse_frame({"error": str(e)})
+        except asyncio.CancelledError:
+            status = 499  # 客户端断开，流未完成
+            raise
+        except GeneratorExit:
+            status = 499  # 服务端主动关闭流（如连接回收），流未完成
+            raise
         except Exception as e:
+            status = 500
             logger.error("路由 %s 流式中断: %s", path, e)
             yield _sse_frame({"error": "内部服务器错误"})
+        finally:
+            duration_ms = (time.perf_counter() - start) * 1000
+            _spawn_call_record(
+                request, worker_type, subject, rc, status, duration_ms,
+                req_bytes=req_bytes,
+            )
 
-    async def _start_stream(payload: Dict[str, Any]) -> StreamingResponse:
+    async def _start_stream(request: Request, payload: Dict[str, Any]) -> StreamingResponse:
+        req_bytes = len(orjson.dumps(payload))
+        start = time.perf_counter()
         try:
             gen = await worker_manager.call_stream(
                 worker_type=worker_type,
@@ -502,27 +650,32 @@ def _make_stream_endpoint(
                 idle_timeout=idle_timeout,
             )
         except RuntimeError as e:
+            duration_ms = (time.perf_counter() - start) * 1000
+            _spawn_call_record(
+                request, worker_type, subject, rc, 503, duration_ms,
+                req_bytes=req_bytes,
+            )
             logger.warning("路由 %s [%s] 流式调用失败: %s", path, worker_type, e)
             raise HTTPException(status_code=503, detail=str(e))
         return StreamingResponse(
-            _sse_body(gen),
+            _sse_body(gen, start, request, req_bytes),
             media_type="text/event-stream",
             headers=_STREAM_HEADERS,
         )
 
     if RequestModel is not None:
         async def endpoint(request: Request, body: RequestModel) -> StreamingResponse:  # type: ignore
-            return await _start_stream(build_payload(request, body=body))
+            return await _start_stream(request, build_payload(request, body=body))
         return endpoint
 
     # GET/DELETE 形态
     if not param_names:
         async def _simple_endpoint(request: Request) -> StreamingResponse:
-            return await _start_stream(build_payload(request, kwargs=None))
+            return await _start_stream(request, build_payload(request, kwargs=None))
         return _simple_endpoint
 
     async def endpoint(request: Request, **kwargs: Any) -> StreamingResponse:
-        return await _start_stream(build_payload(request, kwargs=kwargs))
+        return await _start_stream(request, build_payload(request, kwargs=kwargs))
 
     # 附加类型注解与签名（与 _make_query_endpoint 一致），保证 OpenAPI 参数文档正确
     params: List[Parameter] = [
@@ -550,6 +703,7 @@ def _register_single_route(
     route_conf: Dict[str, Any],
     extra_schemas: Optional[Dict[str, type[BaseModel]]] = None,
     raw_body_models: Optional[Dict[Tuple[str, str], str]] = None,
+    response_models: Optional[Dict[Tuple[str, str], str]] = None,
 ):
     """注册单个路由，生成正确的函数签名以便 OpenAPI 正确生成文档"""
     rc = RouteConfig(route_conf)
@@ -609,6 +763,15 @@ def _register_single_route(
         ResponseModel = generate_response_model(
             rc.response_model, response_model_name
         )
+        # 响应模型仅注册到 components/schemas + responses[200] $ref 注入，
+        # 不传给 add_api_route：端点返回 Response 对象本就走透传（零运行时校验），
+        # 省掉 FastAPI 注册期对 response_model 的反射解析（external 插件频繁
+        # 挂载/摘除时每次重建都会重复这一开销）
+        if ResponseModel is not None:
+            if extra_schemas is not None and response_model_name not in extra_schemas:
+                extra_schemas[response_model_name] = ResponseModel
+            if response_models is not None:
+                response_models[(rc.path, rc.method)] = response_model_name
         RequestModel = generate_request_model(
             param_spec, request_model_name
         )
@@ -627,6 +790,12 @@ def _register_single_route(
         ResponseModel = generate_response_model(
             rc.response_model, response_model_name
         )
+        # 同上：响应模型走 extra_schemas + responses $ref 注入，不传 add_api_route
+        if ResponseModel is not None:
+            if extra_schemas is not None and response_model_name not in extra_schemas:
+                extra_schemas[response_model_name] = ResponseModel
+            if response_models is not None:
+                response_models[(rc.path, rc.method)] = response_model_name
         # 对 GET 端点也生成 Request 模型并注册到 extra_schemas
         if param_spec:
             RequestModel = generate_request_model(
@@ -642,13 +811,14 @@ def _register_single_route(
 
     # 注册路由 - 使用 worker_type 作为 OpenAPI 分组标签
     # 每个 worker 类型自动成为一个独立的分组，显示在 Swagger UI 中
+    # 不传 response_model：端点返回 Response 对象走透传，运行时本就零校验；
+    # 响应文档由 response_models 收集 + _inject_schemas_to_openapi 补 $ref
     router.add_api_route(
         path=rc.path,
         endpoint=endpoint,
         methods=[rc.method],
         summary=rc.summary,
         description=rc.docstring,
-        response_model=ResponseModel,
         tags=[worker_type],
         include_in_schema=True,
         dependencies=dependencies,

src/chongming/app/core/route_registry.py

@@ -19,7 +19,7 @@
 """
 import asyncio
 import hashlib
-import json
+import orjson
 from typing import Any, Dict, List, Optional, Tuple
 
 import zenoh
@@ -37,8 +37,8 @@ DEBOUNCE_SECONDS = 2.0
 
 def _descriptor_hash(descriptor: Dict[str, Any]) -> str:
     """描述符内容 hash（多副本幂等去重用）"""
-    payload = json.dumps(descriptor, sort_keys=True, ensure_ascii=False)
-    return hashlib.sha1(payload.encode("utf-8")).hexdigest()
+    payload = orjson.dumps(descriptor, option=orjson.OPT_SORT_KEYS)
+    return hashlib.sha1(payload).hexdigest()
 
 
 class RouteRegistry:
@@ -65,6 +65,8 @@ class RouteRegistry:
         self._schemas: Dict[str, Any] = {}
         # validate=false 端点的 (path, method) -> 请求模型名（OpenAPI requestBody 注入用）
         self._raw_body_models: Dict[Tuple[str, str], str] = {}
+        # 非流式端点的 (path, method) -> 响应模型名（OpenAPI responses[200] $ref 注入用）
+        self._response_models: Dict[Tuple[str, str], str] = {}
         # worker_type -> 防抖任务
         self._debounce: Dict[str, asyncio.Task] = {}
 
@@ -74,7 +76,12 @@ class RouteRegistry:
         self._inline_types: set = set()
 
         # 将动态 schema 注入 OpenAPI（provider 实时读取本注册表）
-        _inject_schemas_to_openapi(app, lambda: self._schemas, lambda: self._raw_body_models)
+        _inject_schemas_to_openapi(
+            app,
+            lambda: self._schemas,
+            lambda: self._raw_body_models,
+            lambda: self._response_models,
+        )
 
     def set_inline_types(self, types: set):
         """设置 inline 插件类型集合（混合模式防双注册）。"""
@@ -149,8 +156,8 @@ class RouteRegistry:
             return
 
         try:
-            descriptor = json.loads(payload.decode("utf-8"))
-        except (json.JSONDecodeError, UnicodeDecodeError):
+            descriptor = orjson.loads(payload)
+        except orjson.JSONDecodeError:
             logger.warning("丢弃非法路由描述符 [%s]: JSON 解析失败", key)
             return
         if not isinstance(descriptor, dict):
@@ -238,7 +245,7 @@ class RouteRegistry:
         if http_routes:
             self._route_objs[wtype] = register_route_group(
                 self._app, self._wm, wtype, http_routes, self._prefix,
-                self._schemas, self._raw_body_models,
+                self._schemas, self._raw_body_models, self._response_models,
             )
         else:
             logger.info("[%s] 无 HTTP 路由可挂载（%d 个描述符，全部为 MQTT/internal）",

src/chongming/app/core/worker_manager.py

@@ -1,7 +1,7 @@
 import asyncio
+import concurrent.futures
 import hashlib
 import random
-import json
 import orjson
 from typing import Any, AsyncGenerator, Dict, List, Optional
 import zenoh
@@ -12,6 +12,15 @@ logger = get_logger("worker_manager")
 
 _SENTINEL = object()  # 哨兵值，用于标记"No reply received"
 
+# Zenoh RPC 专用线程池：隔离于 asyncio.to_thread 的全局默认池
+# （默认池上限 min(32, cpu+4)，高并发下会被其他阻塞调用挤占导致排队）。
+# zenoh session.get 是同步阻塞调用，每个进行中的查询占用一个线程；
+# wait_for 超时后线程仍阻塞在 zenoh 内部直到其自身超时返回，故池上限即并发上限。
+# 按业务并发量调整（256 ≈ 256 个并发 Zenoh 查询）；线程 daemon=True，进程退出不等待。
+_ZENOH_EXECUTOR = concurrent.futures.ThreadPoolExecutor(
+    max_workers=256, thread_name_prefix="zenoh-rpc"
+)
+
 
 class StreamIdleTimeoutError(TimeoutError):
     """流式调用空闲超时：在 idle_timeout 秒内未收到任何数据块"""
@@ -216,11 +225,13 @@ class WorkerManager:
 
         async def _stream():
             # 逐块消费；zenoh reply 迭代器是同步阻塞的，必须放线程执行，
-            # 避免阻塞事件循环（与 call() 的 _get_first 同模式）
+            # 避免阻塞事件循环（与 call() 的 _get_first 同模式）。
+            # 走专用 _ZENOH_EXECUTOR 池，不与全局默认池的其它阻塞调用争抢
             while True:
                 try:
+                    loop = asyncio.get_running_loop()
                     reply = await asyncio.wait_for(
-                        asyncio.to_thread(_next_reply, reply_iter),
+                        loop.run_in_executor(_ZENOH_EXECUTOR, _next_reply, reply_iter),
                         timeout=idle_timeout,
                     )
                 except asyncio.TimeoutError:
@@ -278,7 +289,7 @@ class WorkerManager:
         Returns:
             Dict[str, Any]: 调用结果（worker 回复解析后的 dict）
         """
-        cache_key = f"{worker_type}:{subject}:{json.dumps(payload, sort_keys=True)}"
+        cache_key = f"{worker_type}:{subject}:{orjson.dumps(payload, option=orjson.OPT_SORT_KEYS).decode()}"
         reply_bytes = await self._call_impl(
             worker_type, subject, orjson.dumps(payload),
             timeout=timeout, retries=retries, cacheable=cacheable, cache_key=cache_key,
@@ -370,8 +381,10 @@ class WorkerManager:
                         return item
                     return _SENTINEL
 
+                # 走专用 _ZENOH_EXECUTOR 池（隔离于全局默认池，避免高并发下排队）
+                loop = asyncio.get_running_loop()
                 first_reply = await asyncio.wait_for(
-                    asyncio.to_thread(_get_first, reply_iter),
+                    loop.run_in_executor(_ZENOH_EXECUTOR, _get_first, reply_iter),
                     timeout=timeout,
                 )
                 if first_reply is _SENTINEL:

src/chongming/app/routes/__init__.py

@@ -1,3 +1,4 @@
 from .storage import router as storage_router
+from .stats import router as stats_router
 
-__all__ = ["storage_router"]
+__all__ = ["storage_router", "stats_router"]

src/chongming/app/routes/stats.py

@@ -0,0 +1,23 @@
+# src/chongming/app/routes/stats.py
+"""
+网关级调用统计查看端点
+
+数据来源为 core.call_stats 的内存统计（进程内视角）：
+动态路由每次调用在后台任务中更新计数/延迟聚合/最近调用缓冲，
+此处提供只读快照，便于观测接口调用量、错误分布与延迟。
+"""
+from fastapi import APIRouter
+
+from chongming_config import get_config
+
+from ..core import call_stats
+
+config = get_config()
+
+router = APIRouter(prefix=config["default"]["prefix"], tags=["stats"])
+
+
+@router.get("/stats", summary="调用统计")
+async def get_call_stats():
+    """返回内存调用统计快照：总量、per worker/subject/status 计数、延迟聚合、最近调用"""
+    return call_stats.summary()

test_call_stats.py

@@ -0,0 +1,333 @@
+"""
+调用统计 + 审计日志（后台任务）测试（依赖隔离：桩掉 chongming 包链、auth、
+chongming_cache 与 worker_manager，加载真实 call_stats 与 route_generator）
+
+用 httpx.ASGITransport 验证:
+  1. CallStats.record/summary 聚合正确性（计数 / 平均 / 最大延迟 / recent 上限）
+  2. spawn/shutdown：后台任务被跟踪、shutdown 后全部取消
+  3. 动态端点请求后：响应与改动前一致（字节透传）+ 统计计数正确 + 审计日志出现 AUDIT 行
+  4. 错误路径（无可用 worker）-> HTTP 503 同样被记录
+  5. 后台任务失败兜底：record_call 内部异常不冒泡，端点响应不受影响
+
+运行: uv run python test_call_stats.py
+"""
+import asyncio
+import importlib.util
+import logging
+import sys
+import types
+from pathlib import Path
+
+import httpx
+import orjson
+from fastapi import FastAPI
+
+REPO_ROOT = Path(__file__).resolve().parent
+sys.path.insert(0, str(REPO_ROOT / "src"))
+sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
+
+# ---- 依赖隔离：桩掉 chongming 包链 / auth / chongming_cache / worker_manager ----
+_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  # 仅用于 Depends，桩掉鉴权
+_cache_stub = types.ModuleType("chongming_cache")
+_cache_stub.Cache = object
+# worker_manager 桩：route_generator 仅用其类型注解与异常类
+_wm_stub = types.ModuleType("chongming.app.core.worker_manager")
+_wm_stub.WorkerManager = object
+_wm_stub.StreamIdleTimeoutError = type("StreamIdleTimeoutError", (RuntimeError,), {})
+for mod in (_pkg, _app, _core, _logger_stub, _auth_stub, _cache_stub, _wm_stub):
+    sys.modules[mod.__name__] = mod
+
+# 加载真实 call_stats（route_generator 依赖其 spawn/record_call）
+_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]
+
+# 加载真实 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)
+_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
+record_call = _cs_mod.record_call
+spawn = _cs_mod.spawn
+
+# ---- 审计日志捕获（桩 logger 直接挂到标准库 audit logger） ----
+_audit_records = []
+
+
+class _AuditHandler(logging.Handler):
+    def emit(self, record):
+        _audit_records.append(record.getMessage())
+
+
+_audit_logger = logging.getLogger("audit")
+_audit_logger.setLevel(logging.INFO)
+_audit_logger.addHandler(_AuditHandler())
+_audit_logger.propagate = False
+
+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:
+    """模拟 WorkerManager：call_raw 返回原始字节；error="no_worker" 时抛 RuntimeError"""
+
+    def __init__(self, reply=b'{"ok": true}', error=None):
+        self.reply = reply
+        self.error = error
+
+    async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False):
+        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):
+        if self.error == "no_worker":
+            raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。")
+
+        async def _gen():
+            yield orjson.dumps({"chunk": 1})
+            yield orjson.dumps({"chunk": 2})
+
+        return _gen()  # 与真实实现一致：async 函数返回异步生成器
+
+
+ROUTES = [
+    # body 端点（validate=true）
+    {"subject": "demo.echo", "path": "/echo", "method": "POST",
+     "params": ["msg: str"], "timeout": 3.0, "auth_required": False},
+    # raw body 端点（validate=false）
+    {"subject": "demo.raw", "path": "/raw/echo", "method": "POST",
+     "params": ["msg: str"], "validate": False, "timeout": 3.0, "auth_required": False},
+    # query 带参端点
+    {"subject": "demo.add", "path": "/add", "method": "GET",
+     "params": ["a: int", "b: int"], "timeout": 3.0, "auth_required": False},
+    # query 无参端点（_simple_endpoint 分支）
+    {"subject": "demo.ping", "path": "/ping", "method": "GET",
+     "timeout": 3.0, "auth_required": False},
+    # 流式端点
+    {"subject": "demo.stream", "path": "/stream", "method": "GET",
+     "stream": True, "timeout": 3.0, "auth_required": False},
+]
+
+
+def build_app(fwm):
+    app = FastAPI()
+    register_route_group(app, fwm, "demo", ROUTES, prefix="/api/v1")
+    return app
+
+
+async def _drain_background():
+    """等待并清理所有后台统计/审计任务（每个异步测试末尾调用）"""
+    tasks = list(_cs_mod._tasks)
+    if tasks:
+        await asyncio.gather(*tasks, return_exceptions=True)
+    await _cs_mod.shutdown()
+
+
+def _reset_stats():
+    """隔离：每个端点测试前重置全局统计实例"""
+    _cs_mod._stats = _cs_mod.CallStats()
+
+
+# ---- 1. CallStats 单元 ----
+
+def test_stats_aggregation():
+    cs = _cs_mod.CallStats(max_recent=10)
+    cs.record({"worker_type": "demo", "subject": "demo.echo", "status": 200, "duration_ms": 10.0})
+    cs.record({"worker_type": "demo", "subject": "demo.echo", "status": 200, "duration_ms": 30.0})
+    cs.record({"worker_type": "demo", "subject": "demo.echo", "status": 503, "duration_ms": 5.0})
+    cs.record({"worker_type": "other", "subject": "other.x", "status": 200, "duration_ms": 2.0})
+    s = cs.summary()
+    check("total_calls", s["total_calls"] == 4, str(s["total_calls"]))
+    by_key = {(c["worker_type"], c["subject"], c["status"]): c["count"] for c in s["calls"]}
+    check(
+        "per-key 计数",
+        by_key == {
+            ("demo", "demo.echo", 200): 2,
+            ("demo", "demo.echo", 503): 1,
+            ("other", "other.x", 200): 1,
+        },
+        str(by_key),
+    )
+    lat = {l["worker_type"]: l for l in s["latency"]}
+    check("延迟聚合 count", lat["demo"]["count"] == 3, str(lat["demo"]))
+    check("延迟聚合 avg", lat["demo"]["avg_ms"] == 15.0, str(lat["demo"]["avg_ms"]))
+    check("延迟聚合 max", lat["demo"]["max_ms"] == 30.0, str(lat["demo"]["max_ms"]))
+    check("recent 保留", len(s["recent"]) == 4, str(len(s["recent"])))
+
+
+def test_recent_bounded():
+    cs = _cs_mod.CallStats(max_recent=5)
+    for _ in range(20):
+        cs.record({"worker_type": "demo", "subject": "demo.x", "status": 200, "duration_ms": 1.0})
+    s = cs.summary()
+    check("recent 环形缓冲上限", len(s["recent"]) == 5, str(len(s["recent"])))
+    check("计数不受缓冲影响", s["total_calls"] == 20, str(s["total_calls"]))
+
+
+# ---- 2. spawn / shutdown ----
+
+def test_spawn_shutdown():
+    async def run():
+        async def never():
+            await asyncio.sleep(3600)
+
+        t = spawn(never())
+        check("任务被跟踪", t in _cs_mod._tasks, str(list(_cs_mod._tasks)))
+        await _cs_mod.shutdown()
+        check("shutdown 后任务取消", t.cancelled(), str(t))
+        check("任务集合清空", len(_cs_mod._tasks) == 0, str(len(_cs_mod._tasks)))
+
+    asyncio.run(run())
+
+
+# ---- 3. 端点集成：统计 + 审计 ----
+
+def test_endpoint_records_stats_and_audit():
+    async def run():
+        _audit_records.clear()
+        _reset_stats()
+        fwm = FakeWorkerManager(reply=b'{"ok": true}')
+        app = build_app(fwm)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
+            r1 = await ac.post("/api/v1/echo", json={"msg": "hi"})
+            r2 = await ac.post("/api/v1/raw/echo", json={"msg": "hi"})
+            r3 = await ac.get("/api/v1/add?a=1&b=2")
+            r4 = await ac.get("/api/v1/ping")
+            r5 = await ac.get("/api/v1/stream")
+        await _drain_background()
+
+        # 响应与改动前一致：worker 原始字节零再序列化
+        check("POST 响应字节透传", r1.status_code == 200 and r1.content == b'{"ok": true}', repr(r1.content))
+        check("raw POST 响应字节透传", r2.status_code == 200 and r2.content == b'{"ok": true}', repr(r2.content))
+        check("GET 响应字节透传", r3.status_code == 200 and r3.content == b'{"ok": true}', repr(r3.content))
+        check("无参 GET 响应字节透传", r4.status_code == 200 and r4.content == b'{"ok": true}', repr(r4.content))
+        check("流式响应 SSE 帧", r5.status_code == 200 and b"data:" in r5.content and b'"chunk":1' in r5.content, repr(r5.content[:80]))
+
+        s = _cs_mod.summary()
+        check("total_calls == 5", s["total_calls"] == 5, str(s["total_calls"]))
+        subjects = {c["subject"] for c in s["calls"]}
+        check(
+            "5 种调用均记录",
+            subjects == {"demo.echo", "demo.raw", "demo.add", "demo.ping", "demo.stream"},
+            str(subjects),
+        )
+        check("全部状态 200", all(c["status"] == 200 for c in s["calls"]), str(s["calls"]))
+
+        check("审计行数 == 5", len(_audit_records) == 5, str(len(_audit_records)))
+        echo_audit = next((a for a in _audit_records if "subject=demo.echo" in a), "")
+        check(
+            "审计行格式",
+            all(k in echo_audit for k in (
+                "AUDIT", "ts=", "worker=demo", "method=POST",
+                "path=/api/v1/echo", "status=200", "req_bytes=", "resp_bytes=",
+            )),
+            echo_audit,
+        )
+        stream_audit = next((a for a in _audit_records if "subject=demo.stream" in a), "")
+        check("流式审计已记录", "status=200" in stream_audit and "path=/api/v1/stream" in stream_audit, stream_audit)
+
+    asyncio.run(run())
+
+
+def test_error_503_recorded():
+    async def run():
+        _audit_records.clear()
+        _reset_stats()
+        fwm = FakeWorkerManager(error="no_worker")
+        app = build_app(fwm)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
+            r = await ac.get("/api/v1/ping")
+        await _drain_background()
+
+        check("503 响应", r.status_code == 503, str(r.status_code))
+        s = _cs_mod.summary()
+        check(
+            "503 计数 1",
+            any(c["subject"] == "demo.ping" and c["status"] == 503 and c["count"] == 1 for c in s["calls"]),
+            str(s["calls"]),
+        )
+        check(
+            "503 审计行",
+            any("subject=demo.ping" in a and "status=503" in a for a in _audit_records),
+            str(_audit_records),
+        )
+
+    asyncio.run(run())
+
+
+# ---- 4. 后台任务失败兜底 ----
+
+def test_background_failure_isolated():
+    async def run():
+        _audit_records.clear()
+        _reset_stats()
+        original = _cs_mod._stats.record
+
+        def boom(entry):
+            raise ValueError("stats broken")
+
+        _cs_mod._stats.record = boom
+        try:
+            # 后台执行体自身不冒泡（统计故障仅记 warning，审计仍写入）
+            await record_call({
+                "ts": 0.0, "worker_type": "demo", "subject": "demo.x", "status": 200,
+                "duration_ms": 1.0, "method": "GET", "path": "/x",
+                "user": None, "client": None, "req_bytes": 0, "resp_bytes": 0,
+            })
+            check("record_call 异常被兜底（不冒泡）", True)
+            check("审计仍写入", any("subject=demo.x" in a for a in _audit_records), str(_audit_records))
+        finally:
+            _cs_mod._stats.record = original
+
+        # 统计故障下端点响应不受影响
+        fwm = FakeWorkerManager(reply=b'{"ok": true}')
+        app = build_app(fwm)
+        _cs_mod._stats.record = boom
+        try:
+            transport = httpx.ASGITransport(app=app)
+            async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
+                r = await ac.get("/api/v1/ping")
+            await _drain_background()
+            check("统计故障下响应仍 200", r.status_code == 200 and r.content == b'{"ok": true}', repr(r.content))
+        finally:
+            _cs_mod._stats.record = original
+
+    asyncio.run(run())
+
+
+# ---- 执行 ----
+test_stats_aggregation()
+test_recent_bounded()
+test_endpoint_records_stats_and_audit()
+test_error_503_recorded()
+test_background_failure_isolated()
+test_spawn_shutdown()  # 最后执行：shutdown 会取消所有未完成任务
+
+print()
+if failures:
+    print(f"共 {len(failures)} 项失败: {failures}")
+    sys.exit(1)
+print("全部通过")

test_jwt_cache.py

@@ -0,0 +1,238 @@
+"""
+JWT 解析共享缓存测试（auth.py _cached_decode / jwt_auth_dependency）
+
+验证:
+  1. 同 token 重复鉴权只解密一次（缓存命中），不同 token 独立缓存
+  2. TTL 过期后重新解码（条目自然失效）
+  3. 缓存条目 exp 已过期：命中即拒绝，不重解密（缓存不延长 token 有效期）
+  4. 无效 token 不缓存失败：每次都走完整验证（decode 调用计数递增）
+  5. 缓存 key 含密钥/算法指纹：同 token 不同密钥不串用
+  6. 返回值为独立对象：调用方修改不影响缓存
+  7. 端到端：真实 chongming_jwt 签发 -> 鉴权成功 -> 二次命中缓存
+
+运行: uv run python test_jwt_cache.py
+"""
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent / "src"))
+sys.path.insert(0, str(Path(__file__).resolve().parent / "utils" / "worker" / "src"))
+
+import orjson
+from fastapi.security import HTTPAuthorizationCredentials
+
+from chongming.app.core import auth as auth_mod
+from chongming.app.core.auth import jwt_auth_dependency
+from chongming_cache import Cache
+from chongming_jwt import JWT
+from chongming_config import get_config
+
+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)
+
+
+config = get_config()
+SECRET = config["security"]["secret_key"]
+ALGORITHM = config["security"]["algorithm"]
+
+# 每个测试使用独立临时 diskcache，隔离真实缓存目录
+_tmp = tempfile.TemporaryDirectory()
+
+
+def fresh_cache():
+    auth_mod._jwt_cache = Cache(backend="diskcache", cache_store_path=_tmp.name)
+    auth_mod._jwt_cache.clear()
+
+
+def make_creds(token):
+    return HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
+
+
+# ---- 1. 缓存命中与隔离 ----
+def test_cache_hit_and_isolation():
+    fresh_cache()
+
+    class CountingJWT:
+        calls = 0
+
+        def __init__(self, secret, algorithm):
+            pass
+
+        def decode(self, token):
+            CountingJWT.calls += 1
+            return orjson.dumps({"sub": token, "exp": int(time.time()) + 3600}).decode()
+
+    auth_mod.JWT = CountingJWT
+    t1, t2 = "token-1", "token-2"
+    p1 = jwt_auth_dependency(credentials=make_creds(t1))
+    p2 = jwt_auth_dependency(credentials=make_creds(t1))
+    p3 = jwt_auth_dependency(credentials=make_creds(t2))
+    check("同 token 二次鉴权命中缓存（只解密一次）", CountingJWT.calls == 2, f"calls={CountingJWT.calls}")
+    check("payload 解析正确", p1["sub"] == "token-1" and p2["sub"] == "token-1", str(p1))
+    check("不同 token 独立缓存", p3["sub"] == "token-2", str(p3))
+    check("缓存中已写入条目", len(auth_mod._jwt_cache) >= 2, str(len(auth_mod._jwt_cache)))
+
+
+# ---- 2. TTL 过期后重新解码 ----
+def test_ttl_expiry_redecodes():
+    fresh_cache()
+    auth_mod._JWT_CACHE_TTL = 1  # 缩短 TTL 加速测试
+
+    class TtlJWT:
+        calls = 0
+
+        def __init__(self, secret, algorithm):
+            pass
+
+        def decode(self, token):
+            TtlJWT.calls += 1
+            return orjson.dumps({"sub": "u1", "exp": int(time.time()) + 3600}).decode()
+
+    auth_mod.JWT = TtlJWT
+    creds = make_creds("ttl-token")
+    p1 = jwt_auth_dependency(credentials=creds)
+    p2 = jwt_auth_dependency(credentials=creds)
+    check("TTL 内命中", TtlJWT.calls == 1 and p2["sub"] == "u1", f"calls={TtlJWT.calls}")
+    time.sleep(1.2)  # 等待 TTL 过期
+    p3 = jwt_auth_dependency(credentials=creds)
+    check("TTL 过期后重新解码", TtlJWT.calls == 2 and p3["sub"] == "u1", f"calls={TtlJWT.calls}")
+
+
+# ---- 3. 缓存条目 exp 已过期：命中即拒绝，不重解密 ----
+def test_cached_expired_rejected():
+    fresh_cache()
+
+    class NeverJWT:
+        calls = 0
+
+        def __init__(self, secret, algorithm):
+            pass
+
+        def decode(self, token):
+            NeverJWT.calls += 1
+            raise AssertionError("不应触发解码（命中缓存即应拒绝）")
+
+    auth_mod.JWT = NeverJWT
+    # 预置一个 exp 已过期的缓存条目（模拟缓存写入后 token 到期）
+    key = auth_mod._cache_key("expired-token", SECRET, ALGORITHM)
+    auth_mod._jwt_cache.set(
+        key, {"payload": {"sub": "u1"}, "exp": int(time.time()) - 10}, expire=60
+    )
+    for i in range(2):
+        try:
+            jwt_auth_dependency(credentials=make_creds("expired-token"))
+            check(f"缓存过期条目第 {i+1} 次应 401", False, "未抛异常")
+        except Exception as e:
+            check(f"缓存过期条目第 {i+1} 次 401", getattr(e, "status_code", None) == 401, str(e))
+    check("过期条目命中后未重解密（两次都直接拒绝）", NeverJWT.calls == 0, f"calls={NeverJWT.calls}")
+    check("过期条目保留在缓存（TTL 内持续拒绝）", key in auth_mod._jwt_cache)
+
+
+# ---- 4. 无效 token 失败不缓存 ----
+def test_invalid_token_not_cached():
+    fresh_cache()
+
+    class BadJWT:
+        calls = 0
+
+        def __init__(self, secret, algorithm):
+            pass
+
+        def decode(self, token):
+            BadJWT.calls += 1
+            raise ValueError("Invalid signature")
+
+    auth_mod.JWT = BadJWT
+    creds = make_creds("bad-token")
+    for _ in range(2):
+        try:
+            jwt_auth_dependency(credentials=creds)
+            check("无效 token 应 401", False)
+        except Exception as e:
+            ok = getattr(e, "status_code", None) == 401
+            check("无效 token 401", ok, str(e))
+    check("无效 token 每次都完整验证（失败不缓存）", BadJWT.calls == 2, f"calls={BadJWT.calls}")
+    check("缓存无失败条目", len(auth_mod._jwt_cache) == 0, str(len(auth_mod._jwt_cache)))
+
+
+# ---- 5. 缓存 key 含密钥/算法指纹 ----
+def test_key_scoped_to_secret():
+    fresh_cache()
+
+    class KeyJWT:
+        calls = 0
+
+        def __init__(self, secret, algorithm):
+            pass
+
+        def decode(self, token):
+            KeyJWT.calls += 1
+            return orjson.dumps({"sub": token, "exp": int(time.time()) + 3600}).decode()
+
+    auth_mod.JWT = KeyJWT
+    creds = make_creds("same-token")
+    jwt_auth_dependency(credentials=creds)                       # secret A
+    jwt_auth_dependency(credentials=creds)                       # secret A -> 命中
+    auth_mod._cached_decode("same-token", "secret-B", ALGORITHM)  # secret B -> miss
+    check("同 token 不同密钥独立缓存", KeyJWT.calls == 2, f"calls={KeyJWT.calls}")
+
+
+# ---- 6. 返回值独立对象：修改不影响缓存 ----
+def test_payload_object_independent():
+    fresh_cache()
+
+    class FixedJWT:
+        def __init__(self, secret, algorithm):
+            pass
+
+        def decode(self, token):
+            return orjson.dumps({"sub": "u1", "role": "admin", "exp": int(time.time()) + 3600}).decode()
+
+    auth_mod.JWT = FixedJWT
+    p1 = jwt_auth_dependency(credentials=make_creds("t-copy"))
+    p1["role"] = "hacker"
+    p2 = jwt_auth_dependency(credentials=make_creds("t-copy"))
+    check("修改返回值不污染缓存", p2["role"] == "admin", str(p2))
+
+
+# ---- 7. 端到端：真实 JWT 签发 -> 鉴权 -> 命中缓存 ----
+def test_e2e_real_jwt():
+    fresh_cache()
+    auth_mod.JWT = JWT  # 恢复真实 JWT 类
+    jwt = JWT(SECRET, ALGORITHM)
+    token = jwt.encode(orjson.dumps({"sub": "1", "username": "alice"}).decode(), expires_in_seconds=3600)
+    creds = make_creds(token)
+    p1 = jwt_auth_dependency(credentials=creds)
+    p2 = jwt_auth_dependency(credentials=creds)
+    check("真实 JWT 鉴权成功", p1["sub"] == "1" and p1["username"] == "alice", str(p1))
+    check("二次调用命中缓存", auth_mod._cache_key(token, SECRET, ALGORITHM) in auth_mod._jwt_cache)
+    try:
+        jwt_auth_dependency(credentials=None)
+        check("缺 header 应 401", False)
+    except Exception as e:
+        check("缺 header 401", getattr(e, "status_code", None) == 401)
+
+
+test_cache_hit_and_isolation()
+test_ttl_expiry_redecodes()
+test_cached_expired_rejected()
+test_invalid_token_not_cached()
+test_key_scoped_to_secret()
+test_payload_object_independent()
+test_e2e_real_jwt()
+
+_tmp.cleanup()
+
+print()
+if failures:
+    print(f"共 {len(failures)} 项失败: {failures}")
+    sys.exit(1)
+print("全部通过")

test_logging_async.py

@@ -0,0 +1,143 @@
+"""
+access 日志采样 + 异步日志队列（QueueHandler + QueueListener）测试
+
+验证:
+  1. _should_log_access：错误（>=400）全量；成功按采样率（确定性计数）；1.0 全量
+  2. QueueHandler + QueueListener：logger 只挂队列 handler，I/O 由后台监听线程落盘
+  3. 队列满不抛异常（丢弃而非阻塞/打断业务）
+  4. 端到端：真实 app 请求 -> access 日志异步落盘（含采样命中）
+
+运行: uv run python test_logging_async.py
+"""
+import importlib.util
+import itertools
+import queue
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent / "src"))
+sys.path.insert(0, str(Path(__file__).resolve().parent / "utils" / "worker" / "src"))
+
+_tmp = tempfile.TemporaryDirectory()
+
+# 预加载真实 logger 模块（不触发 chongming 包链导入），把日志目录重定向到
+# 临时目录，之后 import chongming.app 时 get_logger 即用新路径
+_logger_path = Path(__file__).resolve().parent / "src" / "chongming" / "app" / "core" / "logger.py"
+_logger_spec = importlib.util.spec_from_file_location("chongming.app.core.logger", _logger_path)
+logger_mod = importlib.util.module_from_spec(_logger_spec)
+sys.modules["chongming.app.core.logger"] = logger_mod
+_logger_spec.loader.exec_module(logger_mod)  # type: ignore[union-attr]
+logger_mod.log_config["file"] = _tmp.name
+
+import chongming.app as app_mod
+
+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 _wait_for(pred, timeout=3.0):
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if pred():
+            return True
+        time.sleep(0.05)
+    return False
+
+
+# ---- 1. 采样逻辑 ----
+def test_sampling():
+    app_mod._access_log_counter = itertools.count()
+    check(
+        "错误 4xx/5xx 全量",
+        app_mod._should_log_access(404)
+        and app_mod._should_log_access(500)
+        and app_mod._should_log_access(503),
+    )
+    app_mod._access_log_counter = itertools.count()
+    check("rate>=1.0 全量", app_mod._should_log_access(200, 1.0) and app_mod._should_log_access(200, 1.0))
+    app_mod._access_log_counter = itertools.count()
+    logged = sum(1 for _ in range(10) if app_mod._should_log_access(200, 0.1))
+    check("0.1 采样率 10 个成功请求记 1 条", logged == 1, f"logged={logged}")
+
+
+# ---- 2. 异步队列：logger 只挂队列 handler，监听线程落盘 ----
+def test_async_queue_write():
+    from logging.handlers import QueueHandler
+
+    lg = logger_mod.get_logger("test_async")
+    check("logger 只挂队列 handler", len(lg.handlers) == 1 and isinstance(lg.handlers[0], QueueHandler))
+    check("监听线程已启动", logger_mod._listener is not None and logger_mod._listener._thread.is_alive())
+
+    for i in range(5):
+        lg.info("async log %d", i)  # % 参数化：格式化在监听线程执行
+    ok = _wait_for(
+        lambda: any(
+            "async log 4" in f.read_text(errors="ignore")
+            for f in Path(_tmp.name).glob("test_async-*.log")
+        )
+    )
+    check("日志异步落盘（监听线程写入）", ok)
+
+
+# ---- 3. 队列满：丢弃而非抛异常 ----
+def test_queue_full_drops():
+    old = logger_mod._log_queue
+    logger_mod._log_queue = queue.Queue(maxsize=2)  # 换小队列
+    try:
+        lg = logger_mod.get_logger("test_full")
+        try:
+            for i in range(20):
+                lg.info("flood %d", i)
+            check("队列满不抛异常（丢弃新日志）", True)
+        except Exception as e:
+            check("队列满不抛异常（丢弃新日志）", False, str(e))
+    finally:
+        logger_mod._log_queue = old
+
+
+# ---- 4. 端到端：中间件采样 + 异步落盘 ----
+def test_e2e_middleware():
+    import asyncio
+    import httpx
+
+    app_mod._access_log_counter = itertools.count()  # 重置采样计数
+
+    async def main():
+        transport = httpx.ASGITransport(app=app_mod.app)
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
+            r = await ac.get("/")
+        return r.status_code
+
+    code = asyncio.run(main())
+    check("请求返回 200", code == 200, str(code))
+    ok = _wait_for(
+        lambda: any(
+            "GET / status=200" in f.read_text(errors="ignore")
+            for f in Path(_tmp.name).glob("app-*.log")
+        )
+    )
+    check("access 日志异步落盘（采样命中）", ok)
+
+
+test_sampling()
+test_async_queue_write()
+test_queue_full_drops()
+test_e2e_middleware()
+
+# 关闭监听线程并释放文件句柄，再清理临时目录
+logger_mod.shutdown_logging()
+_tmp.cleanup()
+
+print()
+if failures:
+    print(f"共 {len(failures)} 项失败: {failures}")
+    sys.exit(1)
+print("全部通过")

test_mqtt_auth.py

@@ -10,7 +10,7 @@ MQTT Broker 认证集成测试
   C. MQTTBridge 客户端凭据传递（含特殊字符密码的 URI 编码）
 """
 import asyncio
-import json
+import orjson
 import multiprocessing
 import socket
 import sys
@@ -143,7 +143,7 @@ async def scenario_bridge(port: int):
         assert m.topic == "chongming/demo/op", f"未收到请求: {m.topic}"
         m2 = await asyncio.wait_for(sub.deliver_message(), timeout=10)
         assert m2.topic == "chongming/demo/op/response", f"未收到响应: {m2.topic}"
-        assert json.loads(m2.data) == {"ok": True}, f"响应内容异常: {m2.data}"
+        assert orjson.loads(m2.data) == {"ok": True}, f"响应内容异常: {m2.data}"
         await bridge.stop()
         await pub.disconnect()
         print("PASS [C] MQTTBridge 凭据传递: 连接/路由/响应链路正常（含特殊字符密码）")

test_mqtt_custom_topics.py

@@ -7,7 +7,7 @@ MQTT Bridge 自定义订阅主题单元测试
   3. 内置 chongming/... 主题兜底逻辑不变
 """
 import asyncio
-import json
+import orjson
 import sys
 from pathlib import Path
 from unittest.mock import MagicMock
@@ -24,7 +24,7 @@ from chongming.app.core.mqtt_bridge import (
 def make_msg(topic, payload):
     m = MagicMock()
     m.topic = topic
-    m.data = json.dumps(payload).encode("utf-8")
+    m.data = orjson.dumps(payload)
     return m
 
 

test_mqtt_e2e.py

@@ -12,7 +12,7 @@ MQTT 端到端测试脚本
 """
 
 import asyncio
-import json
+import orjson
 import sys
 import logging
 from amqtt.client import MQTTClient
@@ -51,7 +51,7 @@ async def run_test():
     for i, case in enumerate(TEST_CASES):
         a, b, expected = case["a"], case["b"], case["expected"]
         payload = {"meta": {"method": "MQTT"}, "data": {"a": a, "b": b}}
-        payload_bytes = json.dumps(payload).encode("utf-8")
+        payload_bytes = orjson.dumps(payload)
 
         await client.publish(REQUEST_TOPIC, payload_bytes, QOS_1)
         print(f"📤 [{i+1}/{len(TEST_CASES)}] 发送: {REQUEST_TOPIC} → {payload}")
@@ -59,14 +59,14 @@ async def run_test():
         # 等待响应（超时 5 秒）
         try:
             message = await asyncio.wait_for(client.deliver_message(), timeout=5.0)
-            data = json.loads(message.data.decode("utf-8"))  # type: ignore
+            data = orjson.loads(message.data)  # type: ignore
             actual = data.get("result")
             status = "✅ PASS" if abs(actual - expected) < 0.01 else "❌ FAIL"
             if status == "✅ PASS":
                 PASS += 1
             else:
                 FAIL += 1
-            print(f"📥 [{i+1}/{len(TEST_CASES)}] 响应: {json.dumps(data)} → {status} (expect={expected}, got={actual})")
+            print(f"📥 [{i+1}/{len(TEST_CASES)}] 响应: {orjson.dumps(data).decode()} → {status} (expect={expected}, got={actual})")
         except asyncio.TimeoutError:
             FAIL += 1
             print(f"📥 [{i+1}/{len(TEST_CASES)}] ❌ FAIL: 超时无响应")

test_passthrough_optimization.py

@@ -61,6 +61,15 @@ _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]
 
+# 加载真实 call_stats（route_generator 依赖其 spawn/record_call）
+_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]
+
 # 加载真实 route_generator
 _rg_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "route_generator.py"
 _rg_spec = importlib.util.spec_from_file_location(

test_stream_route.py

@@ -50,6 +50,15 @@ _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]
 
+# 加载真实 call_stats（route_generator 依赖其 spawn/record_call）
+_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]
+
 # 加载真实 route_generator
 _rg_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "route_generator.py"
 _rg_spec = importlib.util.spec_from_file_location(

uv.lock

@@ -486,7 +486,7 @@ source = { git = "http://124.71.68.6:3000/chongming/chongming-stream-manager.git
 [[package]]
 name = "chongming-worker"
 version = "0.3.0"
-source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#c4a998084fc42572e121a27b84082579e66dd3d0" }
+source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#7e4a3f66b30ee7ab56b2b066790a029d4a4ed6e8" }
 dependencies = [
     { name = "eclipse-zenoh" },
     { name = "minio" },
@@ -514,6 +514,7 @@ dependencies = [
     { name = "eclipse-zenoh" },
     { name = "fastapi" },
     { name = "gunicorn" },
+    { name = "httptools" },
     { name = "module-bank" },
     { name = "openai" },
     { name = "orjson" },
@@ -537,8 +538,8 @@ dev = [
 
 [package.metadata]
 requires-dist = [
-    { name = "aiofiles", specifier = ">=25.1.0" },
-    { name = "amqtt", specifier = ">=0.11.3" },
+    { name = "aiofiles" },
+    { name = "amqtt" },
     { name = "bytedance", git = "http://124.71.68.6:3000/chongming/bytedance.git" },
     { name = "chongming-cache", git = "http://124.71.68.6:3000/chongming/chongming-utils-cache.git" },
     { name = "chongming-config", git = "http://124.71.68.6:3000/chongming/chongming-utils-config.git" },
@@ -552,6 +553,7 @@ requires-dist = [
     { name = "eclipse-zenoh" },
     { name = "fastapi", extras = ["uvicorn"] },
     { name = "gunicorn" },
+    { name = "httptools" },
     { name = "module-bank" },
     { name = "openai" },
     { name = "orjson" },
@@ -560,8 +562,8 @@ requires-dist = [
     { name = "tomli" },
     { name = "typing-extensions" },
     { name = "uvicorn" },
-    { name = "uvloop", specifier = ">=0.22.1" },
-    { name = "volcengine-python-sdk", extras = ["ark"], specifier = ">=5.0.41" },
+    { name = "uvloop" },
+    { name = "volcengine-python-sdk", extras = ["ark"] },
 ]
 
 [package.metadata.requires-dev]
@@ -793,6 +795,42 @@ wheels = [
     { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/d2/74/d370e55600d9bcfa0d9794b0166126d49291a3d2b20c268fc98c453a4948/httpcore2-2.12.0-py3-none-any.whl", hash = "sha256:7e04258ce01013d7d615e5b910a3b27fac937d7a95038227e79652b4ba3b4ceb", size = 83074, upload-time = "2026-08-18T13:22:05.854Z" },
 ]
 
+[[package]]
+name = "httptools"
+version = "0.8.0"
+source = { registry = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple" }
+sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/43/e5/d471fcb0e14523fe1c3f4ba58ca52480e7bd70ad7109a3846bc75892f7fb/httptools-0.8.0.tar.gz", hash = "sha256:6b2a32f18d97e16e90827d7a819ffa8dbd8cc245fc4e1fa9d1095b54ef4bd999", size = 271342, upload-time = "2026-05-25T22:17:48.841Z" }
+wheels = [
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/14/88/1d21a36da8f5cb0fa49eafd4b169eba5608d57e75bbcf61845cbc6243216/httptools-0.8.0-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:880490234c10f70a9830743097e8958d6e4b9f5a0ffc24515023afeef984054d", size = 208247, upload-time = "2026-05-25T22:17:07.843Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/a5/42/cc4feea2945cb3051038f090c9b36bd5b8a9d7f5a894a506a8983e33fd1c/httptools-0.8.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:5931891fb7b441b8a3853cf1b85c82c903defce084dd5f6771ca46e31bf862c5", size = 113064, upload-time = "2026-05-25T22:17:09.136Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/e3/a6/febbb8b8db0f58b38e44ad6cb946e6a255ae49b55f2e8543408fb7501ccd/httptools-0.8.0-cp312-cp312-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:b15fc622b0f869d19207c4089a501d9bcc63ca5e071ffdd2f03f922df882dcb2", size = 523851, upload-time = "2026-05-25T22:17:10.106Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/b7/e4/f90a0df0b83beff265b7e3b65f2a4cefd95792d4be0ac3e16049f2acd3c2/httptools-0.8.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:425f83884fd6343828d8c565f046cb72b6d19063f6924093e11bcd8e1548cd09", size = 518842, upload-time = "2026-05-25T22:17:11.218Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/9e/2d/0c9ac76dd2c893841fbf6498d6acec4f2442e1b7067f6e3e316a80e494e8/httptools-0.8.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:ef7c3c97f4311c7be57e2986629df89d49cb434dbff78eafcd48c2bff986b15a", size = 501238, upload-time = "2026-05-25T22:17:12.728Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/ca/42/906adc91ae3a5fa9c59c0a2f21c139725bd7e5b41ae6acd485cd14123ebf/httptools-0.8.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:a1afd7c9fbff0d9f5d489c4ce2768bd09c84a46ddefc7161e6aa82ae35c85745", size = 509567, upload-time = "2026-05-25T22:17:13.842Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/05/0b/4240efeb672751ee5b9b380cb0e3fdc050bc05f68adc7a8aefc4fcd9a69a/httptools-0.8.0-cp312-cp312-win_amd64.whl", hash = "sha256:cd96f29b4bab1d42fa6e3d008711c75e0f79e94e06827330160e3a304227f150", size = 90918, upload-time = "2026-05-25T22:17:15.155Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/5e/e5/8cfcabc5546e8022f168be28bcdaa128a240a0befdd03b59d558b4f18bd6/httptools-0.8.0-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:614ceea8ea606848bece2338ac03b3ce5324bcb4be8dc7d377ed708012fa4db8", size = 205148, upload-time = "2026-05-25T22:17:16.333Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/2a/0e/0fb14848c19a686c8062ff9067c1a48793e3224b47bc5b201535b6036fce/httptools-0.8.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:2d689918c15a013c65ef52d9fd495d766893ab831a2c8d89f2ac5940a5df847c", size = 111368, upload-time = "2026-05-25T22:17:17.586Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/2e/1b/46f1cecf06b9bbde8e4b8c88034ac7908989e5ff7a3a388ef38392949c1f/httptools-0.8.0-cp313-cp313-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:eb3028cca2fc0a6d720e52ef61d8ebb62fcbfeb1de56874546d858d3f25a26b7", size = 486447, upload-time = "2026-05-25T22:17:18.564Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/77/00/258bfc0837221f81d9725c45f9b948a6a6b2994a147a4fb66e85100c668f/httptools-0.8.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:88bdd940f2b5d487b4d032c6afa5489a7dc4694410d43de3c38c4fb3af0dc45d", size = 482448, upload-time = "2026-05-25T22:17:19.912Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/04/ab/d1cef3b5523f4d272a70f42a776c3169a2dddfe3a54de4b2ce4a36341528/httptools-0.8.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:6a43c9dd399758ccc0531acb0a3c4a6c299ee893ee9400e9c893b7bdcfae0681", size = 464460, upload-time = "2026-05-25T22:17:20.882Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/ce/48/5d1d072442277bb2b3434e0e60690b8e8c23840ef7de8b6ea54040a536d3/httptools-0.8.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:0770728beb05094c809b98e814edff5fef69d26ad7d21185f2f6d5884a0ba683", size = 471312, upload-time = "2026-05-25T22:17:22.085Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/0d/66/b96623b27e51a68199ef4efdda0613cced9233fe3062ac74e50749c5ad37/httptools-0.8.0-cp313-cp313-win_amd64.whl", hash = "sha256:7685df791fad561384bfb139e77fde27a1ffd93134e016f95a0db424ffbf77b1", size = 90117, upload-time = "2026-05-25T22:17:23.074Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/1a/12/fa3fbf5f9517b273edea2dc982aa82a8c634091e67c590792b729017bc6f/httptools-0.8.0-cp314-cp314-macosx_10_13_universal2.whl", hash = "sha256:de242a49b5d18e0a8776e654e9f6bf6d89f3875a5c35b425a0e7ce940feb3fd6", size = 206183, upload-time = "2026-05-25T22:17:24.004Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/30/fc/5e7c4cb443370f2090a3aba0453a07384d29ff66b7435bb90e77e1037599/httptools-0.8.0-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:159e9ab5f701ccd42e555a12f1ad8ff69702910fc1c996cf2bb66e5fcb7a231b", size = 112079, upload-time = "2026-05-25T22:17:25.216Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/ba/53/771bd891eb0f236f32145d6a1775777ec85745f3cc983a1f23d1a3b8ddfe/httptools-0.8.0-cp314-cp314-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:c4a9f1707e4823d54dfec6c33fa3697d302aed536ed352a7ebb5a061ddb869d0", size = 481596, upload-time = "2026-05-25T22:17:26.186Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/62/42/94e15bc68ce3d423243c45d7f1b0c7561f13844f97dc52ae23182fb65628/httptools-0.8.0-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:d76ad7b951387e3632c8716a9bb03ac5b45c5f16119aa409db0459520887944e", size = 480865, upload-time = "2026-05-25T22:17:27.542Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/1c/7c/fe2980fc03723272e30f135b62360b075f513dfe7cc73aef36c7f04012bd/httptools-0.8.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:a3b7387147361c3fd47a0bde763c5c91b5b4cd4dc9989b8ece84ff436c99843b", size = 463189, upload-time = "2026-05-25T22:17:28.546Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/15/1b/47fc5fff68acd1bfa20b4734059c9a06cadb88119dcd5258b5b0d21d91c8/httptools-0.8.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:f256d6ce930c52ca1cb2a960b7da03548c454e7d28b06059ad41bfe789036ce0", size = 466610, upload-time = "2026-05-25T22:17:29.816Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/60/bd/07b13c93ffd9bec9546e0d43f8e19378dd696dbd278511406bc07371ef1f/httptools-0.8.0-cp314-cp314-win_amd64.whl", hash = "sha256:19d1ee275bb59ba2643ba9a3a1e51cc0c788caf2b8df506368e03f56fdd08527", size = 92705, upload-time = "2026-05-25T22:17:31.133Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/fd/c4/121648f68ce066d7bd762d6b6d97e620847642d38d54f3d90ff11d947629/httptools-0.8.0-cp314-cp314t-macosx_10_13_universal2.whl", hash = "sha256:de1ed58a974e75d56560acc7e7fed01a454994429456f65209789992e41f2568", size = 215023, upload-time = "2026-05-25T22:17:32.401Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/b9/b0/312a062ae741ae3e8baa8c8bf20be81b2e67337b259ab4349bebc7b6142e/httptools-0.8.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:e93c227b595c6926c1acee96891dd9da4be338cfbe82e5cd3bb9d8dd7dc4ac0b", size = 117405, upload-time = "2026-05-25T22:17:33.742Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/fc/37/fccd705f795386bb05bf413012fecff2a33e5aa8c2f069096de3e9fd8702/httptools-0.8.0-cp314-cp314t-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:2a021c3a8e65cc125390d72f59b968afca3bdcaff25bd67965e0a055a14946ca", size = 558497, upload-time = "2026-05-25T22:17:34.732Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/bd/39/f172e8003576de35f5ba77ff417cf0e34429d35dc014deef15afa337a72c/httptools-0.8.0-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:48774d39cbb70e2b1f71f88852a3087ae1d3a1eb80482bb48c13067ab080c14f", size = 571585, upload-time = "2026-05-25T22:17:35.813Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/3e/b9/f5564760af99f3dbbf3f9104dc00e5da27e96cf433c6bdcf77617f70bf3f/httptools-0.8.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:88eead8ec8680a9f146c655bc88445a325bd7921cfd8194c7337e9467282427d", size = 543297, upload-time = "2026-05-25T22:17:37.08Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/99/67/8d9f2c313618e161b82f3873188e7196126da1d6e29688df40eb3997c77a/httptools-0.8.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:2c032fa028f46871ec7e1fc59fc15e8023eab3e6bbe6ece786a1611719a5d081", size = 539535, upload-time = "2026-05-25T22:17:38.032Z" },
+    { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/48/63/b906c01e53f50d432c0defe43ce52764a111dc1bdd028bafbeb54dcfd008/httptools-0.8.0-cp314-cp314t-win_amd64.whl", hash = "sha256:384c17174464c8e873398b7af24f0b1f44d992c820328413951a625323155d77", size = 108209, upload-time = "2026-05-25T22:17:39.473Z" },
+]
+
 [[package]]
 name = "httpx"
 version = "0.28.1"