fix: run bootstrap via module to avoid stdlib http shadowing

Switch container startup from file execution to module execution so
urllib can import stdlib http.client reliably.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

.env.example

@@ -46,6 +46,11 @@ DEFAULT_MODEL=org_auto
 # 默认模式：chat 或 agent
 DEFAULT_ASK_MODE=chat
 
+# 请求侧 tools/tool_choice 透传到 Lingma（默认开启，可显式关闭）
+TOOL_FORWARD_ENABLED=true
+# 可选：允许透传的工具名白名单，逗号分隔；为空表示不额外限制
+TOOL_ALLOWLIST=
+
 # 专属域（可选）
 DEDICATED_DOMAIN_URL=
 

.gitignore

@@ -7,3 +7,4 @@ data/*
 !data/.gitkeep
 secrets/*
 !secrets/.gitkeep
+.gitnexus

.omc/plans/app-main-split-plan.md

@@ -0,0 +1,353 @@
+# app/main.py 渐进拆分计划
+
+- 日期：2026-04-21
+- 目标文件：`app/main.py`
+- 当前判断：**适合拆分，但不适合一次性大拆；建议按阶段渐进拆分**。
+
+## 1. 目标
+
+把 `app/main.py` 从“单文件总编排”逐步收敛为“组合根 + 路由/辅助模块”，在不破坏以下关键行为的前提下，降低文件复杂度并提高后续维护性：
+
+- OpenAI / Anthropic / Responses 三条协议路径行为一致
+- session cache 命中、回写、失效语义保持不变
+- 单请求固定实例绑定不变
+- streaming 路径中的 in-flight ticket 释放语义不变
+- SSE 帧格式、finish reason / stop reason 行为不变
+- 现有测试尽量少改，尤其避免首轮就大面积修改对 `app.main` 的 patch 点
+
+## 2. 当前结构判断
+
+`app/main.py` 当前可以分成这些职责块：
+
+1. **应用启动与全局装配**
+   - `app/main.py:46-154`
+   - 包括 `settings`、`pool`、`stats_collector`、`chat_guard`、`session_cache`、`lifespan`、middleware
+
+2. **鉴权包装与告警**
+   - `app/main.py:157-196`
+
+3. **健康检查与通用请求辅助逻辑**
+   - `app/main.py:199-353`
+
+4. **共享 tool / stream / bridge helper**
+   - `app/main.py:356-752`
+
+5. **OpenAI Chat 主编排**
+   - `app/main.py:769-1192`
+
+6. **Responses API 适配层**
+   - `app/main.py:1197-1640`
+
+7. **Anthropic Messages 适配层**
+   - `app/main.py:1679-2180`
+
+8. **admin / internal / metrics 路由**
+   - `app/main.py:2183-2356`
+
+## 3. 风险判断
+
+### 3.1 高风险区域（第一阶段不要碰）
+
+以下区域**不建议作为第一刀拆分目标**：
+
+1. `app/main.py:906` 左右的 OpenAI streaming generator
+2. `app/main.py:1886` 左右的 Anthropic streaming generator
+3. `v1_chat_completions` 主编排逻辑
+4. `v1_messages` 主编排逻辑
+5. session cache lookup / write-back / invalidate 的共享编排逻辑
+
+### 3.2 原因
+
+这些区域都同时依赖：
+
+- route-local 状态
+- `pool` / `chat_guard` / `session_cache` / `stats_collector`
+- session continuity
+- 流式 finally 中的 ticket 释放与写回时机
+- OpenAI / Anthropic / Responses 之间的共享行为约束
+
+这类代码即使功能不变，单纯移动位置也容易引发细微回归。
+
+## 4. 建议的目标结构
+
+建议最终逐步演进到以下结构：
+
+```text
+app/
+  main.py                      # 组合根：app 创建、lifespan、router 注册、共享单例
+  http/
+    lifecycle.py               # middleware / startup posture / pool guards（可后置）
+    chat_shared.py             # 跨协议的 prompt/tool/stream helper
+    openai_chat.py             # /v1/chat/completions
+    openai_responses.py        # /responses 与 /v1/responses
+    anthropic_messages.py      # /v1/messages* 与 anthropic helper
+    admin_routes.py            # /internal/*, /metrics, /healthz, /v1/models（按需要划分）
+```
+
+> 注意：这个结构是**目标结构**，不是第一阶段必须一步到位完成的结构。
+
+## 5. 分阶段执行计划
+
+### Phase 0：保护性准备（只做分析，不改行为）
+
+目标：为后续拆分建立安全边界。
+
+动作：
+
+1. 梳理并固定当前回归验证命令
+   - `python3 -m unittest tests/test_tool_call_bridge.py`
+   - `python3 -m unittest discover -s tests -p "test_*.py"`
+
+2. 在实际动代码前，对准备修改的关键符号做 impact analysis
+   - 尤其是：
+     - `v1_chat_completions`
+     - `v1_messages`
+     - `_messages_to_prompt`
+     - `_responses_to_chat_request`
+     - `_openai_tool_call`
+     - `_anthropic_tool_use_block`
+
+3. 先确认测试里对 `app.main` 的 patch 点，避免首轮拆分后直接把测试打碎
+
+完成标准：
+- 有固定回归命令
+- 清楚哪些符号必须在首轮保留兼容出口
+
+---
+
+### Phase 1：提取纯 helper（最低风险）
+
+目标：在不改主路由编排的前提下，先减轻 `app/main.py` 的噪音和长度。
+
+建议新文件：
+
+#### 1) `app/http/tool_bridge.py`
+建议迁移函数：
+- `_json_string`
+- `_openai_forced_tool_name`
+- `_anthropic_forced_tool_name`
+- `_json_object_from_text`
+- `_tool_code_single_arg_name`
+- `_tool_code_object_from_text`
+- `_forced_tool_event_from_text`
+- `_openai_tool_call`
+- `_anthropic_tool_use_block`
+- `_anthropic_tool_result_block`
+
+#### 2) `app/http/responses_adapter.py`
+建议迁移函数：
+- `_responses_input_to_messages`
+- `_responses_to_chat_request`
+- `_responses_id_from_chat_id`
+- `_responses_usage_from_chat`
+- `_responses_non_stream_from_chat_payload`
+- `_sse_data`
+
+#### 3) `app/http/tool_policy.py`（可选）
+如果首轮还想再减一点，可迁移：
+- `_include_usage`
+- `_tool_allowlist`
+- `_openai_tool_name`
+- `_anthropic_tool_name`
+- `_filter_allowed_tools`
+- `_ensure_tool_choice_allowed`
+- `_openai_tool_config`
+- `_anthropic_tool_config`
+- `_openai_has_tooling_context`
+- `_anthropic_content_has_tool_blocks`
+- `_anthropic_has_tooling_context`
+- `_resolve_ask_mode`
+
+首轮兼容策略：
+- `app.main` 中先保留同名导入出口，例如：
+  - `from .http.tool_bridge import _openai_tool_call, ...`
+- 这样即使测试仍然 patch `app.main._openai_tool_call`，改动面也最小
+
+完成标准：
+- `app/main.py` 明显变短
+- 路由逻辑不变
+- 现有测试全过
+- 首轮不改 streaming 主体
+
+---
+
+### Phase 2：提取 Responses 路由（低到中风险）
+
+目标：把 `/responses` 和 `/v1/responses` 的适配层单独放出去。
+
+建议新文件：
+- `app/http/openai_responses.py`
+
+建议包含：
+- `v1_responses`
+- `_responses_stream_from_chat_stream`
+- 以及它依赖的 responses helper（如果 Phase 1 已迁移则直接复用）
+
+注意事项：
+- `v1_responses` 当前是直接包装 `v1_chat_completions`
+- 拆分时优先保持这个关系不变，不要同步重构 chat 主路径
+- 如果测试直接 patch `main.v1_chat_completions`，则需要确保新模块仍从 `app.main` 可拿到兼容入口，或同步最小化调整测试
+
+完成标准：
+- `/responses` 逻辑从 `main.py` 分离
+- `v1_chat_completions` 仍保持原行为
+- responses 相关测试不回归
+
+---
+
+### Phase 3：提取 admin / health / metrics 路由（低风险）
+
+目标：把非核心协议路径先搬走。
+
+建议新文件：
+- `app/http/admin_routes.py`
+
+可迁移内容：
+- `healthz`
+- `v1_models`（可按需一起搬）
+- `/internal/auto-login/*`
+- `/internal/session/export`
+- `/internal/models/raw`
+- `/internal/stats`
+- `/metrics`
+
+注意事项：
+- 这些路由依赖全局 `settings` / `pool` / 鉴权 wrapper
+- 首轮可以通过“从 `main` 注入依赖”或“保留共享单例模块”来降低改动面
+
+完成标准：
+- 运营/admin 路由从主文件剥离
+- 对 chat/messages 主编排零行为影响
+
+---
+
+### Phase 4：提取 Anthropic 路由与 helper（中风险）
+
+目标：将 `/v1/messages*` 独立为单独模块。
+
+建议新文件：
+- `app/http/anthropic_messages.py`
+
+建议迁移：
+- `_anthropic_error`
+- `_anthropic_stop_reason`
+- `v1_messages_count_tokens`
+- `v1_messages`
+
+前提：
+- Phase 1 已把共享 tool / prompt / policy helper 先抽出
+- 已明确哪些共享状态通过参数传入，哪些保持模块共享
+
+注意：
+- 暂时不重构 Anthropic stream generator 内部逻辑，只做“整体迁移”而不是“逻辑改写”
+
+完成标准：
+- Anthropic 适配层从主文件分离
+- 与 OpenAI 的共享行为仍保持一致
+
+---
+
+### Phase 5：最后再考虑提取 OpenAI Chat 主路由（最高风险）
+
+目标：在前几阶段都稳定之后，再处理核心编排。
+
+建议新文件：
+- `app/http/openai_chat.py`
+
+建议迁移：
+- `v1_chat_completions`
+- 仅与其强耦合、且不适合保留在 `main.py` 的少量辅助逻辑
+
+关键原则：
+- 不要在这一阶段同时改 session/cache/streaming 逻辑
+- 只做“位置迁移 + 依赖显式化”
+- 如需引入 service 层，也要在这个阶段之后再单独评估，不要和文件拆分绑定进行
+
+完成标准：
+- `app/main.py` 基本收敛为组合根
+- 主编排仍行为一致
+- 全量测试通过
+
+## 6. 每阶段的验证要求
+
+每一阶段完成后，至少执行：
+
+```bash
+python3 -m unittest tests/test_tool_call_bridge.py
+python3 -m unittest discover -s tests -p "test_*.py"
+```
+
+如果本地服务可启动，建议补一轮 smoke：
+
+```bash
+uvicorn app.main:app --reload --port 8317
+curl -s http://127.0.0.1:8317/healthz
+```
+
+如果是改动了 `/responses` 或 `/v1/messages` 路径，应额外做协议 smoke，确认：
+- SSE 帧格式不变
+- stop reason / finish reason 不变
+- tool call / tool_use bridge 不变
+
+## 7. 兼容策略
+
+为减少首轮测试与调用方震荡，建议：
+
+1. **先迁移实现，再从 `app.main` re-export 同名符号**
+   - 例如：`from .http.responses_adapter import _responses_to_chat_request`
+2. 首轮不要改函数名
+3. 首轮不要顺手重命名模块级全局变量
+4. 首轮不要引入新的抽象层（例如 service / manager / context object）
+
+原则：
+- 第一轮目标是“降噪和减重”，不是“顺便重构架构”
+
+## 8. 不建议做的事
+
+以下动作不建议与本次拆分绑定：
+
+- 同时重写 streaming generator 内部结构
+- 同时改 session cache 语义
+- 同时改 pool / guard / stats 注入方式
+- 同时大改测试结构
+- 同时引入新的 service 层 / context 容器 / 抽象基类
+
+这些都应该是后续独立变更，不要混在第一次拆分里。
+
+## 9. 推荐的首个落地 PR 范围
+
+如果要开始实际实施，**建议第一批只做一个小 PR**：
+
+### PR-1：Helper extraction only
+
+内容：
+- 新增 `app/http/tool_bridge.py`
+- 新增 `app/http/responses_adapter.py`
+- `app/main.py` 改为导入这些 helper
+- 保留 `app.main` 的兼容出口
+- 不动 `v1_chat_completions` / `v1_messages` 的主逻辑
+
+预期收益：
+- `app/main.py` 先减少几百行
+- 风险最可控
+- 为后续路由级拆分打基础
+
+## 10. 后续记录方式
+
+建议后续每完成一个 phase，就在本文件底部追加一段进展记录，例如：
+
+```md
+## Progress Log
+- 2026-04-21: 创建拆分计划
+- 2026-04-22: 完成 Phase 1，抽离 responses helper 与 tool bridge helper
+- 2026-04-23: 运行全量 unittest 通过
+```
+
+这样后续可以持续在同一份计划上回填，不需要再重新整理上下文。
+
+## Progress Log
+- 2026-04-21: 创建拆分计划。
+- 2026-04-21: 完成 Phase 1 helper extraction，新增 `app/http/tool_bridge.py`、`app/http/responses_adapter.py`，并在 `app.main` 保留兼容导入出口。
+- 2026-04-21: 修复 Phase 1 后暴露的 tool bridge 回归；放宽 tool event allow 判断，仅在存在显式 tool 列表时做名称过滤，并保留 forced-tool 回退语义。
+- 2026-04-21: 调整 OpenAI 流式 forced-tool 回退，先缓冲 `tool_code` 文本，能解析为结构化 tool call 时只输出 `tool_calls` chunk，不能解析时再回放文本。
+- 2026-04-21: 验证通过：`python3 -m py_compile app/main.py app/http/tool_bridge.py app/http/responses_adapter.py`、`python3 -m unittest tests/test_tool_call_bridge.py`、`python3 -m unittest discover -s tests -p "test_*.py"`。

CLAUDE.md

@@ -93,3 +93,180 @@ Both protocols share the same backend pool, backpressure guard, stats, and sessi
 - Compose mounts:
   - `./data -> /app/data` (persistent Lingma binary/cache/workdirs)
   - `./secrets -> /secrets:ro` (session bundles, secrets)
+
+
+# CLAUDE.md
+
+Behavioral guidelines to reduce common LLM coding mistakes. Merge with project-specific instructions as needed.
+
+**Tradeoff:** These guidelines bias toward caution over speed. For trivial tasks, use judgment.
+
+## 1. Think Before Coding
+
+**Don't assume. Don't hide confusion. Surface tradeoffs.**
+
+Before implementing:
+- State your assumptions explicitly. If uncertain, ask.
+- If multiple interpretations exist, present them - don't pick silently.
+- If a simpler approach exists, say so. Push back when warranted.
+- If something is unclear, stop. Name what's confusing. Ask.
+
+## 2. Simplicity First
+
+**Minimum code that solves the problem. Nothing speculative.**
+
+- No features beyond what was asked.
+- No abstractions for single-use code.
+- No "flexibility" or "configurability" that wasn't requested.
+- No error handling for impossible scenarios.
+- If you write 200 lines and it could be 50, rewrite it.
+
+Ask yourself: "Would a senior engineer say this is overcomplicated?" If yes, simplify.
+
+## 3. Surgical Changes
+
+**Touch only what you must. Clean up only your own mess.**
+
+When editing existing code:
+- Don't "improve" adjacent code, comments, or formatting.
+- Don't refactor things that aren't broken.
+- Match existing style, even if you'd do it differently.
+- If you notice unrelated dead code, mention it - don't delete it.
+
+When your changes create orphans:
+- Remove imports/variables/functions that YOUR changes made unused.
+- Don't remove pre-existing dead code unless asked.
+
+The test: Every changed line should trace directly to the user's request.
+
+## 4. Goal-Driven Execution
+
+**Define success criteria. Loop until verified.**
+
+Transform tasks into verifiable goals:
+- "Add validation" → "Write tests for invalid inputs, then make them pass"
+- "Fix the bug" → "Write a test that reproduces it, then make it pass"
+- "Refactor X" → "Ensure tests pass before and after"
+
+For multi-step tasks, state a brief plan:
+```
+1. [Step] → verify: [check]
+2. [Step] → verify: [check]
+3. [Step] → verify: [check]
+```
+
+Strong success criteria let you loop independently. Weak criteria ("make it work") require constant clarification.
+
+---
+
+**These guidelines are working if:** fewer unnecessary changes in diffs, fewer rewrites due to overcomplication, and clarifying questions come before implementation rather than after mistakes.
+
+# CLAUDE.md
+
+Behavioral guidelines to reduce common LLM coding mistakes. Merge with project-specific instructions as needed.
+
+**Tradeoff:** These guidelines bias toward caution over speed. For trivial tasks, use judgment.
+
+## 1. Think Before Coding
+
+**Don't assume. Don't hide confusion. Surface tradeoffs.**
+
+Before implementing:
+- State your assumptions explicitly. If uncertain, ask.
+- If multiple interpretations exist, present them - don't pick silently.
+- If a simpler approach exists, say so. Push back when warranted.
+- If something is unclear, stop. Name what's confusing. Ask.
+
+## 2. Simplicity First
+
+**Minimum code that solves the problem. Nothing speculative.**
+
+- No features beyond what was asked.
+- No abstractions for single-use code.
+- No "flexibility" or "configurability" that wasn't requested.
+- No error handling for impossible scenarios.
+- If you write 200 lines and it could be 50, rewrite it.
+
+Ask yourself: "Would a senior engineer say this is overcomplicated?" If yes, simplify.
+
+## 3. Surgical Changes
+
+**Touch only what you must. Clean up only your own mess.**
+
+When editing existing code:
+- Don't "improve" adjacent code, comments, or formatting.
+- Don't refactor things that aren't broken.
+- Match existing style, even if you'd do it differently.
+- If you notice unrelated dead code, mention it - don't delete it.
+
+When your changes create orphans:
+- Remove imports/variables/functions that YOUR changes made unused.
+- Don't remove pre-existing dead code unless asked.
+
+The test: Every changed line should trace directly to the user's request.
+
+## 4. Goal-Driven Execution
+
+**Define success criteria. Loop until verified.**
+
+Transform tasks into verifiable goals:
+- "Add validation" → "Write tests for invalid inputs, then make them pass"
+- "Fix the bug" → "Write a test that reproduces it, then make it pass"
+- "Refactor X" → "Ensure tests pass before and after"
+
+For multi-step tasks, state a brief plan:
+```
+1. [Step] → verify: [check]
+2. [Step] → verify: [check]
+3. [Step] → verify: [check]
+```
+
+Strong success criteria let you loop independently. Weak criteria ("make it work") require constant clarification.
+
+---
+
+**These guidelines are working if:** fewer unnecessary changes in diffs, fewer rewrites due to overcomplication, and clarifying questions come before implementation rather than after mistakes.
+
+<!-- gitnexus:start -->
+# GitNexus — Code Intelligence
+
+This project is indexed by GitNexus as **lingma-openai-gateway** (1093 symbols, 2685 relationships, 97 execution flows). Use the GitNexus MCP tools to understand code, assess impact, and navigate safely.
+
+> If any GitNexus tool warns the index is stale, run `npx gitnexus analyze` in terminal first.
+
+## Always Do
+
+- **MUST run impact analysis before editing any symbol.** Before modifying a function, class, or method, run `gitnexus_impact({target: "symbolName", direction: "upstream"})` and report the blast radius (direct callers, affected processes, risk level) to the user.
+- **MUST run `gitnexus_detect_changes()` before committing** to verify your changes only affect expected symbols and execution flows.
+- **MUST warn the user** if impact analysis returns HIGH or CRITICAL risk before proceeding with edits.
+- When exploring unfamiliar code, use `gitnexus_query({query: "concept"})` to find execution flows instead of grepping. It returns process-grouped results ranked by relevance.
+- When you need full context on a specific symbol — callers, callees, which execution flows it participates in — use `gitnexus_context({name: "symbolName"})`.
+
+## Never Do
+
+- NEVER edit a function, class, or method without first running `gitnexus_impact` on it.
+- NEVER ignore HIGH or CRITICAL risk warnings from impact analysis.
+- NEVER rename symbols with find-and-replace — use `gitnexus_rename` which understands the call graph.
+- NEVER commit changes without running `gitnexus_detect_changes()` to check affected scope.
+
+## Resources
+
+| Resource | Use for |
+|----------|---------|
+| `gitnexus://repo/lingma-openai-gateway/context` | Codebase overview, check index freshness |
+| `gitnexus://repo/lingma-openai-gateway/clusters` | All functional areas |
+| `gitnexus://repo/lingma-openai-gateway/processes` | All execution flows |
+| `gitnexus://repo/lingma-openai-gateway/process/{name}` | Step-by-step execution trace |
+
+## CLI
+
+| Task | Read this skill file |
+|------|---------------------|
+| Understand architecture / "How does X work?" | `.claude/skills/gitnexus/gitnexus-exploring/SKILL.md` |
+| Blast radius / "What breaks if I change X?" | `.claude/skills/gitnexus/gitnexus-impact-analysis/SKILL.md` |
+| Trace bugs / "Why is X failing?" | `.claude/skills/gitnexus/gitnexus-debugging/SKILL.md` |
+| Rename / extract / split / refactor | `.claude/skills/gitnexus/gitnexus-refactoring/SKILL.md` |
+| Tools, resources, schema reference | `.claude/skills/gitnexus/gitnexus-guide/SKILL.md` |
+| Index, status, clean, wiki CLI commands | `.claude/skills/gitnexus/gitnexus-cli/SKILL.md` |
+
+<!-- gitnexus:end -->

DESIGN.md

@@ -47,9 +47,9 @@
 
 - **逆向 Lingma 后端协议**：之前评估过（曾经的"B1 终极方案"），需要反编译二进制，维护成本高、政策风险大，放弃。
 - **多租户 / 水平扩缩**：单容器即可；真要大规模部署 → 套层反代 + N 个网关副本就够，不在进程内解决。
-- **请求侧完整 function calling / tools 透传**：OpenAI schema 里保留了字段，但目前不会把 `tools`/`tool_choice` 透传给 Lingma（上游无等价输入协议）。
+- **请求侧完整 function calling / tools 语义**：仍不是当前目标；现阶段仅支持 `tools`/`tool_choice` 在 `TOOL_FORWARD_ENABLED` 开关下灰度透传（默认关闭）。
 - **响应侧工具事件桥接**：若 Lingma 上游产出 tool 事件，网关会向 OpenAI 输出 `tool_calls`，向 Anthropic 输出 `tool_use` / `tool_result`（stream + non-stream）。
-- **多模态**：请求里的 image/audio 会被降级成占位符 `[image]` / `[audio]`，因为 Lingma chat 不支持。
+- **强制工具回退闭环（non-stream）**：当上游未返回 tool 事件且请求为强制 `tool_choice` 时，网关会从文本里解析严格 JSON，合成 OpenAI `tool_calls` 与 Anthropic `tool_use` / `tool_result`。
 
 ---
 
@@ -518,7 +518,7 @@ FastAPI `lifespan` 退出 → `pool.close()` → 每个 `client.close()` → 进
 ### 5.3 session cache 只哈希 user/system/developer 消息
 
 - **问题**：OpenAI 客户端常常会规范化 / 裁剪 assistant 消息（例如 trim 末尾空白、去掉思考内容），导致下一轮的 `messages[:-1]` 跟上一轮的 `messages` 不完全字节相等。
-- **方案**：`hash_user_context` 只对 `system / user / developer` 三种 role 做 SHA1；assistant/tool 不参与。只要**用户输入路径**稳定，哈希就稳定。
+- **方案**：`hash_user_context` 只对 `system / user / developer` 三种 role 做 SHA1；assistant/tool 不参与。只要**用户输入路径**稳定，哈希就稳定。多模态会先在归一化阶段降级为占位符（如 `[image]` / `[audio]`）再参与哈希，因此会保留“模态存在”信号但不保留原始媒体内容。
 - **权衡**：理论上客户端篡改 assistant 语义（比如把模型的回答改成相反的）时，cache 依然命中，但 Lingma 侧自己持有 session 原版历史，下一轮还是按原版继续。对用户意图的偏离不可见。这是 OK 的——客户端本来就不该篡改 assistant 内容。
 
 ### 5.4 session cache 写入用 `write_key = hash(messages)`，查询用 `lookup_key = hash(messages[:-1])`
@@ -592,7 +592,7 @@ FastAPI `lifespan` 退出 → `pool.close()` → 每个 `client.close()` → 进
 | 需求 | 改哪些文件 | 关键入口 |
 |---|---|---|
 | 加一个新的 OpenAI 端点（如 embeddings） | `main.py`, `openai_schema.py` | 仿照 `v1_models` 加 `@app.post("/v1/embeddings", dependencies=[Depends(auth_guard)])` |
-| 扩展 Anthropic 端点（如 count_tokens / tool_use 相关能力） | `main.py::v1_messages`, `anthropic_schema.py` | count_tokens 只读：复用 `estimate_tokens`；响应侧 `tool_use/tool_result` 桥接已支持，若要请求侧 tools 透传仍需改 `lingma_client.py` payload |
+| 扩展 Anthropic 端点（如 count_tokens / tool_use 相关能力） | `main.py::v1_messages`, `anthropic_schema.py` | count_tokens 只读：复用 `estimate_tokens`；响应侧 `tool_use/tool_result` 桥接已支持；请求侧 `tools/tool_choice` 透传由 `TOOL_FORWARD_ENABLED` 控制并经 `lingma_client.py` payload 下发 |
 | 加一种新的实例调度策略（如加权轮询） | `lingma_pool.py::pick()` | 当前是 affinity → least-in-flight → round-robin |
 | 改认证为 JWT / OAuth | `auth.py` | 三个 `require_*` 函数是全部入口；`main.py` 里只有 `*_guard` 代理 |
 | 增加限流（按 api_key 配额） | `concurrency.py` 加 `PerKeyGuard`；`main.py` 在 `chat_guard.try_acquire()` 后再来一层 | 注意 ticket 释放顺序（内层先释放） |
@@ -600,7 +600,7 @@ FastAPI `lifespan` 退出 → `pool.close()` → 每个 `client.close()` → 进
 | 改 Prometheus 指标名 | 所有 `prometheus_lines()` 或 `prometheus_text()` | 注意生态兼容；更名要在 README 留 alias |
 | 接入 Jaeger / OpenTelemetry | `logging_config.py` 加 OTel instrumentation；`main.py::request_id_middleware` 注入 traceid | request_id 可以复用为 span_id |
 | 加一个 Lingma 新方法调用（比如 code/complete） | `lingma_client.py` 仿照 `query_models`：`await self.ensure_ready(); return await self.rpc.request("code/complete", ...)` | 原始上游响应形态需抓包确认 |
-| 支持 function calling（假设 Lingma 将来支持） | `openai_schema.py` 已保留 `tools` / `tool_choice` 字段；`lingma_client.py::_build_payload` 加 `extra.tools` | 上游协议 TBD |
+| 支持 function calling（假设 Lingma 将来支持） | `openai_schema.py` / `anthropic_schema.py` / `main.py` / `lingma_client.py` | 当前仅支持请求侧 `tools/tool_choice` 在开关控制下透传与响应侧桥接；若要完整 function calling 语义仍需按上游协议补齐 |
 | 多模态穿透 | `openai_schema.py::flatten_content` 不再降级；`lingma_client.py` payload 传 url | 前提：Lingma 支持（目前不支持） |
 | 换 session_cache 后端（如 Redis） | 实现同样接口的 `RedisSessionCache`，`main.py` 初始化换实现 | 接口是 `get / put / invalidate / stats / prometheus_lines / build_key / enabled`，内存换远端成本不高 |
 | 多容器副本（水平扩） | 外面套反代 + sticky session（根据 `Authorization` 或 `x-user` 做 hash）；session cache 改 Redis | 或直接接受多副本 cache 独立，轻微浪费 KV cache 命中率 |
@@ -612,7 +612,8 @@ pip install -r requirements.txt
 # 在容器外跑，需要自己准备 Lingma 二进制
 export LINGMA_BIN=/path/to/Lingma
 export API_KEYS=sk-dev
-uvicorn app.main:app --reload --port 8317
+export PORT=8317
+uvicorn app.main:app --reload --port ${PORT}
 ```
 
 主要断点位置：

Dockerfile

@@ -28,4 +28,4 @@ port=os.environ.get('PORT','8317'); \
 r=urllib.request.urlopen(f'http://127.0.0.1:{port}/healthz', timeout=3); \
 sys.exit(0 if json.load(r).get('ok') else 1)" || exit 1
 
-CMD ["sh", "-c", "python /app/app/bootstrap_lingma.py && uvicorn app.main:app --host ${HOST:-0.0.0.0} --port ${PORT:-8317}"]
+CMD ["sh", "-c", "python -m app.bootstrap_lingma && uvicorn app.main:app --host ${HOST:-0.0.0.0} --port ${PORT:-8317}"]

README.md

@@ -1,396 +1,216 @@
 # Lingma OpenAI Gateway
 
-把本地 Lingma 插件封装成 OpenAI 兼容接口。任何能调 OpenAI 的客户端（Cursor、Dify、LangChain、curl…）都能直接接入。
+将 Lingma 封装为 OpenAI / Anthropic 兼容网关，便于现有客户端直接接入。
 
-**支持：**
-- OpenAI 兼容：`GET /v1/models` / `POST /v1/chat/completions`（含 SSE 流式） / Bearer 鉴权
-- **Anthropic 兼容**：`POST /v1/messages`（含 Anthropic SSE 事件流） / `x-api-key` 鉴权
-- Prometheus / 多账号实例池 / 会话复用（跨两种协议共享） / 免浏览器登录态注入
+- OpenAI：`/v1/models`、`/v1/chat/completions`（含 stream）
+- Anthropic：`/v1/messages`、`/v1/messages/count_tokens`（含 stream）
+- 内置：多实例池、会话复用、Prometheus 指标、登录态 bundle 注入
+- 多模态降级：OpenAI `image_url` / `input_image` 转 `[image]`，`input_audio` 转 `[audio]`；Anthropic `image` 转 `[image]`
 
-> 想看架构、模块划分、设计决策、二开路线图 → 直接读 [`DESIGN.md`](./DESIGN.md)。
+> 架构设计与二开细节请看 [`DESIGN.md`](./DESIGN.md)。
 
 ---
 
-## 架构速览
+## 目录
 
-```
- ┌─────────────┐   OpenAI 协议    ┌─────────────────────────────────────────┐
- │  任意客户端  │  ───────────▶    │  FastAPI (app/main.py)                  │
- │  (curl/     │                  │  ├─ auth_guard / admin_guard            │
- │   Cursor/   │                  │  ├─ chat_guard (InFlightGuard 背压)     │
- │   Dify…)    │                  │  ├─ SessionCache (LRU+TTL, KV 复用)     │
- └─────────────┘                  │  └─ StatsCollector + Prometheus         │
-                                  └────────────────┬────────────────────────┘
-                                                   │ 选实例 (least-in-flight + affinity)
-                                  ┌────────────────▼────────────────────────┐
-                                  │  LingmaPool (app/lingma_pool.py)        │
-                                  │  ├─ inst-0  inst-1  inst-N …            │
-                                  │  └─ 启动前自动 restore session bundle   │
-                                  └────────────────┬────────────────────────┘
-                                                   │
-                           ┌───────────────────────┼───────────────────────┐
-                           ▼                       ▼                       ▼
-                ┌────────────────────┐  ┌────────────────────┐  ┌────────────────────┐
-                │ LingmaGatewayClient│  │        …           │  │        …           │
-                │ (LSP over WS)      │  │                    │  │                    │
-                │ ├─ Popen (PID管理) │  │                    │  │                    │
-                │ ├─ reconnect loop  │  │                    │  │                    │
-                │ └─ ws://:PORT      │  │                    │  │                    │
-                └──────────┬─────────┘  └────────────────────┘  └────────────────────┘
-                           │ spawn + ws
-                ┌──────────▼─────────┐
-                │   Lingma 二进制     │
-                │   --workDir /…     │
-                └────────────────────┘
-```
+1. [5 分钟启动](#5-分钟启动)
+2. [常用命令](#常用命令)
+3. [最小 API 示例](#最小-api-示例)
+4. [部署与更新](#部署与更新)
+5. [排障速查](#排障速查)
+6. [文档入口](#文档入口)
 
 ---
 
-## 一、快速开始
+## 5 分钟启动
+
+### 1) 准备配置
 
 ```bash
 git clone <repo>
 cd lingma-openai-gateway
 cp .env.example .env
-# 至少填 API_KEYS + LINGMA_USERNAME + LINGMA_PASSWORD（或 session bundle）
+```
+
+至少配置这些变量（在 `.env`）：
+
+- `API_KEYS`
+- `LINGMA_USERNAME` / `LINGMA_PASSWORD`（或 `LINGMA_SESSION_BUNDLE(_FILE)`）
+
+### 2) Docker 启动（推荐）
+
+```bash
 mkdir -p data secrets
 docker compose up -d --build
-docker compose logs -f   # 看到 "Uvicorn running on..." 就 OK
+docker compose logs -f
 ```
 
-冒烟测试：
+### 3) 冒烟检查
 
 ```bash
+PORT=$(grep '^PORT=' .env | cut -d= -f2)
 API_KEY=$(grep '^API_KEYS=' .env | cut -d= -f2 | cut -d, -f1)
-curl -s http://127.0.0.1:8317/healthz
-curl -s http://127.0.0.1:8317/v1/models -H "Authorization: Bearer $API_KEY"
-curl -s http://127.0.0.1:8317/v1/chat/completions \
-  -H "Authorization: Bearer $API_KEY" \
+
+curl -s "http://127.0.0.1:${PORT}/healthz"
+curl -s "http://127.0.0.1:${PORT}/v1/models" \
+  -H "Authorization: Bearer ${API_KEY}"
+```
+
+---
+
+## 常用命令
+
+### 本地开发运行
+
+```bash
+pip install -r requirements.txt
+uvicorn app.main:app --reload --port 8317
+```
+
+### Docker 常用
+
+```bash
+docker compose up -d --build
+docker compose logs -f
+docker compose ps
+docker compose down
+```
+
+### 测试
+
+```bash
+# 重点回归套件
+python3 -m unittest tests/test_tool_call_bridge.py
+
+# 全量 unittest
+python3 -m unittest discover -s tests -p "test_*.py"
+```
+
+---
+
+## 最小 API 示例
+
+先取 key：
+
+```bash
+PORT=$(grep '^PORT=' .env | cut -d= -f2)
+API_KEY=$(grep '^API_KEYS=' .env | cut -d= -f2 | cut -d, -f1)
+```
+
+### OpenAI：非流式
+
+```bash
+curl -s "http://127.0.0.1:${PORT}/v1/chat/completions" \
+  -H "Authorization: Bearer ${API_KEY}" \
   -H "Content-Type: application/json" \
-  -d '{"model":"org_auto","messages":[{"role":"user","content":"hi"}]}'
-```
-
----
-
-## 二、配置参考
-
-`.env.example` 是权威说明，这里按主题分组。
-
-### 2.1 核心
-
-| 变量 | 默认 | 说明 |
-|---|---|---|
-| `HOST` / `PORT` | `0.0.0.0` / `8317` | 网关监听地址与端口 |
-| `API_KEYS` | — | Bearer key，多个逗号分隔；**留空则 /v1/\* 无鉴权**，启动会 warn |
-| `LOG_LEVEL` | `INFO` | `DEBUG`/`INFO`/`WARNING`/`ERROR`，日志为结构化 JSON，含 `request_id` |
-| `DEFAULT_MODEL` | `org_auto` | 模型无法映射时兜底 |
-| `DEFAULT_ASK_MODE` | `chat` | `chat` 或 `agent`（传 `model: "agent"` 时自动切） |
-| `DEDICATED_DOMAIN_URL` | — | 企业专属域（可空） |
-
-### 2.2 权限分层（生产建议全配）
-
-| 变量 | 默认 | 说明 |
-|---|---|---|
-| `ADMIN_TOKEN` | — | `/internal/*` 专属 token；未配置时 fallback 到 `API_KEYS`（兼容）；都为空 → 503 |
-| `METRICS_TOKEN` | — | `/metrics` 专属 token；未配置时 fallback 到 `API_KEYS` |
-| `METRICS_PUBLIC` | `false` | 显式公开 `/metrics`（仅用于私网采集器） |
-
-> `ADMIN_TOKEN` / `METRICS_TOKEN` / `API_KEYS` 三者都为空时，`/metrics` 和 `/internal/*` 会返回 503（拒绝裸奔）。
-
-### 2.3 并发与背压
-
-| 变量 | 默认 | 说明 |
-|---|---|---|
-| `GATEWAY_MAX_IN_FLIGHT` | `4` | 并发上限；`<=0` 表示不限 |
-| `GATEWAY_QUEUE_TIMEOUT_SEC` | `30` | 排队超时；超时直接返回 `429 + Retry-After` |
-
-### 2.4 Lingma 进程
-
-| 变量 | 默认 | 说明 |
-|---|---|---|
-| `LINGMA_BIN` | `/app/data/bin/Lingma` | 容器内二进制路径 |
-| `LINGMA_SOURCE_TYPE` | `marketplace` | `marketplace` 或 `vsix` |
-| `LINGMA_MARKETPLACE_PUBLISHER` | `Alibaba-Cloud` | Marketplace 发布者 |
-| `LINGMA_MARKETPLACE_EXTENSION` | `tongyi-lingma` | Marketplace 扩展名 |
-| `LINGMA_VSIX_URL` | 官方地址 | 兜底 VSIX 下载地址 |
-| `LINGMA_BOOTSTRAP_ALWAYS` | `true` | 启动时总是尝试刷新二进制 |
-| `LINGMA_FORCE_REFRESH` | `false` | 强制忽略本地缓存重新下载 |
-| `LINGMA_WORK_DIR` | `/app/data/.lingma/vscode/sharedClientCache` | 登录态/缓存所在目录 |
-| `LINGMA_SOCKET_PORT` | `36510` | 单实例模式下的 Lingma WS 端口 |
-| `LINGMA_STARTUP_TIMEOUT` | `40` | 启动超时秒 |
-| `LINGMA_RPC_TIMEOUT` | `30` | 单次 RPC 超时秒 |
-
-### 2.5 多账号 / 多实例池
-
-| 变量 | 默认 | 说明 |
-|---|---|---|
-| `LINGMA_ACCOUNTS` | — | `u1:p1,u2:p2` 或 JSON 数组；配置后每个账号 = 一个独立 Lingma 子进程 |
-| `LINGMA_INSTANCE_COUNT` | 账号数 | 显式指定实例数；不足账号循环复用并打 warn |
-| `LINGMA_USERNAME` / `LINGMA_PASSWORD` | — | 单实例兼容模式（仅 `LINGMA_ACCOUNTS` 为空时生效） |
-
-### 2.6 会话复用（KV cache 优化）
-
-| 变量 | 默认 | 说明 |
-|---|---|---|
-| `SESSION_REUSE_ENABLED` | `true` | 多轮对话命中时只发增量 user 消息 + 复用上游 `sessionId` |
-| `SESSION_CACHE_MAX_ENTRIES` | `256` | LRU 容量 |
-| `SESSION_CACHE_TTL_SEC` | `1800` | TTL（秒），避免命中已回收的 session |
-
-### 2.7 登录态注入（跳过 Playwright）
-
-| 变量 | 默认 | 说明 |
-|---|---|---|
-| `LINGMA_SESSION_BUNDLE` | — | base64 格式的 bundle（inline，适合短字符串） |
-| `LINGMA_SESSION_BUNDLE_FILE` | — | bundle 文件路径（推荐，避免 env 过长） |
-
-### 2.8 自动登录
-
-| 变量 | 默认 | 说明 |
-|---|---|---|
-| `AUTO_LOGIN_ENABLED` | `true` | 未登录时自动启 Playwright |
-| `AUTO_LOGIN_HEADLESS` | `true` | 无头浏览器 |
-| `AUTO_LOGIN_TIMEOUT` | `180` | 登录超时秒 |
-| `AUTO_LOGIN_MAX_RETRY` | `2` | 登录失败重试次数 |
-
----
-
-## 三、API 参考
-
-### 3.1 公共（`API_KEYS`）
-
-| 方法 | 路径 | 说明 |
-|---|---|---|
-| GET | `/healthz` | 免鉴权；返回 `ok` / `pool_size` / `pool_ready` / 每实例状态 |
-| GET | `/v1/models` | OpenAI 兼容；`id` 是 Lingma 原 key，`name` 是可读名 |
-| POST | `/v1/chat/completions` | OpenAI 兼容；`stream=true` 走 SSE；`model: "agent"` 切 agent 模式 |
-| POST | `/v1/messages` | **Anthropic Messages 兼容**；`x-api-key` 或 `Authorization: Bearer`；`stream=true` 走 Anthropic 命名事件 SSE |
-
-**chat 请求示例（非流式）**
-
-```bash
-curl -s http://127.0.0.1:8317/v1/chat/completions \
-  -H "Authorization: Bearer $API_KEY" -H "Content-Type: application/json" \
-  -d '{"model":"dashscope_qmodel","messages":[{"role":"user","content":"你好"}]}'
-```
-
-**chat 请求示例（流式 + usage）**
-
-```bash
-curl -N http://127.0.0.1:8317/v1/chat/completions \
-  -H "Authorization: Bearer $API_KEY" -H "Content-Type: application/json" \
   -d '{
-    "model":"dashscope_qmodel",
-    "stream":true,
-    "stream_options":{"include_usage":true},
-    "messages":[{"role":"user","content":"介绍一下你自己"}]
+    "model": "org_auto",
+    "messages": [{"role": "user", "content": "hi"}],
+    "stream": false
   }'
 ```
 
-**Anthropic Messages 示例（非流式）**
+### OpenAI：流式
 
 ```bash
-curl -s http://127.0.0.1:8317/v1/messages \
-  -H "x-api-key: $API_KEY" \
+curl -N "http://127.0.0.1:${PORT}/v1/chat/completions" \
+  -H "Authorization: Bearer ${API_KEY}" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "model": "org_auto",
+    "messages": [{"role": "user", "content": "say hi"}],
+    "stream": true
+  }'
+```
+
+### Anthropic：非流式
+
+```bash
+curl -s "http://127.0.0.1:${PORT}/v1/messages" \
+  -H "x-api-key: ${API_KEY}" \
   -H "anthropic-version: 2023-06-01" \
   -H "Content-Type: application/json" \
   -d '{
     "model": "claude-3-5-sonnet-20241022",
     "max_tokens": 256,
-    "system":"你是一个简洁的助手",
-    "messages":[{"role":"user","content":"你好"}]
+    "messages": [{"role": "user", "content": "hi"}],
+    "stream": false
   }'
 ```
 
-**Anthropic Messages 示例（流式）**
+### Anthropic：流式
 
 ```bash
-curl -N http://127.0.0.1:8317/v1/messages \
-  -H "x-api-key: $API_KEY" \
+curl -N "http://127.0.0.1:${PORT}/v1/messages" \
+  -H "x-api-key: ${API_KEY}" \
   -H "anthropic-version: 2023-06-01" \
   -H "Content-Type: application/json" \
   -d '{
     "model": "claude-3-5-sonnet-20241022",
     "max_tokens": 256,
-    "stream":true,
-    "messages":[{"role":"user","content":"写一首四行诗"}]
+    "messages": [{"role": "user", "content": "say hi"}],
+    "stream": true
   }'
-# 返回 message_start / content_block_start / content_block_delta* /
-#      content_block_stop / message_delta / message_stop
 ```
 
-说明：
-- **模型名兼容**：客户端可以继续传 `claude-3-*` 等名字；未识别的 model 会回退到 `DEFAULT_MODEL` 对应的 Lingma key，后端实际仍由 Lingma 提供（Qwen 系列）。如需显式选模型，直接传 Lingma key（`dashscope_qmodel` 等）。
-- **会话复用共享**：Anthropic 与 OpenAI 两个端点共用同一 `SessionCache`，只要 API key 相同、对话前缀相同，就会命中同一上游 `sessionId`。
-- **多模态**：`image` 块会被降级为 `[image]` 占位符（Lingma 不支持 vision）。
-- **工具事件桥接**：当 Lingma 上游返回 `tool` 事件时，网关会输出为 OpenAI `tool_calls`（含 stream/non-stream）和 Anthropic `tool_use`/`tool_result` blocks（含 stream/non-stream）；但请求侧 `tools`/`tool_choice` 仍不会透传到 Lingma。
-- **鉴权**：优先 `x-api-key` 头（Anthropic 官方 SDK 默认），回退 `Authorization: Bearer`（方便 curl / OpenAI 风格客户端）。
-
-### 3.2 观测（`METRICS_TOKEN` 或 `API_KEYS`）
-
-| 方法 | 路径 | 说明 |
-|---|---|---|
-| GET | `/metrics` | Prometheus 文本；含池每实例 gauge、并发、session cache 命中率、token 计数 |
-
-### 3.3 管理（`ADMIN_TOKEN` 或 fallback 到 `API_KEYS`）
-
-| 方法 | 路径 | 说明 |
-|---|---|---|
-| GET | `/internal/stats` | JSON：`stats` + `concurrency` + `pool` + `session_cache` |
-| GET | `/internal/auto-login/status` | 每实例登录态与 auto_login 状态 |
-| POST | `/internal/auto-login/start?instance=inst-0` | 主动触发某实例登录（可不传，由 pool.pick 选） |
-| POST | `/internal/session/export?instance=inst-0` | 把已登录实例的 cache 打包成 base64 bundle |
-| GET | `/internal/models/raw?instance=inst-0` | Lingma 原始 `config/queryModels` 响应（displayName / isReasoning / isVl 等） |
-
----
-
-## 四、常用场景
-
-### 4.1 多账号池
-
-```env
-LINGMA_ACCOUNTS=user1:pass1,user2:pass2,user3:pass3
-# LINGMA_INSTANCE_COUNT=3   # 不写默认=账号数
-```
-
-- 每个账号一个独立 Lingma 子进程 + 独立 `workDir`（`data/.lingma/pool/inst-<i>/`）。
-- 路由：同 `user` 字段或同 system prompt 的请求**粘性**分到同一实例；其他按**最小在途**分配。
-- 一个实例挂掉不影响整体，`/healthz.pool_ready` 下降，自动重连。
-
-### 4.2 跳过 Playwright（session bundle）
-
-**从已登录实例导出：**
+### Anthropic：count_tokens
 
 ```bash
-curl -sS -X POST \
-  -H "Authorization: Bearer $ADMIN_TOKEN" \
-  "http://host:port/internal/session/export" \
-  | jq -r '.bundle_b64' > secrets/lingma-session.b64
-chmod 600 secrets/lingma-session.b64
+curl -s "http://127.0.0.1:${PORT}/v1/messages/count_tokens" \
+  -H "x-api-key: ${API_KEY}" \
+  -H "anthropic-version: 2023-06-01" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "model": "claude-3-5-sonnet-20241022",
+    "max_tokens": 64,
+    "messages": [{"role": "user", "content": "count me"}]
+  }'
 ```
 
-**在新部署注入（选一种）：**
+---
 
-```env
-# 文件注入（推荐）—— 需要在 docker-compose.yml 挂载 secrets 目录
-LINGMA_SESSION_BUNDLE_FILE=/secrets/lingma-session.b64
+## 部署与更新
 
-# 或 inline（适合小 bundle）
-LINGMA_SESSION_BUNDLE=H4sIAAAA...
+### 服务器更新到最新 main
 
-# 多账号 JSON 模式，每账号独立 bundle
-LINGMA_ACCOUNTS=[
-  {"username":"u1","password":"p1","session_bundle_file":"/secrets/u1.b64"},
-  {"username":"u2","password":"p2","session_bundle":"H4sIAAAA..."}
-]
+```bash
+cd /root/lingma-openai-gateway
+git fetch origin
+git checkout -B main origin/main
+git reset --hard origin/main
+git clean -fd
+docker compose up -d --build
+docker compose ps
 ```
 
-**行为保证：**
+### 健康检查
 
-- 只在目标 `workDir` 空（`cache/user` 不存在或 empty）时才注入；不会覆盖活跃登录态。
-- 注入失败（损坏/权限）自动 fallback 到 Playwright。
-- bundle 只含 `cache/{id,user,quota,config.json}` 4 个文件；大小上限 4 MiB，实际通常 < 10 KB。
-- **bundle 等同于密钥**，落盘需 `chmod 600`，不要进 git。
-
-### 4.3 Prometheus 接入
-
-```yaml
-# prometheus scrape_configs 片段
-- job_name: lingma-gateway
-  bearer_token: <METRICS_TOKEN>
-  static_configs: [{targets: ['host:8317']}]
-  metrics_path: /metrics
+```bash
+PORT=$(grep '^PORT=' .env | cut -d= -f2)
+curl -s "http://127.0.0.1:${PORT}/healthz"
 ```
 
-关键指标：
+---
 
-| 指标 | 类型 | 意义 |
+## 排障速查
+
+| 现象 | 常见原因 | 处理 |
 |---|---|---|
-| `gateway_in_flight` / `gateway_queued` | gauge | 并发 / 排队 |
-| `gateway_rejected_total` | counter | 背压拒绝（429）累计 |
-| `gateway_pool_instance_ready{name}` | gauge | 每实例是否就绪（0/1） |
-| `gateway_pool_instance_in_flight{name}` | gauge | 每实例在途 |
-| `gateway_session_cache_hit_total` / `_miss_total` | counter | 会话复用命中率原料 |
-| `gateway_chat_requests_success` / `_error` | counter | chat 成功率 |
+| `/v1/*` 返回 401 | 缺失或错误 API key | 检查 `Authorization: Bearer` 或 `x-api-key` |
+| `healthz` 正常但请求失败 | 用错端口 | 以 `.env` 的 `PORT` 为准，`docker compose ps` 再确认 |
+| `git pull` 提示 not on a branch | 处于 detached HEAD | 执行 `git checkout -B main origin/main` |
+| 自动登录不稳定 | 浏览器流程波动 | 优先使用 `LINGMA_SESSION_BUNDLE(_FILE)` |
+| 工具调用未触发 | 模型未选择工具 | 使用 `tool_choice` 强制，必要时约束输出 JSON |
 
 ---
 
-## 五、升级注意事项
+## 文档入口
 
-从旧版本升级时注意**破坏性变更**（每一项都有 fallback，默认不会炸，但建议显式配置）：
-
-| 版本 | 变更 | 应对 |
-|---|---|---|
-| v0.3 | `/metrics` 裸奔时（无 token / 无 key）由公开改为 503 | 显式配 `METRICS_PUBLIC=true` 或 `METRICS_TOKEN` |
-| v0.3 | `/internal/*` 引入 `ADMIN_TOKEN` | 未配置自动 fallback 到 `API_KEYS`，生产建议单独配 |
-| v0.2 | 默认会话复用（多轮对话只发增量） | 如果你的客户端裁剪了历史导致语义不连续，设 `SESSION_REUSE_ENABLED=false` |
-| v0.2 | Chat 请求走 JSON-RPC `notify` 而非 `request`（修复 30s TTFB bug） | 无需行动 |
-| v0.2 | 多实例池（`LINGMA_ACCOUNTS` 存在时启用） | 不配则保持单实例行为 |
-
----
-
-## 六、故障排查（FAQ）
-
-| 症状 | 排查方向 |
-|---|---|
-| `/healthz` 返回 `ok=false` / `pool_ready=0` | 查 `docker logs`，关键字 `lingma spawned` / `state ... -> ready`；若卡在 `starting` → Lingma 二进制或 workDir 权限问题 |
-| 返回 `401` 且带 `Invalid admin token` | 你用了 `API_KEYS` 去打 `/internal/*`，但服务端已设了 `ADMIN_TOKEN`；用 `ADMIN_TOKEN` 或清空 `ADMIN_TOKEN` |
-| 返回 `503 metrics scraping disabled` | 三个 env 全空，按 "权限分层" 章节配任一 |
-| 返回 `429 Too many in-flight` | 并发超过 `GATEWAY_MAX_IN_FLIGHT`；增大或客户端加重试 |
-| 首 token 延迟 2-3 秒 | Lingma 侧常态；多轮对话第二轮起，会话复用命中后 TTFB 明显降低（看 `gateway_session_cache_hit_total`） |
-| Playwright 登录失败 | 导出一个已登录 bundle 注入（见 4.2），彻底跳过浏览器 |
-| 容器重启后 Lingma 要重新登录 | `data/` 没挂在卷上或被清过；确认 `./data:/app/data` 挂载 + bundle fallback |
-| 升级后 `/metrics` 返回 503 | v0.3 默认严格；按表格 5.1 配置 |
-
-开 `LOG_LEVEL=DEBUG` 可以看到 Lingma 子进程的 stderr 输出，便于定位 native 崩溃。
-
----
-
-## 七、开发与二开
-
-项目本身是单仓 FastAPI，3400 行 Python。推荐阅读路径：
-
-1. **先读 [`DESIGN.md`](./DESIGN.md)** —— 架构、模块职责、关键设计决策、二开指引。
-2. 再按需读对应模块：
-   - 想改请求入口 / 路由 → `app/main.py`
-   - 想加实例调度策略 → `app/lingma_pool.py::pick()`
-   - 想改 Lingma 通信协议 → `app/lingma_client.py`
-   - 想扩展会话复用 → `app/session_cache.py` + `main.py` 的 reuse 块
-   - 想做认证改造 → `app/auth.py` + `main.py::*_guard`
-3. 本地跑：`pip install -r requirements.txt && uvicorn app.main:app --reload`。
-
----
-
-## 八、目录结构
-
-```
-lingma-openai-gateway/
-├── app/                        # 主代码（见 DESIGN.md 模块一览）
-│   ├── main.py                 # FastAPI 入口 + 路由
-│   ├── lingma_pool.py          # N 实例池
-│   ├── lingma_client.py        # LSP over WS + 子进程管理
-│   ├── session_cache.py        # 多轮对话 sessionId 复用
-│   ├── session_bundle.py       # 登录态 export/import
-│   ├── concurrency.py          # InFlightGuard 背压
-│   ├── auto_login.py           # Playwright 登录
-│   ├── auth.py                 # Bearer / admin / metrics 三档鉴权
-│   ├── config.py               # 环境变量 → dataclass
-│   ├── model_map.py            # 模型 key ↔ displayName
-│   ├── openai_schema.py        # OpenAI 请求/响应 Pydantic
-│   ├── stats.py                # StatsCollector + Prometheus
-│   ├── logging_config.py       # 结构化 JSON log + request_id 上下文
-│   └── bootstrap_lingma.py     # 启动时下载/提取 Lingma 二进制
-├── data/                       # 持久化（Lingma 二进制 + workDir），不进 git
-├── secrets/                    # 注入的 bundle 等敏感文件，不进 git
-├── Dockerfile                  # Playwright base + HEALTHCHECK
-├── docker-compose.yml
-├── .env.example                # 配置权威文档
-├── requirements.txt
-├── README.md                   # 本文件
-└── DESIGN.md                   # 架构与二开手册
-```
-
----
+- 配置权威：[`/.env.example`](./.env.example)
+- 架构/模块边界/设计决策：[`/DESIGN.md`](./DESIGN.md)
+- 主要入口代码：[`/app/main.py`](./app/main.py)
+- 测试：[`/tests/test_tool_call_bridge.py`](./tests/test_tool_call_bridge.py)
 
 ## License
 
-内部使用，按需调整。
+MIT

app/anthropic_schema.py

@@ -119,10 +119,8 @@ def anthropic_to_internal_messages(req: AnthropicMessagesRequest) -> list[dict]:
     """Project an Anthropic request into the gateway's internal message list.
 
     Internal shape matches what `_messages_to_prompt` already expects:
-    `[{"role": "system"|"user"|"assistant", "content": "..."}]`. This means
-    session-cache hashing is identical across OpenAI and Anthropic callers —
-    a user who migrates between the two endpoints keeps their session affinity
-    as long as they send the same conversation prefix.
+    `[{"role": "system"|"user"|"assistant", "content": "..."}]`. This keeps
+    user-input cache hashing aligned across OpenAI and Anthropic callers.
     """
     out: list[dict] = []
     if req.system:

app/config.py

@@ -5,6 +5,11 @@ import os
 from dataclasses import dataclass, field
 
 
+
+def _csv_env(raw: str) -> list[str]:
+    return [item.strip() for item in (raw or "").replace("\n", ",").split(",") if item.strip()]
+
+
 @dataclass
 class LingmaAccount:
     username: str
@@ -45,6 +50,7 @@ class Settings:
     session_cache_max_entries: int = 256
     session_cache_ttl_sec: float = 1800.0
     tool_forward_enabled: bool = False
+    tool_allowlist: list[str] = field(default_factory=list)
 
 
 def _bool_env(name: str, default: bool) -> bool:
@@ -176,5 +182,6 @@ def load_settings() -> Settings:
         session_reuse_enabled=_bool_env("SESSION_REUSE_ENABLED", True),
         session_cache_max_entries=int(os.getenv("SESSION_CACHE_MAX_ENTRIES", "256")),
         session_cache_ttl_sec=float(os.getenv("SESSION_CACHE_TTL_SEC", "1800")),
-        tool_forward_enabled=_bool_env("TOOL_FORWARD_ENABLED", False),
+        tool_forward_enabled=_bool_env("TOOL_FORWARD_ENABLED", True),
+        tool_allowlist=_csv_env(os.getenv("TOOL_ALLOWLIST", "")),
     )

app/http/responses_adapter.py

@@ -0,0 +1,176 @@
+from __future__ import annotations
+
+import json
+import time
+import uuid
+from typing import Any
+
+from fastapi import HTTPException
+
+from ..openai_schema import ChatCompletionsRequest, ResponsesRequest, flatten_content
+
+
+def _responses_input_to_messages(req: ResponsesRequest) -> list[dict[str, Any]]:
+    messages: list[dict[str, Any]] = []
+    if req.instructions:
+        messages.append({"role": "system", "content": req.instructions})
+
+    raw_input = req.input
+    if raw_input is None:
+        return messages
+
+    valid_roles = {"system", "user", "assistant", "tool", "developer", "function"}
+
+    def _append(role: str, content: Any, *, tool_call_id: str | None = None) -> None:
+        msg: dict[str, Any] = {"role": role, "content": flatten_content(content)}
+        if role == "tool" and tool_call_id:
+            msg["tool_call_id"] = tool_call_id
+        messages.append(msg)
+
+    if isinstance(raw_input, str):
+        _append("user", raw_input)
+        return messages
+
+    raw_items: list[Any]
+    if isinstance(raw_input, dict):
+        raw_items = [raw_input]
+    elif isinstance(raw_input, list):
+        raw_items = list(raw_input)
+    else:
+        _append("user", str(raw_input))
+        return messages
+
+    for item in raw_items:
+        if isinstance(item, str):
+            _append("user", item)
+            continue
+        if not isinstance(item, dict):
+            _append("user", str(item))
+            continue
+
+        role = item.get("role")
+        if isinstance(role, str) and role in valid_roles:
+            tool_call_id = item.get("tool_call_id") or item.get("call_id")
+            _append(role, item.get("content"), tool_call_id=str(tool_call_id) if tool_call_id else None)
+            continue
+
+        if item.get("type") == "function_call_output":
+            output = item.get("output")
+            if isinstance(output, (dict, list)):
+                output = json.dumps(output, ensure_ascii=False)
+            tool_call_id = item.get("call_id")
+            _append("tool", output, tool_call_id=str(tool_call_id) if tool_call_id else None)
+            continue
+
+        if "content" in item:
+            text = flatten_content(item.get("content"))
+        else:
+            text = flatten_content([item])
+        if text:
+            _append("user", text)
+
+    return messages
+
+
+def _responses_to_chat_request(req: ResponsesRequest) -> ChatCompletionsRequest:
+    return ChatCompletionsRequest(
+        model=req.model,
+        messages=_responses_input_to_messages(req),
+        stream=req.stream,
+        temperature=req.temperature,
+        top_p=req.top_p,
+        max_tokens=req.max_output_tokens,
+        user=req.user,
+        tools=req.tools,
+        tool_choice=req.tool_choice,
+    )
+
+
+def _responses_id_from_chat_id(chat_id: Any) -> str:
+    if isinstance(chat_id, str) and chat_id:
+        suffix = chat_id.removeprefix("chatcmpl-")
+        return f"resp_{suffix}"
+    return f"resp_{uuid.uuid4().hex}"
+
+
+def _responses_usage_from_chat(usage: Any) -> dict[str, int]:
+    if not isinstance(usage, dict):
+        return {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
+    input_tokens = int(usage.get("prompt_tokens") or 0)
+    output_tokens = int(usage.get("completion_tokens") or 0)
+    return {
+        "input_tokens": input_tokens,
+        "output_tokens": output_tokens,
+        "total_tokens": int(usage.get("total_tokens") or (input_tokens + output_tokens)),
+    }
+
+
+def _responses_non_stream_from_chat_payload(chat_payload: Any) -> dict[str, Any]:
+    if not isinstance(chat_payload, dict):
+        raise HTTPException(
+            status_code=502,
+            detail={"error": {"message": "invalid upstream response", "type": "upstream_error"}},
+        )
+    choice = {}
+    choices = chat_payload.get("choices")
+    if isinstance(choices, list) and choices:
+        choice = choices[0] if isinstance(choices[0], dict) else {}
+    message = choice.get("message") if isinstance(choice.get("message"), dict) else {}
+
+    output: list[dict[str, Any]] = []
+    content = message.get("content")
+    if isinstance(content, str) and content:
+        output.append(
+            {
+                "type": "message",
+                "id": f"msg_{uuid.uuid4().hex}",
+                "status": "completed",
+                "role": "assistant",
+                "content": [{"type": "output_text", "text": content}],
+            }
+        )
+
+    tool_calls = message.get("tool_calls")
+    if isinstance(tool_calls, list):
+        for idx, tool_call in enumerate(tool_calls):
+            if not isinstance(tool_call, dict):
+                continue
+            fn = tool_call.get("function") if isinstance(tool_call.get("function"), dict) else {}
+            call_id = str(tool_call.get("id") or f"call_{idx}")
+            output.append(
+                {
+                    "type": "function_call",
+                    "id": call_id,
+                    "call_id": call_id,
+                    "name": str(fn.get("name") or "tool"),
+                    "arguments": str(fn.get("arguments") or "{}"),
+                }
+            )
+
+    output_text_parts: list[str] = []
+    for item in output:
+        if item.get("type") == "message":
+            blocks = item.get("content")
+            if isinstance(blocks, list):
+                for block in blocks:
+                    if isinstance(block, dict) and block.get("type") == "output_text":
+                        text = block.get("text")
+                        if isinstance(text, str) and text:
+                            output_text_parts.append(text)
+
+    return {
+        "id": _responses_id_from_chat_id(chat_payload.get("id")),
+        "object": "response",
+        "created_at": int(chat_payload.get("created") or time.time()),
+        "status": "completed",
+        "error": None,
+        "incomplete_details": None,
+        "model": chat_payload.get("model"),
+        "output": output,
+        "output_text": "".join(output_text_parts),
+        "usage": _responses_usage_from_chat(chat_payload.get("usage")),
+    }
+
+
+def _sse_data(payload: dict[str, Any]) -> str:
+    return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"

app/http/tool_bridge.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+import ast
+import json
+import uuid
+from typing import Any
+
+
+def _json_string(value: Any) -> str:
+    if isinstance(value, str):
+        return value
+    try:
+        return json.dumps(value if value is not None else {}, ensure_ascii=False)
+    except Exception:
+        return "{}"
+
+
+def _openai_forced_tool_name(tool_choice: Any) -> str | None:
+    if not isinstance(tool_choice, dict):
+        return None
+    fn = tool_choice.get("function")
+    if isinstance(fn, dict):
+        name = fn.get("name")
+        if isinstance(name, str) and name.strip():
+            return name.strip()
+    return None
+
+
+def _anthropic_forced_tool_name(tool_choice: Any) -> str | None:
+    if not isinstance(tool_choice, dict):
+        return None
+    if tool_choice.get("type") == "tool":
+        name = tool_choice.get("name")
+        if isinstance(name, str) and name.strip():
+            return name.strip()
+    fn = tool_choice.get("function")
+    if isinstance(fn, dict):
+        name = fn.get("name")
+        if isinstance(name, str) and name.strip():
+            return name.strip()
+    return None
+
+
+def _json_object_from_text(text: str) -> dict[str, Any] | None:
+    raw = text.strip()
+    if not raw:
+        return None
+    if raw.startswith("```") and raw.endswith("```"):
+        raw = raw[3:-3].strip()
+        if raw.lower().startswith("json"):
+            raw = raw[4:].strip()
+    try:
+        parsed = json.loads(raw)
+    except Exception:
+        return None
+    return parsed if isinstance(parsed, dict) else None
+
+
+def _tool_code_single_arg_name(tools: list[dict[str, Any]] | None, forced_tool_name: str) -> str | None:
+    if not isinstance(tools, list):
+        return None
+    for tool in tools:
+        if not isinstance(tool, dict):
+            continue
+        schema: dict[str, Any] | None = None
+        if tool.get("type") == "function":
+            fn = tool.get("function")
+            if isinstance(fn, dict) and fn.get("name") == forced_tool_name:
+                params = fn.get("parameters")
+                if isinstance(params, dict):
+                    schema = params
+        elif tool.get("name") == forced_tool_name:
+            input_schema = tool.get("input_schema")
+            if isinstance(input_schema, dict):
+                schema = input_schema
+        if not isinstance(schema, dict):
+            continue
+        properties = schema.get("properties")
+        if not isinstance(properties, dict) or len(properties) != 1:
+            return None
+        only_name = next(iter(properties.keys()), None)
+        if isinstance(only_name, str) and only_name.strip():
+            return only_name
+        return None
+    return None
+
+
+def _tool_code_object_from_text(
+    text: str,
+    forced_tool_name: str,
+    *,
+    single_arg_name: str | None = None,
+) -> dict[str, Any] | None:
+    raw = text.strip()
+    if not raw.startswith("```tool_code") or not raw.endswith("```"):
+        return None
+    lines = raw.splitlines()
+    if len(lines) < 2:
+        return None
+    body = "\n".join(lines[1:-1]).strip()
+    try:
+        parsed = ast.parse(body, mode="eval")
+    except Exception:
+        return None
+    call = parsed.body
+    if not isinstance(call, ast.Call):
+        return None
+    if not isinstance(call.func, ast.Name) or call.func.id != forced_tool_name:
+        return None
+    arguments: dict[str, Any] = {}
+    if call.args:
+        if len(call.args) != 1 or call.keywords or not single_arg_name:
+            return None
+        try:
+            arguments[single_arg_name] = ast.literal_eval(call.args[0])
+        except Exception:
+            return None
+        return {"arguments": arguments}
+    for kw in call.keywords:
+        if kw.arg is None:
+            return None
+        try:
+            arguments[kw.arg] = ast.literal_eval(kw.value)
+        except Exception:
+            return None
+    return {"arguments": arguments}
+
+
+def _forced_tool_event_from_text(
+    text: str,
+    forced_tool_name: str,
+    *,
+    single_arg_name: str | None = None,
+) -> dict[str, Any] | None:
+    parsed = _json_object_from_text(text)
+    if parsed is None:
+        parsed = _tool_code_object_from_text(text, forced_tool_name, single_arg_name=single_arg_name)
+    if parsed is None:
+        return None
+
+    explicit_name: Any = parsed.get("name") or parsed.get("tool")
+    fn = parsed.get("function")
+    if explicit_name is None and isinstance(fn, dict):
+        explicit_name = fn.get("name")
+    if explicit_name is not None and str(explicit_name) != forced_tool_name:
+        return None
+
+    tool_input: Any = None
+    if "input" in parsed:
+        tool_input = parsed.get("input")
+    elif "arguments" in parsed:
+        args = parsed.get("arguments")
+        if isinstance(args, str):
+            try:
+                tool_input = json.loads(args)
+            except Exception:
+                return None
+        else:
+            tool_input = args
+    elif isinstance(fn, dict) and "arguments" in fn:
+        args = fn.get("arguments")
+        if isinstance(args, str):
+            try:
+                tool_input = json.loads(args)
+            except Exception:
+                return None
+        else:
+            tool_input = args
+    else:
+        reserved = {"name", "tool", "function", "arguments", "input", "result"}
+        tool_input = {k: v for k, v in parsed.items() if k not in reserved}
+
+    event: dict[str, Any] = {
+        "name": forced_tool_name,
+        "input": tool_input if tool_input is not None else {},
+    }
+    if "result" in parsed:
+        event["result"] = parsed.get("result")
+    return event
+
+
+def _openai_tool_call(tool: dict[str, Any], *, forced_id: str | None = None) -> dict[str, Any]:
+    return {
+        "id": str(tool.get("id") or forced_id or f"call_{uuid.uuid4().hex}"),
+        "type": "function",
+        "function": {
+            "name": str(tool.get("name") or "tool"),
+            "arguments": _json_string(tool.get("input")),
+        },
+    }
+
+
+def _anthropic_tool_use_block(
+    tool: dict[str, Any], *, forced_id: str | None = None
+) -> dict[str, Any]:
+    return {
+        "type": "tool_use",
+        "id": str(tool.get("id") or forced_id or f"toolu_{uuid.uuid4().hex}"),
+        "name": str(tool.get("name") or "tool"),
+        "input": tool.get("input") if tool.get("input") is not None else {},
+    }
+
+
+def _anthropic_tool_result_block(
+    tool: dict[str, Any], *, forced_id: str | None = None
+) -> dict[str, Any] | None:
+    if "result" not in tool:
+        return None
+    result = tool.get("result")
+    if isinstance(result, str):
+        content: Any = result
+    else:
+        content = _json_string(result)
+    return {
+        "type": "tool_result",
+        "tool_use_id": str(tool.get("id") or forced_id or ""),
+        "content": content,
+    }

app/lingma_client.py

@@ -101,6 +101,7 @@ class LspWsRpcClient:
         self._rx_buffer = b""
         self._chat_streams: dict[str, dict] = {}
         self._tool_stream_map: dict[str, str] = {}
+        self._tool_roundtrip_done: set[str] = set()
         self._on_disconnect = on_disconnect
         self._closed = False
 
@@ -204,6 +205,7 @@ class LspWsRpcClient:
                 stream["chunks"].put_nowait(None)
         self._chat_streams.clear()
         self._tool_stream_map.clear()
+        self._tool_roundtrip_done.clear()
 
     async def _send(self, payload: dict):
         async with self._send_lock:
@@ -320,6 +322,55 @@ class LspWsRpcClient:
         return merged, changed
 
 
+    @staticmethod
+    def _is_tool_roundtrip_method(method: str | None) -> bool:
+        return method in {"tool/call/sync", "tool/invoke"}
+
+    @staticmethod
+    def _build_tool_approve_params(params: dict[str, Any], tool_id: str) -> dict[str, Any] | None:
+        req_id = params.get("requestId")
+        session_id = params.get("sessionId")
+        if not isinstance(req_id, str) or not req_id.strip():
+            return None
+        if not isinstance(session_id, str) or not session_id.strip():
+            return None
+        return {
+            "type": "tool_call",
+            "sessionId": session_id,
+            "requestId": req_id,
+            "toolCallId": tool_id,
+            "approval": True,
+        }
+
+    @staticmethod
+    def _build_tool_invoke_result_params(params: dict[str, Any], tool_event: dict[str, Any], tool_id: str) -> dict[str, Any]:
+        return {
+            "toolCallId": tool_id,
+            "name": str(tool_event.get("name") or params.get("name") or "tool"),
+            "success": True,
+            "errorMessage": "",
+            "result": tool_event.get("result") if "result" in tool_event else {},
+        }
+
+    async def _maybe_emit_tool_roundtrip(self, method: str, params: dict[str, Any], tool_event: dict[str, Any]) -> None:
+        if not self._is_tool_roundtrip_method(method):
+            return
+        tool_id = self._normalize_tool_id(method, params, tool_event)
+        if not tool_id:
+            return
+        if tool_id in self._tool_roundtrip_done:
+            return
+
+        approve_params = self._build_tool_approve_params(params, tool_id)
+        if approve_params is None:
+            return
+
+        self._tool_roundtrip_done.add(tool_id)
+        await self.notify("tool/call/approve", approve_params)
+        invoke_result_params = self._build_tool_invoke_result_params(params, tool_event, tool_id)
+        await self.notify("tool/invokeResult", invoke_result_params)
+
+
     def _resolve_tool_stream(self, method: str, params: dict[str, Any], tool_event: dict[str, Any] | None) -> dict | None:
         req_id = params.get("requestId")
         if isinstance(req_id, str) and req_id.strip():
@@ -363,6 +414,7 @@ class LspWsRpcClient:
                 if not tool_id:
                     logger.warning("drop unroutable tool event: method=%s missing tool id", method)
                 else:
+                    await self._maybe_emit_tool_roundtrip(method, params, tool_event)
                     tool_states = stream["tool_states"]
                     order = stream["tool_order"]
                     existing = tool_states.get(tool_id)
@@ -431,6 +483,7 @@ class LspWsRpcClient:
         for tool_id, mapped_req in list(self._tool_stream_map.items()):
             if mapped_req == request_id:
                 self._tool_stream_map.pop(tool_id, None)
+                self._tool_roundtrip_done.discard(tool_id)
         # Drain queue so no stray future gets stuck if the consumer bailed early.
         if not stream["done"].is_set():
             stream["done"].set()
@@ -442,13 +495,21 @@ class LspWsRpcClient:
         if stream is None:
             return
         start = time.monotonic()
+        last_chunk_at = start
         while True:
             remain = timeout - (time.monotonic() - start)
             if remain <= 0:
-                raise TimeoutError("chat stream timeout")
+                first_chunk_at = stream.get("first_chunk_at")
+                raise TimeoutError(
+                    "chat stream timeout "
+                    f"request_id={request_id} timeout={timeout:.1f}s "
+                    f"first_chunk_at={None if first_chunk_at is None else round(first_chunk_at - start, 3)}s "
+                    f"last_chunk_at={round(last_chunk_at - start, 3)}s"
+                )
             chunk = await asyncio.wait_for(stream["chunks"].get(), timeout=remain)
             if chunk is None:
                 break
+            last_chunk_at = time.monotonic()
             yield chunk
 
     def get_stream_result(self, request_id: str) -> dict:
@@ -843,12 +904,12 @@ class LingmaGatewayClient:
         is_reply: bool = False,
         tool_config: dict[str, Any] | None = None,
     ):
-        session_type = "developer" if ask_mode == "agent" else "chat"
+        session_type = "ask" if ask_mode == "agent" else "chat"
         payload = {
             "requestId": request_id,
             "sessionId": session_id,
             "sessionType": session_type,
-            "chatTask": "FREE_INPUT",
+            "chatTask": "chat" if ask_mode == "agent" else "FREE_INPUT",
             "mode": ask_mode,
             "stream": True,
             "source": 1,

app/main.py

@@ -25,6 +25,26 @@ from .auth import (
 )
 from .concurrency import BackpressureRejected, InFlightGuard
 from .config import Settings, load_settings
+from .http.responses_adapter import (
+    _responses_id_from_chat_id,
+    _responses_input_to_messages,
+    _responses_non_stream_from_chat_payload,
+    _responses_to_chat_request,
+    _responses_usage_from_chat,
+    _sse_data,
+)
+from .http.tool_bridge import (
+    _anthropic_forced_tool_name,
+    _anthropic_tool_result_block,
+    _anthropic_tool_use_block,
+    _forced_tool_event_from_text,
+    _json_object_from_text,
+    _json_string,
+    _openai_forced_tool_name,
+    _openai_tool_call,
+    _tool_code_object_from_text,
+    _tool_code_single_arg_name,
+)
 from .lingma_pool import LingmaPool, PoolInstance
 from .logging_config import configure_logging, get_logger, request_id_var
 from .model_map import build_model_name_map, flatten_model_keys, resolve_model
@@ -34,10 +54,11 @@ from .openai_schema import (
     ChatCompletionsRequest,
     ModelData,
     ModelsResponse,
+    ResponsesRequest,
     flatten_content,
 )
 from .session_bundle import encode_bundle, pack_workdir
-from .session_cache import SessionCache
+from .session_cache import SessionCache, hash_branch_context
 from .stats import StatsCollector, estimate_tokens
 
 
@@ -56,6 +77,12 @@ session_cache = SessionCache(
     ttl_sec=settings.session_cache_ttl_sec,
 )
 
+STREAMING_RESPONSE_HEADERS = {
+    "Cache-Control": "no-cache, no-transform",
+    "X-Accel-Buffering": "no",
+    "Connection": "keep-alive",
+}
+
 
 def _require_pool() -> LingmaPool:
     if pool is None:
@@ -351,6 +378,68 @@ def _include_usage(stream_options: dict | None) -> bool:
     return bool(stream_options.get("include_usage"))
 
 
+def _tool_allowlist() -> set[str]:
+    return {name.strip() for name in settings.tool_allowlist if isinstance(name, str) and name.strip()}
+
+
+def _openai_tool_name(tool: Any) -> str | None:
+    if not isinstance(tool, dict):
+        return None
+    if tool.get("type") == "function":
+        fn = tool.get("function")
+        if isinstance(fn, dict):
+            name = fn.get("name")
+            if isinstance(name, str) and name.strip():
+                return name.strip()
+    name = tool.get("name")
+    if isinstance(name, str) and name.strip():
+        return name.strip()
+    return None
+
+
+def _anthropic_tool_name(tool: Any) -> str | None:
+    if not isinstance(tool, dict):
+        return None
+    name = tool.get("name")
+    if isinstance(name, str) and name.strip():
+        return name.strip()
+    fn = tool.get("function")
+    if isinstance(fn, dict):
+        nested_name = fn.get("name")
+        if isinstance(nested_name, str) and nested_name.strip():
+            return nested_name.strip()
+    return None
+
+
+def _filter_allowed_tools(tools: list[dict[str, Any]], *, provider: str) -> list[dict[str, Any]]:
+    allowlist = _tool_allowlist()
+    if not allowlist:
+        return tools
+    name_fn = _openai_tool_name if provider == "openai" else _anthropic_tool_name
+    return [tool for tool in tools if (name := name_fn(tool)) and name in allowlist]
+
+
+def _ensure_tool_choice_allowed(tool_choice: Any, *, provider: str) -> None:
+    allowlist = _tool_allowlist()
+    if not allowlist:
+        return
+    forced_name = (
+        _openai_forced_tool_name(tool_choice)
+        if provider == "openai"
+        else _anthropic_forced_tool_name(tool_choice)
+    )
+    if forced_name and forced_name not in allowlist:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error": {
+                    "type": "invalid_request_error",
+                    "message": f"tool '{forced_name}' is not allowed",
+                }
+            },
+        )
+
+
 def _openai_tool_config(req: ChatCompletionsRequest) -> dict[str, Any] | None:
     if not settings.tool_forward_enabled:
         return None
@@ -358,9 +447,11 @@ def _openai_tool_config(req: ChatCompletionsRequest) -> dict[str, Any] | None:
     has_choice = req.tool_choice is not None
     if not has_tools and not has_choice:
         return None
+    _ensure_tool_choice_allowed(req.tool_choice, provider="openai")
+    tools = _filter_allowed_tools(req.tools or [], provider="openai")
     return {
         "provider": "openai",
-        "tools": req.tools or [],
+        "tools": tools,
         "tool_choice": req.tool_choice,
     }
 
@@ -372,9 +463,11 @@ def _anthropic_tool_config(req: AnthropicMessagesRequest) -> dict[str, Any] | No
     has_choice = req.tool_choice is not None
     if not has_tools and not has_choice:
         return None
+    _ensure_tool_choice_allowed(req.tool_choice, provider="anthropic")
+    tools = _filter_allowed_tools(req.tools or [], provider="anthropic")
     return {
         "provider": "anthropic",
-        "tools": req.tools or [],
+        "tools": tools,
         "tool_choice": req.tool_choice,
     }
 
@@ -415,6 +508,43 @@ def _anthropic_has_tooling_context(req: AnthropicMessagesRequest) -> bool:
     return False
 
 
+def _resolve_ask_mode(model: str, has_tooling_context: bool) -> str:
+    model_name = (model or "").lower()
+    if model_name in {"lingma-agent", "agent"} or has_tooling_context:
+        return "agent"
+    return settings.default_ask_mode
+
+
+async def _apply_cached_instance_or_invalidate(
+    *,
+    protocol: str,
+    inst: PoolInstance,
+    cached_instance_name: str | None,
+    cached_session_id: str | None,
+    lookup_key: str | None,
+) -> str | None:
+    if cached_instance_name and inst.name != cached_instance_name:
+        logger.info(
+            "%s session cache instance %s unhealthy, falling back to %s",
+            protocol,
+            cached_instance_name,
+            inst.name,
+        )
+        if lookup_key:
+            await session_cache.invalidate(lookup_key)
+        return None
+    return cached_session_id
+
+
+
+def _streaming_response(event_stream) -> StreamingResponse:
+    return StreamingResponse(
+        event_stream,
+        media_type="text/event-stream",
+        headers=STREAMING_RESPONSE_HEADERS,
+    )
+
+
 def _stream_event_type(event: Any) -> str:
     if isinstance(event, dict):
         t = event.get("type")
@@ -443,54 +573,6 @@ def _stream_tool_event(event: Any) -> dict[str, Any] | None:
     return None
 
 
-def _json_string(value: Any) -> str:
-    if isinstance(value, str):
-        return value
-    try:
-        return json.dumps(value if value is not None else {}, ensure_ascii=False)
-    except Exception:
-        return "{}"
-
-
-def _openai_tool_call(tool: dict[str, Any], *, forced_id: str | None = None) -> dict[str, Any]:
-    return {
-        "id": str(tool.get("id") or forced_id or f"call_{uuid.uuid4().hex}"),
-        "type": "function",
-        "function": {
-            "name": str(tool.get("name") or "tool"),
-            "arguments": _json_string(tool.get("input")),
-        },
-    }
-
-
-def _anthropic_tool_use_block(
-    tool: dict[str, Any], *, forced_id: str | None = None
-) -> dict[str, Any]:
-    return {
-        "type": "tool_use",
-        "id": str(tool.get("id") or forced_id or f"toolu_{uuid.uuid4().hex}"),
-        "name": str(tool.get("name") or "tool"),
-        "input": tool.get("input") if tool.get("input") is not None else {},
-    }
-
-
-def _anthropic_tool_result_block(
-    tool: dict[str, Any], *, forced_id: str | None = None
-) -> dict[str, Any] | None:
-    if "result" not in tool:
-        return None
-    result = tool.get("result")
-    if isinstance(result, str):
-        content: Any = result
-    else:
-        content = _json_string(result)
-    return {
-        "type": "tool_result",
-        "tool_use_id": str(tool.get("id") or forced_id or ""),
-        "content": content,
-    }
-
-
 @app.post("/v1/chat/completions", dependencies=[Depends(auth_guard)])
 async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
     p = _require_pool()
@@ -504,13 +586,11 @@ async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
     #   1. Reuse the upstream sessionId so Lingma/Qwen hits its KV cache.
     #   2. Send only the new user message instead of the whole history.
     #   3. Stick the request to the pool instance that originally served it.
-    ask_mode = settings.default_ask_mode
-    if req.model.lower() in {"lingma-agent", "agent"}:
-        ask_mode = "agent"
-
     tool_config = _openai_tool_config(req)
     has_tooling_context = _openai_has_tooling_context(req, messages_dump)
 
+    ask_mode = _resolve_ask_mode(req.model, has_tooling_context)
+
     reuse_eligible = (
         session_cache.enabled
         and ask_mode == "chat"
@@ -522,29 +602,38 @@ async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
     cached_session_id: str | None = None
     cached_instance_name: str | None = None
     if reuse_eligible:
-        lookup_key = session_cache.build_key(api_key, messages_dump[:-1], tool_config=tool_config)
-        write_key = session_cache.build_key(api_key, messages_dump, tool_config=tool_config)
+        prefix_branch_context = hash_branch_context(messages_dump[:-1])
+        lookup_key = session_cache.build_key(
+            api_key,
+            messages_dump[:-1],
+            tool_config=tool_config,
+            branch_context=prefix_branch_context,
+        )
+        write_key = session_cache.build_key(
+            api_key,
+            messages_dump,
+            tool_config=tool_config,
+            branch_context=hash_branch_context(messages_dump),
+        )
         entry = await session_cache.get(lookup_key)
+        if entry is None:
+            legacy_lookup_key = session_cache.build_key(api_key, messages_dump[:-1], tool_config=tool_config)
+            entry = await session_cache.get(legacy_lookup_key)
+            if entry is not None:
+                lookup_key = legacy_lookup_key
         if entry is not None:
             cached_session_id = entry.session_id
             cached_instance_name = entry.instance_name or None
-
-    # Instance selection: prefer cached instance for continuity, else normal affinity.
     affinity = cached_instance_name or _affinity_key_for(req)
     inst = p.pick(affinity_key=affinity)
 
-    # If cache pointed at a specific instance that's no longer healthy, we already
-    # fell back via pool.pick -> drop the cached session since Lingma on a
-    # different process won't know about it.
-    if cached_instance_name and inst.name != cached_instance_name:
-        logger.info(
-            "session cache instance %s unhealthy, falling back to %s (dropping cached session)",
-            cached_instance_name,
-            inst.name,
+    cached_session_id = await _apply_cached_instance_or_invalidate(
+        protocol="chat",
+        inst=inst,
+        cached_instance_name=cached_instance_name,
+        cached_session_id=cached_session_id,
+        lookup_key=lookup_key,
     )
-        cached_session_id = None
-        if lookup_key:
-            await session_cache.invalidate(lookup_key)
 
     await _ensure_instance_logged_in(inst)
 
@@ -618,11 +707,31 @@ async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
             completion_id = f"chatcmpl-{uuid.uuid4().hex}"
             completion_tokens_holder = {"n": 0}
             stream_meta: dict = {}
+            forced_tool_name = _openai_forced_tool_name(req.tool_choice)
+            forced_tool_single_arg_name = _tool_code_single_arg_name(req.tools, forced_tool_name) if forced_tool_name else None
 
             async def event_stream(_ticket=ticket, _inst=inst, _meta=stream_meta):
                 success = False
                 tool_call_indexes: dict[str, int] = {}
                 saw_tool_call = False
+                buffered_text_parts: list[str] = []
+
+                def _text_payload(text: str) -> str:
+                    payload = {
+                        "id": completion_id,
+                        "object": "chat.completion.chunk",
+                        "created": created,
+                        "model": model,
+                        "choices": [
+                            {
+                                "index": 0,
+                                "delta": {"content": text},
+                                "finish_reason": None,
+                            }
+                        ],
+                    }
+                    return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+
                 try:
                     async for chunk in _inst.client.chat_stream(
                         prompt,
@@ -637,6 +746,25 @@ async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
                             tool = _stream_tool_event(chunk)
                             if not tool:
                                 continue
+
+                            tool_name = str(tool.get("name") or "")
+                            allowed = True
+                            if tool_config and isinstance(tool_config.get("tools"), list) and tool_config.get("tools"):
+                                allowed = False
+                                for t in tool_config.get("tools"):
+                                    if tool_name == _anthropic_tool_name(t) or tool_name == _openai_tool_name(t):
+                                        allowed = True
+                                        break
+                                if not allowed and forced_tool_name and tool_name == forced_tool_name:
+                                    allowed = True
+                            if not allowed:
+                                continue
+
+                            if buffered_text_parts:
+                                for buffered_text in buffered_text_parts:
+                                    yield _text_payload(buffered_text)
+                                buffered_text_parts.clear()
+
                             tool_id = str(tool.get("id") or "")
                             if not tool_id:
                                 tool_id = f"call_{len(tool_call_indexes)}"
@@ -671,7 +799,24 @@ async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
                         text = _stream_text(chunk)
                         if not text:
                             continue
+                        buffered_text_parts.append(text)
                         completion_tokens_holder["n"] += estimate_tokens(text)
+                        if forced_tool_name and not saw_tool_call:
+                            continue
+                        yield _text_payload(text)
+
+                    if buffered_text_parts and not saw_tool_call and forced_tool_name:
+                        fallback_event = _forced_tool_event_from_text(
+                            "".join(buffered_text_parts),
+                            forced_tool_name,
+                            single_arg_name=forced_tool_single_arg_name,
+                        )
+                        if fallback_event is not None:
+                            saw_tool_call = True
+                            tool_id = "call_fallback_0"
+                            idx = 0
+                            tool_call_indexes[tool_id] = idx
+                            fallback_tool_call = _openai_tool_call(fallback_event, forced_id=tool_id)
                             payload = {
                                 "id": completion_id,
                                 "object": "chat.completion.chunk",
@@ -680,13 +825,26 @@ async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
                                 "choices": [
                                     {
                                         "index": 0,
-                                    "delta": {"content": text},
+                                        "delta": {
+                                            "tool_calls": [
+                                                {
+                                                    "index": idx,
+                                                    **fallback_tool_call,
+                                                }
+                                            ]
+                                        },
                                         "finish_reason": None,
                                     }
                                 ],
                             }
+                            buffered_text_parts.clear()
                             yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
 
+                    if buffered_text_parts:
+                        for buffered_text in buffered_text_parts:
+                            yield _text_payload(buffered_text)
+                        buffered_text_parts.clear()
+
                     done_payload = {
                         "id": completion_id,
                         "object": "chat.completion.chunk",
@@ -702,7 +860,6 @@ async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
                     }
                     yield f"data: {json.dumps(done_payload, ensure_ascii=False)}\n\n"
 
-
                     if include_usage:
                         usage_payload = {
                             "id": completion_id,
@@ -721,14 +878,22 @@ async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
                     yield "data: [DONE]\n\n"
                     success = True
                 except asyncio.CancelledError:
-                    logger.info("chat.stream cancelled by client (inst=%s)", _inst.name)
+                    logger.info(
+                        "chat.stream cancelled by client (inst=%s, session_id=%s)",
+                        _inst.name,
+                        cached_session_id,
+                    )
                     raise
                 except Exception as exc:
-                    logger.warning("chat.stream error (inst=%s): %s", _inst.name, exc)
+                    logger.warning(
+                        "chat.stream error (inst=%s, session_id=%s, prompt_tokens=%s, completion_tokens=%s): %s",
+                        _inst.name,
+                        cached_session_id,
+                        prompt_tokens,
+                        completion_tokens_holder["n"],
+                        exc,
+                    )
                 finally:
-                    # Persist upstream sessionId only on a clean chat/finish.
-                    # Partial streams (cancelled, timed out) leave Lingma's
-                    # session in an indeterminate state, so we must not reuse.
                     if success and write_key:
                         sid = _meta.get("session_id")
                         if sid:
@@ -743,15 +908,7 @@ async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
                     _ticket.release()
 
             ticket_transferred = True
-            return StreamingResponse(
-                event_stream(),
-                media_type="text/event-stream",
-                headers={
-                    "Cache-Control": "no-cache, no-transform",
-                    "X-Accel-Buffering": "no",
-                    "Connection": "keep-alive",
-                },
-            )
+            return _streaming_response(event_stream())
 
         try:
             result = await inst.client.chat_complete(
@@ -794,12 +951,37 @@ async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
         message_content = result.get("text") or ""
         tool_calls: list[dict[str, Any]] = []
         saw_tool_call = False
+        forced_tool_name = _openai_forced_tool_name(req.tool_choice)
         if isinstance(tool_events, list):
             for idx, item in enumerate(tool_events):
                 if isinstance(item, dict):
+                    tool_name = str(item.get("name") or "")
+                    allowed = True
+                    if tool_config and isinstance(tool_config.get("tools"), list) and tool_config.get("tools"):
+                        allowed = False
+                        for t in tool_config.get("tools"):
+                            if tool_name == _anthropic_tool_name(t) or tool_name == _openai_tool_name(t):
+                                allowed = True
+                                break
+                        if not allowed and forced_tool_name and tool_name == forced_tool_name:
+                            allowed = True
+                    if not allowed:
+                        continue
+
                     tool_id = str(item.get("id") or f"call_{idx}")
                     tool_calls.append(_openai_tool_call(item, forced_id=tool_id))
                     saw_tool_call = True
+        if not saw_tool_call:
+            if forced_tool_name:
+                fallback_event = _forced_tool_event_from_text(
+                    message_content,
+                    forced_tool_name,
+                    single_arg_name=_tool_code_single_arg_name(req.tools, forced_tool_name),
+                )
+                if fallback_event is not None:
+                    tool_calls.append(_openai_tool_call(fallback_event, forced_id="call_fallback_0"))
+                    saw_tool_call = True
+                    message_content = ""
         response = ChatCompletionResponse(
             id=f"chatcmpl-{uuid.uuid4().hex}",
             created=int(time.time()),
@@ -836,6 +1018,318 @@ async def v1_chat_completions(req: ChatCompletionsRequest, request: Request):
             ticket.release()
 
 
+
+
+async def _responses_stream_from_chat_stream(
+    chat_stream: StreamingResponse,
+    *,
+    response_id: str,
+    model: str,
+):
+    created_at = int(time.time())
+    usage: dict[str, int] = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
+    completed_sent = False
+    output_item_id = f"msg_{uuid.uuid4().hex}"
+    output_index = 0
+    content_index = 0
+    output_text_parts: list[str] = []
+    function_call_items: list[dict[str, Any]] = []
+    function_call_index_by_id: dict[str, int] = {}
+    function_call_arguments_by_id: dict[str, str] = {}
+    function_call_name_by_id: dict[str, str] = {}
+    function_call_id_by_upstream_index: dict[int, str] = {}
+
+    def _message_item(status: str) -> dict[str, Any]:
+        return {
+            "id": output_item_id,
+            "type": "message",
+            "role": "assistant",
+            "status": status,
+            "content": [
+                {
+                    "type": "output_text",
+                    "text": "".join(output_text_parts),
+                }
+            ],
+        }
+
+    def _function_call_item(call_id: str, *, status: str, name: str, arguments: str) -> dict[str, Any]:
+        return {
+            "id": call_id,
+            "type": "function_call",
+            "call_id": call_id,
+            "name": name,
+            "arguments": arguments,
+            "status": status,
+        }
+
+    def _completed_output_items() -> list[dict[str, Any]]:
+        return [_message_item("completed"), *function_call_items]
+
+    def _completed_frame() -> str:
+        return _sse_data(
+            {
+                "type": "response.completed",
+                "response": {
+                    "id": response_id,
+                    "object": "response",
+                    "created_at": created_at,
+                    "status": "completed",
+                    "model": model,
+                    "output": _completed_output_items(),
+                    "usage": usage,
+                },
+            }
+        )
+
+    def _finish_output_item_frames() -> list[str]:
+        frames = [
+            _sse_data(
+                {
+                    "type": "response.output_text.done",
+                    "response_id": response_id,
+                    "item_id": output_item_id,
+                    "output_index": output_index,
+                    "content_index": content_index,
+                    "text": "".join(output_text_parts),
+                }
+            ),
+            _sse_data(
+                {
+                    "type": "response.output_item.done",
+                    "response_id": response_id,
+                    "output_index": output_index,
+                    "item": _message_item("completed"),
+                }
+            ),
+        ]
+        for idx, item in enumerate(function_call_items, start=1):
+            frames.append(
+                _sse_data(
+                    {
+                        "type": "response.function_call_arguments.done",
+                        "response_id": response_id,
+                        "item_id": item["id"],
+                        "output_index": idx,
+                        "arguments": item["arguments"],
+                    }
+                )
+            )
+            frames.append(
+                _sse_data(
+                    {
+                        "type": "response.output_item.done",
+                        "response_id": response_id,
+                        "output_index": idx,
+                        "item": item,
+                    }
+                )
+            )
+        return frames
+
+    def _ensure_function_call_item(call_id: str) -> list[str]:
+        existing_index = function_call_index_by_id.get(call_id)
+        name = function_call_name_by_id.get(call_id, "tool")
+        arguments = function_call_arguments_by_id.get(call_id, "")
+        if existing_index is not None:
+            function_call_items[existing_index] = _function_call_item(
+                call_id,
+                status="completed",
+                name=name,
+                arguments=arguments,
+            )
+            return []
+        item = _function_call_item(
+            call_id,
+            status="completed",
+            name=name,
+            arguments=arguments,
+        )
+        function_call_items.append(item)
+        item_index = len(function_call_items) - 1
+        function_call_index_by_id[call_id] = item_index
+        return [
+            _sse_data(
+                {
+                    "type": "response.output_item.added",
+                    "response_id": response_id,
+                    "output_index": item_index + 1,
+                    "item": _function_call_item(
+                        call_id,
+                        status="in_progress",
+                        name=name,
+                        arguments="",
+                    ),
+                }
+            )
+        ]
+
+    yield _sse_data(
+        {
+            "type": "response.created",
+            "response": {
+                "id": response_id,
+                "object": "response",
+                "created_at": created_at,
+                "status": "in_progress",
+                "model": model,
+                "output": [],
+            },
+        }
+    )
+    yield _sse_data(
+        {
+            "type": "response.output_item.added",
+            "response_id": response_id,
+            "output_index": output_index,
+            "item": _message_item("in_progress"),
+        }
+    )
+
+    try:
+        async for part in chat_stream.body_iterator:
+            chunk = part.decode("utf-8") if isinstance(part, bytes) else str(part)
+            for frame in chunk.split("\n\n"):
+                frame = frame.strip()
+                if not frame or not frame.startswith("data:"):
+                    continue
+                body = frame[len("data:") :].strip()
+                if body == "[DONE]":
+                    for event in _finish_output_item_frames():
+                        yield event
+                    yield _completed_frame()
+                    yield "data: [DONE]\n\n"
+                    completed_sent = True
+                    return
+
+                try:
+                    payload = json.loads(body)
+                except Exception:
+                    continue
+
+                frame_usage = _responses_usage_from_chat(payload.get("usage"))
+                if any(frame_usage.values()):
+                    usage = frame_usage
+
+                choices = payload.get("choices")
+                if not isinstance(choices, list) or not choices:
+                    continue
+                choice = choices[0] if isinstance(choices[0], dict) else {}
+                delta = choice.get("delta") if isinstance(choice.get("delta"), dict) else {}
+
+                text = delta.get("content")
+                if isinstance(text, str) and text:
+                    output_text_parts.append(text)
+                    yield _sse_data(
+                        {
+                            "type": "response.output_text.delta",
+                            "response_id": response_id,
+                            "item_id": output_item_id,
+                            "output_index": output_index,
+                            "content_index": content_index,
+                            "delta": text,
+                        }
+                    )
+
+                tool_calls = delta.get("tool_calls")
+                if isinstance(tool_calls, list):
+                    for idx, tool_call in enumerate(tool_calls):
+                        if not isinstance(tool_call, dict):
+                            continue
+                        fn = tool_call.get("function") if isinstance(tool_call.get("function"), dict) else {}
+                        upstream_index_raw = tool_call.get("index")
+                        upstream_index = upstream_index_raw if isinstance(upstream_index_raw, int) else idx
+                        call_id = str(
+                            tool_call.get("id")
+                            or function_call_id_by_upstream_index.get(upstream_index)
+                            or f"call_{upstream_index}"
+                        )
+                        function_call_id_by_upstream_index[upstream_index] = call_id
+                        name = str(fn.get("name") or function_call_name_by_id.get(call_id) or "tool")
+                        function_call_name_by_id[call_id] = name
+                        arguments_delta = str(fn.get("arguments") or "")
+                        accumulated_arguments = (
+                            function_call_arguments_by_id.get(call_id, "") + arguments_delta
+                        )
+                        function_call_arguments_by_id[call_id] = accumulated_arguments
+                        for event in _ensure_function_call_item(call_id):
+                            yield event
+                        if arguments_delta:
+                            yield _sse_data(
+                                {
+                                    "type": "response.function_call_arguments.delta",
+                                    "response_id": response_id,
+                                    "item_id": call_id,
+                                    "output_index": function_call_index_by_id[call_id] + 1,
+                                    "delta": arguments_delta,
+                                }
+                            )
+    except asyncio.CancelledError:
+        if not completed_sent:
+            for event in _finish_output_item_frames():
+                yield event
+            yield _completed_frame()
+            yield "data: [DONE]\n\n"
+            completed_sent = True
+        return
+    except Exception:
+        if not completed_sent:
+            for event in _finish_output_item_frames():
+                yield event
+            yield _completed_frame()
+            yield "data: [DONE]\n\n"
+            completed_sent = True
+        return
+
+    if not completed_sent:
+        for event in _finish_output_item_frames():
+            yield event
+        yield _completed_frame()
+        yield "data: [DONE]\n\n"
+
+
+
+@app.post("/responses", dependencies=[Depends(auth_guard)])
+@app.post("/v1/responses", dependencies=[Depends(auth_guard)])
+async def v1_responses(req: ResponsesRequest, request: Request):
+    chat_req = _responses_to_chat_request(req)
+    chat_response = await v1_chat_completions(chat_req, request)
+
+    if isinstance(chat_response, StreamingResponse):
+        response_id = f"resp_{uuid.uuid4().hex}"
+        return StreamingResponse(
+            _responses_stream_from_chat_stream(
+                chat_response,
+                response_id=response_id,
+                model=req.model,
+            ),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache, no-transform",
+                "X-Accel-Buffering": "no",
+                "Connection": "keep-alive",
+            },
+        )
+
+    invalid_upstream_error = {
+        "error": {"message": "invalid upstream response", "type": "upstream_error"}
+    }
+    try:
+        chat_payload = json.loads(chat_response.body)
+    except Exception:
+        raise HTTPException(
+            status_code=502,
+            detail=invalid_upstream_error,
+        )
+    if not isinstance(chat_payload, dict):
+        raise HTTPException(
+            status_code=502,
+            detail=invalid_upstream_error,
+        )
+    return JSONResponse(content=_responses_non_stream_from_chat_payload(chat_payload))
+
+
+
 def _anthropic_error(status_code: int, error_type: str, message: str) -> JSONResponse:
     """Build an Anthropic-shaped error response (`type:error` envelope)."""
     return JSONResponse(
@@ -912,12 +1406,17 @@ async def v1_messages(req: AnthropicMessagesRequest, request: Request):
     )
 
     # ------------------------------------------------------------- session reuse
-    # Anthropic clients don't expose an ask_mode, so we always run in "chat".
-    ask_mode = "chat"
-
+    try:
         tool_config = _anthropic_tool_config(req)
+    except HTTPException as exc:
+        detail = exc.detail if isinstance(exc.detail, dict) else {}
+        error = detail.get("error") if isinstance(detail.get("error"), dict) else {}
+        message = error.get("message") or str(detail) or "invalid tool configuration"
+        return _anthropic_error(exc.status_code, "invalid_request_error", message)
     has_tooling_context = _anthropic_has_tooling_context(req)
 
+    ask_mode = _resolve_ask_mode(req.model, has_tooling_context)
+
     reuse_eligible = (
         session_cache.enabled and ask_mode == "chat" and len(messages_dump) >= 2 and not has_tooling_context
     )
@@ -926,9 +1425,25 @@ async def v1_messages(req: AnthropicMessagesRequest, request: Request):
     cached_session_id: str | None = None
     cached_instance_name: str | None = None
     if reuse_eligible:
-        lookup_key = session_cache.build_key(api_key, messages_dump[:-1], tool_config=tool_config)
-        write_key = session_cache.build_key(api_key, messages_dump, tool_config=tool_config)
+        prefix_branch_context = hash_branch_context(messages_dump[:-1])
+        lookup_key = session_cache.build_key(
+            api_key,
+            messages_dump[:-1],
+            tool_config=tool_config,
+            branch_context=prefix_branch_context,
+        )
+        write_key = session_cache.build_key(
+            api_key,
+            messages_dump,
+            tool_config=tool_config,
+            branch_context=hash_branch_context(messages_dump),
+        )
         entry = await session_cache.get(lookup_key)
+        if entry is None:
+            legacy_lookup_key = session_cache.build_key(api_key, messages_dump[:-1], tool_config=tool_config)
+            entry = await session_cache.get(legacy_lookup_key)
+            if entry is not None:
+                lookup_key = legacy_lookup_key
         if entry is not None:
             cached_session_id = entry.session_id
             cached_instance_name = entry.instance_name or None
@@ -1071,6 +1586,21 @@ async def v1_messages(req: AnthropicMessagesRequest, request: Request):
                             tool = _stream_tool_event(chunk)
                             if not tool:
                                 continue
+                            
+                            tool_name = str(tool.get("name") or "")
+                            allowed = True
+                            if tool_config and isinstance(tool_config.get("tools"), list) and tool_config.get("tools"):
+                                allowed = False
+                                for t in tool_config.get("tools"):
+                                    if tool_name == _anthropic_tool_name(t) or tool_name == _openai_tool_name(t):
+                                        allowed = True
+                                        break
+                                forced_tool_name = _anthropic_forced_tool_name(req.tool_choice)
+                                if not allowed and forced_tool_name and tool_name == forced_tool_name:
+                                    allowed = True
+                            if not allowed:
+                                continue
+
                             tool_id = str(tool.get("id") or f"toolu_stream_{block_index}")
 
                             tool_use_block = _anthropic_tool_use_block(tool, forced_id=tool_id)
@@ -1198,15 +1728,8 @@ async def v1_messages(req: AnthropicMessagesRequest, request: Request):
                     _ticket.release()
 
             ticket_transferred = True
-            return StreamingResponse(
-                event_stream(),
-                media_type="text/event-stream",
-                headers={
-                    "Cache-Control": "no-cache, no-transform",
-                    "X-Accel-Buffering": "no",
-                    "Connection": "keep-alive",
-                },
-            )
+            return _streaming_response(event_stream())
+
 
         # ------------------------------------------------------------- non-stream
         try:
@@ -1248,10 +1771,27 @@ async def v1_messages(req: AnthropicMessagesRequest, request: Request):
             content_blocks.append({"type": "text", "text": text})
         tool_events = result.get("toolEvents") or []
         saw_pending_tool_use = False
+        saw_tool_event = False
         if isinstance(tool_events, list):
             for idx, item in enumerate(tool_events):
                 if not isinstance(item, dict):
                     continue
+                
+                tool_name = str(item.get("name") or "")
+                allowed = True
+                if tool_config and isinstance(tool_config.get("tools"), list) and tool_config.get("tools"):
+                    allowed = False
+                    for t in tool_config.get("tools"):
+                        if tool_name == _anthropic_tool_name(t) or tool_name == _openai_tool_name(t):
+                            allowed = True
+                            break
+                    forced_tool_name = _anthropic_forced_tool_name(req.tool_choice)
+                    if not allowed and forced_tool_name and tool_name == forced_tool_name:
+                        allowed = True
+                if not allowed:
+                    continue
+
+                saw_tool_event = True
                 tool_id = str(item.get("id") or f"toolu_nonstream_{idx}")
                 content_blocks.append(_anthropic_tool_use_block(item, forced_id=tool_id))
                 tool_result = _anthropic_tool_result_block(item, forced_id=tool_id)
@@ -1260,7 +1800,25 @@ async def v1_messages(req: AnthropicMessagesRequest, request: Request):
                 else:
                     saw_pending_tool_use = True
 
+        if not saw_tool_event:
+            forced_tool_name = _anthropic_forced_tool_name(req.tool_choice)
+            if forced_tool_name:
+                fallback_event = _forced_tool_event_from_text(
+                    text,
+                    forced_tool_name,
+                    single_arg_name=_tool_code_single_arg_name(req.tools, forced_tool_name),
+                )
+                if fallback_event is not None:
+                    content_blocks = []
+                    tool_id = "toolu_fallback_0"
+                    content_blocks.append(_anthropic_tool_use_block(fallback_event, forced_id=tool_id))
+                    tool_result = _anthropic_tool_result_block(fallback_event, forced_id=tool_id)
+                    saw_pending_tool_use = tool_result is None
+                    if tool_result is not None:
+                        content_blocks.append(tool_result)
+
         response_body: dict = {
+
             "id": message_id,
             "type": "message",
             "role": "assistant",

app/openai_schema.py

@@ -32,6 +32,19 @@ class ChatCompletionsRequest(BaseModel):
     tool_choice: Any | None = None
 
 
+class ResponsesRequest(BaseModel):
+    model: str
+    input: Any | None = None
+    stream: bool = False
+    temperature: float | None = None
+    top_p: float | None = None
+    max_output_tokens: int | None = None
+    user: str | None = None
+    tools: list[dict[str, Any]] | None = None
+    tool_choice: Any | None = None
+    instructions: str | None = None
+
+
 class ModelData(BaseModel):
     id: str
     name: str | None = None

app/session_cache.py

@@ -26,7 +26,7 @@ class SessionEntry:
 def hash_user_context(messages: list[dict]) -> str:
     """Hash the user/system/developer turns of a message list.
 
-    We deliberately skip `assistant`/`tool` messages because:
+    We deliberately skip `assistant`/`tool` messages here because:
     - Clients may subtly reformat or trim assistant replies between turns,
       breaking exact-match keying.
     - Only the *inputs* are stable, and they're sufficient to identify a
@@ -43,6 +43,28 @@ def hash_user_context(messages: list[dict]) -> str:
     return h.hexdigest()
 
 
+def hash_branch_context(messages: list[dict]) -> str:
+    """Hash assistant/tool turns to reduce branch collisions."""
+    h = hashlib.sha1()
+    for m in messages:
+        role = m.get("role", "")
+        if role not in ("assistant", "tool"):
+            continue
+        content = m.get("content")
+        text = content if isinstance(content, str) else flatten_content(content)
+        tool_calls = m.get("tool_calls")
+        if tool_calls is not None:
+            try:
+                tool_calls_text = json.dumps(tool_calls, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+            except Exception:
+                tool_calls_text = str(tool_calls)
+        else:
+            tool_calls_text = ""
+        tool_call_id = m.get("tool_call_id") or ""
+        h.update(f"{role}\x1f{text or ''}\x1f{tool_calls_text}\x1f{tool_call_id}\x1e".encode("utf-8"))
+    return h.hexdigest()
+
+
 def _tool_fingerprint(tool_config: dict | None) -> str:
     if not isinstance(tool_config, dict):
         return "-"
@@ -90,11 +112,21 @@ class SessionCache:
     def enabled(self) -> bool:
         return self.max > 0
 
-    def build_key(self, api_key: str, messages: list[dict], *, tool_config: dict | None = None) -> str:
+    def build_key(
+        self,
+        api_key: str,
+        messages: list[dict],
+        *,
+        tool_config: dict | None = None,
+        branch_context: str | None = None,
+    ) -> str:
         # API key scoping prevents cross-tenant session leakage even when
         # different clients happen to produce identical histories.
         key_scope = hashlib.sha1((api_key or "-").encode("utf-8")).hexdigest()[:12]
-        return f"{key_scope}:{hash_user_context(messages)}:{_tool_fingerprint(tool_config)}"
+        base = f"{key_scope}:{hash_user_context(messages)}:{_tool_fingerprint(tool_config)}"
+        if not branch_context:
+            return base
+        return f"{base}:{branch_context}"
 
     async def get(self, key: str) -> SessionEntry | None:
         if not self.enabled:

tests/TEST_PLAN.md

@@ -0,0 +1,53 @@
+# lingma-openai-gateway 测试计划（tests）
+
+## 1. 目标
+- 覆盖网关核心稳定性路径：认证、并发限流、会话复用、协议内容规范化。
+- 在不引入外部依赖（Lingma 进程/Playwright）的前提下，使用 `unittest` 完成可重复回归。
+- 与现有 `tests/test_tool_call_bridge.py` 互补：该文件聚焦工具桥接，本计划补齐基础模块行为。
+
+## 2. 范围与优先级
+- **P0（必须）**
+  1) 认证行为（`app/auth.py`）
+  2) 并发守卫行为（`app/concurrency.py`）
+  3) 会话缓存与工具配置指纹（`app/session_cache.py`）
+- **P1（应覆盖）**
+  4) OpenAI/Anthropic 内容规范化（`app/openai_schema.py`, `app/anthropic_schema.py`）
+
+## 3. 用例矩阵
+| 用例ID | 优先级 | 模块 | 场景 | 预期 |
+|---|---|---|---|---|
+| TC-AUTH-01 | P0 | auth | Bearer 正确 token | 认证通过 |
+| TC-AUTH-02 | P0 | auth | 缺失/错误 Authorization | 401 + `invalid_api_key` |
+| TC-AUTH-03 | P0 | auth | Anthropic `x-api-key` 与 Bearer 兜底 | 正确 key 通过，缺失时报 `AnthropicAuthError` |
+| TC-AUTH-04 | P0 | auth | metrics 在未配置 token 且非 public | 503 + `metrics_disabled` |
+| TC-CONC-01 | P0 | concurrency | `max_in_flight<=0` 无限制模式 | 获取/释放计数正确，release 幂等 |
+| TC-CONC-02 | P0 | concurrency | 单槽占用后第二请求超时 | 抛 `BackpressureRejected`，rejected 计数+1 |
+| TC-SESS-01 | P0 | session_cache | `hash_user_context` 忽略 assistant/tool | 哈希不受 assistant/tool 变化影响 |
+| TC-SESS-02 | P0 | session_cache | key 包含 tool_config 指纹 | 同语义配置同 key，配置变化 key 变化 |
+| TC-SESS-03 | P0 | session_cache | LRU 淘汰 | 超限后旧项淘汰，`evict_total` 增加 |
+| TC-SESS-04 | P0 | session_cache | TTL 过期 | 读取 miss，`expire_total` 增加 |
+| TC-SCHEMA-01 | P1 | openai_schema | 多类型 content flatten | 文本合并，图片/音频占位 |
+| TC-SCHEMA-02 | P1 | anthropic_schema | tool_use/tool_result flatten | 生成可读文本片段 |
+| TC-SCHEMA-03 | P1 | anthropic_schema | `anthropic_to_internal_messages` | system + messages 正确映射 |
+| TC-SCHEMA-04 | P1 | anthropic_schema | `affinity_key_for_anthropic` 优先级 | `metadata.user_id` 优先，fallback 为 hash 前缀 |
+
+## 4. 测试文件落地
+- 既有：`tests/test_tool_call_bridge.py`
+- 新增：
+  - `tests/test_auth_concurrency.py`
+  - `tests/test_session_cache_tooling.py`
+  - `tests/test_schema_normalization.py`
+
+## 5. 执行步骤
+1. 定点执行新增测试文件。
+2. 全量执行 `tests/` 下 `test_*.py`。
+3. 汇总通过率与失败项（若失败，给出定位与修复建议）。
+
+## 6. 执行命令
+```bash
+python3 -m unittest tests/test_auth_concurrency.py
+python3 -m unittest tests/test_session_cache_tooling.py
+python3 -m unittest tests/test_schema_normalization.py
+python3 -m unittest tests/test_tool_call_bridge.py
+python3 -m unittest discover -s tests -p "test_*.py"
+```

tests/test_auth_concurrency.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import asyncio
+import unittest
+
+from fastapi import HTTPException
+from starlette.requests import Request
+
+from app.auth import AnthropicAuthError, require_anthropic_key, require_bearer, require_metrics_access
+from app.concurrency import BackpressureRejected, InFlightGuard
+
+
+def _req(headers: dict[str, str] | None = None) -> Request:
+    pairs = []
+    for k, v in (headers or {}).items():
+        pairs.append((k.lower().encode("latin-1"), v.encode("latin-1")))
+    scope = {
+        "type": "http",
+        "http_version": "1.1",
+        "method": "GET",
+        "scheme": "http",
+        "path": "/x",
+        "raw_path": b"/x",
+        "query_string": b"",
+        "headers": pairs,
+        "client": ("test", 1),
+        "server": ("test", 80),
+        "root_path": "",
+    }
+    return Request(scope)
+
+
+class AuthAndConcurrencyTests(unittest.IsolatedAsyncioTestCase):
+    def test_require_bearer_accepts_valid_token(self) -> None:
+        request = _req({"authorization": "Bearer good"})
+        require_bearer(request, ["good"])
+
+    def test_require_bearer_rejects_invalid_token(self) -> None:
+        request = _req({"authorization": "Bearer bad"})
+        with self.assertRaises(HTTPException) as ctx:
+            require_bearer(request, ["good"])
+        self.assertEqual(ctx.exception.status_code, 401)
+        self.assertEqual(ctx.exception.detail["error"]["code"], "invalid_api_key")
+
+    def test_require_anthropic_key_accepts_x_api_key_or_bearer(self) -> None:
+        request_x = _req({"x-api-key": "k1"})
+        require_anthropic_key(request_x, ["k1"])
+
+        request_b = _req({"authorization": "Bearer k2"})
+        require_anthropic_key(request_b, ["k2"])
+
+    def test_require_anthropic_key_raises_on_missing(self) -> None:
+        request = _req()
+        with self.assertRaises(AnthropicAuthError) as ctx:
+            require_anthropic_key(request, ["k"])
+        self.assertEqual(ctx.exception.status_code, 401)
+        self.assertEqual(ctx.exception.error_type, "authentication_error")
+
+    def test_require_metrics_access_503_when_no_tokens_configured(self) -> None:
+        request = _req({"authorization": "Bearer any"})
+        with self.assertRaises(HTTPException) as ctx:
+            require_metrics_access(request, api_keys=[], metrics_token="", public=False)
+        self.assertEqual(ctx.exception.status_code, 503)
+        self.assertEqual(ctx.exception.detail["error"]["code"], "metrics_disabled")
+
+    async def test_inflight_guard_unlimited_and_release_idempotent(self) -> None:
+        guard = InFlightGuard(max_in_flight=0, queue_timeout_sec=0.01)
+        ticket = await guard.try_acquire()
+        self.assertEqual(guard.in_flight, 1)
+        ticket.release()
+        ticket.release()
+        self.assertEqual(guard.in_flight, 0)
+        self.assertEqual(guard.accepted_total, 1)
+
+    async def test_inflight_guard_rejects_when_queue_timeout(self) -> None:
+        guard = InFlightGuard(max_in_flight=1, queue_timeout_sec=0.01)
+        first = await guard.try_acquire()
+        with self.assertRaises(BackpressureRejected):
+            await guard.try_acquire()
+        self.assertEqual(guard.rejected_total, 1)
+        first.release()
+        self.assertEqual(guard.in_flight, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pool_stats_config.py

@@ -0,0 +1,216 @@
+from __future__ import annotations
+
+import json
+import os
+import sys
+import types
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+# app.lingma_pool imports auto_login; tests here don't execute Playwright paths.
+# Stub module import so test environments without playwright can import pool code.
+_playwright = types.ModuleType("playwright")
+_playwright_async = types.ModuleType("playwright.async_api")
+
+
+class _StubPlaywrightTimeoutError(Exception):
+    pass
+
+
+async def _stub_async_playwright():
+    raise RuntimeError("playwright is stubbed in unit tests")
+
+
+_playwright_async.TimeoutError = _StubPlaywrightTimeoutError
+_playwright_async.async_playwright = _stub_async_playwright
+sys.modules.setdefault("playwright", _playwright)
+sys.modules.setdefault("playwright.async_api", _playwright_async)
+
+from app.config import _parse_accounts, load_settings
+from app.lingma_pool import LingmaPool
+from app.stats import StatsCollector, estimate_tokens
+
+
+def _affinity_key_for_bucket(pool_size: int, bucket_index: int) -> str:
+    for i in range(20000):
+        key = f"k-{i}"
+        if abs(hash(key)) % pool_size == bucket_index:
+            return key
+    raise RuntimeError("failed to find affinity key")
+
+
+class _FakeInstance:
+    def __init__(self, idx: int, *, healthy: bool, in_flight: int):
+        self.name = f"inst-{idx}"
+        self.cfg = SimpleNamespace(index=idx)
+        self._healthy = healthy
+        self.in_flight = in_flight
+
+    @property
+    def healthy(self) -> bool:
+        return self._healthy
+
+
+class LingmaPoolRoutingTests(unittest.TestCase):
+    def test_pool_pick_prefers_healthy_affinity_bucket(self) -> None:
+        inst0 = _FakeInstance(0, healthy=True, in_flight=0)
+        inst1 = _FakeInstance(1, healthy=True, in_flight=9)
+        pool = LingmaPool([inst0, inst1])
+
+        key = _affinity_key_for_bucket(2, 1)
+        picked = pool.pick(affinity_key=key)
+
+        self.assertIs(picked, inst1)
+
+    def test_pool_pick_falls_back_to_least_in_flight_when_affinity_unhealthy(self) -> None:
+        inst0 = _FakeInstance(0, healthy=True, in_flight=1)
+        inst1 = _FakeInstance(1, healthy=False, in_flight=0)
+        inst2 = _FakeInstance(2, healthy=True, in_flight=1)
+        pool = LingmaPool([inst0, inst1, inst2])
+
+        key = _affinity_key_for_bucket(3, 1)
+        picked = pool.pick(affinity_key=key)
+
+        self.assertIs(picked, inst0)
+
+    def test_pool_pick_round_robin_when_all_unhealthy(self) -> None:
+        inst0 = _FakeInstance(0, healthy=False, in_flight=0)
+        inst1 = _FakeInstance(1, healthy=False, in_flight=0)
+        inst2 = _FakeInstance(2, healthy=False, in_flight=0)
+        pool = LingmaPool([inst0, inst1, inst2])
+
+        self.assertIs(pool.pick(), inst0)
+        self.assertIs(pool.pick(), inst1)
+        self.assertIs(pool.pick(), inst2)
+        self.assertIs(pool.pick(), inst0)
+
+    def test_pool_prometheus_lines_include_required_metrics(self) -> None:
+        inst0 = _FakeInstance(0, healthy=True, in_flight=2)
+        inst1 = _FakeInstance(1, healthy=False, in_flight=5)
+        pool = LingmaPool([inst0, inst1])
+
+        text = "\n".join(pool.prometheus_lines())
+
+        self.assertIn("# TYPE gateway_pool_instance_in_flight gauge", text)
+        self.assertIn("# TYPE gateway_pool_instance_ready gauge", text)
+        self.assertIn('gateway_pool_instance_in_flight{name="inst-0",idx="0"} 2', text)
+        self.assertIn('gateway_pool_instance_ready{name="inst-0",idx="0"} 1', text)
+        self.assertIn('gateway_pool_instance_ready{name="inst-1",idx="1"} 0', text)
+
+
+class StatsCollectorTests(unittest.IsolatedAsyncioTestCase):
+    def test_estimate_tokens_empty_short_utf8(self) -> None:
+        self.assertEqual(estimate_tokens(""), 0)
+        self.assertGreaterEqual(estimate_tokens("a"), 1)
+        self.assertEqual(estimate_tokens("你好世界"), 3)
+
+    async def test_record_chat_updates_counters_and_clamps_negative_tokens(self) -> None:
+        s = StatsCollector()
+
+        await s.record_chat(stream=True, success=True, prompt_tokens=-3, completion_tokens=5)
+        await s.record_chat(stream=False, success=False, prompt_tokens=2, completion_tokens=-7)
+        snap = await s.snapshot()
+
+        self.assertEqual(snap["chat_requests_total"], 2)
+        self.assertEqual(snap["chat_requests_success"], 1)
+        self.assertEqual(snap["chat_requests_error"], 1)
+        self.assertEqual(snap["chat_stream_requests"], 1)
+        self.assertEqual(snap["chat_non_stream_requests"], 1)
+        self.assertEqual(snap["prompt_tokens_estimated_total"], 2)
+        self.assertEqual(snap["completion_tokens_estimated_total"], 5)
+
+    async def test_snapshot_and_prometheus_text_consistency(self) -> None:
+        s = StatsCollector()
+
+        await s.record_chat(stream=True, success=True, prompt_tokens=3, completion_tokens=4)
+        snap = await s.snapshot()
+        text = await s.prometheus_text()
+
+        self.assertEqual(snap["total_tokens_estimated"], 7)
+        self.assertIn("gateway_total_tokens_estimated 7", text)
+        self.assertIn("gateway_chat_requests_total 1", text)
+        self.assertTrue(text.endswith("\n"))
+
+
+class ConfigParsingTests(unittest.TestCase):
+    def test_parse_accounts_accepts_json_csv_newline_formats(self) -> None:
+        raw_json = json.dumps([
+            {"username": "u1", "password": "p1"},
+            {"username": "u2", "password": "p2"},
+        ])
+        parsed_json = _parse_accounts(raw_json)
+        self.assertEqual([a.username for a in parsed_json], ["u1", "u2"])
+
+        parsed_csv = _parse_accounts("u3:p3,u4:p4")
+        self.assertEqual([a.username for a in parsed_csv], ["u3", "u4"])
+
+        parsed_nl = _parse_accounts("u5:p5\nu6:p6")
+        self.assertEqual([a.username for a in parsed_nl], ["u5", "u6"])
+
+    def test_parse_accounts_allows_bundle_only_in_json(self) -> None:
+        raw = json.dumps([{"session_bundle": "abc"}])
+        parsed = _parse_accounts(raw)
+
+        self.assertEqual(len(parsed), 1)
+        self.assertEqual(parsed[0].username, "")
+        self.assertEqual(parsed[0].password, "")
+        self.assertEqual(parsed[0].session_bundle_b64, "abc")
+
+    def test_parse_accounts_csv_splits_only_first_colon(self) -> None:
+        parsed = _parse_accounts("u:p:with:colon")
+
+        self.assertEqual(len(parsed), 1)
+        self.assertEqual(parsed[0].username, "u")
+        self.assertEqual(parsed[0].password, "p:with:colon")
+
+    def test_load_settings_creates_bundle_only_account_without_credentials(self) -> None:
+        with patch.dict(os.environ, {"LINGMA_SESSION_BUNDLE": "abc"}, clear=True):
+            settings = load_settings()
+
+        self.assertEqual(len(settings.accounts), 1)
+        self.assertEqual(settings.accounts[0].username, "")
+        self.assertEqual(settings.accounts[0].password, "")
+        self.assertEqual(settings.accounts[0].session_bundle_b64, "abc")
+
+    def test_load_settings_invalid_instance_count_fallback(self) -> None:
+        with patch.dict(
+            os.environ,
+            {"LINGMA_ACCOUNTS": "u1:p1,u2:p2", "LINGMA_INSTANCE_COUNT": "not-a-number"},
+            clear=True,
+        ):
+            settings_with_accounts = load_settings()
+
+        self.assertEqual(settings_with_accounts.instance_count, 2)
+
+        with patch.dict(os.environ, {"LINGMA_INSTANCE_COUNT": "not-a-number"}, clear=True):
+            settings_without_accounts = load_settings()
+
+        self.assertEqual(settings_without_accounts.instance_count, 1)
+    def test_load_settings_parses_tool_allowlist_csv(self) -> None:
+        with patch.dict(os.environ, {"TOOL_ALLOWLIST": " lookup , write_file ,,search_docs "}, clear=True):
+            settings = load_settings()
+
+        self.assertEqual(settings.tool_allowlist, ["lookup", "write_file", "search_docs"])
+
+    def test_load_settings_defaults_tool_forward_enabled_true(self) -> None:
+        with patch.dict(os.environ, {}, clear=True):
+            settings = load_settings()
+
+        self.assertTrue(settings.tool_forward_enabled)
+
+    def test_load_settings_respects_tool_forward_enabled_false(self) -> None:
+        with patch.dict(os.environ, {"TOOL_FORWARD_ENABLED": "false"}, clear=True):
+            settings = load_settings()
+
+        self.assertFalse(settings.tool_forward_enabled)
+
+    def test_load_settings_empty_tool_allowlist(self) -> None:
+        with patch.dict(os.environ, {"TOOL_ALLOWLIST": "  , ,  "}, clear=True):
+            settings = load_settings()
+
+        self.assertEqual(settings.tool_allowlist, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_schema_normalization.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import unittest
+
+from app.anthropic_schema import (
+    AnthropicMessagesRequest,
+    affinity_key_for_anthropic,
+    anthropic_to_internal_messages,
+    flatten_anthropic_content,
+)
+from app.openai_schema import flatten_content
+
+
+class SchemaNormalizationTests(unittest.TestCase):
+    def test_openai_flatten_content_with_multimodal_parts(self) -> None:
+        out = flatten_content(
+            [
+                {"type": "text", "text": "hello"},
+                {"type": "image_url", "image_url": {"url": "x"}},
+                {"type": "input_image", "image_url": {"url": "y"}},
+                {"type": "input_audio", "input_audio": {"data": "x"}},
+                {"type": "text", "text": "world"},
+            ]
+        )
+        self.assertEqual(out, "hello\n[image]\n[image]\n[audio]\nworld")
+
+    def test_anthropic_flatten_content_with_tool_blocks(self) -> None:
+        out = flatten_anthropic_content(
+            [
+                {"type": "text", "text": "before"},
+                {"type": "tool_use", "name": "search", "input": {"q": "hi"}},
+                {"type": "tool_result", "content": "ok"},
+            ]
+        )
+        self.assertIn("before", out)
+        self.assertIn("[tool_use]", out)
+        self.assertIn("[tool_result] ok", out)
+
+    def test_anthropic_to_internal_messages_maps_system_and_messages(self) -> None:
+        req = AnthropicMessagesRequest(
+            model="org_auto",
+            max_tokens=64,
+            system="sys",
+            messages=[
+                {"role": "user", "content": "u1"},
+                {"role": "assistant", "content": "a1"},
+            ],
+        )
+        out = anthropic_to_internal_messages(req)
+        self.assertEqual(out[0], {"role": "system", "content": "sys"})
+        self.assertEqual(out[1], {"role": "user", "content": "u1"})
+        self.assertEqual(out[2], {"role": "assistant", "content": "a1"})
+
+    def test_affinity_key_for_anthropic_priority(self) -> None:
+        req_user = AnthropicMessagesRequest(
+            model="org_auto",
+            max_tokens=64,
+            metadata={"user_id": "u-1"},
+            messages=[{"role": "user", "content": "hello"}],
+        )
+        self.assertEqual(affinity_key_for_anthropic(req_user), "u-1")
+
+        req_fallback = AnthropicMessagesRequest(
+            model="org_auto",
+            max_tokens=64,
+            messages=[{"role": "user", "content": "hello"}],
+        )
+        key = affinity_key_for_anthropic(req_fallback)
+        self.assertIsInstance(key, str)
+        self.assertTrue(key.startswith("first:"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_session_cache_tooling.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import unittest
+
+from app.session_cache import SessionCache, hash_branch_context, hash_user_context
+
+
+class SessionCacheToolingTests(unittest.IsolatedAsyncioTestCase):
+    def test_hash_user_context_ignores_assistant_and_tool(self) -> None:
+        base = [
+            {"role": "system", "content": "S"},
+            {"role": "user", "content": "U"},
+        ]
+        with_extra = base + [
+            {"role": "assistant", "content": "A1"},
+            {"role": "tool", "content": "T1"},
+        ]
+        self.assertEqual(hash_user_context(base), hash_user_context(with_extra))
+
+    def test_hash_branch_context_distinguishes_assistant_tool_branch(self) -> None:
+        base = [
+            {"role": "system", "content": "S"},
+            {"role": "user", "content": "U"},
+            {"role": "assistant", "content": "A1"},
+            {"role": "tool", "content": "T1", "tool_call_id": "call-1"},
+        ]
+        changed = [
+            {"role": "system", "content": "S"},
+            {"role": "user", "content": "U"},
+            {"role": "assistant", "content": "A2"},
+            {"role": "tool", "content": "T1", "tool_call_id": "call-1"},
+        ]
+        self.assertNotEqual(hash_branch_context(base), hash_branch_context(changed))
+
+    def test_build_key_changes_with_tool_config(self) -> None:
+        cache = SessionCache(max_entries=8, ttl_sec=60)
+        msgs = [{"role": "user", "content": "hi"}]
+        key1 = cache.build_key("k", msgs, tool_config={"a": 1, "b": 2})
+        key2 = cache.build_key("k", msgs, tool_config={"b": 2, "a": 1})
+        key3 = cache.build_key("k", msgs, tool_config={"a": 1})
+        self.assertEqual(key1, key2)
+        self.assertNotEqual(key1, key3)
+
+    def test_build_key_keeps_legacy_shape_without_branch_context(self) -> None:
+        cache = SessionCache(max_entries=8, ttl_sec=60)
+        msgs = [{"role": "user", "content": "hi"}]
+        legacy = cache.build_key("k", msgs)
+        with_branch = cache.build_key("k", msgs, branch_context="abc")
+        self.assertEqual(legacy.count(":"), 2)
+        self.assertEqual(with_branch.count(":"), 3)
+
+    async def test_lru_evicts_oldest(self) -> None:
+        cache = SessionCache(max_entries=2, ttl_sec=600)
+        await cache.put("k1", "s1")
+        await cache.put("k2", "s2")
+        await cache.put("k3", "s3")
+        self.assertIsNone(await cache.get("k1"))
+        self.assertEqual(cache.evict_total, 1)
+
+    async def test_ttl_expiry_increments_expire_counter(self) -> None:
+        cache = SessionCache(max_entries=4, ttl_sec=0.001)
+        await cache.put("k1", "s1")
+        await __import__("asyncio").sleep(0.01)
+        self.assertIsNone(await cache.get("k1"))
+        self.assertEqual(cache.expire_total, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()