perf: 优化请求鉴权链路并批量化统计/调度查询

- 为 Pipeline/Context 增加按需读取请求体能力，支持 async 懒加载 JSON body
- 为 chat/cli/video/claude/openai-cli 适配器关闭默认预读，减少无效 body 读取与超时风险
- 将本地登录、JWT 用户加载、API Key 鉴权迁移到线程池隔离会话执行，避免阻塞事件循环
- 为 API Key 鉴权返回结构化余额结果，并在主请求会话中重新绑定 user/api_key 后再校验状态、过期和锁定信息
- 为 management/user token 前缀认证引入独立会话与结果回绑，避免跨会话对象写入失效

- 为 Usage 余额检查补充结构化返回，统一透出 remaining 与欠费/不可用文案映射
- 为用户与管理端活跃请求查询增加 maintain_status 开关，避免轮询指定 id 时误触发状态修复
- 重写 user_me usage 汇总逻辑，支持 group_by=None 的粗粒度聚合
- 修正 provider 维度成功率与平均响应时间统计，基于 success_count 和成功响应耗时汇总计算
- 前端 Usage 轮询由 setInterval 改为串行 setTimeout，避免并发轮询叠加

- 为 StatsAggregator 增加按本地日期批量计算百分位能力，替代逐天 fan-out 查询
- 为混合统计查询合并连续实时日期区间，并批量读取 StatsDaily，减少逐日查询次数
- 为用户日统计增加批量聚合入口，替代逐用户循环聚合
- 为系统配置导出改用 selectinload 预加载 provider 关联数据，减少 N+1 查询
- 为管理员用户列表增加钱包批量查询，避免逐用户回表

- 为调度器增加 provider 轻量引用预过滤，先按 allowed_providers 缩小范围再加载完整 provider 图
- 为 CandidateBuilder 增加 provider refs/provider_ids 查询能力，保留分页顺序
- 为模型缓存增加 provider_model_mappings 索引缓存与 model_mappings 规则缓存，减少重复全量扫描
- 为请求候选中间态改为 flush/batch commit，降低 pending/streaming 状态切换的事务往返
- 为钱包访问结果补充 balance_snapshot，并抽取余额快照复用逻辑

- 补充 pipeline、auth、admin users、user_me usage、stats aggregator、model cache、
  scheduler、wallet、request candidate 等回归与契约测试

tests/unit/test_api_request_context.py

@@ -11,6 +11,13 @@ from src.api.base.context import ApiRequestContext
 
 
 def _build_request(headers: dict[str, str] | None = None) -> Request:
+    return _build_request_with_body(b"", headers=headers)
+
+
+def _build_request_with_body(
+    body: bytes,
+    headers: dict[str, str] | None = None,
+) -> Request:
     header_items = [
         (str(key).encode("latin-1"), str(value).encode("latin-1"))
         for key, value in (headers or {}).items()
@@ -28,8 +35,14 @@ def _build_request(headers: dict[str, str] | None = None) -> Request:
         "server": ("testserver", 80),
     }
 
+    received = False
+
     async def receive() -> dict[str, object]:
-        return {"type": "http.request", "body": b"", "more_body": False}
+        nonlocal received
+        if received:
+            return {"type": "http.request", "body": b"", "more_body": False}
+        received = True
+        return {"type": "http.request", "body": body, "more_body": False}
 
     request = Request(scope, receive)
     request.state.perf_metrics = {}
@@ -89,3 +102,22 @@ class TestApiRequestContextEnsureJsonBody:
 
         assert context.client_content_encoding == "gzip"
         assert context.client_accept_encoding == "gzip, deflate"
+
+    @pytest.mark.asyncio
+    async def test_ensure_json_body_async_loads_body_lazily(self) -> None:
+        payload = {"message": "hello", "count": 2}
+        request = _build_request_with_body(json.dumps(payload).encode("utf-8"))
+        context = ApiRequestContext.build(
+            request=request,
+            db=None,  # type: ignore[arg-type]
+            user=None,
+            api_key=None,
+            raw_body=None,
+        )
+
+        assert context.raw_body is None
+
+        result = await context.ensure_json_body_async()
+
+        assert result == payload
+        assert context.raw_body == json.dumps(payload).encode("utf-8")

tests/unit/test_request_candidate_intermediate_status.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from src.services.request.candidate import RequestCandidateService
+
+
+def _build_db_with_candidate(candidate: SimpleNamespace) -> MagicMock:
+    query = MagicMock()
+    query.filter.return_value.first.return_value = candidate
+
+    db = MagicMock()
+    db.query.return_value = query
+    db.info = {"managed_by_middleware": True}
+    return db
+
+
+def test_mark_candidate_started_flushes_without_immediate_commit() -> None:
+    candidate = SimpleNamespace(status="available", started_at=None)
+    db = _build_db_with_candidate(candidate)
+
+    RequestCandidateService.mark_candidate_started(db, "candidate-1")
+
+    assert candidate.status == "pending"
+    assert candidate.started_at is not None
+    db.flush.assert_called_once()
+    db.commit.assert_not_called()
+
+
+def test_mark_candidate_streaming_flushes_without_immediate_commit() -> None:
+    candidate = SimpleNamespace(status="pending", concurrent_requests=None)
+    db = _build_db_with_candidate(candidate)
+
+    RequestCandidateService.mark_candidate_streaming(db, "candidate-1", concurrent_requests=3)
+
+    assert candidate.status == "streaming"
+    assert candidate.concurrent_requests == 3
+    db.flush.assert_called_once()
+    db.commit.assert_not_called()