diff --git a/nanobot-patches/README.md b/nanobot-patches/README.md
new file mode 100644
index 0000000..8011771
--- /dev/null
+++ b/nanobot-patches/README.md
@@ -0,0 +1,128 @@
+# nanobot 图片识别修复补丁
+
+> 适用对象：`nanobot`（uv 工具安装，位于 `~/.local/share/uv/tools/nanobot-ai/`）
+> 修复日期：2026-08-21
+> 备份目录：本目录 `/home/ubuntu/nanocode/nanobot-patches/`
+
+---
+
+## 一、问题现象
+
+使用支持视觉的模型（如 `deepseek-v4-flash-vision-exp`）时，通过 `read_file` 读取图片后，模型**看不到图片**，只收到类似 `(Image file: xxx.png)` 的占位符文本，导致无法识别图片内容。
+
+## 二、根因
+
+`nanobot/providers/openai_compat_provider.py` 的 `_sanitize_messages()` 中有一行逻辑：
+
+```python
+force_string_content = bool(self._spec and self._spec.name == "deepseek")
+```
+
+只要 provider 名字是 `deepseek`，它就把**所有**消息的 `content` 强制转为纯字符串。而 `content` 一旦是字符串，里面 `image_url` / `input_image` / `image` / `file` 等视觉块就会被剥离，图片信息丢失。
+
+> 该逻辑本意是兼容 DeepSeek 纯文本模型的"只收字符串 content"，但**一刀切**地也作用到了视觉模型上，导致视觉内容被丢弃。
+
+## 三、修复方法
+
+在 `openai_compat_provider.py` 中：
+
+1. 新增一个静态方法，判断 `content` 是否包含视觉块：
+
+```python
+@staticmethod
+def _content_has_vision_blocks(content: Any) -> bool:
+    """Return True when content is a block list containing image/file blocks.
+
+    Used to keep vision content intact for multimodal models (e.g.
+    ``deepseek-v4-flash-vision-exp``) instead of coercing it to plain text.
+    """
+    if not isinstance(content, list):
+        return False
+    for item in content:
+        mapping = OpenAICompatProvider._maybe_mapping(item)
+        if not mapping:
+            continue
+        if mapping.get("type") in ("image_url", "input_image", "image", "file"):
+            return True
+    return False
+```
+
+2. 在 `_sanitize_messages()` 的强制转字符串条件里，增加"含视觉块则跳过"的判断：
+
+```python
+if (
+    force_string_content
+    and not (clean.get("role") == "assistant" and clean.get("tool_calls"))
+    and not OpenAICompatProvider._content_has_vision_blocks(clean.get("content"))
+):
+    clean["content"] = self._coerce_content_to_string(clean.get("content"))
+```
+
+> 关键点：`_content_has_vision_blocks()` 判断在 `_coerce_content_to_string()` **之前**生效，视觉块因此被保留。纯文本消息仍会走原字符串转换逻辑，不受影响。
+
+## 四、涉及文件与路径
+
+- 被修改文件：
+  `/home/ubuntu/.local/share/uv/tools/nanobot-ai/lib/python3.11/site-packages/nanobot/providers/openai_compat_provider.py`
+
+- 本目录备份：
+  - `openai_compat_provider.py.before_patch` —— 修复**前**（旧）文件
+  - `openai_compat_provider.py.after_patch` —— 修复**后**（当前）文件
+
+## 五、⚠️ 重要：uv 升级会覆盖补丁
+
+`nanobot-ai` 是通过 `uv tool` 安装的。一旦执行升级（例如 `uv tool upgrade nanobot-ai` 或重新安装），`site-packages` 下的文件会被**覆盖**，补丁随之丢失。
+
+**重新打补丁的方法**（升级后执行）：
+
+```bash
+# 1. 确认当前文件与 before_patch 是否一致（若一致说明补丁已丢失）
+diff /home/ubuntu/nanocode/nanobot-patches/openai_compat_provider.py.before_patch \
+     /home/ubuntu/.local/share/uv/tools/nanobot-ai/lib/python3.11/site-packages/nanobot/providers/openai_compat_provider.py
+
+# 2. 若补丁丢失，用 after_patch 覆盖回去
+cp /home/ubuntu/nanocode/nanobot-patches/openai_compat_provider.py.after_patch \
+   /home/ubuntu/.local/share/uv/tools/nanobot-ai/lib/python3.11/site-packages/nanobot/providers/openai_compat_provider.py
+
+# 3. 重启 nanobot 服务使补丁生效
+sudo systemctl restart nanobot
+```
+
+## 六、验证方法
+
+用 nanobot 自身的 venv Python 跑一次完整链路（构造图片块 → `_sanitize_messages` → `chat`），确认 `content` 仍为 `list` 且保留 `image_url` 块，并能返回图片描述：
+
+```bash
+PY=/home/ubuntu/.local/share/uv/tools/nanobot-ai/bin/python
+$PY -c "
+import asyncio, base64
+from nanobot.providers.registry import PROVIDERS
+from nanobot.providers.openai_compat_provider import OpenAICompatProvider
+
+spec = next(s for s in PROVIDERS if s.name=='deepseek')
+key = 'sk-xxxx'  # 替换为 DeepSeek API Key
+model = 'deepseek-v4-flash-vision-exp'
+b64 = base64.b64encode(open('/path/to/img.png','rb').read()).decode()
+messages = [{'role':'user','content':[
+    {'type':'image_url','image_url':{'url':'data:image/png;base64,'+b64}},
+    {'type':'text','text':'请描述这张图片'}
+]}]
+prov = OpenAICompatProvider(api_key=key, spec=spec, default_model=model)
+async def main():
+    s = prov._sanitize_messages(messages)
+    print('blocks:', [b.get('type') for b in s[0]['content']])  # 期望 ['image_url','text']
+    resp = await prov.chat(messages, model=model)
+    print(resp.text)
+asyncio.run(main())
+"
+```
+
+## 七、关于"识别能力"的说明
+
+- **图片识别能力本身**是模型自带的（如 `deepseek-v4-flash-vision-exp`、Kimi 等视觉模型都支持），**不需要**记忆到 nanobot 的长期记忆里。
+- **需要防丢失的**是"代码补丁"这一工程事实：它属于 `site-packages` 的源码改动，会随 `uv` 升级被覆盖。因此用**文档 + 前后备份**来保护，而不是当作模型能力去记忆。
+
+## 八、关联历史
+
+- 修改前备份文件：`/home/ubuntu/backup_openai_compat_provider.py.1787316174`
+- 历史记忆（memory）中的补丁位置与"uv 升级会覆盖、需重新打"的记录与此文档一致。
diff --git a/nanobot-patches/openai_compat_provider.py.after_patch b/nanobot-patches/openai_compat_provider.py.after_patch
new file mode 100644
index 0000000..82726ea
--- /dev/null
+++ b/nanobot-patches/openai_compat_provider.py.after_patch
@@ -0,0 +1,1764 @@
+"""OpenAI-compatible provider for all non-Anthropic LLM APIs."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import importlib.util
+import json
+import os
+import re
+import secrets
+import string
+import time
+import uuid
+from collections import deque
+from collections.abc import Awaitable, Callable
+from ipaddress import ip_address
+from typing import TYPE_CHECKING, Any
+from urllib.parse import urlparse
+
+from loguru import logger
+from pydantic.alias_generators import to_snake
+
+from nanobot.providers.base import (
+    LLMProvider,
+    LLMResponse,
+    ToolCallRequest,
+    parse_tool_arguments,
+    resolve_stream_idle_timeout_s,
+    tool_arguments_json_for_replay,
+)
+from nanobot.providers.openai_responses import (
+    consume_sdk_stream,
+    convert_messages,
+    convert_tools,
+    parse_response_output,
+)
+
+if TYPE_CHECKING:
+    from openai import AsyncOpenAI as AsyncOpenAIType
+
+    from nanobot.providers.registry import ProviderSpec
+
+# Module-level placeholder — set lazily by _ensure_client on first real
+# use, or replaced by tests via ``patch(...)``.  Kept as a plain name so
+# that ``unittest.mock.patch`` can find and replace it.
+AsyncOpenAI: Any = None
+
+_ALLOWED_MSG_KEYS = frozenset({
+    "role", "content", "tool_calls", "tool_call_id", "name",
+    "reasoning_content", "extra_content",
+})
+_ALNUM = string.ascii_letters + string.digits
+
+_STANDARD_TC_KEYS = frozenset({"id", "type", "index", "function"})
+_STANDARD_FN_KEYS = frozenset({"name", "arguments"})
+_DEFAULT_OPENROUTER_HEADERS = {
+    "HTTP-Referer": "https://github.com/HKUDS/nanobot",
+    "X-OpenRouter-Title": "nanobot",
+    "X-OpenRouter-Categories": "cli-agent,personal-agent",
+}
+_KIMI_K3_MODEL = "kimi-k3"
+_KIMI_THINKING_MODELS: frozenset[str] = frozenset({
+    "kimi-k2.5",
+    "kimi-k2.6",
+    "kimi-k2.7",
+    "kimi-k2.7-code",
+    "kimi-k2.7-code-highspeed",
+    "k2.6-code-preview",
+})
+_KIMI_ALWAYS_THINKING_MODELS: frozenset[str] = frozenset({
+    "kimi-k2.7-code",
+    "kimi-k2.7-code-highspeed",
+})
+_KIMI_SERVER_MANAGED_TEMPERATURE_MODELS: frozenset[str] = frozenset({
+    "kimi-k2.5",
+    "kimi-k2.6",
+})
+_TEXT_TOOL_CALL_RE = re.compile(r"<tool_call>\s*(.*?)\s*</tool_call>", re.DOTALL)
+# Thinking-capable MiMo models per Xiaomi docs (see
+# tests/providers/test_xiaomi_mimo_thinking.py). mimo-v2-flash is omitted
+# because it does not support thinking.
+_MIMO_THINKING_MODELS: frozenset[str] = frozenset({
+    "mimo-v2.5-pro",
+    "mimo-v2.5",
+    "mimo-v2-pro",
+    "mimo-v2-omni",
+})
+_OPENAI_COMPAT_REQUEST_TIMEOUT_S = 120.0
+
+# Maps ProviderSpec.thinking_style → extra_body builder.
+# Each builder takes a bool (thinking_enabled) and returns the dict to
+# merge into extra_body, keeping the style→wire-format mapping in one place.
+_THINKING_STYLE_MAP: dict[str, Any] = {
+    "thinking_type": lambda on: {"thinking": {"type": "enabled" if on else "disabled"}},
+    "enable_thinking": lambda on: {"enable_thinking": on},
+    "reasoning_split": lambda on: {"reasoning_split": on},
+}
+_GATEWAY_REASONING_STYLE_MAP: dict[str, Any] = {
+    "reasoning_effort": lambda effort: {"reasoning": {"effort": effort}},
+}
+_QWEN_THINKING_MODELS: frozenset[str] = frozenset({
+    "qwen3.7-max",
+    "qwen3.7-plus",
+    "qwen3.6-max-preview",
+    "qwen3.6-plus",
+    "qwen3.6-flash",
+    "qwen3.5-plus",
+    "qwen3.5-flash",
+})
+
+_MODEL_THINKING_STYLES: dict[str, str] = {
+    **dict.fromkeys(_KIMI_THINKING_MODELS, "thinking_type"),
+    **dict.fromkeys(_MIMO_THINKING_MODELS, "thinking_type"),
+    **dict.fromkeys(_QWEN_THINKING_MODELS, "enable_thinking"),
+}
+
+
+def _model_slug(model_name: str) -> str:
+    return model_name.lower().rsplit("/", 1)[-1]
+
+
+def _provider_prefix_key(name: str) -> str:
+    return to_snake(name.replace("-", "_")).lower()
+
+
+def _requires_max_completion_tokens(model_name: str) -> bool:
+    """Return True for models that require ``max_completion_tokens``."""
+    slug = _model_slug(model_name)
+    return slug == _KIMI_K3_MODEL or "gpt-5" in slug or any(
+        slug == p or slug.startswith((p + "-", p + ".")) for p in ("o1", "o3", "o4")
+    )
+
+
+def _model_thinking_style(model_name: str) -> str:
+    return _MODEL_THINKING_STYLES.get(_model_slug(model_name), "")
+
+
+def _thinking_styles_for(spec: ProviderSpec | None, model_name: str) -> list[str]:
+    styles: list[str] = []
+    if spec and spec.thinking_style:
+        styles.append(spec.thinking_style)
+    model_style = _model_thinking_style(model_name)
+    if model_style and model_style not in styles:
+        styles.append(model_style)
+    return styles
+
+
+def _thinking_extra_body(style: str, thinking_enabled: bool) -> dict[str, Any] | None:
+    builder = _THINKING_STYLE_MAP.get(style)
+    return builder(thinking_enabled) if builder else None
+
+
+def _gateway_reasoning_extra_body(style: str, effort: str | None) -> dict[str, Any] | None:
+    if not effort:
+        return None
+    builder = _GATEWAY_REASONING_STYLE_MAP.get(style)
+    return builder(effort) if builder else None
+
+
+def _openai_compat_timeout_s() -> float:
+    """Return the bounded request timeout used for OpenAI-compatible providers."""
+    return _float_env("NANOBOT_OPENAI_COMPAT_TIMEOUT_S", _OPENAI_COMPAT_REQUEST_TIMEOUT_S)
+
+
+def _float_env(name: str, default: float) -> float:
+    raw = os.environ.get(name)
+    if raw is None or not raw.strip():
+        return default
+    try:
+        value = float(raw)
+    except (TypeError, ValueError):
+        logger.warning("Ignoring invalid {}={!r}; using {}", name, raw, default)
+        return default
+    if value <= 0:
+        logger.warning("Ignoring non-positive {}={!r}; using {}", name, raw, default)
+        return default
+    return value
+
+
+def _short_tool_id() -> str:
+    """9-char alphanumeric ID compatible with all providers (incl. Mistral)."""
+    return "".join(secrets.choice(_ALNUM) for _ in range(9))
+
+
+def _strip_json_fence(text: str) -> str:
+    stripped = text.strip()
+    if not stripped.startswith("```") or not stripped.endswith("```"):
+        return stripped
+    lines = stripped.splitlines()
+    if len(lines) < 2:
+        return stripped
+    return "\n".join(lines[1:-1]).strip()
+
+
+def _extract_text_tool_calls(content: str | None) -> tuple[str | None, list[ToolCallRequest]]:
+    """Normalize common text-format tool call blocks into structured calls."""
+    if not content or "<tool_call>" not in content:
+        return content, []
+
+    tool_calls: list[ToolCallRequest] = []
+    spans: list[tuple[int, int]] = []
+    for match in _TEXT_TOOL_CALL_RE.finditer(content):
+        try:
+            payload = json.loads(_strip_json_fence(match.group(1)))
+        except Exception:
+            continue
+        if not isinstance(payload, dict):
+            continue
+
+        nested = payload.get("tool_call")
+        if isinstance(nested, dict):
+            payload = nested
+        function = payload.get("function")
+        if not isinstance(function, dict):
+            function = payload
+        name = function.get("name")
+        if not isinstance(name, str) or not name:
+            continue
+
+        arguments = function.get("arguments", payload.get("arguments", {}))
+        tool_calls.append(ToolCallRequest(
+            id=str(payload.get("id") or _short_tool_id()),
+            name=name,
+            arguments=parse_tool_arguments(arguments),
+        ))
+        spans.append(match.span())
+
+    if not tool_calls:
+        return content, []
+
+    visible_parts: list[str] = []
+    last = 0
+    for start, end in spans:
+        visible_parts.append(content[last:start])
+        last = end
+    visible_parts.append(content[last:])
+    visible_content = "".join(visible_parts).strip() or None
+    return visible_content, tool_calls
+
+
+def _get(obj: Any, key: str) -> Any:
+    """Get a value from dict or object attribute, returning None if absent."""
+    if isinstance(obj, dict):
+        return obj.get(key)
+    return getattr(obj, key, None)
+
+
+def _coerce_dict(value: Any) -> dict[str, Any] | None:
+    """Try to coerce *value* to a dict; return None if not possible or empty."""
+    if value is None:
+        return None
+    if isinstance(value, dict):
+        return value if value else None
+    model_dump = getattr(value, "model_dump", None)
+    if callable(model_dump):
+        dumped = model_dump()
+        if isinstance(dumped, dict) and dumped:
+            return dumped
+    return None
+
+
+def _extract_tc_extras(tc: Any) -> tuple[
+    dict[str, Any] | None,
+    dict[str, Any] | None,
+    dict[str, Any] | None,
+]:
+    """Extract (extra_content, provider_specific_fields, fn_provider_specific_fields).
+
+    Works for both SDK objects and dicts.  Captures Gemini ``extra_content``
+    verbatim and any non-standard keys on the tool-call / function.
+    """
+    extra_content = _coerce_dict(_get(tc, "extra_content"))
+
+    tc_dict = _coerce_dict(tc)
+    prov = None
+    fn_prov = None
+    if tc_dict is not None:
+        leftover = {k: v for k, v in tc_dict.items()
+                    if k not in _STANDARD_TC_KEYS and k != "extra_content" and v is not None}
+        if leftover:
+            prov = leftover
+        fn = _coerce_dict(tc_dict.get("function"))
+        if fn is not None:
+            fn_leftover = {k: v for k, v in fn.items()
+                          if k not in _STANDARD_FN_KEYS and v is not None}
+            if fn_leftover:
+                fn_prov = fn_leftover
+    else:
+        prov = _coerce_dict(_get(tc, "provider_specific_fields"))
+        fn_obj = _get(tc, "function")
+        if fn_obj is not None:
+            fn_prov = _coerce_dict(_get(fn_obj, "provider_specific_fields"))
+
+    return extra_content, prov, fn_prov
+
+
+def _uses_openrouter_attribution(spec: "ProviderSpec | None", api_base: str | None) -> bool:
+    """Apply Nanobot attribution headers to OpenRouter requests by default."""
+    if spec and spec.name == "openrouter":
+        return True
+    return bool(api_base and "openrouter" in api_base.lower())
+
+
+_RESPONSES_FAILURE_THRESHOLD = 3
+_RESPONSES_PROBE_INTERVAL_S = 300  # 5 minutes
+
+
+def _is_local_endpoint(
+    spec: "ProviderSpec | None",
+    api_base: str | None,
+) -> bool:
+    """Return True when the endpoint is a local or LAN model server.
+
+    Matches either the provider spec's ``is_local`` flag or common private-
+    network patterns in the base URL (localhost, 127.x, 192.168.x, 10.x,
+    172.16-31.x, Docker ``host.docker.internal``).
+    """
+    if spec and spec.is_local:
+        return True
+    if not api_base:
+        return False
+    raw = api_base.strip().lower()
+    parsed = urlparse(raw if "://" in raw else f"//{raw}")
+    try:
+        host = parsed.hostname
+    except ValueError:
+        return False
+    if host in {"localhost", "host.docker.internal"}:
+        return True
+    if not host:
+        return False
+    try:
+        addr = ip_address(host)
+    except ValueError:
+        return False
+    return addr.is_loopback or addr.is_private
+
+
+def _is_direct_openai_base(api_base: str | None) -> bool:
+    """Return True for direct OpenAI endpoints, not generic OpenAI-compatible gateways."""
+    if not api_base:
+        return True
+    normalized = api_base.strip().lower().rstrip("/")
+    return "api.openai.com" in normalized and "openrouter" not in normalized
+
+
+def _responses_circuit_key(
+    model: str | None,
+    default_model: str,
+    reasoning_effort: str | None,
+) -> str:
+    model_name = (model or default_model).lower()
+    effort = reasoning_effort.lower() if isinstance(reasoning_effort, str) else ""
+    return f"{model_name}:{effort}"
+
+
+def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
+    """Recursively merge *override* into *base*, returning a new dict.
+
+    Nested dicts are merged key-by-key; all other types in *override*
+    replace the corresponding key in *base*.
+    """
+    merged = dict(base)
+    for key, value in override.items():
+        if (
+            key in merged
+            and isinstance(merged[key], dict)
+            and isinstance(value, dict)
+        ):
+            merged[key] = _deep_merge(merged[key], value)
+        else:
+            merged[key] = value
+    return merged
+
+
+def _merge_unique_list(base: Any, override: Any) -> Any:
+    """Append list values while preserving order and removing duplicates."""
+    if not isinstance(base, list) or not isinstance(override, list):
+        return override
+    result: list[Any] = []
+    seen: set[str] = set()
+    for value in [*base, *override]:
+        try:
+            key = json.dumps(value, sort_keys=True, ensure_ascii=False)
+        except Exception:
+            key = repr(value)
+        if key in seen:
+            continue
+        seen.add(key)
+        result.append(value)
+    return result
+
+
+def _merge_responses_extra_body(
+    body: dict[str, Any],
+    extra_body: dict[str, Any],
+) -> dict[str, Any]:
+    """Merge configured Responses API body fields without clobbering tools."""
+    reserved = {"include", "tools"}
+    regular_extra = {key: value for key, value in extra_body.items() if key not in reserved}
+    merged = _deep_merge(body, regular_extra)
+
+    if "include" in extra_body:
+        merged["include"] = _merge_unique_list(body.get("include"), extra_body["include"])
+
+    if "tools" in extra_body:
+        current_tools = body.get("tools")
+        configured_tools = extra_body["tools"]
+        if isinstance(current_tools, list) and isinstance(configured_tools, list):
+            merged["tools"] = [*current_tools, *configured_tools]
+        else:
+            merged["tools"] = configured_tools
+
+    return merged
+
+
+class OpenAICompatProvider(LLMProvider):
+    """Unified provider for all OpenAI-compatible APIs.
+
+    Receives a resolved ``ProviderSpec`` from the caller — no internal
+    registry lookups needed.
+    """
+
+    def __init__(
+        self,
+        api_key: str | None = None,
+        api_base: str | None = None,
+        default_model: str = "gpt-4o",
+        extra_headers: dict[str, str] | None = None,
+        spec: ProviderSpec | None = None,
+        extra_body: dict[str, Any] | None = None,
+        api_type: str = "auto",
+        extra_query: dict[str, str] | None = None,
+        proxy: str | None = None,
+    ):
+        super().__init__(api_key, api_base)
+        self.default_model = default_model
+        self.extra_headers = extra_headers or {}
+        self._spec = spec
+        self._extra_body = extra_body or {}
+        self._api_type = api_type if spec and spec.name == "openai" else "auto"
+        self._extra_query = extra_query or {}
+        self._proxy = proxy or None
+
+        if api_key and spec and spec.env_key:
+            self._setup_env(api_key, api_base)
+
+        effective_base = api_base or (spec.default_api_base if spec else None) or None
+        self._effective_base = effective_base
+        self._default_headers = {"x-session-affinity": uuid.uuid4().hex}
+        if _uses_openrouter_attribution(spec, effective_base):
+            self._default_headers.update(_DEFAULT_OPENROUTER_HEADERS)
+        if extra_headers:
+            self._default_headers.update(extra_headers)
+        self._api_key_for_client = api_key or "no-key"
+        self._is_local = _is_local_endpoint(spec, effective_base)
+
+        # Lazy-init: the OpenAI client and its httpx transport are expensive
+        # to create (~700 ms on Windows). Defer until first use.
+        self._client: AsyncOpenAIType | None = None
+        self._client_lock = asyncio.Lock()
+
+        # Responses API circuit breaker: skip after repeated failures,
+        # probe again after _RESPONSES_PROBE_INTERVAL_S seconds.
+        self._responses_failures: dict[str, int] = {}
+        self._responses_tripped_at: dict[str, float] = {}
+
+    def _build_client(self) -> None:
+        """Create the OpenAI client using the current module-level AsyncOpenAI."""
+        import httpx
+
+        timeout_s = _openai_compat_timeout_s()
+        http_client: httpx.AsyncClient | None = None
+        if self._proxy:
+            http_client = httpx.AsyncClient(
+                timeout=timeout_s,
+                proxy=self._proxy,
+                trust_env=False,
+                follow_redirects=True,
+            )
+        elif self._is_local:
+            # Local model servers (Ollama, llama.cpp, vLLM) often close idle
+            # HTTP connections before the client-side keepalive expires. When
+            # two LLM calls happen seconds apart (e.g. heartbeat _decide then
+            # process_direct), the second call may grab a now-dead pooled
+            # connection, causing a transient APIConnectionError on every first
+            # attempt. Disabling keepalive for local endpoints avoids this by
+            # opening a fresh connection for each request, which is cheap on a
+            # LAN. Cloud providers benefit from keepalive, so we leave the
+            # default pool settings for them.
+            #
+            # Also disable proxy for local endpoints: when the host has
+            # HTTP_PROXY / HTTPS_PROXY / ALL_PROXY set, httpx would try to
+            # route local traffic through the proxy, which typically cannot
+            # reach localhost or LAN addresses.
+            _local_limits = httpx.Limits(keepalive_expiry=0)
+            http_client = httpx.AsyncClient(
+                limits=_local_limits,
+                timeout=timeout_s,
+                transport=httpx.AsyncHTTPTransport(proxy=None, limits=_local_limits),
+            )
+        # else: http_client stays None → SDK creates DefaultAsyncHttpxClient
+        # which already reads proxy env vars via trust_env=True, has proper
+        # connection limits, and follows redirects.
+        self._client = AsyncOpenAI(
+            api_key=self._api_key_for_client,
+            base_url=self._effective_base,
+            default_headers=self._default_headers,
+            default_query=self._extra_query or None,
+            max_retries=0,
+            timeout=timeout_s,
+            http_client=http_client,
+        )
+
+    async def _ensure_client(self):
+        """Return the shared OpenAI client, creating it on first call."""
+        if self._client is not None:
+            return self._client
+        async with self._client_lock:
+            if self._client is not None:
+                return self._client
+            global AsyncOpenAI
+            if AsyncOpenAI is None:
+                if os.environ.get("LANGFUSE_SECRET_KEY") and importlib.util.find_spec("langfuse"):
+                    from langfuse.openai import AsyncOpenAI as _AsyncOpenAI
+                else:
+                    if os.environ.get("LANGFUSE_SECRET_KEY"):
+                        logger.warning(
+                            "LANGFUSE_SECRET_KEY is set but langfuse is not installed; "
+                            "install with `pip install langfuse` to enable tracing"
+                        )
+                    from openai import AsyncOpenAI as _AsyncOpenAI
+                AsyncOpenAI = _AsyncOpenAI
+
+            self._build_client()
+            return self._client
+
+    def _setup_env(self, api_key: str, api_base: str | None) -> None:
+        """Set environment variables based on provider spec."""
+        spec = self._spec
+        if not spec or not spec.env_key:
+            return
+        if spec.is_gateway:
+            os.environ[spec.env_key] = api_key
+        else:
+            os.environ.setdefault(spec.env_key, api_key)
+        effective_base = api_base or spec.default_api_base
+        for env_name, env_val in spec.env_extras:
+            resolved = env_val.replace("{api_key}", api_key).replace("{api_base}", effective_base)
+            os.environ.setdefault(env_name, resolved)
+
+    @classmethod
+    def _apply_cache_control(
+        cls,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None,
+    ) -> tuple[list[dict[str, Any]], list[dict[str, Any]] | None]:
+        """Inject cache_control markers for prompt caching."""
+        cache_marker = {"type": "ephemeral"}
+        new_messages = list(messages)
+
+        def _mark(msg: dict[str, Any]) -> dict[str, Any]:
+            content = msg.get("content")
+            if isinstance(content, str):
+                return {**msg, "content": [
+                    {"type": "text", "text": content, "cache_control": cache_marker},
+                ]}
+            if isinstance(content, list) and content:
+                nc = list(content)
+                nc[-1] = {**nc[-1], "cache_control": cache_marker}
+                return {**msg, "content": nc}
+            return msg
+
+        if new_messages and new_messages[0].get("role") == "system":
+            new_messages[0] = _mark(new_messages[0])
+        if len(new_messages) >= 3:
+            new_messages[-2] = _mark(new_messages[-2])
+
+        new_tools = tools
+        if tools:
+            new_tools = list(tools)
+            for idx in cls._tool_cache_marker_indices(new_tools):
+                new_tools[idx] = {**new_tools[idx], "cache_control": cache_marker}
+        return new_messages, new_tools
+
+    @staticmethod
+    def _normalize_tool_call_id(tool_call_id: Any) -> Any:
+        """Normalize to a provider-safe 9-char alphanumeric form."""
+        if not isinstance(tool_call_id, str):
+            return tool_call_id
+        if len(tool_call_id) == 9 and tool_call_id.isalnum():
+            return tool_call_id
+        return hashlib.sha1(tool_call_id.encode()).hexdigest()[:9]
+
+    def _should_normalize_tool_call_ids(self) -> bool:
+        """Return True for providers that reject normal OpenAI tool call IDs."""
+        return bool(self._spec and self._spec.name == "mistral")
+
+    @staticmethod
+    def _content_has_vision_blocks(content: Any) -> bool:
+        """Return True when content is a block list containing image/file blocks.
+
+        Used to keep vision content intact for multimodal models (e.g.
+        ``deepseek-v4-flash-vision-exp``) instead of coercing it to plain text.
+        """
+        if not isinstance(content, list):
+            return False
+        for item in content:
+            mapping = OpenAICompatProvider._maybe_mapping(item)
+            if not mapping:
+                continue
+            if mapping.get("type") in ("image_url", "input_image", "image", "file"):
+                return True
+        return False
+
+    @staticmethod
+    def _coerce_content_to_string(content: Any) -> str | None:
+        """Coerce block/list content into plain text for strict string-only APIs."""
+        if content is None or isinstance(content, str):
+            return content
+        text = OpenAICompatProvider._extract_text_content(content)
+        if isinstance(text, str) and text:
+            return text
+        try:
+            dumped = json.dumps(content, ensure_ascii=False)
+        except Exception:
+            dumped = str(content)
+        return dumped or "(empty)"
+
+    def _sanitize_messages(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """Strip non-standard keys, normalize tool_call IDs."""
+        sanitized = LLMProvider._sanitize_request_messages(messages, _ALLOWED_MSG_KEYS)
+        id_map: dict[str, str] = {}
+        pending_tool_ids: dict[str, deque[str]] = {}
+        force_string_content = bool(self._spec and self._spec.name == "deepseek")
+        normalize_tool_ids = self._should_normalize_tool_call_ids()
+        strip_reasoning = bool(
+            self._spec
+            and getattr(self._spec, "strip_history_reasoning_content", False)
+        )
+        if strip_reasoning:
+            for msg in sanitized:
+                msg.pop("reasoning_content", None)
+
+        def map_id(value: Any) -> Any:
+            if not isinstance(value, str):
+                return value
+            if not normalize_tool_ids:
+                return value
+            return id_map.setdefault(value, self._normalize_tool_call_id(value))
+
+        def unique_tool_id(value: Any, used_ids: set[str], idx: int) -> str:
+            if isinstance(value, str) and value:
+                base = map_id(value)
+            else:
+                base = _short_tool_id()
+            if not isinstance(base, str) or not base:
+                base = _short_tool_id()
+            if base not in used_ids:
+                return base
+            seed = value if isinstance(value, str) and value else base
+            salt = 1
+            while True:
+                candidate = self._normalize_tool_call_id(f"{seed}:{idx}:{salt}")
+                if isinstance(candidate, str) and candidate not in used_ids:
+                    return candidate
+                salt += 1
+
+        def map_tool_result_id(value: Any) -> Any:
+            if not isinstance(value, str):
+                return value
+            queue = pending_tool_ids.get(value)
+            if queue:
+                mapped = queue.popleft()
+                if not queue:
+                    pending_tool_ids.pop(value, None)
+                return mapped
+            return map_id(value)
+
+        for clean in sanitized:
+            if isinstance(clean.get("tool_calls"), list):
+                normalized = []
+                used_ids: set[str] = set()
+                for idx, tc in enumerate(clean["tool_calls"]):
+                    if not isinstance(tc, dict):
+                        normalized.append(tc)
+                        continue
+                    tc_clean = dict(tc)
+                    raw_id = tc_clean.get("id")
+                    mapped_id = unique_tool_id(raw_id, used_ids, idx)
+                    tc_clean["id"] = mapped_id
+                    used_ids.add(mapped_id)
+                    if isinstance(raw_id, str) and raw_id:
+                        pending_tool_ids.setdefault(raw_id, deque()).append(mapped_id)
+                    function = tc_clean.get("function")
+                    if isinstance(function, dict):
+                        function_clean = dict(function)
+                        if "arguments" in function_clean:
+                            function_clean["arguments"] = tool_arguments_json_for_replay(
+                                function_clean.get("arguments")
+                            )
+                        else:
+                            function_clean["arguments"] = "{}"
+                        tc_clean["function"] = function_clean
+                    normalized.append(tc_clean)
+                clean["tool_calls"] = normalized
+                if clean.get("role") == "assistant":
+                    # Some OpenAI-compatible gateways reject assistant messages
+                    # that mix non-empty content with tool_calls.
+                    clean["content"] = None
+            if "tool_call_id" in clean and clean["tool_call_id"]:
+                clean["tool_call_id"] = map_tool_result_id(clean["tool_call_id"])
+            if (
+                force_string_content
+                and not (clean.get("role") == "assistant" and clean.get("tool_calls"))
+                and not OpenAICompatProvider._content_has_vision_blocks(clean.get("content"))
+            ):
+                clean["content"] = self._coerce_content_to_string(clean.get("content"))
+        return self._enforce_role_alternation(sanitized)
+
+    # ------------------------------------------------------------------
+    # Build kwargs
+    # ------------------------------------------------------------------
+
+    def _request_model_name(self, model_name: str) -> str:
+        spec = self._spec
+        if not spec or "/" not in model_name:
+            return model_name
+        if spec.strip_model_prefix:
+            return model_name.split("/")[-1]
+
+        route_prefixes = getattr(spec, "strip_model_prefixes", ())
+        if not isinstance(route_prefixes, tuple) or not route_prefixes:
+            return model_name
+        model_prefix, routed_model = model_name.split("/", 1)
+        model_prefix_key = _provider_prefix_key(model_prefix)
+        if any(_provider_prefix_key(prefix) == model_prefix_key for prefix in route_prefixes):
+            return routed_model
+        return model_name
+
+    @staticmethod
+    def _supports_temperature(
+        model_name: str,
+        reasoning_effort: str | None = None,
+    ) -> bool:
+        """Return True when the model accepts a temperature parameter.
+
+        Kimi K3 uses a fixed temperature that should be omitted. GPT-5 family
+        and reasoning models (o1/o3/o4) reject temperature when
+        reasoning_effort is set to anything other than ``"none"``.
+        """
+        if _model_slug(model_name) == _KIMI_K3_MODEL:
+            return False
+        if reasoning_effort and reasoning_effort.lower() != "none":
+            return False
+        name = model_name.lower()
+        return not any(token in name for token in ("gpt-5", "o1", "o3", "o4"))
+
+    def _build_kwargs(
+        self,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None,
+        model: str | None,
+        max_tokens: int,
+        temperature: float,
+        reasoning_effort: str | None,
+        tool_choice: str | dict[str, Any] | None,
+    ) -> dict[str, Any]:
+        model_name = model or self.default_model
+        spec = self._spec
+
+        if spec and spec.supports_prompt_caching:
+            model_name = model or self.default_model
+            if any(model_name.lower().startswith(k) for k in ("anthropic/", "claude")):
+                messages, tools = self._apply_cache_control(messages, tools)
+
+        model_name = self._request_model_name(model_name)
+
+        kwargs: dict[str, Any] = {
+            "model": model_name,
+            "messages": self._sanitize_messages(self._sanitize_empty_content(messages)),
+        }
+
+        # GPT-5 and reasoning models (o1/o3/o4) reject temperature when
+        # reasoning_effort is active.  Only include it when safe.
+        if self._supports_temperature(model_name, reasoning_effort):
+            kwargs["temperature"] = temperature
+
+        if (
+            spec and getattr(spec, "supports_max_completion_tokens", False)
+        ) or _requires_max_completion_tokens(model_name):
+            kwargs["max_completion_tokens"] = max(1, max_tokens)
+        else:
+            kwargs["max_tokens"] = max(1, max_tokens)
+
+        if spec:
+            model_lower = model_name.lower()
+            for pattern, overrides in spec.model_overrides:
+                if pattern in model_lower:
+                    kwargs.update(overrides)
+                    break
+
+        # Moonshot selects the only valid temperature from the K2.5/K2.6 thinking mode:
+        # 1.0 when enabled and 0.6 when disabled. Omitting the parameter lets the API
+        # apply the matching value for both its default and explicit thinking controls.
+        if (
+            spec
+            and spec.name == "moonshot"
+            and _model_slug(model_name) in _KIMI_SERVER_MANAGED_TEMPERATURE_MODELS
+        ):
+            kwargs.pop("temperature", None)
+
+        # Normalize reasoning_effort into a semantic form (OpenAI vocab)
+        # used for internal decisions, and a wire form actually sent out.
+        # "minimum" is accepted as a DashScope-native alias for "minimal".
+        semantic_effort: str | None = None
+        if isinstance(reasoning_effort, str):
+            semantic_effort = reasoning_effort.lower()
+            if semantic_effort == "minimum":
+                semantic_effort = "minimal"
+
+        wire_effort = reasoning_effort
+        slug = _model_slug(model_name)
+        if slug == _KIMI_K3_MODEL and semantic_effort is not None:
+            # K3 always reasons and currently accepts only the top-level
+            # reasoning_effort="max". Preserve disabled/default semantics by
+            # omitting the field; normalize older enabled presets to "max" so
+            # switching from a K2.x model does not send an unsupported value.
+            if semantic_effort in ("none", "minimal"):
+                wire_effort = None
+            else:
+                semantic_effort = "max"
+                wire_effort = "max"
+        if spec and spec.name == "dashscope" and semantic_effort == "minimal":
+            # DashScope accepts none/minimum/low/medium/high/xhigh; "minimal" 400s.
+            wire_effort = "minimum"
+
+        # Magistral and other providers where reasoning is implicit reject the
+        # reasoning_effort kwarg entirely. Strip it before the remap so we don't
+        # accidentally send "none"/"high" to a model that always reasons.
+        strip_effort = False
+        if spec and getattr(spec, "implicit_reasoning_models", ()):
+            model_lower = model_name.lower()
+            strip_effort = any(
+                pat in model_lower for pat in spec.implicit_reasoning_models
+            )
+
+        # Some providers accept a constrained reasoning_effort vocabulary
+        # (Mistral: only "high"/"none"). Remap from OpenAI vocab to the
+        # provider's accepted set; an empty mapped value means "omit".
+        if (
+            not strip_effort
+            and spec
+            and getattr(spec, "reasoning_effort_remap", ())
+            and isinstance(semantic_effort, str)
+        ):
+            remap = dict(spec.reasoning_effort_remap)
+            mapped = remap.get(semantic_effort)
+            if mapped is not None:
+                wire_effort = mapped or None
+                semantic_effort = mapped or "none"
+
+        if strip_effort:
+            wire_effort = None
+        elif wire_effort and semantic_effort != "none":
+            kwargs["reasoning_effort"] = wire_effort
+
+        # Only send thinking controls when reasoning_effort is explicit so
+        # omitting the config preserves each provider's default.
+        if reasoning_effort is not None:
+            thinking_enabled = semantic_effort not in ("none", "minimal")
+            for thinking_style in _thinking_styles_for(spec, model_name):
+                if not thinking_enabled and slug in _KIMI_ALWAYS_THINKING_MODELS:
+                    continue
+                extra = _thinking_extra_body(thinking_style, thinking_enabled)
+                if extra:
+                    kwargs.setdefault("extra_body", {}).update(extra)
+            gateway_style = getattr(spec, "gateway_reasoning_style", "") if spec else ""
+            if (
+                gateway_style
+                and _model_thinking_style(model_name)
+                and (thinking_enabled or slug not in _KIMI_ALWAYS_THINKING_MODELS)
+            ):
+                extra = _gateway_reasoning_extra_body(gateway_style, semantic_effort)
+                if extra:
+                    kwargs.setdefault("extra_body", {}).update(extra)
+
+            # Moonshot rejects requests that carry both 'reasoning_effort'
+            # and the native 'thinking' param.  We already expressed the
+            # user's intent via the provider-native shape, so drop the
+            # redundant wire-level kwarg.  Only kimi models need this —
+            # Xiaomi's API accepts both params.
+            if slug in _KIMI_THINKING_MODELS:
+                kwargs.pop("reasoning_effort", None)
+
+        if tools:
+            kwargs["tools"] = tools
+            kwargs["tool_choice"] = tool_choice or "auto"
+
+        # Backfill reasoning_content="" on assistants missing it: DeepSeek
+        # thinking mode rejects history otherwise (#3554, #3584); "" reads
+        # as "no thinking that turn". DeepSeek-V4/reasoner reason natively,
+        # so backfill even without explicit reasoning_effort.
+        explicit_thinking = (
+            reasoning_effort is not None
+            and semantic_effort not in ("none", "minimal")
+            and (
+                (spec and spec.thinking_style)
+                or _model_thinking_style(model_name)
+            )
+        )
+        implicit_deepseek_thinking = (
+            spec is not None
+            and spec.name == "deepseek"
+            and semantic_effort not in ("none", "minimal", "minimum")
+            and any(t in model_name.lower() for t in ("deepseek-v4", "deepseek-reasoner"))
+        )
+        if explicit_thinking or implicit_deepseek_thinking:
+            for msg in kwargs["messages"]:
+                if msg.get("role") == "assistant" and "reasoning_content" not in msg:
+                    msg["reasoning_content"] = ""
+
+        # Merge user-configured extra_body last so it can override or
+        # extend provider-specific defaults (e.g. chat_template_kwargs,
+        # guided_json, repetition_penalty).  Uses recursive merge so
+        # nested dicts like {"chat_template_kwargs": {"enable_thinking": false}}
+        # do not clobber sibling keys already set by thinking-style logic.
+        if self._extra_body:
+            existing = kwargs.get("extra_body", {})
+            kwargs["extra_body"] = _deep_merge(existing, self._extra_body)
+
+        return kwargs
+
+    def _should_use_responses_api(
+        self,
+        model: str | None,
+        reasoning_effort: str | None,
+    ) -> bool:
+        """Use Responses API only for direct OpenAI requests that benefit from it."""
+        if self._api_type == "chat_completions":
+            return False
+        if self._spec and self._spec.name not in ("openai", "github_copilot"):
+            return False
+        if self._api_type == "responses":
+            # Explicit configuration means Responses is mandatory; do not
+            # consult the circuit breaker or fall back to Chat Completions.
+            return True
+        if self._spec is None or self._spec.name != "github_copilot":
+            if not _is_direct_openai_base(self._effective_base):
+                return False
+
+        model_name = (model or self.default_model).lower()
+        wants = False
+        if reasoning_effort and reasoning_effort.lower() != "none":
+            wants = True
+        elif any(token in model_name for token in ("gpt-5", "o1", "o3", "o4")):
+            wants = True
+        if not wants:
+            return False
+
+        return self._responses_circuit_allows_probe(model, reasoning_effort)
+
+    def _responses_circuit_allows_probe(
+        self,
+        model: str | None,
+        reasoning_effort: str | None,
+    ) -> bool:
+        """Return False when the Responses API circuit breaker is open."""
+        key = _responses_circuit_key(model, self.default_model, reasoning_effort)
+        failures = self._responses_failures.get(key, 0)
+        if failures >= _RESPONSES_FAILURE_THRESHOLD:
+            tripped = self._responses_tripped_at.get(key, 0.0)
+            if (time.monotonic() - tripped) < _RESPONSES_PROBE_INTERVAL_S:
+                return False
+            # Half-open: allow one probe attempt
+        return True
+
+    def _record_responses_failure(self, model: str | None, reasoning_effort: str | None) -> None:
+        key = _responses_circuit_key(model, self.default_model, reasoning_effort)
+        count = self._responses_failures.get(key, 0) + 1
+        self._responses_failures[key] = count
+        if count >= _RESPONSES_FAILURE_THRESHOLD:
+            self._responses_tripped_at[key] = time.monotonic()
+            logger.warning(
+                "Responses API circuit open for {} — falling back to Chat Completions",
+                key,
+            )
+
+    def _record_responses_success(self, model: str | None, reasoning_effort: str | None) -> None:
+        key = _responses_circuit_key(model, self.default_model, reasoning_effort)
+        self._responses_failures.pop(key, None)
+        self._responses_tripped_at.pop(key, None)
+
+    @staticmethod
+    def _should_fallback_from_responses_error(e: Exception) -> bool:
+        """Fallback only for likely Responses API compatibility errors."""
+        response = getattr(e, "response", None)
+        status_code = getattr(e, "status_code", None)
+        if status_code is None and response is not None:
+            status_code = getattr(response, "status_code", None)
+        if status_code not in {400, 404, 422}:
+            return False
+
+        body = (
+            getattr(e, "body", None)
+            or getattr(e, "doc", None)
+            or getattr(response, "text", None)
+        )
+        body_text = str(body).lower() if body is not None else ""
+        compatibility_markers = (
+            "responses",
+            "response api",
+            "max_output_tokens",
+            "instructions",
+            "previous_response",
+            "unsupported",
+            "not supported",
+            "unknown parameter",
+            "unrecognized request argument",
+        )
+        return any(marker in body_text for marker in compatibility_markers)
+
+    def _build_responses_body(
+        self,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None,
+        model: str | None,
+        max_tokens: int,
+        temperature: float,
+        reasoning_effort: str | None,
+        tool_choice: str | dict[str, Any] | None,
+    ) -> dict[str, Any]:
+        """Build a Responses API body for direct OpenAI requests."""
+        model_name = model or self.default_model
+        model_name = self._request_model_name(model_name)
+        sanitized_messages = self._sanitize_messages(self._sanitize_empty_content(messages))
+        instructions, input_items = convert_messages(sanitized_messages)
+
+        body: dict[str, Any] = {
+            "model": model_name,
+            "instructions": instructions or None,
+            "input": input_items,
+            "max_output_tokens": max(1, max_tokens),
+            "store": False,
+            "stream": False,
+        }
+
+        if self._supports_temperature(model_name, reasoning_effort):
+            body["temperature"] = temperature
+
+        if reasoning_effort and reasoning_effort.lower() != "none":
+            body["reasoning"] = {"effort": reasoning_effort}
+            body["include"] = ["reasoning.encrypted_content"]
+
+        if tools:
+            body["tools"] = convert_tools(tools)
+            body["tool_choice"] = tool_choice or "auto"
+
+        extra_body = getattr(self, "_extra_body", {})
+        if extra_body:
+            body = _merge_responses_extra_body(body, extra_body)
+
+        return body
+
+    # ------------------------------------------------------------------
+    # Response parsing
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _maybe_mapping(value: Any) -> dict[str, Any] | None:
+        if isinstance(value, dict):
+            return value
+        model_dump = getattr(value, "model_dump", None)
+        if callable(model_dump):
+            dumped = model_dump()
+            if isinstance(dumped, dict):
+                return dumped
+        return None
+
+    @classmethod
+    def _extract_text_content(cls, value: Any) -> str | None:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return value
+        if isinstance(value, list):
+            parts: list[str] = []
+            for item in value:
+                item_map = cls._maybe_mapping(item)
+                if item_map:
+                    # Skip Mistral-style {"type":"thinking","thinking":[...]}
+                    # blocks: their text belongs in reasoning_content.
+                    if item_map.get("type") == "thinking":
+                        continue
+                    text = item_map.get("text")
+                    if isinstance(text, str):
+                        parts.append(text)
+                        continue
+                text = getattr(item, "text", None)
+                if isinstance(text, str):
+                    parts.append(text)
+                    continue
+                if isinstance(item, str):
+                    parts.append(item)
+            return "".join(parts) or None
+        return str(value)
+
+    @classmethod
+    def _extract_thinking_content(cls, value: Any) -> str | None:
+        """Extract reasoning text from Mistral-style thinking blocks.
+
+        Mistral returns content as a list mixing
+        ``{"type":"thinking","thinking":[{"type":"text","text":...}]}`` and
+        ``{"type":"text","text":...}``. The thinking text belongs in
+        ``reasoning_content`` so the agent can surface it as a reasoning
+        trace rather than as the assistant's reply.
+        """
+        if not isinstance(value, list):
+            return None
+        parts: list[str] = []
+        for item in value:
+            item_map = cls._maybe_mapping(item)
+            if not item_map:
+                continue
+            if item_map.get("type") != "thinking":
+                continue
+            inner = item_map.get("thinking")
+            text = cls._extract_text_content(inner)
+            if text:
+                parts.append(text)
+        return "".join(parts) or None
+
+    @classmethod
+    def _extract_usage(cls, response: Any) -> dict[str, int]:
+        """Extract token usage from an OpenAI-compatible response.
+
+        Handles both dict-based (raw JSON) and object-based (SDK Pydantic)
+        responses.  Provider-specific ``cached_tokens`` fields are normalised
+        under a single key; see the priority chain inside for details.
+        """
+        # --- resolve usage object ---
+        usage_obj = None
+        response_map = cls._maybe_mapping(response)
+        if response_map is not None:
+            usage_obj = response_map.get("usage")
+        elif hasattr(response, "usage") and response.usage:
+            usage_obj = response.usage
+
+        usage_map = cls._maybe_mapping(usage_obj)
+        if usage_map is not None:
+            result = {
+                "prompt_tokens": int(usage_map.get("prompt_tokens") or 0),
+                "completion_tokens": int(usage_map.get("completion_tokens") or 0),
+                "total_tokens": int(usage_map.get("total_tokens") or 0),
+            }
+        elif usage_obj:
+            result = {
+                "prompt_tokens": getattr(usage_obj, "prompt_tokens", 0) or 0,
+                "completion_tokens": getattr(usage_obj, "completion_tokens", 0) or 0,
+                "total_tokens": getattr(usage_obj, "total_tokens", 0) or 0,
+            }
+        else:
+            return {}
+
+        # --- cached_tokens (normalised across providers) ---
+        # Try nested paths first (dict), fall back to attribute (SDK object).
+        # Priority order ensures the most specific field wins.
+        for path in (
+            ("prompt_tokens_details", "cached_tokens"),  # OpenAI/Zhipu/MiniMax/Qwen/Mistral/xAI
+            ("cached_tokens",),                          # StepFun/Moonshot (top-level)
+            ("prompt_cache_hit_tokens",),                # DeepSeek/SiliconFlow
+        ):
+            cached = cls._get_nested_int(usage_map, path)
+            if not cached and usage_obj:
+                cached = cls._get_nested_int(usage_obj, path)
+            if cached:
+                result["cached_tokens"] = cached
+                break
+
+        return result
+
+    @staticmethod
+    def _get_nested_int(obj: Any, path: tuple[str, ...]) -> int:
+        """Drill into *obj* by *path* segments and return an ``int`` value.
+
+        Supports both dict-key access and attribute access so it works
+        uniformly with raw JSON dicts **and** SDK Pydantic models.
+        """
+        current = obj
+        for segment in path:
+            if current is None:
+                return 0
+            if isinstance(current, dict):
+                current = current.get(segment)
+            else:
+                current = getattr(current, segment, None)
+        return int(current or 0) if current is not None else 0
+
+    def _parse(self, response: Any) -> LLMResponse:
+        if isinstance(response, str):
+            return LLMResponse(content=response, finish_reason="stop")
+
+        response_map = self._maybe_mapping(response)
+        if response_map is not None:
+            choices = response_map.get("choices") or []
+            if not choices:
+                content = self._extract_text_content(
+                    response_map.get("content") or response_map.get("output_text")
+                )
+                reasoning_content = self._extract_text_content(
+                    response_map.get("reasoning_content")
+                )
+                if content is not None:
+                    return LLMResponse(
+                        content=content,
+                        reasoning_content=reasoning_content,
+                        finish_reason=str(response_map.get("finish_reason") or "stop"),
+                        usage=self._extract_usage(response_map),
+                    )
+                return LLMResponse(
+                    content="Error: API returned empty choices.",
+                    finish_reason="error",
+                    error_kind="empty",
+                )
+
+            choice0 = self._maybe_mapping(choices[0]) or {}
+            msg0 = self._maybe_mapping(choice0.get("message")) or {}
+            content = self._extract_text_content(msg0.get("content"))
+            finish_reason = str(choice0.get("finish_reason") or "stop")
+
+            raw_tool_calls: list[Any] = []
+            # StepFun: fallback to reasoning field when content is empty
+            if not content and msg0.get("reasoning") and self._spec and self._spec.reasoning_as_content:
+                content = self._extract_text_content(msg0.get("reasoning"))
+            reasoning_content = msg0.get("reasoning_content")
+            if reasoning_content is None and msg0.get("reasoning"):
+                reasoning_content = self._extract_text_content(msg0.get("reasoning"))
+            # Mistral reasoning models return thinking text inside the content
+            # array; lift it into reasoning_content so the runner records it
+            # under the reasoning trace.
+            spec = getattr(self, "_spec", None)
+            if reasoning_content is None and getattr(spec, "extract_thinking_blocks", False):
+                reasoning_content = self._extract_thinking_content(msg0.get("content"))
+            for ch in choices:
+                ch_map = self._maybe_mapping(ch) or {}
+                m = self._maybe_mapping(ch_map.get("message")) or {}
+                tool_calls = m.get("tool_calls")
+                if isinstance(tool_calls, list) and tool_calls:
+                    raw_tool_calls.extend(tool_calls)
+                    if ch_map.get("finish_reason") in ("tool_calls", "stop"):
+                        finish_reason = str(ch_map["finish_reason"])
+                if not content:
+                    content = self._extract_text_content(m.get("content"))
+                if reasoning_content is None:
+                    reasoning_content = m.get("reasoning_content")
+
+            # Deduplicate tool call IDs (same pattern as streaming path)
+            # Some providers reuse the same ID for parallel tool calls.
+            _seen_tc_ids: set[str] = set()
+            parsed_tool_calls = []
+            for tc in raw_tool_calls:
+                tc_map = self._maybe_mapping(tc) or {}
+                fn = self._maybe_mapping(tc_map.get("function")) or {}
+                args = parse_tool_arguments(fn.get("arguments", {}))
+                ec, prov, fn_prov = _extract_tc_extras(tc)
+                raw_id = str(tc_map.get("id") or _short_tool_id())
+                if not raw_id or raw_id in _seen_tc_ids:
+                    raw_id = _short_tool_id()
+                _seen_tc_ids.add(raw_id)
+                parsed_tool_calls.append(ToolCallRequest(
+                    id=raw_id,
+                    name=str(fn.get("name") or ""),
+                    arguments=args,
+                    extra_content=ec,
+                    provider_specific_fields=prov,
+                    function_provider_specific_fields=fn_prov,
+                ))
+            if not parsed_tool_calls:
+                content, parsed_tool_calls = _extract_text_tool_calls(content)
+
+            return LLMResponse(
+                content=content,
+                tool_calls=parsed_tool_calls,
+                finish_reason=finish_reason,
+                usage=self._extract_usage(response_map),
+                reasoning_content=reasoning_content if isinstance(reasoning_content, str) else None,
+            )
+
+        if not response.choices:
+            return LLMResponse(
+                content="Error: API returned empty choices.",
+                finish_reason="error",
+                error_kind="empty",
+            )
+
+        choice = response.choices[0]
+        msg = choice.message
+        content = msg.content
+        finish_reason = choice.finish_reason
+
+        raw_tool_calls: list[Any] = []
+        for ch in response.choices:
+            m = ch.message
+            if hasattr(m, "tool_calls") and m.tool_calls:
+                raw_tool_calls.extend(m.tool_calls)
+                if ch.finish_reason in ("tool_calls", "stop"):
+                    finish_reason = ch.finish_reason
+            if not content and m.content:
+                content = m.content
+            if not content and getattr(m, "reasoning", None) and self._spec and self._spec.reasoning_as_content:
+                content = m.reasoning
+
+        tool_calls = []
+        for tc in raw_tool_calls:
+            args = parse_tool_arguments(tc.function.arguments)
+            ec, prov, fn_prov = _extract_tc_extras(tc)
+            tool_calls.append(ToolCallRequest(
+                id=str(getattr(tc, "id", None) or _short_tool_id()),
+                name=tc.function.name,
+                arguments=args,
+                extra_content=ec,
+                provider_specific_fields=prov,
+                function_provider_specific_fields=fn_prov,
+            ))
+        if not tool_calls:
+            content, tool_calls = _extract_text_tool_calls(content)
+
+        reasoning_content = getattr(msg, "reasoning_content", None)
+        if reasoning_content is None and getattr(msg, "reasoning", None):
+            reasoning_content = msg.reasoning
+
+        return LLMResponse(
+            content=content,
+            tool_calls=tool_calls,
+            finish_reason=finish_reason or "stop",
+            usage=self._extract_usage(response),
+            reasoning_content=reasoning_content,
+        )
+
+    @classmethod
+    def _parse_chunks(cls, chunks: list[Any]) -> LLMResponse:
+        content_parts: list[str] = []
+        reasoning_parts: list[str] = []
+        tc_bufs: dict[int, dict[str, Any]] = {}
+        finish_reason = "stop"
+        usage: dict[str, int] = {}
+
+        def _accum_tc(tc: Any, idx_hint: int) -> None:
+            """Accumulate one streaming tool-call delta into *tc_bufs*."""
+            tc_index: int = _get(tc, "index") if _get(tc, "index") is not None else idx_hint
+            buf = tc_bufs.setdefault(tc_index, {
+                "id": "", "name": "", "arguments": "",
+                "extra_content": None, "prov": None, "fn_prov": None,
+            })
+            tc_id = _get(tc, "id")
+            if tc_id:
+                buf["id"] = str(tc_id)
+            fn = _get(tc, "function")
+            if fn is not None:
+                fn_name = _get(fn, "name")
+                if fn_name:
+                    buf["name"] = str(fn_name)
+                fn_args = _get(fn, "arguments")
+                if fn_args:
+                    buf["arguments"] += str(fn_args)
+            ec, prov, fn_prov = _extract_tc_extras(tc)
+            if ec:
+                buf["extra_content"] = ec
+            if prov:
+                buf["prov"] = prov
+            if fn_prov:
+                buf["fn_prov"] = fn_prov
+
+        def _accum_legacy_function_call(function_call: Any) -> None:
+            """Accumulate legacy ``delta.function_call`` streaming chunks."""
+            if not function_call:
+                return
+            buf = tc_bufs.setdefault(0, {
+                "id": "", "name": "", "arguments": "",
+                "extra_content": None, "prov": None, "fn_prov": None,
+            })
+            fn_name = _get(function_call, "name")
+            if fn_name:
+                buf["name"] = str(fn_name)
+            fn_args = _get(function_call, "arguments")
+            if fn_args:
+                buf["arguments"] += str(fn_args)
+
+        for chunk in chunks:
+            if isinstance(chunk, str):
+                content_parts.append(chunk)
+                continue
+
+            chunk_map = cls._maybe_mapping(chunk)
+            if chunk_map is not None:
+                choices = chunk_map.get("choices") or []
+                if not choices:
+                    usage = cls._extract_usage(chunk_map) or usage
+                    text = cls._extract_text_content(
+                        chunk_map.get("content") or chunk_map.get("output_text")
+                    )
+                    if text:
+                        content_parts.append(text)
+                    continue
+                choice = cls._maybe_mapping(choices[0]) or {}
+                if choice.get("finish_reason"):
+                    finish_reason = str(choice["finish_reason"])
+                delta = cls._maybe_mapping(choice.get("delta")) or {}
+                raw_delta_content = delta.get("content")
+                text = cls._extract_text_content(raw_delta_content)
+                if text:
+                    content_parts.append(text)
+                text = cls._extract_text_content(delta.get("reasoning_content"))
+                if not text:
+                    text = cls._extract_text_content(delta.get("reasoning"))
+                if not text:
+                    # Mistral streams thinking inside the content array as
+                    # {"type":"thinking", thinking:[{"type":"text", ...}]}.
+                    text = cls._extract_thinking_content(raw_delta_content)
+                if text:
+                    reasoning_parts.append(text)
+                for idx, tc in enumerate(delta.get("tool_calls") or []):
+                    _accum_tc(tc, idx)
+                _accum_legacy_function_call(delta.get("function_call"))
+                usage = cls._extract_usage(chunk_map) or usage
+                continue
+
+            if not chunk.choices:
+                usage = cls._extract_usage(chunk) or usage
+                continue
+            choice = chunk.choices[0]
+            if choice.finish_reason:
+                finish_reason = choice.finish_reason
+            delta = choice.delta
+            if delta and delta.content:
+                text = cls._extract_text_content(delta.content)
+                if text:
+                    content_parts.append(text)
+                thinking_text = cls._extract_thinking_content(delta.content)
+                if thinking_text:
+                    reasoning_parts.append(thinking_text)
+            if delta:
+                reasoning = getattr(delta, "reasoning_content", None)
+                if not reasoning:
+                    reasoning = getattr(delta, "reasoning", None)
+                if reasoning:
+                    text = cls._extract_text_content(reasoning)
+                    if text:
+                        reasoning_parts.append(text)
+            for tc in (getattr(delta, "tool_calls", None) or []) if delta else []:
+                _accum_tc(tc, getattr(tc, "index", 0))
+            if delta:
+                _accum_legacy_function_call(getattr(delta, "function_call", None))
+
+        # Some providers (e.g. Zhipu/GLM) reuse the same tool_call id for
+        # parallel tool calls in streaming mode. Deduplicate before building
+        # the response so downstream tool messages don't collide.
+        _seen_tc_ids: set[str] = set()
+        for b in tc_bufs.values():
+            if not b["id"] or b["id"] in _seen_tc_ids:
+                b["id"] = _short_tool_id()
+            _seen_tc_ids.add(b["id"])
+
+        content = "".join(content_parts) or None
+        tool_calls = [
+            ToolCallRequest(
+                id=b["id"] or _short_tool_id(),
+                name=b["name"],
+                arguments=parse_tool_arguments(b["arguments"]),
+                extra_content=b.get("extra_content"),
+                provider_specific_fields=b.get("prov"),
+                function_provider_specific_fields=b.get("fn_prov"),
+            )
+            for b in tc_bufs.values()
+        ]
+        if not tool_calls:
+            content, tool_calls = _extract_text_tool_calls(content)
+
+        return LLMResponse(
+            content=content,
+            tool_calls=tool_calls,
+            finish_reason=finish_reason,
+            usage=usage,
+            reasoning_content="".join(reasoning_parts) or None,
+        )
+
+    @classmethod
+    def _extract_error_metadata(cls, e: Exception) -> dict[str, Any]:
+        response = getattr(e, "response", None)
+        headers = getattr(response, "headers", None)
+        payload = (
+            getattr(e, "body", None)
+            or getattr(e, "doc", None)
+            or getattr(response, "text", None)
+        )
+        if payload is None and response is not None:
+            response_json = getattr(response, "json", None)
+            if callable(response_json):
+                try:
+                    payload = response_json()
+                except Exception:
+                    payload = None
+        error_type, error_code = LLMProvider._extract_error_type_code(payload)
+
+        status_code = getattr(e, "status_code", None)
+        if status_code is None and response is not None:
+            status_code = getattr(response, "status_code", None)
+
+        should_retry: bool | None = None
+        if headers is not None:
+            raw = headers.get("x-should-retry")
+            if isinstance(raw, str):
+                lowered = raw.strip().lower()
+                if lowered == "true":
+                    should_retry = True
+                elif lowered == "false":
+                    should_retry = False
+
+        error_kind: str | None = None
+        error_name = e.__class__.__name__.lower()
+        if "timeout" in error_name:
+            error_kind = "timeout"
+        elif "connection" in error_name:
+            error_kind = "connection"
+
+        return {
+            "error_status_code": int(status_code) if status_code is not None else None,
+            "error_kind": error_kind,
+            "error_type": error_type,
+            "error_code": error_code,
+            "error_retry_after_s": cls._extract_retry_after_from_headers(headers),
+            "error_should_retry": should_retry,
+        }
+
+    @staticmethod
+    def _handle_error(
+        e: Exception,
+        *,
+        spec: ProviderSpec | None = None,
+        api_base: str | None = None,
+    ) -> LLMResponse:
+        body = (
+            getattr(e, "doc", None)
+            or getattr(e, "body", None)
+            or getattr(getattr(e, "response", None), "text", None)
+        )
+        body_text = body if isinstance(body, str) else str(body) if body is not None else ""
+        msg = f"Error: {body_text.strip()[:500]}" if body_text.strip() else f"Error calling LLM: {e}"
+
+        text = f"{body_text} {e}".lower()
+        if spec and spec.is_local and ("502" in text or "connection" in text or "refused" in text):
+            msg += (
+                "\nHint: this is a local model endpoint. Check that the local server is reachable at "
+                f"{api_base or spec.default_api_base}, and if you are using a proxy/tunnel, make sure it "
+                "can reach your local Ollama/vLLM service instead of routing localhost through the remote host."
+            )
+
+        response = getattr(e, "response", None)
+        retry_after = LLMProvider._extract_retry_after_from_headers(getattr(response, "headers", None))
+        if retry_after is None:
+            retry_after = LLMProvider._extract_retry_after(msg)
+        return LLMResponse(
+            content=msg,
+            finish_reason="error",
+            retry_after=retry_after,
+            **OpenAICompatProvider._extract_error_metadata(e),
+        )
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    async def chat(
+        self,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None = None,
+        model: str | None = None,
+        max_tokens: int = 4096,
+        temperature: float = 0.7,
+        reasoning_effort: str | None = None,
+        tool_choice: str | dict[str, Any] | None = None,
+    ) -> LLMResponse:
+        await self._ensure_client()
+        try:
+            if self._should_use_responses_api(model, reasoning_effort):
+                try:
+                    body = self._build_responses_body(
+                        messages, tools, model, max_tokens, temperature,
+                        reasoning_effort, tool_choice,
+                    )
+                    result = parse_response_output(await self._client.responses.create(**body))
+                    self._record_responses_success(model, reasoning_effort)
+                    return result
+                except Exception as responses_error:
+                    if self._spec and self._spec.name == "github_copilot":
+                        # Copilot gateway exposes GPT-5/o-series only via /responses;
+                        # falling back to /chat/completions cannot succeed and would
+                        # hide the real error.
+                        raise
+                    if self._api_type == "responses":
+                        raise
+                    if not self._should_fallback_from_responses_error(responses_error):
+                        raise
+                    self._record_responses_failure(model, reasoning_effort)
+
+            kwargs = self._build_kwargs(
+                messages, tools, model, max_tokens, temperature,
+                reasoning_effort, tool_choice,
+            )
+            return self._parse(await self._client.chat.completions.create(**kwargs))
+        except Exception as e:
+            return self._handle_error(e, spec=self._spec, api_base=self.api_base)
+
+    async def chat_stream(
+        self,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None = None,
+        model: str | None = None,
+        max_tokens: int = 4096,
+        temperature: float = 0.7,
+        reasoning_effort: str | None = None,
+        tool_choice: str | dict[str, Any] | None = None,
+        on_content_delta: Callable[[str], Awaitable[None]] | None = None,
+        on_thinking_delta: Callable[[str], Awaitable[None]] | None = None,
+        on_tool_call_delta: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
+    ) -> LLMResponse:
+        await self._ensure_client()
+        idle_timeout_s = resolve_stream_idle_timeout_s()
+        try:
+            if self._should_use_responses_api(model, reasoning_effort):
+                try:
+                    body = self._build_responses_body(
+                        messages, tools, model, max_tokens, temperature,
+                        reasoning_effort, tool_choice,
+                    )
+                    body["stream"] = True
+                    stream = await self._client.responses.create(**body)
+
+                    async def _timed_stream():
+                        stream_iter = stream.__aiter__()
+                        while True:
+                            try:
+                                yield await asyncio.wait_for(
+                                    stream_iter.__anext__(),
+                                    timeout=idle_timeout_s,
+                                )
+                            except StopAsyncIteration:
+                                break
+
+                    (
+                        content,
+                        tool_calls,
+                        finish_reason,
+                        usage,
+                        reasoning_content,
+                    ) = await consume_sdk_stream(
+                        _timed_stream(),
+                        on_content_delta,
+                        on_tool_call_delta=on_tool_call_delta,
+                    )
+                    self._record_responses_success(model, reasoning_effort)
+                    return LLMResponse(
+                        content=content or None,
+                        tool_calls=tool_calls,
+                        finish_reason=finish_reason,
+                        usage=usage,
+                        reasoning_content=reasoning_content,
+                    )
+                except Exception as responses_error:
+                    if self._spec and self._spec.name == "github_copilot":
+                        # Copilot gateway exposes GPT-5/o-series only via /responses;
+                        # falling back to /chat/completions cannot succeed and would
+                        # hide the real error.
+                        raise
+                    if self._api_type == "responses":
+                        raise
+                    if not self._should_fallback_from_responses_error(responses_error):
+                        raise
+                    self._record_responses_failure(model, reasoning_effort)
+
+            kwargs = self._build_kwargs(
+                messages, tools, model, max_tokens, temperature,
+                reasoning_effort, tool_choice,
+            )
+            if self._spec and self._spec.name == "zhipu" and tools and on_tool_call_delta:
+                # Z.AI/GLM keeps streaming tool-call arguments behind an
+                # explicit provider flag.  Pass it through the OpenAI SDK's
+                # extra_body escape hatch so the usual delta.tool_calls path
+                # can surface live file-edit progress.
+                kwargs.setdefault("extra_body", {})["tool_stream"] = True
+            kwargs["stream"] = True
+            kwargs["stream_options"] = {"include_usage": True}
+            stream = await self._client.chat.completions.create(**kwargs)
+            chunks: list[Any] = []
+            stream_iter = stream.__aiter__()
+            while True:
+                try:
+                    chunk = await asyncio.wait_for(
+                        stream_iter.__anext__(),
+                        timeout=idle_timeout_s,
+                    )
+                except StopAsyncIteration:
+                    break
+                chunks.append(chunk)
+                if chunk.choices:
+                    delta_obj = chunk.choices[0].delta
+                    raw_delta_content = getattr(delta_obj, "content", None)
+                    if on_content_delta:
+                        # Mistral streams content as a list of {"type":"thinking",
+                        # ...} + {"type":"text",...} blocks. Extract just the
+                        # text portion before invoking the callback so callers
+                        # never see non-string content.
+                        text = self._extract_text_content(raw_delta_content)
+                        if text:
+                            await on_content_delta(text)
+                    if on_thinking_delta:
+                        reasoning = getattr(delta_obj, "reasoning_content", None) or getattr(
+                            delta_obj, "reasoning", None,
+                        )
+                        r_text = self._extract_text_content(reasoning)
+                        if not r_text:
+                            # Mistral keeps the thinking trace inside the
+                            # content array rather than a separate field.
+                            r_text = self._extract_thinking_content(raw_delta_content)
+                        if r_text:
+                            await on_thinking_delta(r_text)
+                    if on_tool_call_delta:
+                        for idx, tool_delta in enumerate(
+                            getattr(delta_obj, "tool_calls", None) or []
+                        ):
+                            fn = _get(tool_delta, "function")
+                            tool_index = _get(tool_delta, "index")
+                            await on_tool_call_delta({
+                                "index": tool_index if tool_index is not None else idx,
+                                "call_id": str(_get(tool_delta, "id") or ""),
+                                "name": str(_get(fn, "name") or "") if fn is not None else "",
+                                "arguments_delta": (
+                                    str(_get(fn, "arguments") or "") if fn is not None else ""
+                                ),
+                            })
+                        function_call = getattr(delta_obj, "function_call", None)
+                        if function_call:
+                            await on_tool_call_delta({
+                                "index": 0,
+                                "call_id": "",
+                                "name": str(_get(function_call, "name") or ""),
+                                "arguments_delta": str(_get(function_call, "arguments") or ""),
+                            })
+            return self._parse_chunks(chunks)
+        except asyncio.TimeoutError:
+            return LLMResponse(
+                content=(
+                    f"Error calling LLM: stream stalled for more than "
+                    f"{idle_timeout_s:g} seconds"
+                ),
+                finish_reason="error",
+                error_kind="timeout",
+            )
+        except Exception as e:
+            return self._handle_error(e, spec=self._spec, api_base=self.api_base)
+
+    def get_default_model(self) -> str:
+        return self.default_model
diff --git a/nanobot-patches/openai_compat_provider.py.before_patch b/nanobot-patches/openai_compat_provider.py.before_patch
new file mode 100644
index 0000000..c8bad3c
--- /dev/null
+++ b/nanobot-patches/openai_compat_provider.py.before_patch
@@ -0,0 +1,1746 @@
+"""OpenAI-compatible provider for all non-Anthropic LLM APIs."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import importlib.util
+import json
+import os
+import re
+import secrets
+import string
+import time
+import uuid
+from collections import deque
+from collections.abc import Awaitable, Callable
+from ipaddress import ip_address
+from typing import TYPE_CHECKING, Any
+from urllib.parse import urlparse
+
+from loguru import logger
+from pydantic.alias_generators import to_snake
+
+from nanobot.providers.base import (
+    LLMProvider,
+    LLMResponse,
+    ToolCallRequest,
+    parse_tool_arguments,
+    resolve_stream_idle_timeout_s,
+    tool_arguments_json_for_replay,
+)
+from nanobot.providers.openai_responses import (
+    consume_sdk_stream,
+    convert_messages,
+    convert_tools,
+    parse_response_output,
+)
+
+if TYPE_CHECKING:
+    from openai import AsyncOpenAI as AsyncOpenAIType
+
+    from nanobot.providers.registry import ProviderSpec
+
+# Module-level placeholder — set lazily by _ensure_client on first real
+# use, or replaced by tests via ``patch(...)``.  Kept as a plain name so
+# that ``unittest.mock.patch`` can find and replace it.
+AsyncOpenAI: Any = None
+
+_ALLOWED_MSG_KEYS = frozenset({
+    "role", "content", "tool_calls", "tool_call_id", "name",
+    "reasoning_content", "extra_content",
+})
+_ALNUM = string.ascii_letters + string.digits
+
+_STANDARD_TC_KEYS = frozenset({"id", "type", "index", "function"})
+_STANDARD_FN_KEYS = frozenset({"name", "arguments"})
+_DEFAULT_OPENROUTER_HEADERS = {
+    "HTTP-Referer": "https://github.com/HKUDS/nanobot",
+    "X-OpenRouter-Title": "nanobot",
+    "X-OpenRouter-Categories": "cli-agent,personal-agent",
+}
+_KIMI_K3_MODEL = "kimi-k3"
+_KIMI_THINKING_MODELS: frozenset[str] = frozenset({
+    "kimi-k2.5",
+    "kimi-k2.6",
+    "kimi-k2.7",
+    "kimi-k2.7-code",
+    "kimi-k2.7-code-highspeed",
+    "k2.6-code-preview",
+})
+_KIMI_ALWAYS_THINKING_MODELS: frozenset[str] = frozenset({
+    "kimi-k2.7-code",
+    "kimi-k2.7-code-highspeed",
+})
+_KIMI_SERVER_MANAGED_TEMPERATURE_MODELS: frozenset[str] = frozenset({
+    "kimi-k2.5",
+    "kimi-k2.6",
+})
+_TEXT_TOOL_CALL_RE = re.compile(r"<tool_call>\s*(.*?)\s*</tool_call>", re.DOTALL)
+# Thinking-capable MiMo models per Xiaomi docs (see
+# tests/providers/test_xiaomi_mimo_thinking.py). mimo-v2-flash is omitted
+# because it does not support thinking.
+_MIMO_THINKING_MODELS: frozenset[str] = frozenset({
+    "mimo-v2.5-pro",
+    "mimo-v2.5",
+    "mimo-v2-pro",
+    "mimo-v2-omni",
+})
+_OPENAI_COMPAT_REQUEST_TIMEOUT_S = 120.0
+
+# Maps ProviderSpec.thinking_style → extra_body builder.
+# Each builder takes a bool (thinking_enabled) and returns the dict to
+# merge into extra_body, keeping the style→wire-format mapping in one place.
+_THINKING_STYLE_MAP: dict[str, Any] = {
+    "thinking_type": lambda on: {"thinking": {"type": "enabled" if on else "disabled"}},
+    "enable_thinking": lambda on: {"enable_thinking": on},
+    "reasoning_split": lambda on: {"reasoning_split": on},
+}
+_GATEWAY_REASONING_STYLE_MAP: dict[str, Any] = {
+    "reasoning_effort": lambda effort: {"reasoning": {"effort": effort}},
+}
+_QWEN_THINKING_MODELS: frozenset[str] = frozenset({
+    "qwen3.7-max",
+    "qwen3.7-plus",
+    "qwen3.6-max-preview",
+    "qwen3.6-plus",
+    "qwen3.6-flash",
+    "qwen3.5-plus",
+    "qwen3.5-flash",
+})
+
+_MODEL_THINKING_STYLES: dict[str, str] = {
+    **dict.fromkeys(_KIMI_THINKING_MODELS, "thinking_type"),
+    **dict.fromkeys(_MIMO_THINKING_MODELS, "thinking_type"),
+    **dict.fromkeys(_QWEN_THINKING_MODELS, "enable_thinking"),
+}
+
+
+def _model_slug(model_name: str) -> str:
+    return model_name.lower().rsplit("/", 1)[-1]
+
+
+def _provider_prefix_key(name: str) -> str:
+    return to_snake(name.replace("-", "_")).lower()
+
+
+def _requires_max_completion_tokens(model_name: str) -> bool:
+    """Return True for models that require ``max_completion_tokens``."""
+    slug = _model_slug(model_name)
+    return slug == _KIMI_K3_MODEL or "gpt-5" in slug or any(
+        slug == p or slug.startswith((p + "-", p + ".")) for p in ("o1", "o3", "o4")
+    )
+
+
+def _model_thinking_style(model_name: str) -> str:
+    return _MODEL_THINKING_STYLES.get(_model_slug(model_name), "")
+
+
+def _thinking_styles_for(spec: ProviderSpec | None, model_name: str) -> list[str]:
+    styles: list[str] = []
+    if spec and spec.thinking_style:
+        styles.append(spec.thinking_style)
+    model_style = _model_thinking_style(model_name)
+    if model_style and model_style not in styles:
+        styles.append(model_style)
+    return styles
+
+
+def _thinking_extra_body(style: str, thinking_enabled: bool) -> dict[str, Any] | None:
+    builder = _THINKING_STYLE_MAP.get(style)
+    return builder(thinking_enabled) if builder else None
+
+
+def _gateway_reasoning_extra_body(style: str, effort: str | None) -> dict[str, Any] | None:
+    if not effort:
+        return None
+    builder = _GATEWAY_REASONING_STYLE_MAP.get(style)
+    return builder(effort) if builder else None
+
+
+def _openai_compat_timeout_s() -> float:
+    """Return the bounded request timeout used for OpenAI-compatible providers."""
+    return _float_env("NANOBOT_OPENAI_COMPAT_TIMEOUT_S", _OPENAI_COMPAT_REQUEST_TIMEOUT_S)
+
+
+def _float_env(name: str, default: float) -> float:
+    raw = os.environ.get(name)
+    if raw is None or not raw.strip():
+        return default
+    try:
+        value = float(raw)
+    except (TypeError, ValueError):
+        logger.warning("Ignoring invalid {}={!r}; using {}", name, raw, default)
+        return default
+    if value <= 0:
+        logger.warning("Ignoring non-positive {}={!r}; using {}", name, raw, default)
+        return default
+    return value
+
+
+def _short_tool_id() -> str:
+    """9-char alphanumeric ID compatible with all providers (incl. Mistral)."""
+    return "".join(secrets.choice(_ALNUM) for _ in range(9))
+
+
+def _strip_json_fence(text: str) -> str:
+    stripped = text.strip()
+    if not stripped.startswith("```") or not stripped.endswith("```"):
+        return stripped
+    lines = stripped.splitlines()
+    if len(lines) < 2:
+        return stripped
+    return "\n".join(lines[1:-1]).strip()
+
+
+def _extract_text_tool_calls(content: str | None) -> tuple[str | None, list[ToolCallRequest]]:
+    """Normalize common text-format tool call blocks into structured calls."""
+    if not content or "<tool_call>" not in content:
+        return content, []
+
+    tool_calls: list[ToolCallRequest] = []
+    spans: list[tuple[int, int]] = []
+    for match in _TEXT_TOOL_CALL_RE.finditer(content):
+        try:
+            payload = json.loads(_strip_json_fence(match.group(1)))
+        except Exception:
+            continue
+        if not isinstance(payload, dict):
+            continue
+
+        nested = payload.get("tool_call")
+        if isinstance(nested, dict):
+            payload = nested
+        function = payload.get("function")
+        if not isinstance(function, dict):
+            function = payload
+        name = function.get("name")
+        if not isinstance(name, str) or not name:
+            continue
+
+        arguments = function.get("arguments", payload.get("arguments", {}))
+        tool_calls.append(ToolCallRequest(
+            id=str(payload.get("id") or _short_tool_id()),
+            name=name,
+            arguments=parse_tool_arguments(arguments),
+        ))
+        spans.append(match.span())
+
+    if not tool_calls:
+        return content, []
+
+    visible_parts: list[str] = []
+    last = 0
+    for start, end in spans:
+        visible_parts.append(content[last:start])
+        last = end
+    visible_parts.append(content[last:])
+    visible_content = "".join(visible_parts).strip() or None
+    return visible_content, tool_calls
+
+
+def _get(obj: Any, key: str) -> Any:
+    """Get a value from dict or object attribute, returning None if absent."""
+    if isinstance(obj, dict):
+        return obj.get(key)
+    return getattr(obj, key, None)
+
+
+def _coerce_dict(value: Any) -> dict[str, Any] | None:
+    """Try to coerce *value* to a dict; return None if not possible or empty."""
+    if value is None:
+        return None
+    if isinstance(value, dict):
+        return value if value else None
+    model_dump = getattr(value, "model_dump", None)
+    if callable(model_dump):
+        dumped = model_dump()
+        if isinstance(dumped, dict) and dumped:
+            return dumped
+    return None
+
+
+def _extract_tc_extras(tc: Any) -> tuple[
+    dict[str, Any] | None,
+    dict[str, Any] | None,
+    dict[str, Any] | None,
+]:
+    """Extract (extra_content, provider_specific_fields, fn_provider_specific_fields).
+
+    Works for both SDK objects and dicts.  Captures Gemini ``extra_content``
+    verbatim and any non-standard keys on the tool-call / function.
+    """
+    extra_content = _coerce_dict(_get(tc, "extra_content"))
+
+    tc_dict = _coerce_dict(tc)
+    prov = None
+    fn_prov = None
+    if tc_dict is not None:
+        leftover = {k: v for k, v in tc_dict.items()
+                    if k not in _STANDARD_TC_KEYS and k != "extra_content" and v is not None}
+        if leftover:
+            prov = leftover
+        fn = _coerce_dict(tc_dict.get("function"))
+        if fn is not None:
+            fn_leftover = {k: v for k, v in fn.items()
+                          if k not in _STANDARD_FN_KEYS and v is not None}
+            if fn_leftover:
+                fn_prov = fn_leftover
+    else:
+        prov = _coerce_dict(_get(tc, "provider_specific_fields"))
+        fn_obj = _get(tc, "function")
+        if fn_obj is not None:
+            fn_prov = _coerce_dict(_get(fn_obj, "provider_specific_fields"))
+
+    return extra_content, prov, fn_prov
+
+
+def _uses_openrouter_attribution(spec: "ProviderSpec | None", api_base: str | None) -> bool:
+    """Apply Nanobot attribution headers to OpenRouter requests by default."""
+    if spec and spec.name == "openrouter":
+        return True
+    return bool(api_base and "openrouter" in api_base.lower())
+
+
+_RESPONSES_FAILURE_THRESHOLD = 3
+_RESPONSES_PROBE_INTERVAL_S = 300  # 5 minutes
+
+
+def _is_local_endpoint(
+    spec: "ProviderSpec | None",
+    api_base: str | None,
+) -> bool:
+    """Return True when the endpoint is a local or LAN model server.
+
+    Matches either the provider spec's ``is_local`` flag or common private-
+    network patterns in the base URL (localhost, 127.x, 192.168.x, 10.x,
+    172.16-31.x, Docker ``host.docker.internal``).
+    """
+    if spec and spec.is_local:
+        return True
+    if not api_base:
+        return False
+    raw = api_base.strip().lower()
+    parsed = urlparse(raw if "://" in raw else f"//{raw}")
+    try:
+        host = parsed.hostname
+    except ValueError:
+        return False
+    if host in {"localhost", "host.docker.internal"}:
+        return True
+    if not host:
+        return False
+    try:
+        addr = ip_address(host)
+    except ValueError:
+        return False
+    return addr.is_loopback or addr.is_private
+
+
+def _is_direct_openai_base(api_base: str | None) -> bool:
+    """Return True for direct OpenAI endpoints, not generic OpenAI-compatible gateways."""
+    if not api_base:
+        return True
+    normalized = api_base.strip().lower().rstrip("/")
+    return "api.openai.com" in normalized and "openrouter" not in normalized
+
+
+def _responses_circuit_key(
+    model: str | None,
+    default_model: str,
+    reasoning_effort: str | None,
+) -> str:
+    model_name = (model or default_model).lower()
+    effort = reasoning_effort.lower() if isinstance(reasoning_effort, str) else ""
+    return f"{model_name}:{effort}"
+
+
+def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
+    """Recursively merge *override* into *base*, returning a new dict.
+
+    Nested dicts are merged key-by-key; all other types in *override*
+    replace the corresponding key in *base*.
+    """
+    merged = dict(base)
+    for key, value in override.items():
+        if (
+            key in merged
+            and isinstance(merged[key], dict)
+            and isinstance(value, dict)
+        ):
+            merged[key] = _deep_merge(merged[key], value)
+        else:
+            merged[key] = value
+    return merged
+
+
+def _merge_unique_list(base: Any, override: Any) -> Any:
+    """Append list values while preserving order and removing duplicates."""
+    if not isinstance(base, list) or not isinstance(override, list):
+        return override
+    result: list[Any] = []
+    seen: set[str] = set()
+    for value in [*base, *override]:
+        try:
+            key = json.dumps(value, sort_keys=True, ensure_ascii=False)
+        except Exception:
+            key = repr(value)
+        if key in seen:
+            continue
+        seen.add(key)
+        result.append(value)
+    return result
+
+
+def _merge_responses_extra_body(
+    body: dict[str, Any],
+    extra_body: dict[str, Any],
+) -> dict[str, Any]:
+    """Merge configured Responses API body fields without clobbering tools."""
+    reserved = {"include", "tools"}
+    regular_extra = {key: value for key, value in extra_body.items() if key not in reserved}
+    merged = _deep_merge(body, regular_extra)
+
+    if "include" in extra_body:
+        merged["include"] = _merge_unique_list(body.get("include"), extra_body["include"])
+
+    if "tools" in extra_body:
+        current_tools = body.get("tools")
+        configured_tools = extra_body["tools"]
+        if isinstance(current_tools, list) and isinstance(configured_tools, list):
+            merged["tools"] = [*current_tools, *configured_tools]
+        else:
+            merged["tools"] = configured_tools
+
+    return merged
+
+
+class OpenAICompatProvider(LLMProvider):
+    """Unified provider for all OpenAI-compatible APIs.
+
+    Receives a resolved ``ProviderSpec`` from the caller — no internal
+    registry lookups needed.
+    """
+
+    def __init__(
+        self,
+        api_key: str | None = None,
+        api_base: str | None = None,
+        default_model: str = "gpt-4o",
+        extra_headers: dict[str, str] | None = None,
+        spec: ProviderSpec | None = None,
+        extra_body: dict[str, Any] | None = None,
+        api_type: str = "auto",
+        extra_query: dict[str, str] | None = None,
+        proxy: str | None = None,
+    ):
+        super().__init__(api_key, api_base)
+        self.default_model = default_model
+        self.extra_headers = extra_headers or {}
+        self._spec = spec
+        self._extra_body = extra_body or {}
+        self._api_type = api_type if spec and spec.name == "openai" else "auto"
+        self._extra_query = extra_query or {}
+        self._proxy = proxy or None
+
+        if api_key and spec and spec.env_key:
+            self._setup_env(api_key, api_base)
+
+        effective_base = api_base or (spec.default_api_base if spec else None) or None
+        self._effective_base = effective_base
+        self._default_headers = {"x-session-affinity": uuid.uuid4().hex}
+        if _uses_openrouter_attribution(spec, effective_base):
+            self._default_headers.update(_DEFAULT_OPENROUTER_HEADERS)
+        if extra_headers:
+            self._default_headers.update(extra_headers)
+        self._api_key_for_client = api_key or "no-key"
+        self._is_local = _is_local_endpoint(spec, effective_base)
+
+        # Lazy-init: the OpenAI client and its httpx transport are expensive
+        # to create (~700 ms on Windows). Defer until first use.
+        self._client: AsyncOpenAIType | None = None
+        self._client_lock = asyncio.Lock()
+
+        # Responses API circuit breaker: skip after repeated failures,
+        # probe again after _RESPONSES_PROBE_INTERVAL_S seconds.
+        self._responses_failures: dict[str, int] = {}
+        self._responses_tripped_at: dict[str, float] = {}
+
+    def _build_client(self) -> None:
+        """Create the OpenAI client using the current module-level AsyncOpenAI."""
+        import httpx
+
+        timeout_s = _openai_compat_timeout_s()
+        http_client: httpx.AsyncClient | None = None
+        if self._proxy:
+            http_client = httpx.AsyncClient(
+                timeout=timeout_s,
+                proxy=self._proxy,
+                trust_env=False,
+                follow_redirects=True,
+            )
+        elif self._is_local:
+            # Local model servers (Ollama, llama.cpp, vLLM) often close idle
+            # HTTP connections before the client-side keepalive expires. When
+            # two LLM calls happen seconds apart (e.g. heartbeat _decide then
+            # process_direct), the second call may grab a now-dead pooled
+            # connection, causing a transient APIConnectionError on every first
+            # attempt. Disabling keepalive for local endpoints avoids this by
+            # opening a fresh connection for each request, which is cheap on a
+            # LAN. Cloud providers benefit from keepalive, so we leave the
+            # default pool settings for them.
+            #
+            # Also disable proxy for local endpoints: when the host has
+            # HTTP_PROXY / HTTPS_PROXY / ALL_PROXY set, httpx would try to
+            # route local traffic through the proxy, which typically cannot
+            # reach localhost or LAN addresses.
+            _local_limits = httpx.Limits(keepalive_expiry=0)
+            http_client = httpx.AsyncClient(
+                limits=_local_limits,
+                timeout=timeout_s,
+                transport=httpx.AsyncHTTPTransport(proxy=None, limits=_local_limits),
+            )
+        # else: http_client stays None → SDK creates DefaultAsyncHttpxClient
+        # which already reads proxy env vars via trust_env=True, has proper
+        # connection limits, and follows redirects.
+        self._client = AsyncOpenAI(
+            api_key=self._api_key_for_client,
+            base_url=self._effective_base,
+            default_headers=self._default_headers,
+            default_query=self._extra_query or None,
+            max_retries=0,
+            timeout=timeout_s,
+            http_client=http_client,
+        )
+
+    async def _ensure_client(self):
+        """Return the shared OpenAI client, creating it on first call."""
+        if self._client is not None:
+            return self._client
+        async with self._client_lock:
+            if self._client is not None:
+                return self._client
+            global AsyncOpenAI
+            if AsyncOpenAI is None:
+                if os.environ.get("LANGFUSE_SECRET_KEY") and importlib.util.find_spec("langfuse"):
+                    from langfuse.openai import AsyncOpenAI as _AsyncOpenAI
+                else:
+                    if os.environ.get("LANGFUSE_SECRET_KEY"):
+                        logger.warning(
+                            "LANGFUSE_SECRET_KEY is set but langfuse is not installed; "
+                            "install with `pip install langfuse` to enable tracing"
+                        )
+                    from openai import AsyncOpenAI as _AsyncOpenAI
+                AsyncOpenAI = _AsyncOpenAI
+
+            self._build_client()
+            return self._client
+
+    def _setup_env(self, api_key: str, api_base: str | None) -> None:
+        """Set environment variables based on provider spec."""
+        spec = self._spec
+        if not spec or not spec.env_key:
+            return
+        if spec.is_gateway:
+            os.environ[spec.env_key] = api_key
+        else:
+            os.environ.setdefault(spec.env_key, api_key)
+        effective_base = api_base or spec.default_api_base
+        for env_name, env_val in spec.env_extras:
+            resolved = env_val.replace("{api_key}", api_key).replace("{api_base}", effective_base)
+            os.environ.setdefault(env_name, resolved)
+
+    @classmethod
+    def _apply_cache_control(
+        cls,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None,
+    ) -> tuple[list[dict[str, Any]], list[dict[str, Any]] | None]:
+        """Inject cache_control markers for prompt caching."""
+        cache_marker = {"type": "ephemeral"}
+        new_messages = list(messages)
+
+        def _mark(msg: dict[str, Any]) -> dict[str, Any]:
+            content = msg.get("content")
+            if isinstance(content, str):
+                return {**msg, "content": [
+                    {"type": "text", "text": content, "cache_control": cache_marker},
+                ]}
+            if isinstance(content, list) and content:
+                nc = list(content)
+                nc[-1] = {**nc[-1], "cache_control": cache_marker}
+                return {**msg, "content": nc}
+            return msg
+
+        if new_messages and new_messages[0].get("role") == "system":
+            new_messages[0] = _mark(new_messages[0])
+        if len(new_messages) >= 3:
+            new_messages[-2] = _mark(new_messages[-2])
+
+        new_tools = tools
+        if tools:
+            new_tools = list(tools)
+            for idx in cls._tool_cache_marker_indices(new_tools):
+                new_tools[idx] = {**new_tools[idx], "cache_control": cache_marker}
+        return new_messages, new_tools
+
+    @staticmethod
+    def _normalize_tool_call_id(tool_call_id: Any) -> Any:
+        """Normalize to a provider-safe 9-char alphanumeric form."""
+        if not isinstance(tool_call_id, str):
+            return tool_call_id
+        if len(tool_call_id) == 9 and tool_call_id.isalnum():
+            return tool_call_id
+        return hashlib.sha1(tool_call_id.encode()).hexdigest()[:9]
+
+    def _should_normalize_tool_call_ids(self) -> bool:
+        """Return True for providers that reject normal OpenAI tool call IDs."""
+        return bool(self._spec and self._spec.name == "mistral")
+
+    @staticmethod
+    def _coerce_content_to_string(content: Any) -> str | None:
+        """Coerce block/list content into plain text for strict string-only APIs."""
+        if content is None or isinstance(content, str):
+            return content
+        text = OpenAICompatProvider._extract_text_content(content)
+        if isinstance(text, str) and text:
+            return text
+        try:
+            dumped = json.dumps(content, ensure_ascii=False)
+        except Exception:
+            dumped = str(content)
+        return dumped or "(empty)"
+
+    def _sanitize_messages(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """Strip non-standard keys, normalize tool_call IDs."""
+        sanitized = LLMProvider._sanitize_request_messages(messages, _ALLOWED_MSG_KEYS)
+        id_map: dict[str, str] = {}
+        pending_tool_ids: dict[str, deque[str]] = {}
+        force_string_content = bool(self._spec and self._spec.name == "deepseek")
+        normalize_tool_ids = self._should_normalize_tool_call_ids()
+        strip_reasoning = bool(
+            self._spec
+            and getattr(self._spec, "strip_history_reasoning_content", False)
+        )
+        if strip_reasoning:
+            for msg in sanitized:
+                msg.pop("reasoning_content", None)
+
+        def map_id(value: Any) -> Any:
+            if not isinstance(value, str):
+                return value
+            if not normalize_tool_ids:
+                return value
+            return id_map.setdefault(value, self._normalize_tool_call_id(value))
+
+        def unique_tool_id(value: Any, used_ids: set[str], idx: int) -> str:
+            if isinstance(value, str) and value:
+                base = map_id(value)
+            else:
+                base = _short_tool_id()
+            if not isinstance(base, str) or not base:
+                base = _short_tool_id()
+            if base not in used_ids:
+                return base
+            seed = value if isinstance(value, str) and value else base
+            salt = 1
+            while True:
+                candidate = self._normalize_tool_call_id(f"{seed}:{idx}:{salt}")
+                if isinstance(candidate, str) and candidate not in used_ids:
+                    return candidate
+                salt += 1
+
+        def map_tool_result_id(value: Any) -> Any:
+            if not isinstance(value, str):
+                return value
+            queue = pending_tool_ids.get(value)
+            if queue:
+                mapped = queue.popleft()
+                if not queue:
+                    pending_tool_ids.pop(value, None)
+                return mapped
+            return map_id(value)
+
+        for clean in sanitized:
+            if isinstance(clean.get("tool_calls"), list):
+                normalized = []
+                used_ids: set[str] = set()
+                for idx, tc in enumerate(clean["tool_calls"]):
+                    if not isinstance(tc, dict):
+                        normalized.append(tc)
+                        continue
+                    tc_clean = dict(tc)
+                    raw_id = tc_clean.get("id")
+                    mapped_id = unique_tool_id(raw_id, used_ids, idx)
+                    tc_clean["id"] = mapped_id
+                    used_ids.add(mapped_id)
+                    if isinstance(raw_id, str) and raw_id:
+                        pending_tool_ids.setdefault(raw_id, deque()).append(mapped_id)
+                    function = tc_clean.get("function")
+                    if isinstance(function, dict):
+                        function_clean = dict(function)
+                        if "arguments" in function_clean:
+                            function_clean["arguments"] = tool_arguments_json_for_replay(
+                                function_clean.get("arguments")
+                            )
+                        else:
+                            function_clean["arguments"] = "{}"
+                        tc_clean["function"] = function_clean
+                    normalized.append(tc_clean)
+                clean["tool_calls"] = normalized
+                if clean.get("role") == "assistant":
+                    # Some OpenAI-compatible gateways reject assistant messages
+                    # that mix non-empty content with tool_calls.
+                    clean["content"] = None
+            if "tool_call_id" in clean and clean["tool_call_id"]:
+                clean["tool_call_id"] = map_tool_result_id(clean["tool_call_id"])
+            if (
+                force_string_content
+                and not (clean.get("role") == "assistant" and clean.get("tool_calls"))
+            ):
+                clean["content"] = self._coerce_content_to_string(clean.get("content"))
+        return self._enforce_role_alternation(sanitized)
+
+    # ------------------------------------------------------------------
+    # Build kwargs
+    # ------------------------------------------------------------------
+
+    def _request_model_name(self, model_name: str) -> str:
+        spec = self._spec
+        if not spec or "/" not in model_name:
+            return model_name
+        if spec.strip_model_prefix:
+            return model_name.split("/")[-1]
+
+        route_prefixes = getattr(spec, "strip_model_prefixes", ())
+        if not isinstance(route_prefixes, tuple) or not route_prefixes:
+            return model_name
+        model_prefix, routed_model = model_name.split("/", 1)
+        model_prefix_key = _provider_prefix_key(model_prefix)
+        if any(_provider_prefix_key(prefix) == model_prefix_key for prefix in route_prefixes):
+            return routed_model
+        return model_name
+
+    @staticmethod
+    def _supports_temperature(
+        model_name: str,
+        reasoning_effort: str | None = None,
+    ) -> bool:
+        """Return True when the model accepts a temperature parameter.
+
+        Kimi K3 uses a fixed temperature that should be omitted. GPT-5 family
+        and reasoning models (o1/o3/o4) reject temperature when
+        reasoning_effort is set to anything other than ``"none"``.
+        """
+        if _model_slug(model_name) == _KIMI_K3_MODEL:
+            return False
+        if reasoning_effort and reasoning_effort.lower() != "none":
+            return False
+        name = model_name.lower()
+        return not any(token in name for token in ("gpt-5", "o1", "o3", "o4"))
+
+    def _build_kwargs(
+        self,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None,
+        model: str | None,
+        max_tokens: int,
+        temperature: float,
+        reasoning_effort: str | None,
+        tool_choice: str | dict[str, Any] | None,
+    ) -> dict[str, Any]:
+        model_name = model or self.default_model
+        spec = self._spec
+
+        if spec and spec.supports_prompt_caching:
+            model_name = model or self.default_model
+            if any(model_name.lower().startswith(k) for k in ("anthropic/", "claude")):
+                messages, tools = self._apply_cache_control(messages, tools)
+
+        model_name = self._request_model_name(model_name)
+
+        kwargs: dict[str, Any] = {
+            "model": model_name,
+            "messages": self._sanitize_messages(self._sanitize_empty_content(messages)),
+        }
+
+        # GPT-5 and reasoning models (o1/o3/o4) reject temperature when
+        # reasoning_effort is active.  Only include it when safe.
+        if self._supports_temperature(model_name, reasoning_effort):
+            kwargs["temperature"] = temperature
+
+        if (
+            spec and getattr(spec, "supports_max_completion_tokens", False)
+        ) or _requires_max_completion_tokens(model_name):
+            kwargs["max_completion_tokens"] = max(1, max_tokens)
+        else:
+            kwargs["max_tokens"] = max(1, max_tokens)
+
+        if spec:
+            model_lower = model_name.lower()
+            for pattern, overrides in spec.model_overrides:
+                if pattern in model_lower:
+                    kwargs.update(overrides)
+                    break
+
+        # Moonshot selects the only valid temperature from the K2.5/K2.6 thinking mode:
+        # 1.0 when enabled and 0.6 when disabled. Omitting the parameter lets the API
+        # apply the matching value for both its default and explicit thinking controls.
+        if (
+            spec
+            and spec.name == "moonshot"
+            and _model_slug(model_name) in _KIMI_SERVER_MANAGED_TEMPERATURE_MODELS
+        ):
+            kwargs.pop("temperature", None)
+
+        # Normalize reasoning_effort into a semantic form (OpenAI vocab)
+        # used for internal decisions, and a wire form actually sent out.
+        # "minimum" is accepted as a DashScope-native alias for "minimal".
+        semantic_effort: str | None = None
+        if isinstance(reasoning_effort, str):
+            semantic_effort = reasoning_effort.lower()
+            if semantic_effort == "minimum":
+                semantic_effort = "minimal"
+
+        wire_effort = reasoning_effort
+        slug = _model_slug(model_name)
+        if slug == _KIMI_K3_MODEL and semantic_effort is not None:
+            # K3 always reasons and currently accepts only the top-level
+            # reasoning_effort="max". Preserve disabled/default semantics by
+            # omitting the field; normalize older enabled presets to "max" so
+            # switching from a K2.x model does not send an unsupported value.
+            if semantic_effort in ("none", "minimal"):
+                wire_effort = None
+            else:
+                semantic_effort = "max"
+                wire_effort = "max"
+        if spec and spec.name == "dashscope" and semantic_effort == "minimal":
+            # DashScope accepts none/minimum/low/medium/high/xhigh; "minimal" 400s.
+            wire_effort = "minimum"
+
+        # Magistral and other providers where reasoning is implicit reject the
+        # reasoning_effort kwarg entirely. Strip it before the remap so we don't
+        # accidentally send "none"/"high" to a model that always reasons.
+        strip_effort = False
+        if spec and getattr(spec, "implicit_reasoning_models", ()):
+            model_lower = model_name.lower()
+            strip_effort = any(
+                pat in model_lower for pat in spec.implicit_reasoning_models
+            )
+
+        # Some providers accept a constrained reasoning_effort vocabulary
+        # (Mistral: only "high"/"none"). Remap from OpenAI vocab to the
+        # provider's accepted set; an empty mapped value means "omit".
+        if (
+            not strip_effort
+            and spec
+            and getattr(spec, "reasoning_effort_remap", ())
+            and isinstance(semantic_effort, str)
+        ):
+            remap = dict(spec.reasoning_effort_remap)
+            mapped = remap.get(semantic_effort)
+            if mapped is not None:
+                wire_effort = mapped or None
+                semantic_effort = mapped or "none"
+
+        if strip_effort:
+            wire_effort = None
+        elif wire_effort and semantic_effort != "none":
+            kwargs["reasoning_effort"] = wire_effort
+
+        # Only send thinking controls when reasoning_effort is explicit so
+        # omitting the config preserves each provider's default.
+        if reasoning_effort is not None:
+            thinking_enabled = semantic_effort not in ("none", "minimal")
+            for thinking_style in _thinking_styles_for(spec, model_name):
+                if not thinking_enabled and slug in _KIMI_ALWAYS_THINKING_MODELS:
+                    continue
+                extra = _thinking_extra_body(thinking_style, thinking_enabled)
+                if extra:
+                    kwargs.setdefault("extra_body", {}).update(extra)
+            gateway_style = getattr(spec, "gateway_reasoning_style", "") if spec else ""
+            if (
+                gateway_style
+                and _model_thinking_style(model_name)
+                and (thinking_enabled or slug not in _KIMI_ALWAYS_THINKING_MODELS)
+            ):
+                extra = _gateway_reasoning_extra_body(gateway_style, semantic_effort)
+                if extra:
+                    kwargs.setdefault("extra_body", {}).update(extra)
+
+            # Moonshot rejects requests that carry both 'reasoning_effort'
+            # and the native 'thinking' param.  We already expressed the
+            # user's intent via the provider-native shape, so drop the
+            # redundant wire-level kwarg.  Only kimi models need this —
+            # Xiaomi's API accepts both params.
+            if slug in _KIMI_THINKING_MODELS:
+                kwargs.pop("reasoning_effort", None)
+
+        if tools:
+            kwargs["tools"] = tools
+            kwargs["tool_choice"] = tool_choice or "auto"
+
+        # Backfill reasoning_content="" on assistants missing it: DeepSeek
+        # thinking mode rejects history otherwise (#3554, #3584); "" reads
+        # as "no thinking that turn". DeepSeek-V4/reasoner reason natively,
+        # so backfill even without explicit reasoning_effort.
+        explicit_thinking = (
+            reasoning_effort is not None
+            and semantic_effort not in ("none", "minimal")
+            and (
+                (spec and spec.thinking_style)
+                or _model_thinking_style(model_name)
+            )
+        )
+        implicit_deepseek_thinking = (
+            spec is not None
+            and spec.name == "deepseek"
+            and semantic_effort not in ("none", "minimal", "minimum")
+            and any(t in model_name.lower() for t in ("deepseek-v4", "deepseek-reasoner"))
+        )
+        if explicit_thinking or implicit_deepseek_thinking:
+            for msg in kwargs["messages"]:
+                if msg.get("role") == "assistant" and "reasoning_content" not in msg:
+                    msg["reasoning_content"] = ""
+
+        # Merge user-configured extra_body last so it can override or
+        # extend provider-specific defaults (e.g. chat_template_kwargs,
+        # guided_json, repetition_penalty).  Uses recursive merge so
+        # nested dicts like {"chat_template_kwargs": {"enable_thinking": false}}
+        # do not clobber sibling keys already set by thinking-style logic.
+        if self._extra_body:
+            existing = kwargs.get("extra_body", {})
+            kwargs["extra_body"] = _deep_merge(existing, self._extra_body)
+
+        return kwargs
+
+    def _should_use_responses_api(
+        self,
+        model: str | None,
+        reasoning_effort: str | None,
+    ) -> bool:
+        """Use Responses API only for direct OpenAI requests that benefit from it."""
+        if self._api_type == "chat_completions":
+            return False
+        if self._spec and self._spec.name not in ("openai", "github_copilot"):
+            return False
+        if self._api_type == "responses":
+            # Explicit configuration means Responses is mandatory; do not
+            # consult the circuit breaker or fall back to Chat Completions.
+            return True
+        if self._spec is None or self._spec.name != "github_copilot":
+            if not _is_direct_openai_base(self._effective_base):
+                return False
+
+        model_name = (model or self.default_model).lower()
+        wants = False
+        if reasoning_effort and reasoning_effort.lower() != "none":
+            wants = True
+        elif any(token in model_name for token in ("gpt-5", "o1", "o3", "o4")):
+            wants = True
+        if not wants:
+            return False
+
+        return self._responses_circuit_allows_probe(model, reasoning_effort)
+
+    def _responses_circuit_allows_probe(
+        self,
+        model: str | None,
+        reasoning_effort: str | None,
+    ) -> bool:
+        """Return False when the Responses API circuit breaker is open."""
+        key = _responses_circuit_key(model, self.default_model, reasoning_effort)
+        failures = self._responses_failures.get(key, 0)
+        if failures >= _RESPONSES_FAILURE_THRESHOLD:
+            tripped = self._responses_tripped_at.get(key, 0.0)
+            if (time.monotonic() - tripped) < _RESPONSES_PROBE_INTERVAL_S:
+                return False
+            # Half-open: allow one probe attempt
+        return True
+
+    def _record_responses_failure(self, model: str | None, reasoning_effort: str | None) -> None:
+        key = _responses_circuit_key(model, self.default_model, reasoning_effort)
+        count = self._responses_failures.get(key, 0) + 1
+        self._responses_failures[key] = count
+        if count >= _RESPONSES_FAILURE_THRESHOLD:
+            self._responses_tripped_at[key] = time.monotonic()
+            logger.warning(
+                "Responses API circuit open for {} — falling back to Chat Completions",
+                key,
+            )
+
+    def _record_responses_success(self, model: str | None, reasoning_effort: str | None) -> None:
+        key = _responses_circuit_key(model, self.default_model, reasoning_effort)
+        self._responses_failures.pop(key, None)
+        self._responses_tripped_at.pop(key, None)
+
+    @staticmethod
+    def _should_fallback_from_responses_error(e: Exception) -> bool:
+        """Fallback only for likely Responses API compatibility errors."""
+        response = getattr(e, "response", None)
+        status_code = getattr(e, "status_code", None)
+        if status_code is None and response is not None:
+            status_code = getattr(response, "status_code", None)
+        if status_code not in {400, 404, 422}:
+            return False
+
+        body = (
+            getattr(e, "body", None)
+            or getattr(e, "doc", None)
+            or getattr(response, "text", None)
+        )
+        body_text = str(body).lower() if body is not None else ""
+        compatibility_markers = (
+            "responses",
+            "response api",
+            "max_output_tokens",
+            "instructions",
+            "previous_response",
+            "unsupported",
+            "not supported",
+            "unknown parameter",
+            "unrecognized request argument",
+        )
+        return any(marker in body_text for marker in compatibility_markers)
+
+    def _build_responses_body(
+        self,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None,
+        model: str | None,
+        max_tokens: int,
+        temperature: float,
+        reasoning_effort: str | None,
+        tool_choice: str | dict[str, Any] | None,
+    ) -> dict[str, Any]:
+        """Build a Responses API body for direct OpenAI requests."""
+        model_name = model or self.default_model
+        model_name = self._request_model_name(model_name)
+        sanitized_messages = self._sanitize_messages(self._sanitize_empty_content(messages))
+        instructions, input_items = convert_messages(sanitized_messages)
+
+        body: dict[str, Any] = {
+            "model": model_name,
+            "instructions": instructions or None,
+            "input": input_items,
+            "max_output_tokens": max(1, max_tokens),
+            "store": False,
+            "stream": False,
+        }
+
+        if self._supports_temperature(model_name, reasoning_effort):
+            body["temperature"] = temperature
+
+        if reasoning_effort and reasoning_effort.lower() != "none":
+            body["reasoning"] = {"effort": reasoning_effort}
+            body["include"] = ["reasoning.encrypted_content"]
+
+        if tools:
+            body["tools"] = convert_tools(tools)
+            body["tool_choice"] = tool_choice or "auto"
+
+        extra_body = getattr(self, "_extra_body", {})
+        if extra_body:
+            body = _merge_responses_extra_body(body, extra_body)
+
+        return body
+
+    # ------------------------------------------------------------------
+    # Response parsing
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _maybe_mapping(value: Any) -> dict[str, Any] | None:
+        if isinstance(value, dict):
+            return value
+        model_dump = getattr(value, "model_dump", None)
+        if callable(model_dump):
+            dumped = model_dump()
+            if isinstance(dumped, dict):
+                return dumped
+        return None
+
+    @classmethod
+    def _extract_text_content(cls, value: Any) -> str | None:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return value
+        if isinstance(value, list):
+            parts: list[str] = []
+            for item in value:
+                item_map = cls._maybe_mapping(item)
+                if item_map:
+                    # Skip Mistral-style {"type":"thinking","thinking":[...]}
+                    # blocks: their text belongs in reasoning_content.
+                    if item_map.get("type") == "thinking":
+                        continue
+                    text = item_map.get("text")
+                    if isinstance(text, str):
+                        parts.append(text)
+                        continue
+                text = getattr(item, "text", None)
+                if isinstance(text, str):
+                    parts.append(text)
+                    continue
+                if isinstance(item, str):
+                    parts.append(item)
+            return "".join(parts) or None
+        return str(value)
+
+    @classmethod
+    def _extract_thinking_content(cls, value: Any) -> str | None:
+        """Extract reasoning text from Mistral-style thinking blocks.
+
+        Mistral returns content as a list mixing
+        ``{"type":"thinking","thinking":[{"type":"text","text":...}]}`` and
+        ``{"type":"text","text":...}``. The thinking text belongs in
+        ``reasoning_content`` so the agent can surface it as a reasoning
+        trace rather than as the assistant's reply.
+        """
+        if not isinstance(value, list):
+            return None
+        parts: list[str] = []
+        for item in value:
+            item_map = cls._maybe_mapping(item)
+            if not item_map:
+                continue
+            if item_map.get("type") != "thinking":
+                continue
+            inner = item_map.get("thinking")
+            text = cls._extract_text_content(inner)
+            if text:
+                parts.append(text)
+        return "".join(parts) or None
+
+    @classmethod
+    def _extract_usage(cls, response: Any) -> dict[str, int]:
+        """Extract token usage from an OpenAI-compatible response.
+
+        Handles both dict-based (raw JSON) and object-based (SDK Pydantic)
+        responses.  Provider-specific ``cached_tokens`` fields are normalised
+        under a single key; see the priority chain inside for details.
+        """
+        # --- resolve usage object ---
+        usage_obj = None
+        response_map = cls._maybe_mapping(response)
+        if response_map is not None:
+            usage_obj = response_map.get("usage")
+        elif hasattr(response, "usage") and response.usage:
+            usage_obj = response.usage
+
+        usage_map = cls._maybe_mapping(usage_obj)
+        if usage_map is not None:
+            result = {
+                "prompt_tokens": int(usage_map.get("prompt_tokens") or 0),
+                "completion_tokens": int(usage_map.get("completion_tokens") or 0),
+                "total_tokens": int(usage_map.get("total_tokens") or 0),
+            }
+        elif usage_obj:
+            result = {
+                "prompt_tokens": getattr(usage_obj, "prompt_tokens", 0) or 0,
+                "completion_tokens": getattr(usage_obj, "completion_tokens", 0) or 0,
+                "total_tokens": getattr(usage_obj, "total_tokens", 0) or 0,
+            }
+        else:
+            return {}
+
+        # --- cached_tokens (normalised across providers) ---
+        # Try nested paths first (dict), fall back to attribute (SDK object).
+        # Priority order ensures the most specific field wins.
+        for path in (
+            ("prompt_tokens_details", "cached_tokens"),  # OpenAI/Zhipu/MiniMax/Qwen/Mistral/xAI
+            ("cached_tokens",),                          # StepFun/Moonshot (top-level)
+            ("prompt_cache_hit_tokens",),                # DeepSeek/SiliconFlow
+        ):
+            cached = cls._get_nested_int(usage_map, path)
+            if not cached and usage_obj:
+                cached = cls._get_nested_int(usage_obj, path)
+            if cached:
+                result["cached_tokens"] = cached
+                break
+
+        return result
+
+    @staticmethod
+    def _get_nested_int(obj: Any, path: tuple[str, ...]) -> int:
+        """Drill into *obj* by *path* segments and return an ``int`` value.
+
+        Supports both dict-key access and attribute access so it works
+        uniformly with raw JSON dicts **and** SDK Pydantic models.
+        """
+        current = obj
+        for segment in path:
+            if current is None:
+                return 0
+            if isinstance(current, dict):
+                current = current.get(segment)
+            else:
+                current = getattr(current, segment, None)
+        return int(current or 0) if current is not None else 0
+
+    def _parse(self, response: Any) -> LLMResponse:
+        if isinstance(response, str):
+            return LLMResponse(content=response, finish_reason="stop")
+
+        response_map = self._maybe_mapping(response)
+        if response_map is not None:
+            choices = response_map.get("choices") or []
+            if not choices:
+                content = self._extract_text_content(
+                    response_map.get("content") or response_map.get("output_text")
+                )
+                reasoning_content = self._extract_text_content(
+                    response_map.get("reasoning_content")
+                )
+                if content is not None:
+                    return LLMResponse(
+                        content=content,
+                        reasoning_content=reasoning_content,
+                        finish_reason=str(response_map.get("finish_reason") or "stop"),
+                        usage=self._extract_usage(response_map),
+                    )
+                return LLMResponse(
+                    content="Error: API returned empty choices.",
+                    finish_reason="error",
+                    error_kind="empty",
+                )
+
+            choice0 = self._maybe_mapping(choices[0]) or {}
+            msg0 = self._maybe_mapping(choice0.get("message")) or {}
+            content = self._extract_text_content(msg0.get("content"))
+            finish_reason = str(choice0.get("finish_reason") or "stop")
+
+            raw_tool_calls: list[Any] = []
+            # StepFun: fallback to reasoning field when content is empty
+            if not content and msg0.get("reasoning") and self._spec and self._spec.reasoning_as_content:
+                content = self._extract_text_content(msg0.get("reasoning"))
+            reasoning_content = msg0.get("reasoning_content")
+            if reasoning_content is None and msg0.get("reasoning"):
+                reasoning_content = self._extract_text_content(msg0.get("reasoning"))
+            # Mistral reasoning models return thinking text inside the content
+            # array; lift it into reasoning_content so the runner records it
+            # under the reasoning trace.
+            spec = getattr(self, "_spec", None)
+            if reasoning_content is None and getattr(spec, "extract_thinking_blocks", False):
+                reasoning_content = self._extract_thinking_content(msg0.get("content"))
+            for ch in choices:
+                ch_map = self._maybe_mapping(ch) or {}
+                m = self._maybe_mapping(ch_map.get("message")) or {}
+                tool_calls = m.get("tool_calls")
+                if isinstance(tool_calls, list) and tool_calls:
+                    raw_tool_calls.extend(tool_calls)
+                    if ch_map.get("finish_reason") in ("tool_calls", "stop"):
+                        finish_reason = str(ch_map["finish_reason"])
+                if not content:
+                    content = self._extract_text_content(m.get("content"))
+                if reasoning_content is None:
+                    reasoning_content = m.get("reasoning_content")
+
+            # Deduplicate tool call IDs (same pattern as streaming path)
+            # Some providers reuse the same ID for parallel tool calls.
+            _seen_tc_ids: set[str] = set()
+            parsed_tool_calls = []
+            for tc in raw_tool_calls:
+                tc_map = self._maybe_mapping(tc) or {}
+                fn = self._maybe_mapping(tc_map.get("function")) or {}
+                args = parse_tool_arguments(fn.get("arguments", {}))
+                ec, prov, fn_prov = _extract_tc_extras(tc)
+                raw_id = str(tc_map.get("id") or _short_tool_id())
+                if not raw_id or raw_id in _seen_tc_ids:
+                    raw_id = _short_tool_id()
+                _seen_tc_ids.add(raw_id)
+                parsed_tool_calls.append(ToolCallRequest(
+                    id=raw_id,
+                    name=str(fn.get("name") or ""),
+                    arguments=args,
+                    extra_content=ec,
+                    provider_specific_fields=prov,
+                    function_provider_specific_fields=fn_prov,
+                ))
+            if not parsed_tool_calls:
+                content, parsed_tool_calls = _extract_text_tool_calls(content)
+
+            return LLMResponse(
+                content=content,
+                tool_calls=parsed_tool_calls,
+                finish_reason=finish_reason,
+                usage=self._extract_usage(response_map),
+                reasoning_content=reasoning_content if isinstance(reasoning_content, str) else None,
+            )
+
+        if not response.choices:
+            return LLMResponse(
+                content="Error: API returned empty choices.",
+                finish_reason="error",
+                error_kind="empty",
+            )
+
+        choice = response.choices[0]
+        msg = choice.message
+        content = msg.content
+        finish_reason = choice.finish_reason
+
+        raw_tool_calls: list[Any] = []
+        for ch in response.choices:
+            m = ch.message
+            if hasattr(m, "tool_calls") and m.tool_calls:
+                raw_tool_calls.extend(m.tool_calls)
+                if ch.finish_reason in ("tool_calls", "stop"):
+                    finish_reason = ch.finish_reason
+            if not content and m.content:
+                content = m.content
+            if not content and getattr(m, "reasoning", None) and self._spec and self._spec.reasoning_as_content:
+                content = m.reasoning
+
+        tool_calls = []
+        for tc in raw_tool_calls:
+            args = parse_tool_arguments(tc.function.arguments)
+            ec, prov, fn_prov = _extract_tc_extras(tc)
+            tool_calls.append(ToolCallRequest(
+                id=str(getattr(tc, "id", None) or _short_tool_id()),
+                name=tc.function.name,
+                arguments=args,
+                extra_content=ec,
+                provider_specific_fields=prov,
+                function_provider_specific_fields=fn_prov,
+            ))
+        if not tool_calls:
+            content, tool_calls = _extract_text_tool_calls(content)
+
+        reasoning_content = getattr(msg, "reasoning_content", None)
+        if reasoning_content is None and getattr(msg, "reasoning", None):
+            reasoning_content = msg.reasoning
+
+        return LLMResponse(
+            content=content,
+            tool_calls=tool_calls,
+            finish_reason=finish_reason or "stop",
+            usage=self._extract_usage(response),
+            reasoning_content=reasoning_content,
+        )
+
+    @classmethod
+    def _parse_chunks(cls, chunks: list[Any]) -> LLMResponse:
+        content_parts: list[str] = []
+        reasoning_parts: list[str] = []
+        tc_bufs: dict[int, dict[str, Any]] = {}
+        finish_reason = "stop"
+        usage: dict[str, int] = {}
+
+        def _accum_tc(tc: Any, idx_hint: int) -> None:
+            """Accumulate one streaming tool-call delta into *tc_bufs*."""
+            tc_index: int = _get(tc, "index") if _get(tc, "index") is not None else idx_hint
+            buf = tc_bufs.setdefault(tc_index, {
+                "id": "", "name": "", "arguments": "",
+                "extra_content": None, "prov": None, "fn_prov": None,
+            })
+            tc_id = _get(tc, "id")
+            if tc_id:
+                buf["id"] = str(tc_id)
+            fn = _get(tc, "function")
+            if fn is not None:
+                fn_name = _get(fn, "name")
+                if fn_name:
+                    buf["name"] = str(fn_name)
+                fn_args = _get(fn, "arguments")
+                if fn_args:
+                    buf["arguments"] += str(fn_args)
+            ec, prov, fn_prov = _extract_tc_extras(tc)
+            if ec:
+                buf["extra_content"] = ec
+            if prov:
+                buf["prov"] = prov
+            if fn_prov:
+                buf["fn_prov"] = fn_prov
+
+        def _accum_legacy_function_call(function_call: Any) -> None:
+            """Accumulate legacy ``delta.function_call`` streaming chunks."""
+            if not function_call:
+                return
+            buf = tc_bufs.setdefault(0, {
+                "id": "", "name": "", "arguments": "",
+                "extra_content": None, "prov": None, "fn_prov": None,
+            })
+            fn_name = _get(function_call, "name")
+            if fn_name:
+                buf["name"] = str(fn_name)
+            fn_args = _get(function_call, "arguments")
+            if fn_args:
+                buf["arguments"] += str(fn_args)
+
+        for chunk in chunks:
+            if isinstance(chunk, str):
+                content_parts.append(chunk)
+                continue
+
+            chunk_map = cls._maybe_mapping(chunk)
+            if chunk_map is not None:
+                choices = chunk_map.get("choices") or []
+                if not choices:
+                    usage = cls._extract_usage(chunk_map) or usage
+                    text = cls._extract_text_content(
+                        chunk_map.get("content") or chunk_map.get("output_text")
+                    )
+                    if text:
+                        content_parts.append(text)
+                    continue
+                choice = cls._maybe_mapping(choices[0]) or {}
+                if choice.get("finish_reason"):
+                    finish_reason = str(choice["finish_reason"])
+                delta = cls._maybe_mapping(choice.get("delta")) or {}
+                raw_delta_content = delta.get("content")
+                text = cls._extract_text_content(raw_delta_content)
+                if text:
+                    content_parts.append(text)
+                text = cls._extract_text_content(delta.get("reasoning_content"))
+                if not text:
+                    text = cls._extract_text_content(delta.get("reasoning"))
+                if not text:
+                    # Mistral streams thinking inside the content array as
+                    # {"type":"thinking", thinking:[{"type":"text", ...}]}.
+                    text = cls._extract_thinking_content(raw_delta_content)
+                if text:
+                    reasoning_parts.append(text)
+                for idx, tc in enumerate(delta.get("tool_calls") or []):
+                    _accum_tc(tc, idx)
+                _accum_legacy_function_call(delta.get("function_call"))
+                usage = cls._extract_usage(chunk_map) or usage
+                continue
+
+            if not chunk.choices:
+                usage = cls._extract_usage(chunk) or usage
+                continue
+            choice = chunk.choices[0]
+            if choice.finish_reason:
+                finish_reason = choice.finish_reason
+            delta = choice.delta
+            if delta and delta.content:
+                text = cls._extract_text_content(delta.content)
+                if text:
+                    content_parts.append(text)
+                thinking_text = cls._extract_thinking_content(delta.content)
+                if thinking_text:
+                    reasoning_parts.append(thinking_text)
+            if delta:
+                reasoning = getattr(delta, "reasoning_content", None)
+                if not reasoning:
+                    reasoning = getattr(delta, "reasoning", None)
+                if reasoning:
+                    text = cls._extract_text_content(reasoning)
+                    if text:
+                        reasoning_parts.append(text)
+            for tc in (getattr(delta, "tool_calls", None) or []) if delta else []:
+                _accum_tc(tc, getattr(tc, "index", 0))
+            if delta:
+                _accum_legacy_function_call(getattr(delta, "function_call", None))
+
+        # Some providers (e.g. Zhipu/GLM) reuse the same tool_call id for
+        # parallel tool calls in streaming mode. Deduplicate before building
+        # the response so downstream tool messages don't collide.
+        _seen_tc_ids: set[str] = set()
+        for b in tc_bufs.values():
+            if not b["id"] or b["id"] in _seen_tc_ids:
+                b["id"] = _short_tool_id()
+            _seen_tc_ids.add(b["id"])
+
+        content = "".join(content_parts) or None
+        tool_calls = [
+            ToolCallRequest(
+                id=b["id"] or _short_tool_id(),
+                name=b["name"],
+                arguments=parse_tool_arguments(b["arguments"]),
+                extra_content=b.get("extra_content"),
+                provider_specific_fields=b.get("prov"),
+                function_provider_specific_fields=b.get("fn_prov"),
+            )
+            for b in tc_bufs.values()
+        ]
+        if not tool_calls:
+            content, tool_calls = _extract_text_tool_calls(content)
+
+        return LLMResponse(
+            content=content,
+            tool_calls=tool_calls,
+            finish_reason=finish_reason,
+            usage=usage,
+            reasoning_content="".join(reasoning_parts) or None,
+        )
+
+    @classmethod
+    def _extract_error_metadata(cls, e: Exception) -> dict[str, Any]:
+        response = getattr(e, "response", None)
+        headers = getattr(response, "headers", None)
+        payload = (
+            getattr(e, "body", None)
+            or getattr(e, "doc", None)
+            or getattr(response, "text", None)
+        )
+        if payload is None and response is not None:
+            response_json = getattr(response, "json", None)
+            if callable(response_json):
+                try:
+                    payload = response_json()
+                except Exception:
+                    payload = None
+        error_type, error_code = LLMProvider._extract_error_type_code(payload)
+
+        status_code = getattr(e, "status_code", None)
+        if status_code is None and response is not None:
+            status_code = getattr(response, "status_code", None)
+
+        should_retry: bool | None = None
+        if headers is not None:
+            raw = headers.get("x-should-retry")
+            if isinstance(raw, str):
+                lowered = raw.strip().lower()
+                if lowered == "true":
+                    should_retry = True
+                elif lowered == "false":
+                    should_retry = False
+
+        error_kind: str | None = None
+        error_name = e.__class__.__name__.lower()
+        if "timeout" in error_name:
+            error_kind = "timeout"
+        elif "connection" in error_name:
+            error_kind = "connection"
+
+        return {
+            "error_status_code": int(status_code) if status_code is not None else None,
+            "error_kind": error_kind,
+            "error_type": error_type,
+            "error_code": error_code,
+            "error_retry_after_s": cls._extract_retry_after_from_headers(headers),
+            "error_should_retry": should_retry,
+        }
+
+    @staticmethod
+    def _handle_error(
+        e: Exception,
+        *,
+        spec: ProviderSpec | None = None,
+        api_base: str | None = None,
+    ) -> LLMResponse:
+        body = (
+            getattr(e, "doc", None)
+            or getattr(e, "body", None)
+            or getattr(getattr(e, "response", None), "text", None)
+        )
+        body_text = body if isinstance(body, str) else str(body) if body is not None else ""
+        msg = f"Error: {body_text.strip()[:500]}" if body_text.strip() else f"Error calling LLM: {e}"
+
+        text = f"{body_text} {e}".lower()
+        if spec and spec.is_local and ("502" in text or "connection" in text or "refused" in text):
+            msg += (
+                "\nHint: this is a local model endpoint. Check that the local server is reachable at "
+                f"{api_base or spec.default_api_base}, and if you are using a proxy/tunnel, make sure it "
+                "can reach your local Ollama/vLLM service instead of routing localhost through the remote host."
+            )
+
+        response = getattr(e, "response", None)
+        retry_after = LLMProvider._extract_retry_after_from_headers(getattr(response, "headers", None))
+        if retry_after is None:
+            retry_after = LLMProvider._extract_retry_after(msg)
+        return LLMResponse(
+            content=msg,
+            finish_reason="error",
+            retry_after=retry_after,
+            **OpenAICompatProvider._extract_error_metadata(e),
+        )
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    async def chat(
+        self,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None = None,
+        model: str | None = None,
+        max_tokens: int = 4096,
+        temperature: float = 0.7,
+        reasoning_effort: str | None = None,
+        tool_choice: str | dict[str, Any] | None = None,
+    ) -> LLMResponse:
+        await self._ensure_client()
+        try:
+            if self._should_use_responses_api(model, reasoning_effort):
+                try:
+                    body = self._build_responses_body(
+                        messages, tools, model, max_tokens, temperature,
+                        reasoning_effort, tool_choice,
+                    )
+                    result = parse_response_output(await self._client.responses.create(**body))
+                    self._record_responses_success(model, reasoning_effort)
+                    return result
+                except Exception as responses_error:
+                    if self._spec and self._spec.name == "github_copilot":
+                        # Copilot gateway exposes GPT-5/o-series only via /responses;
+                        # falling back to /chat/completions cannot succeed and would
+                        # hide the real error.
+                        raise
+                    if self._api_type == "responses":
+                        raise
+                    if not self._should_fallback_from_responses_error(responses_error):
+                        raise
+                    self._record_responses_failure(model, reasoning_effort)
+
+            kwargs = self._build_kwargs(
+                messages, tools, model, max_tokens, temperature,
+                reasoning_effort, tool_choice,
+            )
+            return self._parse(await self._client.chat.completions.create(**kwargs))
+        except Exception as e:
+            return self._handle_error(e, spec=self._spec, api_base=self.api_base)
+
+    async def chat_stream(
+        self,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None = None,
+        model: str | None = None,
+        max_tokens: int = 4096,
+        temperature: float = 0.7,
+        reasoning_effort: str | None = None,
+        tool_choice: str | dict[str, Any] | None = None,
+        on_content_delta: Callable[[str], Awaitable[None]] | None = None,
+        on_thinking_delta: Callable[[str], Awaitable[None]] | None = None,
+        on_tool_call_delta: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
+    ) -> LLMResponse:
+        await self._ensure_client()
+        idle_timeout_s = resolve_stream_idle_timeout_s()
+        try:
+            if self._should_use_responses_api(model, reasoning_effort):
+                try:
+                    body = self._build_responses_body(
+                        messages, tools, model, max_tokens, temperature,
+                        reasoning_effort, tool_choice,
+                    )
+                    body["stream"] = True
+                    stream = await self._client.responses.create(**body)
+
+                    async def _timed_stream():
+                        stream_iter = stream.__aiter__()
+                        while True:
+                            try:
+                                yield await asyncio.wait_for(
+                                    stream_iter.__anext__(),
+                                    timeout=idle_timeout_s,
+                                )
+                            except StopAsyncIteration:
+                                break
+
+                    (
+                        content,
+                        tool_calls,
+                        finish_reason,
+                        usage,
+                        reasoning_content,
+                    ) = await consume_sdk_stream(
+                        _timed_stream(),
+                        on_content_delta,
+                        on_tool_call_delta=on_tool_call_delta,
+                    )
+                    self._record_responses_success(model, reasoning_effort)
+                    return LLMResponse(
+                        content=content or None,
+                        tool_calls=tool_calls,
+                        finish_reason=finish_reason,
+                        usage=usage,
+                        reasoning_content=reasoning_content,
+                    )
+                except Exception as responses_error:
+                    if self._spec and self._spec.name == "github_copilot":
+                        # Copilot gateway exposes GPT-5/o-series only via /responses;
+                        # falling back to /chat/completions cannot succeed and would
+                        # hide the real error.
+                        raise
+                    if self._api_type == "responses":
+                        raise
+                    if not self._should_fallback_from_responses_error(responses_error):
+                        raise
+                    self._record_responses_failure(model, reasoning_effort)
+
+            kwargs = self._build_kwargs(
+                messages, tools, model, max_tokens, temperature,
+                reasoning_effort, tool_choice,
+            )
+            if self._spec and self._spec.name == "zhipu" and tools and on_tool_call_delta:
+                # Z.AI/GLM keeps streaming tool-call arguments behind an
+                # explicit provider flag.  Pass it through the OpenAI SDK's
+                # extra_body escape hatch so the usual delta.tool_calls path
+                # can surface live file-edit progress.
+                kwargs.setdefault("extra_body", {})["tool_stream"] = True
+            kwargs["stream"] = True
+            kwargs["stream_options"] = {"include_usage": True}
+            stream = await self._client.chat.completions.create(**kwargs)
+            chunks: list[Any] = []
+            stream_iter = stream.__aiter__()
+            while True:
+                try:
+                    chunk = await asyncio.wait_for(
+                        stream_iter.__anext__(),
+                        timeout=idle_timeout_s,
+                    )
+                except StopAsyncIteration:
+                    break
+                chunks.append(chunk)
+                if chunk.choices:
+                    delta_obj = chunk.choices[0].delta
+                    raw_delta_content = getattr(delta_obj, "content", None)
+                    if on_content_delta:
+                        # Mistral streams content as a list of {"type":"thinking",
+                        # ...} + {"type":"text",...} blocks. Extract just the
+                        # text portion before invoking the callback so callers
+                        # never see non-string content.
+                        text = self._extract_text_content(raw_delta_content)
+                        if text:
+                            await on_content_delta(text)
+                    if on_thinking_delta:
+                        reasoning = getattr(delta_obj, "reasoning_content", None) or getattr(
+                            delta_obj, "reasoning", None,
+                        )
+                        r_text = self._extract_text_content(reasoning)
+                        if not r_text:
+                            # Mistral keeps the thinking trace inside the
+                            # content array rather than a separate field.
+                            r_text = self._extract_thinking_content(raw_delta_content)
+                        if r_text:
+                            await on_thinking_delta(r_text)
+                    if on_tool_call_delta:
+                        for idx, tool_delta in enumerate(
+                            getattr(delta_obj, "tool_calls", None) or []
+                        ):
+                            fn = _get(tool_delta, "function")
+                            tool_index = _get(tool_delta, "index")
+                            await on_tool_call_delta({
+                                "index": tool_index if tool_index is not None else idx,
+                                "call_id": str(_get(tool_delta, "id") or ""),
+                                "name": str(_get(fn, "name") or "") if fn is not None else "",
+                                "arguments_delta": (
+                                    str(_get(fn, "arguments") or "") if fn is not None else ""
+                                ),
+                            })
+                        function_call = getattr(delta_obj, "function_call", None)
+                        if function_call:
+                            await on_tool_call_delta({
+                                "index": 0,
+                                "call_id": "",
+                                "name": str(_get(function_call, "name") or ""),
+                                "arguments_delta": str(_get(function_call, "arguments") or ""),
+                            })
+            return self._parse_chunks(chunks)
+        except asyncio.TimeoutError:
+            return LLMResponse(
+                content=(
+                    f"Error calling LLM: stream stalled for more than "
+                    f"{idle_timeout_s:g} seconds"
+                ),
+                finish_reason="error",
+                error_kind="timeout",
+            )
+        except Exception as e:
+            return self._handle_error(e, spec=self._spec, api_base=self.api_base)
+
+    def get_default_model(self) -> str:
+        return self.default_model