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>
fix: enable tool forwarding by default and add config regression tests
2026-04-21 13:57:44 +08:00 · 2026-04-21 13:41:41 +08:00 · 2026-04-21 08:05:09 +08:00 · 2026-04-20 19:24:02 +08:00 · 2026-04-20 15:08:16 +08:00 · 2026-04-20 14:55:32 +08:00
22 changed files with 4747 additions and 455 deletions
--- a/.env.example
+++ b/.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=
--- a/.gitignore
+++ b/.gitignore
@@ -7,3 +7,4 @@ data/*
 !data/.gitkeep
 secrets/*
 !secrets/.gitkeep
 .gitnexus
--- a/.omc/plans/app-main-split-plan.md
+++ b/.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"`。
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -0,0 +1,272 @@
 # CLAUDE.md
 This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
 ## Primary docs to read first
 - `README.md` (runtime commands, env model, API examples)
 - `DESIGN.md` (architecture decisions, module boundaries, request lifecycle)
 - `.env.example` (authoritative env var reference)
 No Cursor/Copilot rule files were found in this repo (`.cursorrules`, `.cursor/rules/`, `.github/copilot-instructions.md`).
 ## Common development commands
 ### Start locally
 ```bash
 pip install -r requirements.txt
 uvicorn app.main:app --reload --port 8317
 ```
 ### Start with Docker Compose
 ```bash
 cp .env.example .env
 mkdir -p data secrets
 docker compose up -d --build
 docker compose logs -f
 ```
 ### Run tests
 ```bash
 # current focused suite
 python3 -m unittest tests/test_tool_call_bridge.py
 # discover all unittest tests under tests/
 python3 -m unittest discover -s tests -p "test_*.py"
 # run a single test method
 python3 -m unittest tests.test_tool_call_bridge.ToolCallBridgeTests.test_openai_non_stream_bridges_tool_calls
 ```
 ### Smoke-check running gateway
 ```bash
 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"
 ```
 ### Linting/type-checking status
 - There is currently no repo-configured lint/type command (no `ruff`/`flake8`/`mypy` config found).
 - Do not invent tooling commands; if linting is needed, add tooling in a dedicated change first.
 ## Architecture (big picture)
 ### What this service is
 A FastAPI gateway that fronts Lingma and exposes:
 - OpenAI-compatible API (`/v1/models`, `/v1/chat/completions`)
 - Anthropic Messages-compatible API (`/v1/messages`, `/v1/messages/count_tokens`)
 Both protocols share the same backend pool, backpressure guard, stats, and session reuse logic.
 ### Request lifecycle (important for most changes)
 1. Authenticate request (`app/auth.py`)
 2. Normalize inbound protocol payload to internal message shape (`openai_schema.py` / `anthropic_schema.py`)
 3. Session-cache lookup (`app/session_cache.py`) for prefix-based reuse
 4. Pick backend instance (`app/lingma_pool.py`) with affinity + least-in-flight
 5. Acquire concurrency ticket (`app/concurrency.py`)
 6. Call Lingma via websocket/LSP client (`app/lingma_client.py`)
 7. Map upstream result/stream back to wire protocol in `app/main.py`
 8. Record stats and release ticket (including stream-finally paths)
 ### Core module boundaries
 - `app/main.py`: API entrypoint + orchestration + wire-format adapters
 - `app/lingma_pool.py`: multi-instance lifecycle, selection, health-aware fallback
 - `app/lingma_client.py`: subprocess + LSP-over-WebSocket transport to Lingma
 - `app/session_cache.py`: LRU+TTL cache of conversation-prefix -> upstream session id (+ instance binding)
 - `app/concurrency.py`: in-flight guard and queue timeout/backpressure behavior
 - `app/stats.py`: usage counters and Prometheus text
 ### Protocol-specific notes
 - Anthropic and OpenAI endpoints are separate adapters over shared internals.
 - Response-side tool bridge is implemented: upstream Lingma tool events are surfaced as:
  - OpenAI: `tool_calls` (stream + non-stream)
  - Anthropic: `tool_use` / `tool_result` blocks (stream + non-stream)
 - Request-side `tools` / `tool_choice` are accepted by schemas but not forwarded to Lingma.
 ### Operational invariants to preserve
 - One request must stay on one Lingma instance for session continuity.
 - Session cache entries include instance identity; invalidate on unhealthy instance mismatch.
 - Streaming paths must always release in-flight tickets in `finally`.
 - Multi-instance mode must use isolated workdirs per instance.
 ### Deployment/runtime model
 - Container startup runs `python /app/app/bootstrap_lingma.py` before uvicorn.
 - 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 -->
--- a/DESIGN.md
+++ b/DESIGN.md
@@ -47,8 +47,9 @@
 - **逆向 Lingma 后端协议**：之前评估过（曾经的"B1 终极方案"），需要反编译二进制，维护成本高、政策风险大，放弃。
 - **多租户 / 水平扩缩**：单容器即可；真要大规模部署 → 套层反代 + N 个网关副本就够，不在进程内解决。
- **完整 function calling / tools**：OpenAI schema 里保留了字段，但目前不透传给 Lingma（Lingma 侧没有等价能力）。
+- **请求侧完整 function calling / tools 语义**：仍不是当前目标；现阶段仅支持 `tools`/`tool_choice` 在 `TOOL_FORWARD_ENABLED` 开关下灰度透传（默认关闭）。
- **多模态**：请求里的 image/audio 会被降级成占位符 `[image]` / `[audio]`，因为 Lingma chat 不支持。
+- **响应侧工具事件桥接**：若 Lingma 上游产出 tool 事件，网关会向 OpenAI 输出 `tool_calls`，向 Anthropic 输出 `tool_use` / `tool_result`（stream + non-stream）。
 - **强制工具回退闭环（non-stream）**：当上游未返回 tool 事件且请求为强制 `tool_choice` 时，网关会从文本里解析严格 JSON，合成 OpenAI `tool_calls` 与 Anthropic `tool_use` / `tool_result`。
 ---
@@ -517,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])`
@@ -591,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 需要 Lingma 上游支持，payload 转发点在 `chat_stream` / `chat_complete` |
+| 扩展 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 释放顺序（内层先释放） |
@@ -599,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 命中率 |
@@ -611,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}
 ```
 主要断点位置：
@@ -627,7 +629,7 @@ uvicorn app.main:app --reload --port 8317
 | 标签 | 描述 | 影响 | 计划 |
 |---|---|---|---|
 | D1 | `config.py` 还是纯 `dataclass` + `os.getenv`，未迁 `pydantic-settings` | 类型校验靠自己 cast | 低优，收益有限，有精力再做 |
-| D3 | 无单元测试骨架 | 重构要靠 deploy 验证 | 想加 CI 时优先补 |
+| D3 | 已有基础单测覆盖 tool-call bridge（OpenAI/Anthropic，stream + non-stream），但整体测试矩阵仍不完整 | 回归仍依赖手工验证与定向测试 | 后续补充会话复用、背压、鉴权和异常路径用例 |
 | Docker non-root | 容器还是 root 跑 | 容器逃逸时影响宿主 | 需要加 `gosu` + chown entrypoint，涉及数据迁移，谨慎推进 |
 | ADMIN_TOKEN 轮换 | 没有过期机制，只能重启 | 自用场景不影响 | 接 Vault / sops 时一并做 |
 | Lingma 版本漂移 | 新版 Lingma 改 LSP 方法或新增必需 cache 文件时会无声崩 | 注入失败会 fallback，但 chat 不回话题型的错误不易定位 | 加一个 `/internal/smoke` 端点做端到端自检 |
@@ -707,6 +709,45 @@ uvicorn app.main:app --reload --port 8317
 | → | `chat/ask` (notify!) | 见 `_build_payload` | 不回 result；通过 server push 下推 |
 | ← | `chat/answer` | `{requestId, text, content}` | 流式 token |
 | ← | `chat/finish` | `{requestId, sessionId, ...其它元数据}` | 结束信号，含上游真实 sessionId |
 | ← | `tool/call/sync` | `{requestId?, toolCallId, toolCallStatus, parameters, results?}` | 工具状态与结果回流 |
 | ← | `tool/invoke` | `{requestId?, toolCallId, ...}` | 工具调用中间事件（兼容旧链路） |
 | ← | `tool/call/approve` | `{requestId?, toolCallId, approval, ...}` | 工具审批事件 |
 | ← | `tool/invokeResult` | `{requestId?, toolCallId, name, success, errorMessage, result}` | 工具执行结果事件 |
 ### 9.1 Tool call 监控 SOP（VSCode 真实环境）
 目标：拿到 Lingma 扩展真实 method/字段，避免猜测协议。
 1. 确认入口文件
   - `~/.vscode/extensions/alibaba-cloud.tongyi-lingma-*/package.json`
   - 查 `main`（当前是 `dist/extension.js`）
 2. 在发送侧打点
   - 在 `sendRequest` / `sendNotification` 处记录 method 与参数 keys
   - 优先写文件，不依赖 console
 3. 在入站 `tool/call/sync` handler 打点
   - 记录 `toolCallId`、`toolCallStatus`、是否包含 `results`
 4. 用真实交互触发
   - VSCode 内发起会话并触发工具
   - 点击 Accept/Reject，观察事件闭环
 5. 验证闭环
   - `tool/call/sync(pending|processing)`
   - `tool/call/approve`
   - `tool/invokeResult`
   - `tool/call/sync(results)`
 6. 回滚
   - 用备份文件恢复 `dist/extension.js`
   - 避免长期携带探针到日常环境
 **建议日志位置**：
 - `~/.lingma/vscode/sharedClientCache/logs/lingma-probe.log`
 - `~/.lingma/vscode/sharedClientCache/logs/lingma-extension.log`
 **注意**：优先使用 VSCode，不混用 Cursor 扩展环境；`pipe` 连接模式下，扩展层探针最稳定。
 **`chat/ask` payload 关键字段**：
--- a/2
+++ b/2
@@ -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}"]
--- a/README.md
+++ b/README.md
@@ -1,395 +1,216 @@
 # Lingma OpenAI Gateway
-把本地 Lingma 插件封装成 OpenAI 兼容接口。任何能调 OpenAI 的客户端（Cursor、Dify、LangChain、curl…）都能直接接入。
+将 Lingma 封装为 OpenAI / Anthropic 兼容网关，便于现有客户端直接接入。
-**支持：**
+- OpenAI：`/v1/models`、`/v1/chat/completions`（含 stream）
- OpenAI 兼容：`GET /v1/models` / `POST /v1/chat/completions`（含 SSE 流式） / Bearer 鉴权
+- Anthropic：`/v1/messages`、`/v1/messages/count_tokens`（含 stream）
- **Anthropic 兼容**：`POST /v1/messages`（含 Anthropic SSE 事件流） / `x-api-key` 鉴权
+- 内置：多实例池、会话复用、Prometheus 指标、登录态 bundle 注入
- Prometheus / 多账号实例池 / 会话复用（跨两种协议共享） / 免浏览器登录态注入
+- 多模态降级：OpenAI `image_url` / `input_image` 转 `[image]`，`input_audio` 转 `[audio]`；Anthropic `image` 转 `[image]`
-> 想看架构、模块划分、设计决策、二开路线图 → 直接读 [`DESIGN.md`](./DESIGN.md)。
+> 架构设计与二开细节请看 [`DESIGN.md`](./DESIGN.md)。
 ---
-## 架构速览
+## 目录
-```
+1. [5 分钟启动](#5-分钟启动)
- ┌─────────────┐   OpenAI 协议    ┌─────────────────────────────────────────┐
+2. [常用命令](#常用命令)
- │  任意客户端  │  ───────────▶    │  FastAPI (app/main.py)                  │
+3. [最小 API 示例](#最小-api-示例)
- │  (curl/     │                  │  ├─ auth_guard / admin_guard            │
+4. [部署与更新](#部署与更新)
- │   Cursor/   │                  │  ├─ chat_guard (InFlightGuard 背压)     │
+5. [排障速查](#排障速查)
- │   Dify…)    │                  │  ├─ SessionCache (LRU+TTL, KV 复用)     │
+6. [文档入口](#文档入口)
 └─────────────┘                  │  └─ 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 /…     │
                └────────────────────┘
 ```
 ---
-## 一、快速开始
+## 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:${PORT}/healthz"
-curl -s http://127.0.0.1:8317/v1/chat/completions \
+curl -s "http://127.0.0.1:${PORT}/v1/models" \
-  -H "Authorization: Bearer $API_KEY" \
+  -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",
+    "model": "org_auto",
-    "stream":true,
+    "messages": [{"role": "user", "content": "hi"}],
-    "stream_options":{"include_usage":true},
+    "stream": false
    "messages":[{"role":"user","content":"介绍一下你自己"}]
  }'
 ```
-**Anthropic Messages 示例（非流式）**
+### OpenAI：流式
 ```bash
-curl -s http://127.0.0.1:8317/v1/messages \
+curl -N "http://127.0.0.1:${PORT}/v1/chat/completions" \
-  -H "x-api-key: $API_KEY" \
+  -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",
+    "model": "claude-3-5-sonnet-20241022",
-    "max_tokens":256,
+    "max_tokens": 256,
-    "system":"你是一个简洁的助手",
+    "messages": [{"role": "user", "content": "hi"}],
-    "messages":[{"role":"user","content":"你好"}]
+    "stream": false
  }'
 ```
-**Anthropic Messages 示例（流式）**
+### Anthropic：流式
 ```bash
-curl -N http://127.0.0.1:8317/v1/messages \
+curl -N "http://127.0.0.1:${PORT}/v1/messages" \
-  -H "x-api-key: $API_KEY" \
+  -H "x-api-key: ${API_KEY}" \
  -H "anthropic-version: 2023-06-01" \
  -H "Content-Type: application/json" \
  -d '{
-    "model":"claude-3-5-sonnet-20241022",
+    "model": "claude-3-5-sonnet-20241022",
-    "max_tokens":256,
+    "max_tokens": 256,
-    "stream":true,
+    "messages": [{"role": "user", "content": "say hi"}],
-    "messages":[{"role":"user","content":"写一首四行诗"}]
+    "stream": true
  }'
 # 返回 message_start / content_block_start / content_block_delta* /
 #      content_block_stop / message_delta / message_stop
 ```
-说明：
+### Anthropic：count_tokens
 - **模型名兼容**：客户端可以继续传 `claude-3-*` 等名字；未识别的 model 会回退到 `DEFAULT_MODEL` 对应的 Lingma key，后端实际仍由 Lingma 提供（Qwen 系列）。如需显式选模型，直接传 Lingma key（`dashscope_qmodel` 等）。
 - **会话复用共享**：Anthropic 与 OpenAI 两个端点共用同一 `SessionCache`，只要 API key 相同、对话前缀相同，就会命中同一上游 `sessionId`。
 - **多模态**：`image` 块会被降级为 `[image]` 占位符（Lingma 不支持 vision）；`tool_use` / `tool_result` 会以纯文本形式保留语义。
 - **鉴权**：优先 `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）
 **从已登录实例导出：**
 ```bash
-curl -sS -X POST \
+curl -s "http://127.0.0.1:${PORT}/v1/messages/count_tokens" \
-  -H "Authorization: Bearer $ADMIN_TOKEN" \
+  -H "x-api-key: ${API_KEY}" \
-  "http://host:port/internal/session/export" \
+  -H "anthropic-version: 2023-06-01" \
-  | jq -r '.bundle_b64' > secrets/lingma-session.b64
+  -H "Content-Type: application/json" \
-chmod 600 secrets/lingma-session.b64
+  -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）
+### 服务器更新到最新 main
 LINGMA_SESSION_BUNDLE=H4sIAAAA...
-# 多账号 JSON 模式，每账号独立 bundle
+```bash
-LINGMA_ACCOUNTS=[
+cd /root/lingma-openai-gateway
-  {"username":"u1","password":"p1","session_bundle_file":"/secrets/u1.b64"},
+git fetch origin
-  {"username":"u2","password":"p2","session_bundle":"H4sIAAAA..."}
+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）时才注入；不会覆盖活跃登录态。
+```bash
- 注入失败（损坏/权限）自动 fallback 到 Playwright。
+PORT=$(grep '^PORT=' .env | cut -d= -f2)
- bundle 只含 `cache/{id,user,quota,config.json}` 4 个文件；大小上限 4 MiB，实际通常 < 10 KB。
+curl -s "http://127.0.0.1:${PORT}/healthz"
 - **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
 ```
-关键指标：
+---
-| 指标 | 类型 | 意义 |
+## 排障速查
 | 现象 | 常见原因 | 处理 |
 |---|---|---|
-| `gateway_in_flight` / `gateway_queued` | gauge | 并发 / 排队 |
+| `/v1/*` 返回 401 | 缺失或错误 API key | 检查 `Authorization: Bearer` 或 `x-api-key` |
-| `gateway_rejected_total` | counter | 背压拒绝（429）累计 |
+| `healthz` 正常但请求失败 | 用错端口 | 以 `.env` 的 `PORT` 为准，`docker compose ps` 再确认 |
-| `gateway_pool_instance_ready{name}` | gauge | 每实例是否就绪（0/1） |
+| `git pull` 提示 not on a branch | 处于 detached HEAD | 执行 `git checkout -B main origin/main` |
-| `gateway_pool_instance_in_flight{name}` | gauge | 每实例在途 |
+| 自动登录不稳定 | 浏览器流程波动 | 优先使用 `LINGMA_SESSION_BUNDLE(_FILE)` |
-| `gateway_session_cache_hit_total` / `_miss_total` | counter | 会话复用命中率原料 |
+| 工具调用未触发 | 模型未选择工具 | 使用 `tool_choice` 强制，必要时约束输出 JSON |
 | `gateway_chat_requests_success` / `_error` | counter | chat 成功率 |
 ---
-## 五、升级注意事项
+## 文档入口
-从旧版本升级时注意**破坏性变更**（每一项都有 fallback，默认不会炸，但建议显式配置）：
+- 配置权威：[`/.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)
 | 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                   # 架构与二开手册
 ```
 ---
 ## License
-内部使用，按需调整。
+MIT
--- a/app/anthropic_schema.py
+++ b/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
+    `[{"role": "system"|"user"|"assistant", "content": "..."}]`. This keeps
-    session-cache hashing is identical across OpenAI and Anthropic callers —
+    user-input cache hashing aligned 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.
    """
    out: list[dict] = []
    if req.system:
--- a/app/config.py
+++ b/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
@@ -44,6 +49,8 @@ class Settings:
    session_reuse_enabled: bool = True
    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:
@@ -175,4 +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", True),
        tool_allowlist=_csv_env(os.getenv("TOOL_ALLOWLIST", "")),
    )
--- a/app/http/init.py
+++ b/app/http/init.py
--- a/app/http/responses_adapter.py
+++ b/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"
--- a/app/http/tool_bridge.py
+++ b/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,
    }
--- a/app/lingma_client.py
+++ b/app/lingma_client.py
@@ -9,7 +9,7 @@ import subprocess
 import time
 import uuid
 from pathlib import Path
-from typing import AsyncIterator, Callable, Optional
+from typing import Any, AsyncIterator, Callable, Optional
 import websockets
@@ -100,9 +100,90 @@ class LspWsRpcClient:
        self._reader_task: asyncio.Task | None = None
        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
    @staticmethod
    def _extract_tool_event(params: dict[str, Any]) -> dict[str, Any] | None:
        candidates: list[dict[str, Any]] = []
        def add_candidate(obj: Any) -> None:
            if isinstance(obj, dict):
                candidates.append(obj)
        add_candidate(params.get("toolCall"))
        add_candidate(params.get("tool_call"))
        add_candidate(params.get("tool"))
        data = params.get("data")
        if isinstance(data, dict):
            add_candidate(data.get("toolCall"))
            add_candidate(data.get("tool_call"))
            add_candidate(data.get("tool"))
        results = params.get("results")
        if isinstance(results, list):
            for item in results:
                add_candidate(item)
        if not candidates:
            fallback_id = params.get("toolCallId") or params.get("tool_call_id")
            if not fallback_id:
                return None
            return {
                "id": str(fallback_id),
                "name": str(params.get("name") or "tool"),
                "input": params.get("parameters") or {},
                "result": params.get("result"),
            }
        raw = candidates[0]
        tool_id = (
            raw.get("toolCallId")
            or raw.get("tool_call_id")
            or raw.get("id")
            or params.get("toolCallId")
            or params.get("tool_call_id")
        )
        name = (
            raw.get("name")
            or raw.get("toolName")
            or raw.get("tool_name")
            or params.get("name")
        )
        call_input = raw.get("input")
        if call_input is None:
            call_input = raw.get("arguments")
        if call_input is None:
            call_input = raw.get("args")
        if call_input is None:
            call_input = raw.get("parameters")
        if call_input is None:
            call_input = params.get("parameters")
        result_payload = raw.get("result")
        if result_payload is None:
            result_payload = params.get("result")
        if result_payload is None and isinstance(data, dict):
            result_payload = data.get("result")
        if result_payload is None and isinstance(raw.get("results"), list):
            result_payload = raw.get("results")
        if not tool_id:
            return None
        event: dict[str, Any] = {
            "id": str(tool_id),
            "name": str(name or "tool"),
            "input": call_input if call_input is not None else {},
        }
        if result_payload is not None:
            event["result"] = result_payload
        return event
    async def start(self):
        self._reader_task = asyncio.create_task(self._reader_loop())
@@ -123,6 +204,8 @@ class LspWsRpcClient:
                stream["done"].set()
                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:
@@ -172,6 +255,141 @@ class LspWsRpcClient:
                except Exception:
                    logger.exception("on_disconnect callback failed")
    @staticmethod
    def _normalize_tool_id(method: str, params: dict[str, Any], tool_event: dict[str, Any] | None) -> str | None:
        event_id = None
        if isinstance(tool_event, dict):
            event_id = tool_event.get("id")
        if isinstance(event_id, str) and event_id.strip():
            return event_id.strip()
        fallback_id = params.get("toolCallId") or params.get("tool_call_id")
        if isinstance(fallback_id, str) and fallback_id.strip():
            return fallback_id.strip()
        req_id = params.get("requestId")
        name = None
        if isinstance(tool_event, dict):
            name = tool_event.get("name")
        if not name:
            name = params.get("name")
        if isinstance(req_id, str) and req_id.strip() and isinstance(name, str) and name.strip():
            return f"{req_id.strip()}:tool:{name.strip()}"
        if isinstance(req_id, str) and req_id.strip():
            return f"{req_id.strip()}:tool"
        return None
    @staticmethod
    def _merge_tool_event(existing: dict[str, Any] | None, incoming: dict[str, Any]) -> tuple[dict[str, Any], bool]:
        merged = dict(existing or {})
        changed = False
        val = incoming.get("id")
        if val and merged.get("id") != val:
            merged["id"] = val
            changed = True
        name = incoming.get("name")
        if name:
            existing_name = merged.get("name")
            if not existing_name:
                merged["name"] = name
                changed = True
            else:
                existing_norm = str(existing_name).strip().lower()
                incoming_norm = str(name).strip().lower()
                if existing_norm == "tool" and incoming_norm != "tool":
                    merged["name"] = name
                    changed = True
                elif existing_norm != "tool" and incoming_norm == "tool":
                    pass
                elif merged.get("name") != name:
                    merged["name"] = name
                    changed = True
        if "input" in incoming and incoming.get("input") is not None:
            incoming_input = incoming.get("input")
            should_update_input = incoming_input != {} or "input" not in merged
            if should_update_input and merged.get("input") != incoming_input:
                merged["input"] = incoming_input
                changed = True
        if "result" in incoming and incoming.get("result") is not None:
            if merged.get("result") != incoming.get("result"):
                merged["result"] = incoming.get("result")
                changed = True
        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():
            stream = self._chat_streams.get(req_id)
            if stream is not None and tool_event is not None:
                tool_id = self._normalize_tool_id(method, params, tool_event)
                if tool_id:
                    self._tool_stream_map[tool_id] = req_id
            return stream
        if tool_event is not None:
            tool_id = self._normalize_tool_id(method, params, tool_event)
            if tool_id:
                mapped_req = self._tool_stream_map.get(tool_id)
                if mapped_req:
                    return self._chat_streams.get(mapped_req)
        return None
    async def _handle_server_message(self, msg: dict):
        method = msg.get("method")
        params = msg.get("params") or {}
@@ -185,7 +403,34 @@ class LspWsRpcClient:
                    stream["parts"].append(text)
                    if stream["first_chunk_at"] is None:
                        stream["first_chunk_at"] = time.monotonic()
-                    stream["chunks"].put_nowait(text)
+                    stream["chunks"].put_nowait({"type": "text", "text": text})
        if method in {"tool/call/sync", "tool/invoke", "tool/call/approve", "tool/invokeResult"}:
            tool_event = self._extract_tool_event(params)
            stream = self._resolve_tool_stream(method, params, tool_event)
            if stream is not None and tool_event is not None:
                tool_id = self._normalize_tool_id(method, params, tool_event)
                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)
                    merged, changed = self._merge_tool_event(existing, tool_event)
                    if not existing:
                        if "id" not in merged or not merged.get("id"):
                            merged["id"] = tool_id
                        tool_states[tool_id] = merged
                        order.append(tool_id)
                        stream["chunks"].put_nowait({"type": "tool", "tool": merged})
                    elif changed:
                        tool_states[tool_id] = merged
                        stream["chunks"].put_nowait({"type": "tool", "tool": merged})
            elif tool_event is not None:
                logger.warning("drop unroutable tool event: method=%s requestId=%s", method, params.get("requestId"))
        if method == "chat/finish":
            req_id = params.get("requestId")
@@ -224,6 +469,8 @@ class LspWsRpcClient:
            "chunks": asyncio.Queue(),
            "done": asyncio.Event(),
            "finish": None,
            "tool_states": {},
            "tool_order": [],
            "started_at": time.monotonic(),
            "first_chunk_at": None,
            "finish_at": None,
@@ -233,24 +480,36 @@ class LspWsRpcClient:
        stream = self._chat_streams.pop(request_id, None)
        if stream is None:
            return
        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()
            with contextlib.suppress(Exception):
                stream["chunks"].put_nowait(None)
-    async def consume_stream(self, request_id: str, timeout: float) -> AsyncIterator[str]:
+    async def consume_stream(self, request_id: str, timeout: float) -> AsyncIterator[dict[str, Any]]:
        stream = self._chat_streams.get(request_id)
        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:
@@ -261,11 +520,20 @@ class LspWsRpcClient:
            first_ms = int((stream["first_chunk_at"] - stream["started_at"]) * 1000)
        if stream.get("finish_at") is not None:
            total_ms = int((stream["finish_at"] - stream["started_at"]) * 1000)
        ordered_tool_events: list[dict[str, Any]] = []
        tool_states = stream.get("tool_states") or {}
        for tool_id in stream.get("tool_order") or []:
            event = tool_states.get(tool_id)
            if isinstance(event, dict):
                ordered_tool_events.append(event)
        return {
            "text": "".join(stream.get("parts") or []),
            "finish": stream.get("finish") or {},
            "firstTokenLatencyMs": first_ms,
            "totalLatencyMs": total_ms,
            "toolEvents": ordered_tool_events,
        }
@@ -634,13 +902,14 @@ class LingmaGatewayClient:
        request_id: str,
        *,
        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"
-        return {
+        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,
@@ -665,6 +934,9 @@ class LingmaGatewayClient:
                "localeLang": "zh-CN",
            },
        }
        if tool_config is not None:
            payload["toolConfig"] = tool_config
        return payload
    async def _kick_chat_ask(self, payload: dict) -> None:
        """Fire chat/ask as a notification.
@@ -685,12 +957,19 @@ class LingmaGatewayClient:
        *,
        session_id: str | None = None,
        is_reply: bool = False,
        tool_config: dict[str, Any] | None = None,
    ) -> dict:
        await self.ensure_ready()
        request_id = str(uuid.uuid4())
        sid = session_id or str(uuid.uuid4())
        payload = self._build_payload(
-            prompt, model_key, ask_mode, sid, request_id, is_reply=is_reply
+            prompt,
            model_key,
            ask_mode,
            sid,
            request_id,
            is_reply=is_reply,
            tool_config=tool_config,
        )
        self.rpc.create_stream(request_id)
        try:
@@ -721,9 +1000,14 @@ class LingmaGatewayClient:
        *,
        session_id: str | None = None,
        is_reply: bool = False,
        tool_config: dict[str, Any] | None = None,
        out_meta: dict | None = None,
-    ) -> AsyncIterator[str]:
+    ) -> AsyncIterator[dict[str, Any]]:
-        """Stream `chat/answer` chunks.
+        """Stream chat events.
        Yields structured events:
          * {"type": "text", "text": "..."}
          * {"type": "tool", "tool": {...}}
        If `out_meta` is provided, the final `chat/finish` payload's sessionId
        (and the raw finish dict) is written into it when the stream ends or is
@@ -734,15 +1018,21 @@ class LingmaGatewayClient:
        request_id = str(uuid.uuid4())
        sid = session_id or str(uuid.uuid4())
        payload = self._build_payload(
-            prompt, model_key, ask_mode, sid, request_id, is_reply=is_reply
+            prompt,
            model_key,
            ask_mode,
            sid,
            request_id,
            is_reply=is_reply,
            tool_config=tool_config,
        )
        self.rpc.create_stream(request_id)
        try:
            await self._kick_chat_ask(payload)
-            async for chunk in self.rpc.consume_stream(
+            async for event in self.rpc.consume_stream(
                request_id, timeout=max(60.0, self.rpc_timeout + 60.0)
            ):
-                yield chunk
+                yield event
        finally:
            # Runs on normal completion, exception, or consumer GeneratorExit (client disconnect).
            if out_meta is not None:
@@ -753,6 +1043,7 @@ class LingmaGatewayClient:
                    out_meta["finish"] = finish
                    out_meta["request_id"] = request_id
                    out_meta["chars"] = len(stream_result.get("text") or "")
                    out_meta["tool_events"] = stream_result.get("toolEvents") or []
                except Exception:
                    pass
            self.rpc.pop_stream(request_id)
--- a/app/main.py
+++ b/app/main.py
--- a/app/openai_schema.py
+++ b/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
--- a/app/session_cache.py
+++ b/app/session_cache.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 import asyncio
 import hashlib
 import json
 import time
 from collections import OrderedDict
 from dataclasses import dataclass
@@ -25,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
@@ -42,6 +43,38 @@ 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 "-"
    try:
        canonical = json.dumps(tool_config, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    except Exception:
        canonical = str(tool_config)
    return hashlib.sha1(canonical.encode("utf-8")).hexdigest()[:16]
 class SessionCache:
    """LRU + TTL cache: conversation-prefix hash -> upstream Lingma sessionId.
@@ -79,11 +112,21 @@ class SessionCache:
    def enabled(self) -> bool:
        return self.max > 0
-    def build_key(self, api_key: str, messages: list[dict]) -> 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)}"
+        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:
--- a/tests/TEST_PLAN.md
+++ b/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"
 ```
--- a/tests/test_auth_concurrency.py
+++ b/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()
--- a/tests/test_pool_stats_config.py
+++ b/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()
--- a/tests/test_schema_normalization.py
+++ b/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()
--- a/tests/test_session_cache_tooling.py
+++ b/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()
--- a/tests/test_tool_call_bridge.py
+++ b/tests/test_tool_call_bridge.py
Author	SHA1	Message	Date
mmc	4748432501	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>	2026-04-21 13:57:44 +08:00
mmc	83d69097c9	fix: enable tool forwarding by default and add config regression tests Switch TOOL_FORWARD_ENABLED default to true in runtime config and .env.example, and add regression tests covering default-on and explicit false behavior. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-21 13:41:41 +08:00
GitHub Actions	0e146e60d9	refactor: extract Phase 1 gateway helpers Move tool bridge and responses adapter helpers out of app.main so the main entrypoint can shrink without changing route orchestration behavior. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-21 08:05:09 +08:00
mmc	d0df089282	fix: harden responses streaming and tool-call fallback Ensure /v1/responses streams always terminate with response.completed and normalize Lingma tool_code fallbacks into structured tool calls, including single-argument forms. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-20 19:24:02 +08:00
mmc	866a212573	fix: restore proper SSE frame delimiters Emit real newline-delimited SSE frames for /v1/responses so clients can parse response.completed before the stream closes. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-20 15:08:16 +08:00
mmc	5e6c1c1a63	fix: harden responses stream termination Ensure /v1/responses streaming always emits completion frames on upstream EOF, errors, and cancellation, and add targeted diagnostics for interrupted Lingma streams. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-20 14:55:32 +08:00
GitHub Actions	12a4d9584e	feat: harden cache reuse semantics and expand protocol regressions Stabilize cross-protocol ask-mode/streaming behavior and reduce session-reuse branch collisions, then add focused docs/tests for multimodal normalization and pool/stats/config paths. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-20 14:26:11 +08:00
GitHub Actions	b96b91e5b7	test: add baseline gateway regression suites Add focused unittest coverage for auth/concurrency, schema normalization, and session-cache tooling behavior, and ignore local .gitnexus index artifacts. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-20 13:25:36 +08:00
GitHub Actions	c08dea89a2	fix: ensure responses stream always completes Emit a fallback response.completed and [DONE] when upstream SSE closes early so OpenAI /v1/responses clients do not fail on incomplete streams. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-20 13:23:43 +08:00
GitHub Actions	c9bd71f727	feat: add OpenAI /v1/responses adapter via chat flow Implement a thin responses layer that reuses existing chat/completions execution so auth, pooling, streaming, tool passthrough, and error semantics stay aligned across APIs. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-20 13:11:00 +08:00
GitHub Actions	56c57a4901	docs: sync DESIGN with current tooling behavior Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-20 08:31:45 +08:00
GitHub Actions	df80a86310	docs: refocus README on quickstart and runbook flow Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-20 08:11:00 +08:00
GitHub Actions	15cd5e8770	fix: close forced tool-choice with structured fallback Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-20 07:18:01 +08:00
GitHub Actions	63583712a8	fix: fallback agent payload source to numeric value Keep Lingma chat/ask payload source as numeric 1 for agent mode A/B validation against remote upstream timeout behavior. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-20 06:36:07 +08:00
GitHub Actions	c67a9c3d61	fix: align agent payload semantics with VSCode tool flow Force OpenAI tooling-context requests into agent mode and align Lingma ask payload fields for agent requests so server-side tool path matches VSCode semantics. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-19 23:19:52 +08:00
GitHub Actions	e208025f35	fix: emit Lingma tool approve/invoke roundtrip Forward tool/call/sync and tool/invoke events to Lingma with auto-approve and invokeResult so tool calls can complete end-to-end. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-19 21:35:05 +08:00
GitHub Actions	3498b81fa2	fix: enable anthropic agent mode for tooling requests Use agent ask_mode for Anthropic messages with tooling context so tool/write flows are executed, and add regression coverage plus docs/env updates for TOOL_FORWARD_ENABLED. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-19 20:15:14 +08:00
GitHub Actions	e600bae27c	fix: harden tooling session reuse and event routing Ensure session reuse is disabled for tooling contexts, include tool config in cache keys, and stabilize tool event merge/routing with expanded bridge tests. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-19 19:29:30 +08:00
GitHub Actions	5aa7fbfae5	fix: align Lingma tool event lifecycle handling Handle tool/invokeResult and richer tool/call/sync payloads in the client, and document/retest the verified VSCode monitoring workflow for tool events. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-19 09:49:01 +08:00
GitHub Actions	1c7b86e2c0	feat: bridge Lingma tool events to OpenAI/Anthropic responses Add structured tool event propagation from Lingma stream/finish metadata and map it to OpenAI tool_calls and Anthropic tool_use/tool_result in both streaming and non-streaming responses. Add focused bridge tests and update docs/design notes to match current behavior. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>	2026-04-18 22:34:43 +08:00