providers.py

"""
Multi-provider support for Dulus.

Supported providers:
  anthropic  — Claude (claude-opus-4-6, claude-sonnet-4-6, ...)
  openai     — GPT (gpt-4o, o3-mini, ...)
  gemini     — Google Gemini (gemini-2.0-flash, gemini-1.5-pro, ...)
  kimi       — Moonshot AI (kimi-k2.5, moonshot-v1-8k/32k/128k)
  kimi-code  — Kimi Code (kimi-for-coding, membership API from kimi.com/code)
               Supports fallback chain: KIMI_CODE_API_KEY → KIMI_CODE2_API_KEY → KIMI_CODE3_API_KEY
  qwen       — Alibaba DashScope (qwen-max, qwen-plus, ...)
  modelstudio — Alibaba Cloud Model Studio, Singapore workspace (OpenAI-compatible)
  amd        — AMD Developer Cloud / ROCm vLLM server (OpenAI-compatible)
  zhipu      — Zhipu GLM (glm-4, glm-4-plus, ...)
  deepseek   — DeepSeek (deepseek-chat, deepseek-reasoner, ...)
  minimax    — MiniMax (MiniMax-Text-01, abab6.5s-chat, ...)
  ollama     — Local Ollama (llama3.3, qwen2.5-coder, ...)
  lmstudio   — Local LM Studio (any loaded model)
  custom     — Any OpenAI-compatible endpoint

Model string formats:
  "claude-opus-4-6"          auto-detected → anthropic
  "gpt-4o"                   auto-detected → openai
  "ollama/qwen2.5-coder"     explicit provider prefix
  "custom/my-model"          uses CUSTOM_BASE_URL from config
"""
from __future__ import annotations
import json
import urllib.request
import urllib.parse
import urllib.error
import requests
import re
import time
import random
import functools
import subprocess
import platform
from typing import Generator, Any, Callable


# ── TLS: trust the OS certificate store ──────────────────────────────────
# Cloudflare WARP / Cloudflare One does TLS inspection with its own root CA,
# which is installed in the OS trust store (Windows cert store / macOS Keychain
# / Linux ca-certificates) but is NOT in Python's bundled `certifi`. Without
# this, every HTTPS call fails with SSLCertVerificationError while WARP is on.
# `truststore` makes Python's ssl module use the OS store, so the WARP CA is
# trusted automatically — and normal browsing (WARP off) keeps working too.
# This covers httpx (OpenAI + Anthropic SDKs), requests, and urllib at once.
#
# Optional manual override: set DULUS_CA_BUNDLE (or SSL_CERT_FILE) to a PEM
# file with the Cloudflare cert if you prefer pinning it explicitly.
import os as _os_tls
_ca_bundle = _os_tls.environ.get("DULUS_CA_BUNDLE") or _os_tls.environ.get("SSL_CERT_FILE")
if _ca_bundle and _os_tls.path.exists(_ca_bundle):
    # Point the common HTTP stacks at the explicit bundle.
    _os_tls.environ.setdefault("SSL_CERT_FILE", _ca_bundle)
    _os_tls.environ.setdefault("REQUESTS_CA_BUNDLE", _ca_bundle)
else:
    try:
        import truststore as _truststore
        _truststore.inject_into_ssl()  # use OS trust store everywhere
    except Exception:
        # truststore not installed or injection failed — fall back to certifi.
        # Install with:  pip install truststore   (Python 3.10+)
        pass


# ── Provider resilience: retry with exponential backoff + jitter ─────────

class _ProviderRetry:
    """Lightweight retry wrapper for provider streaming calls.

    Retries on: timeout, connection errors, 429 (rate limit), 5xx.
    Does NOT retry on: 4xx (client errors), auth failures.

    MAX_RETRIES = 0 by design (Kev, 2026-09-03): automatic silent retries made
    models "freeze out of nowhere" — the terminal stalled in a hidden sleep
    while the user had no clue a transient error happened. We now surface the
    error immediately and stop; the user re-sends or switches model. Bump this
    back up ONLY if you also keep the retry visible (it now yields a warning).
    """
    MAX_RETRIES: int = 0
    BASE_DELAY: float = 0.5
    MAX_DELAY: float = 5.0

    @classmethod
    def is_retryable(cls, exc: Exception) -> bool:
        """Return True if the exception is worth retrying."""
        msg = str(exc).lower()
        # Quota exhaustion (NVIDIA "ResourceExhausted", Gemini "quota exceeded")
        # will not recover within a retry window — surface it immediately.
        if "resourceexhausted" in msg or "resource_exhausted" in msg or "resource exhausted" in msg:
            return False
        if "quota" in msg or "request limit reached" in msg:
            return False
        # Rate limit / server overload
        if "429" in msg or "rate limit" in msg or "too many requests" in msg:
            return True
        # Server errors
        if "500" in msg or "502" in msg or "503" in msg or "504" in msg:
            return True
        # Timeouts / connection issues
        if "timeout" in msg or "connection" in msg or "timed out" in msg:
            return True
        if "chunked encoding" in msg or "broken pipe" in msg:
            return True
        return False

    @classmethod
    def sleep_for_attempt(cls, attempt: int) -> float:
        """Exponential backoff with full jitter."""
        exp = cls.BASE_DELAY * (2 ** attempt)
        jitter = random.random() * exp
        return min(jitter, cls.MAX_DELAY)

    @classmethod
    def wrap_generator(cls, fn: Callable, *args, **kwargs) -> Generator:
        """Wrap a generator function with retry logic.

        If the request fails BEFORE yielding anything, waits and retries up to
        MAX_RETRIES times. Once a stream has emitted output it is unsafe to
        retry (would duplicate partial output), so a failure after the first
        chunk always propagates.
        """
        last_exc: Exception | None = None
        for attempt in range(cls.MAX_RETRIES + 1):
            emitted = False
            try:
                for chunk in fn(*args, **kwargs):
                    emitted = True
                    yield chunk
                return
            except Exception as exc:
                last_exc = exc
                if (
                    emitted
                    or attempt >= cls.MAX_RETRIES
                    or not cls.is_retryable(exc)
                ):
                    raise
                delay = cls.sleep_for_attempt(attempt)
                # Announce the retry instead of freezing silently. A bare
                # time.sleep() here is exactly what makes the terminal look
                # "frozen out of nowhere" — the user has no idea a transient
                # error (timeout / 429 / 5xx) happened and we're waiting.
                try:
                    _reason = friendly_api_error(exc)
                except Exception:
                    _reason = f"{type(exc).__name__}: {str(exc)[:120]}"
                yield TextChunk(
                    f"\n⚠️  Provider hiccup ({_reason}). "
                    f"Retrying in {delay:.1f}s "
                    f"(attempt {attempt + 2}/{cls.MAX_RETRIES + 1})…\n"
                )
                time.sleep(delay)
        # Should never reach here, but just in case
        if last_exc:
            raise last_exc


_ANTHROPIC_PARAM_RE = re.compile(
    r'<parameter\s+name="(?P<key>[^"]+)"\s*>(?P<val>.*?)</parameter>',
    re.DOTALL,
)
_TOOLCALL_NAME_RE = re.compile(r'"name"\s*:\s*"(?P<name>[^"]+)"')


def _parse_tool_call_payload(payload: str):
    """Best-effort extraction of (tool_name, input_dict) from the body of a
    `<tool_call>...</tool_call>` block.

    Why this exists: models sometimes leak Anthropic's `<function_calls>` /
    `<parameter>` syntax INSIDE our `<tool_call>` block, which corrupts the
    JSON. Without a recovery path the call is silently dropped by
    `try: json.loads(...); except: pass` and the user sees their request
    vanish into thin air. We try three strategies, easiest first.

    Returns (name, input_dict) or None.
    """
    payload = (payload or "").strip()
    if not payload:
        return None

    # Strategy 1 — clean JSON ("name" + "input"). Vast majority of calls.
    try:
        data = json.loads(payload)
        if isinstance(data, dict):
            name = data.get("name") or (
                data.get("function", {}).get("name")
                if isinstance(data.get("function"), dict) else None
            )
            if name:
                inp = data.get("input") or (
                    data.get("function", {}).get("arguments")
                    if isinstance(data.get("function"), dict) else None
                ) or {}
                if isinstance(inp, str):
                    try:
                        inp = json.loads(inp)
                    except Exception:
                        inp = {}
                return name, (inp if isinstance(inp, dict) else {})
    except Exception:
        pass

    # Strategy 2 — Anthropic-syntax leak: `{"name": "X">  <parameter ...>` etc.
    # Pull the tool name from the leading JSON-ish prefix and the params from
    # all `<parameter name="K">V</parameter>` blocks. Coerce numeric/bool
    # strings into proper JSON types where obvious.
    if "<parameter" in payload and "</parameter>" in payload:
        m = _TOOLCALL_NAME_RE.search(payload)
        if m:
            name = m.group("name")
            inp: dict = {}
            for pm in _ANTHROPIC_PARAM_RE.finditer(payload):
                key = pm.group("key")
                val = pm.group("val")
                # Coerce common scalar forms; leave everything else as string.
                sval = val.strip()
                if sval.lower() == "true":
                    inp[key] = True
                elif sval.lower() == "false":
                    inp[key] = False
                elif sval.lower() == "null":
                    inp[key] = None
                else:
                    try:
                        if sval.isdigit() or (sval.startswith("-") and sval[1:].isdigit()):
                            inp[key] = int(sval)
                        elif sval.replace(".", "", 1).lstrip("-").isdigit():
                            inp[key] = float(sval)
                        elif sval.startswith("[") or sval.startswith("{"):
                            inp[key] = json.loads(sval)
                        else:
                            inp[key] = val
                    except Exception:
                        inp[key] = val
            if inp:
                return name, inp

    # Strategy 3 — last-ditch: maybe the JSON is just unterminated. Try to
    # find balanced braces from the first `{` and parse that.
    first = payload.find("{")
    if first != -1:
        depth = 0
        end = -1
        in_str = False
        esc = False
        for i in range(first, len(payload)):
            c = payload[i]
            if esc:
                esc = False; continue
            if c == "\\":
                esc = True; continue
            if c == '"':
                in_str = not in_str; continue
            if in_str:
                continue
            if c == "{":
                depth += 1
            elif c == "}":
                depth -= 1
                if depth == 0:
                    end = i + 1
                    break
        if end != -1:
            try:
                data = json.loads(payload[first:end])
                if isinstance(data, dict):
                    name = data.get("name")
                    if name:
                        return name, (data.get("input") or {})
            except Exception:
                pass

    return None


def _decode_webchat_entities(text: str) -> str:
    """Decode the HTML entities a chat UI (e.g. Claude.ai webchat) may apply to a
    `<tool_call>` block when it renders the assistant turn, so the parser still
    finds the tags + JSON. Only entities relevant to tag/JSON detection are
    decoded; everything else is left intact. (`&amp;` is decoded last to handle
    single-level double-encoding like `&amp;lt;` → `&lt;`.)
    """
    if "&" not in text:
        return text
    return (text.replace("&lt;", "<").replace("&gt;", ">")
                .replace("&quot;", '"').replace("&#34;", '"')
                .replace("&#39;", "'").replace("&#x27;", "'")
                .replace("&amp;", "&"))


class WebToolParser:
    """Shared parser for prompt-based tool calls in XML format.
    Also supports auto-wrapping raw JSON tool calls if auto_wrap_json=True.
    """
    def __init__(self, auto_wrap_json: bool = False):
        self._in_call = False
        self._call_buf = ""
        self._raw_buf = ""
        self._auto_wrap_json = auto_wrap_json
        self.tool_calls = []

    def parse_chunk(self, chunk: str) -> str:
        """Parse chunk, return display text and accumulate tool calls."""
        if not chunk: return ""
        self._raw_buf += chunk
        # Claude.ai (and other chat UIs) may HTML-encode our <tool_call> tags or
        # wrap them in ``` fences when rendering. Decode the relevant entities so
        # the tags + JSON stay detectable (calls inside ``` fences parse as-is).
        if "&" in self._raw_buf:
            self._raw_buf = _decode_webchat_entities(self._raw_buf)
        display = ""

        while True:
            if not self._in_call:
                # Look for start tag
                pos = self._raw_buf.find("<tool_call>")
                if pos == -1:
                    # No start tag. Check for partial start tag at the very end
                    last_lt = self._raw_buf.rfind("<")
                    if last_lt != -1 and "<tool_call>".startswith(self._raw_buf[last_lt:]):
                        display += self._raw_buf[:last_lt]
                        self._raw_buf = self._raw_buf[last_lt:]
                    else:
                        display += self._raw_buf
                        self._raw_buf = ""
                    break
                else:
                    # Found start tag: everything before is text
                    display += self._raw_buf[:pos]
                    self._in_call = True
                    self._raw_buf = self._raw_buf[pos + len("<tool_call>"):]
                    continue # Look for end tag in the rest of buffer
            else:
                # Inside a tag: look for end tag
                pos = self._raw_buf.find("</tool_call>")
                if pos == -1:
                    # End tag not found yet. BUT: if we already buffered a full
                    # Anthropic-style `<function_calls>` block leak (i.e. JSON
                    # opener + at least one `</parameter>`), the model is never
                    # going to close with `</tool_call>` — recover now instead
                    # of waiting forever and dumping as text on flush().
                    self._call_buf += self._raw_buf
                    self._raw_buf = ""
                    if "</parameter>" in self._call_buf and "<parameter" in self._call_buf:
                        parsed = _parse_tool_call_payload(self._call_buf.strip())
                        if parsed:
                            name, inp = parsed
                            self.tool_calls.append({
                                "id": f"call_pt_{len(self.tool_calls)}",
                                "name": name,
                                "input": inp,
                            })
                            self._call_buf = ""
                            self._in_call = False
                            continue
                    break
                else:
                    # Found end tag: extract JSON and continue
                    self._call_buf += self._raw_buf[:pos]
                    self._raw_buf = self._raw_buf[pos + len("</tool_call>"):]
                    parsed = _parse_tool_call_payload(self._call_buf.strip())
                    if parsed:
                        name, inp = parsed
                        self.tool_calls.append({
                            "id": f"call_pt_{len(self.tool_calls)}",
                            "name": name,
                            "input": inp,
                        })
                    self._call_buf = ""
                    self._in_call = False
                    continue # Look for more tags in the rest of buffer

        # 2. Raw JSON Fallback (only if enabled and NOT inside a tag)
        if self._auto_wrap_json and not self._in_call and "{" in display:
            search_pos = 0
            while True:
                start = display.find("{", search_pos)
                if start == -1: break
                
                snippet = display[start:start+500]
                if '"name"' in snippet and ('"input"' in snippet or '"arguments"' in snippet):
                    brace_count = 0
                    end_pos = -1
                    for j in range(start, len(display)):
                        if display[j] == "{": brace_count += 1
                        elif display[j] == "}":
                            brace_count -= 1
                            if brace_count == 0:
                                end_pos = j + 1
                                break
                    if end_pos != -1:
                        try:
                            json_str = display[start:end_pos]
                            data = json.loads(json_str)
                            name = data.get("name") or (data.get("function", {}).get("name") if isinstance(data.get("function"), dict) else None)
                            if name:
                                self.tool_calls.append({
                                    "id": f"call_pt_{len(self.tool_calls)}",
                                    "name": name,
                                    "input": data.get("input") or data.get("function", {}).get("arguments") or {},
                                })
                                display = display[:start] + display[end_pos:]
                                search_pos = start
                                continue
                        except: pass
                search_pos = start + 1
        
        return display

    def flush(self) -> str:
        """Return any remaining text in the buffer."""
        res = self._raw_buf
        self._raw_buf = ""
        # If we were in a call but it never ended, we should probably output the partial call?
        # But for now, just the raw text.
        if self._in_call:
            res = "<tool_call>" + self._call_buf + res
            self._call_buf = ""
            self._in_call = False
        return res


def _format_web_tool_manifest(tool_schemas: list, config: dict, messages: list) -> str:
    """Format tools as a prompt hint for web models.
    First turn → full manifest with strong instructions + tool list.
    Continuation turns → short format reminder (always injected, cheap).
    Disable entirely with config["no_tools"] = True.
    """
    if not tool_schemas or config.get("no_tools"):
        return ""

    is_first_turn = len([m for m in messages if m.get("role") == "user"]) <= 1

    # Web providers (claude.ai, qwen.ai, etc.) keep the conversation server-side,
    # so the turn-1 manifest is still in the model's context on every later turn.
    # Re-injecting wastes tokens. Skip unless the user explicitly opted in.
    if not is_first_turn and not config.get("always_inject_tools"):
        return ""

    manifest = [
        "\n\n[TOOL USE — READ CAREFULLY]",
        "You are running inside an agent harness that can EXECUTE tools for you.",
        "When you need information, file contents, or to run an action — DO NOT describe what you would do; CALL the tool.",
        "",
        "EXACT format (any deviation = the call is ignored):",
        '  <tool_call>{"name": "ToolName", "input": {"key": "value"}}</tool_call>',
        "",
        "Rules:",
        "1. The <tool_call> tag MUST be on its own line, with valid JSON inside.",
        "2. Use ONLY tool names from the list below. Do NOT invent tools (no `SleepTimer`, no `WaitFor`, no fake names — `Reminder` is the real one if you need a delayed wake-up).",
        "3. To call multiple tools, emit multiple <tool_call> blocks in the SAME response — do not wait for results between them.",
        "4. After tool results come back, you may call more tools or give a final answer.",
        "5. If no tool is needed, just answer normally — no tool_call tag.",
        "",
        "Example (correct):",
        '  <tool_call>{"name": "Read", "input": {"file_path": "/tmp/foo.txt"}}</tool_call>',
        "",
        "Available Tools:",
    ]
    for s in tool_schemas:
        manifest.append(f"- {s['name']}: {s.get('description', '')}")
        manifest.append(f"  Inputs: {json.dumps(s.get('parameters', {}).get('properties', {}), separators=(',', ':'))}")

    return "\n".join(manifest)


def _consolidate_web_history(messages: list, manifest: str = "") -> str:
    """Consolidate history since last assistant turn into one prompt string.
    This ensures tool results and system notifications are correctly perceived
    by web-based models that take a single prompt string.
    """
    if not messages:
        return manifest
    
    # Find last assistant message that actually has text or was saved
    last_ast = -1
    for i in range(len(messages) - 1, -1, -1):
        if messages[i].get("role") == "assistant":
            last_ast = i
            break
    
    parts = []
    relevant = messages[last_ast + 1:] if last_ast != -1 else messages
    
    for m in relevant:
        role = m.get("role", "user")
        content = m.get("content", "")
        
        # We only skip empty content if it's NOT a tool result.
        # Tool results must be sent even if empty so the model knows they ran.
        if role != "tool" and not content:
            continue
            
        header = f"--- [{role.upper()}] ---"
        if role == "tool":
            header = f"--- [Tool Result: {m.get('name', 'Unknown')}] ---"
            if not content:
                content = "(No output / Empty result)"
        
        parts.append(f"{header}\n{content}")
    
    prompt = "\n\n".join(parts).strip()
    if manifest:
        prompt = manifest + "\n\n" + prompt
        
    return prompt.strip()

# ── Provider registry ──────────────────────────────────────────────────────

PROVIDERS: dict[str, dict] = {
    "anthropic": {
        "type":       "anthropic",
        "api_key_env": "ANTHROPIC_API_KEY",
        "context_limit": 200000,
        "models": [
            "claude-opus-4-6", "claude-sonnet-4-6", "claude-haiku-4-5-20251001",
            "claude-opus-4-5", "claude-sonnet-4-5",
            "claude-3-5-sonnet-20241022", "claude-3-5-haiku-20241022",
        ],
    },
    "openai": {
        "type":       "openai",
        "api_key_env": "OPENAI_API_KEY",
        "base_url":   "https://api.openai.com/v1",
        "context_limit": 128000,
        "max_completion_tokens": 16384,  # safe cap across gpt-4o/gpt-4.1 family
        "models": [
            "gpt-4o", "gpt-4o-mini", "gpt-4-turbo", "gpt-4.1", "gpt-4.1-mini",
            "o3-mini", "o1", "o1-mini",
        ],
    },
    "gemini": {
        "type":       "openai",
        "api_key_env": "GEMINI_API_KEY",
        "base_url":   "https://generativelanguage.googleapis.com/v1beta/openai/",
        "context_limit": 1000000,
        "max_completion_tokens": 65536,  # Gemini 2.x supports up to 65k output tokens
        "models": [
            "gemini-2.5-pro-preview-03-25",
            "gemini-2.0-flash", "gemini-2.0-flash-lite",
            "gemini-1.5-pro", "gemini-1.5-flash",
        ],
    },
    "kimi": {
        "type":       "openai",
        "api_key_env": "MOONSHOT_API_KEY",
        "base_url":   "https://api.moonshot.ai/v1",
        "context_limit": 250000,
        "models": [
            "kimi-k2.5", "kimi-latest",
            "moonshot-v1-8k", "moonshot-v1-32k", "moonshot-v1-128k",
        ],
    },
    "kimi-code": {
        "type":       "openai",
        "api_key_env": "KIMI_CODE_API_KEY",
        "base_url":   "https://api.kimi.com/coding/v1",
        "context_limit": 1_000_000,
        # Official Kimi Code model IDs (kimi.com/code/docs → Model Configuration):
        #   k3 (up to 1M), k3-256k, kimi-for-coding (K2.7 Code, 256k).
        "models": [
            "k3", "k3-256k", "kimi-for-coding",
        ],
    },
    # Kimi membership OAuth (`/login kimi`) — same api.kimi.com/coding/v1 endpoint
    # as kimi-code, but bills against the user's Kimi membership (no API key).
    "kimi-oauth": {
        "type":       "kimi-oauth",
        "api_key_env": None,
        "base_url":   "https://api.kimi.com/coding/v1",
        "context_limit": 1_000_000,
        "models": [
            "k3", "k3-256k", "kimi-for-coding",
        ],
    },
    "qwen": {
        "type":       "openai",
        "api_key_env": "DASHSCOPE_API_KEY",
        "base_url":   "https://dashscope.aliyuncs.com/compatible-mode/v1",
        "context_limit": 1000000,
        "models": [
            "qwen-max", "qwen-plus", "qwen-turbo", "qwen-long",
            "qwen2.5-72b-instruct", "qwen2.5-coder-32b-instruct",
            "qwq-32b",
        ],
    },
    # Alibaba Cloud Model Studio (Singapore / ap-southeast-1). OpenAI-compatible
    # endpoint scoped to a workspace. The base_url embeds your Workspace ID:
    #   https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1
    # Override the workspace without editing code via env/config:
    #   MODELSTUDIO_WORKSPACE_ID=ws-xxxx   (or MODELSTUDIO_BASE_URL=<full url>)
    #   MODELSTUDIO_API_KEY=<key>          (falls back to DASHSCOPE_API_KEY)
    # Pick models as 'modelstudio/<model>' e.g. 'modelstudio/qwen-max'.
    "modelstudio": {
        "type":       "openai",
        "api_key_env": "MODELSTUDIO_API_KEY",
        "base_url":   "https://ws-1qcqvxk37njsah79.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1",
        "context_limit": 1000000,
        # qwen rejects max_tokens > 65536 (400 invalid_parameter). Cap output so a
        # large config max_tokens (e.g. 1,000,000) doesn't blow up every call.
        "max_completion_tokens": 32768,
        # Ordered fallback chain (strongest → cheaper) of the free-quota LLM
        # models in the Singapore plan. Override per-account with
        #   /config modelstudio_fallback_chain=<id>,<id>,...
        # A wrong/retired id just falls through to the next one automatically.
        "models": [
            "qwen3-max", "qwen3.7-max", "qwen3.7-plus", "qwen3.6-plus",
            "qwen3.5-plus-2026-04-20", "qwen3.6-flash", "qwen3.5-flash",
            "qwen3.5-122b-a10b", "deepseek-v4-pro", "deepseek-v3.2", "glm-5.1",
        ],
    },
    # AMD Developer Cloud / ROCm. Serve any open model with vLLM (or TGI) on an
    # AMD GPU instance — both expose an OpenAI-compatible /v1 endpoint. Point
    # Dulus at it without editing code:
    #   AMD_BASE_URL=http://<instance-ip>:8000/v1   (your vLLM server)
    #   AMD_API_KEY=<token>   (optional; vLLM accepts any value by default)
    # Pick models as 'amd/<served-model-name>' e.g. 'amd/Qwen2.5-72B-Instruct'.
    # The served name must match what you launched vLLM with (--served-model-name).
    "amd": {
        "type":       "openai",
        "api_key_env": "AMD_API_KEY",
        "base_url":   "",  # resolved at call time from AMD_BASE_URL / config
        "context_limit": 131072,
        "models": [
            "Qwen2.5-72B-Instruct", "Qwen2.5-Coder-32B-Instruct",
            "Llama-3.3-70B-Instruct", "Mixtral-8x7B-Instruct",
        ],
    },
    "zhipu": {
        "type":       "openai",
        "api_key_env": "ZHIPU_API_KEY",
        "base_url":   "https://api.z.ai/api/coding/paas/v4",
        "context_limit": 128000,
        "models": [
            "glm-4-plus", "glm-4", "glm-4-flash", "glm-4-air",
            "glm-z1-flash", "GLM-4.7", "GLM-4.5-AIR",
        ],
    },
    "dulus": {
        # Fuel-backed models on the Dulus control plane. Sign in with
        # `/login dulus` (OAuth PKCE — no API key) or mint a dulus_sk_* key
        # with `/login dulus key` for CI/servers. The control plane maps the
        # dulus-* tiers to upstream models and meters Fuel per token; the
        # client never sees the upstream model name and holds no provider key.
        "type":       "dulus-oauth",
        "api_key_env": None,
        "base_url":   "https://control.dulus.ai/v1",
        "context_limit": 262144,
        # 8k was too small: with thinking:max the model burns the whole
        # completion budget on reasoning and the stream ends on
        # finish_reason="length" with no answer and no tool call.
        "max_completion_tokens": 32768,
        "models": [
            "dulus-a-9b", "dulus-b-27b", "dulus-x-397b",
            "dulus-mistral-7b", "dulus-mistral-nemo", "dulus-gpt-oss-20b",
            "dulus-mistral-small-24b", "dulus-qwen-32b", "dulus-coder-30b",
            "dulus-llama-70b", "dulus-vl-72b", "dulus-gpt-oss-120b",
        ],
    },
    "deepseek": {
        "type":       "openai",
        "api_key_env": "DEEPSEEK_API_KEY",
        "base_url":   "https://api.deepseek.com/v1",
        "context_limit": 64000,
        "models": [
            "deepseek-chat", "deepseek-coder", "deepseek-reasoner",
            "deepseek-v3", "deepseek-r1",
        ],
    },
    # Azure OpenAI (v1 OpenAI-compatible endpoint). Deployment name == model
    # name. The endpoint is per-resource, so set it via your environment:
    #   AZURE_OPENAI_ENDPOINT=https://<your-resource>.cognitiveservices.azure.com
    #   AZURE_OPENAI_KEY=<your-key>     (or: /config azure_api_key=...)
    # Pick models as 'azure/<deployment-name>'. Also serves Kimi/other
    # deployments hosted on Azure AI Foundry.
    "azure": {
        "type":       "openai",
        "api_key_env": "AZURE_OPENAI_KEY",
        "base_url":   "",  # resolved at call time from AZURE_OPENAI_ENDPOINT / config
        "context_limit": 128000,
        "max_completion_tokens": 16384,
        "models": [
            "gpt-4.1-nano", "gpt-4.1-mini", "gpt-4.1", "gpt-4o", "gpt-4o-mini",
        ],
    },
    # LiteLLM unified gateway. ONE provider entry that fans out to 100+
    # underlying backends via prefixed model strings:
    #     openrouter/anthropic/claude-3-5-sonnet
    #     groq/llama-3.3-70b-versatile
    #     together_ai/meta-llama/Llama-3-70b-chat-hf
    #     bedrock/anthropic.claude-3-sonnet-20240229-v1:0
    #     vertex_ai/gemini-1.5-pro
    #     cohere/command-r-plus
    #     perplexity/sonar-large-online
    #     xai/grok-2-latest
    #     mistral/mistral-large-latest
    #     fireworks_ai/...   anyscale/...   replicate/...   azure/...
    # LiteLLM auto-reads per-backend env vars (OPENROUTER_API_KEY,
    # GROQ_API_KEY, TOGETHER_API_KEY, …). User picks the model string in
    # the welcome wizard; the right env var must exist for that backend.
    "litellm": {
        "type":       "litellm",
        "api_key_env": None,   # backend-specific; LiteLLM resolves the right one
        "context_limit": 200000,   # safe default; LiteLLM has accurate per-model values
        "models": [
            # A curated, useful slice — LiteLLM supports ~1000 model strings.
            # The user can type any of them; these are just suggestions for /model picker.
            "openrouter/anthropic/claude-3-5-sonnet",
            "openrouter/openai/gpt-4o",
            "openrouter/google/gemini-pro-1.5",
            "openrouter/meta-llama/llama-3.3-70b-instruct",
            "openrouter/x-ai/grok-2-1212",
            "groq/llama-3.3-70b-versatile",
            "groq/mixtral-8x7b-32768",
            "together_ai/meta-llama/Llama-3-70b-chat-hf",
            "perplexity/sonar-large-online",
            "cohere/command-r-plus",
            "mistral/mistral-large-latest",
            "fireworks_ai/accounts/fireworks/models/llama-v3p3-70b-instruct",
        ],
    },
    "minimax": {
        "type":       "openai",
        "api_key_env": "MINIMAX_API_KEY",
        "base_url":   "https://api.minimaxi.chat/v1",
        "context_limit": 1000000,
        "models": [
            "MiniMax-Text-01", "MiniMax-VL-01",
            "abab6.5s-chat", "abab6.5-chat",
            "abab5.5s-chat", "abab5.5-chat",
        ],
    },
    "ollama": {
        "type":       "ollama",
        "api_key_env": None,
        "base_url":   "http://localhost:11434",
        "api_key":    "ollama",
        "context_limit": 250000,
        "models": [
            "gemma4", "qwen3", "qwen2.5-coder", "llama3.3",
            "deepseek-r1", "mistral",
        ],
    },
    "lmstudio": {
        "type":       "openai",
        "api_key_env": None,
        "base_url":   "http://localhost:1234/v1",
        "api_key":    "lm-studio",
        "context_limit": 128000,
        "models": [],   # dynamic, depends on loaded model
    },
    "edge": {
        # Local / on-device small models over an OpenAI-compatible endpoint. The
        # provider is backend-agnostic — ANY server that speaks OpenAI on
        # 127.0.0.1 works. Three ways to feed it, easiest first:
        #   • llama.cpp:  llama-server -m gemma-3-1b-it-Q4_K_M.gguf --port 8080
        #                 (works today, runs anywhere: desktop, VM, Termux/phone)
        #   • Ollama:     ollama serve
        #                 (then /config edge_base_url=http://127.0.0.1:11434/v1)
        #   • Dulus Edge Bridge APK — on-device Gemma on Android via AICore
        #     (NPU-accelerated, zero-download). WORK IN PROGRESS: the Android
        #     build takes time; llama.cpp/Ollama is the path that ships now.
        # The requested model name is just a label — llama-server serves whatever
        # weights you loaded with -m. Override host/port (Termux, a phone on your
        # LAN) with DULUS_EDGE_BASE_URL or `/config edge_base_url=http://HOST:PORT/v1`.
        "type":        "openai",
        "api_key_env": None,
        "base_url":    "http://127.0.0.1:8080/v1",
        "api_key":     "dulus-edge",
        "context_limit": 32000,
        "models": [
            # llama.cpp / Ollama GGUFs — the path that works today.
            "gemma-3n-e4b", "gemma-3n-e2b",
            # Dulus Edge Bridge APK (on-device Android) — WIP.
            "gemma-4-e4b", "gemma-4-e2b", "gemini-nano",
        ],
    },
    "nvidia-web": {
        "type":       "openai",
        "api_key_env": "NVIDIA_API_KEY",
        "base_url":   "https://integrate.api.nvidia.com/v1",
        "context_limit": 128000,
        "max_completion_tokens": 16384,
        "models": [
            "deepseek-ai/deepseek-v4-flash",
            "deepseek-ai/deepseek-r1",
            "meta/llama-3.3-70b-instruct",
            "nvidia/llama-3.1-nemotron-70b-instruct",
            "mistralai/mixtral-8x22b-instruct-v0.1",
            "microsoft/phi-3-medium-128k-instruct",
            "stepfun-ai/step-3.5-flash",
            "qwen/qwen2.5-72b-instruct",
            "google/gemma-2-27b-it",
        ],
    },
    "gcloud": {
        "type":       "gcloud",
        "api_key_env": None,
        "context_limit": 1000000,
        "max_completion_tokens": 65536,
        "models": [
            "gemini-2.5-pro",
            "gemini-2.0-flash",
            "gemini-1.5-pro",
        ],
    },
    "xai-oauth": {
        "type":       "xai-oauth",
        "api_key_env": None,
        "context_limit": 128000,
        "models": [
            "grok-4", "grok-3", "grok-2-latest", "grok-beta", "grok-build",
        ],
    },
    # ChatGPT / Codex subscription OAuth (`/login chatgpt` or reuse ~/.codex/auth.json).
    # Hits chatgpt.com/backend-api/codex (Responses API) — NOT api.openai.com.
    # Usage bills against the ChatGPT plan, not platform API credits.
    "chatgpt-oauth": {
        "type":       "chatgpt-oauth",
        "api_key_env": None,
        "base_url":   "https://chatgpt.com/backend-api/codex",
        "context_limit": 200000,
        "models": [
            # Keep model IDs bare. The picker adds `chatgpt-oauth/`; embedding
            # `chatgpt/` here produced malformed `chatgpt-oauth/chatgpt/...`.
            "gpt-5.6-sol", "gpt-5.6-sol-pro", "gpt-5.6-terra", "gpt-5.6-luna",
            "gpt-5.5", "gpt-5.4", "gpt-5.4-mini",
            "gpt-5.1-codex", "gpt-5.1-codex-mini",
            "codex-mini-latest", "o3", "o4-mini",
        ],
    },
    "xiaomi": {
        "type": "openai_compat",
        "api_key_env": "XIAOMI_API_KEY",
        "base_url": "https://api.xiaomimimo.com/v1",
        "context_limit": 128000,
        "models": [
            "mimo-v2.5", "mimo-v2.5-pro", "mimo-v2-omni", "MiMo-V2-Pro",
        ],
        # Quirks ripped from open Hermes Xiaomi provider:
        # - supports_vision=True but supports_vision_tool_messages=False (rejects list tool content)
        # - health check /v1/models returns 401 even with key
        # - thinking mode support on some variants
    },
    "sakana": {
        "type":       "openai",
        "api_key_env": "SAKANA_API_KEY",
        "base_url":   "https://api.sakana.ai/v1",
        "context_limit": 128000,
        "models": [
            "fugu-mini", "fugu-ultra",
        ],
    },
    # Meta AI (Muse Spark) — Responses-API-style SSE endpoint at api.meta.ai.
    # Key via /config meta_api_key=... or export META_API_KEY.
    # Pick models as 'meta/muse-spark-1.3' (or bare 'muse-spark-1.3').
    "meta": {
        "type":       "meta",
        "api_key_env": "META_API_KEY",
        "base_url":   "https://api.meta.ai/v1",
        "context_limit": 128000,
        "max_completion_tokens": 32000,
        "models": [
            "muse-spark-1.3", "muse-spark-1.3-contributor",
        ],
    },
}

# Cost per million tokens (approximate, fallback to 0 for unknown)
COSTS = {
    "claude-opus-4-6":          (15.0, 75.0),
    "claude-sonnet-4-6":        (3.0,  15.0),
    "claude-haiku-4-5-20251001": (0.8,  4.0),
    "gpt-4o":                   (2.5,  10.0),
    "gpt-4o-mini":              (0.15,  0.6),
    "o3-mini":                  (1.1,   4.4),
    "gemini-2.0-flash":         (0.075, 0.3),
    "gemini-1.5-pro":           (1.25,  5.0),
    "gemini-2.5-pro-preview-03-25": (1.25, 10.0),
    "moonshot-v1-8k":           (1.0,   3.0),
    "moonshot-v1-32k":          (2.4,   7.0),
    "moonshot-v1-128k":         (8.0,  24.0),
    "qwen-max":                 (2.4,   9.6),
    "qwen-plus":                (0.4,   1.2),
    "deepseek-chat":            (0.27,  1.1),
    "deepseek-reasoner":        (0.55,  2.19),
    "glm-4-plus":               (0.7,   0.7),
    "GLM-4.7":                  (0.7,   0.7),
    "GLM-4.5-AIR":              (0.5,   0.5),
    "MiniMax-Text-01":          (0.7,   2.1),
    "abab6.5s-chat":            (0.1,   0.1),
    "abab6.5-chat":             (0.5,   0.5),
    "gcloud/gemini-2.5-pro":    (1.25, 10.0),
    "gcloud/gemini-2.0-flash":  (0.075, 0.3),
    "gcloud/gemini-1.5-pro":    (1.25, 5.0),
    # Edge / local (llama.cpp, Ollama, on-device APK) — no per-token cost.
    "gemma-3n-e4b":             (0.0, 0.0),
    "gemma-3n-e2b":             (0.0, 0.0),
    "gemma-4-e4b":              (0.0, 0.0),
    "gemma-4-e2b":              (0.0, 0.0),
    "gemini-nano":              (0.0, 0.0),
}

# Auto-detection: prefix → provider name
_PREFIXES = [
    # Dulus account models (Fuel-metered control plane, /login dulus) — MUST
    # precede the generic mistral/qwen/llama→ollama rules below, or ids like
    # dulus-mistral-7b would route to a local Ollama instead of the Dulus
    # control plane. Order matters: first match wins.
    ("dulus/",        "dulus"),
    ("dulus-",        "dulus"),
    # ChatGPT/Codex subscription (OAuth) — MUST precede the generic "gpt-"→openai
    # rule below, or gpt-5.x / codex-* ids route to the paid API instead of the
    # user's ChatGPT plan. Order matters: first match wins.
    ("chatgpt/",      "chatgpt-oauth"),
    ("chatgpt-",      "chatgpt-oauth"),
    ("codex/",        "chatgpt-oauth"),
    ("codex-",        "chatgpt-oauth"),
    ("gpt-5.6",       "chatgpt-oauth"),
    ("gpt-5.5",       "chatgpt-oauth"),
    ("gpt-5.4",       "chatgpt-oauth"),
    ("gpt-5.1-codex", "chatgpt-oauth"),
    ("gpt-5-codex",   "chatgpt-oauth"),
    ("codex-mini",    "chatgpt-oauth"),
    ("codex-auto",    "chatgpt-oauth"),
    # Meta AI (Muse Spark) — api.meta.ai Responses-style SSE, key via META_API_KEY.
    ("meta/",         "meta"),
    ("muse-",         "meta"),
    ("claude-",       "anthropic"),
    ("gpt-",          "openai"),
    ("o1",            "openai"),
    ("o3",            "openai"),
    # Edge / local small models over an OpenAI-compatible server (llama.cpp
    # `llama-server`, Ollama, or the WIP Dulus Edge Bridge APK). MUST precede the
    # gemini-/gemma cloud+ollama rules below so these win. Arbitrary GGUFs are
    # reachable with the explicit `edge/<name>` prefix.
    ("edge/",         "edge"),
    ("gemini-nano",   "edge"),
    ("gemma-3n",      "edge"),
    ("gemma-4",       "edge"),
    ("gemini-",       "gemini"),
    ("kimi-code/",    "kimi-code"),
    ("kimi-for-coding", "kimi-code"),
    ("kimi",          "kimi"),  # matches 'kimi-' and 'kimi'
    ("moonshot-",     "kimi"),
    ("moonshot",       "kimi"), 
    ("qwen",          "qwen"),  # qwen-max, qwen2.5-...
    ("qwq-",          "qwen"),
    ("glm-",          "zhipu"),
    ("GLM-",          "zhipu"),
    ("deepseek-",     "deepseek"),
    ("minimax-",      "minimax"),
    ("MiniMax-",      "minimax"),
    ("abab",          "minimax"),
    ("llama",         "ollama"),
    ("mistral",       "ollama"),
    ("phi",           "ollama"),
    ("gemma",         "ollama"),
    ("gcloud/",       "gcloud"),
    ("gcloud-",       "gcloud"),
    ("grok-",         "xai-oauth"),
    ("grok-build",    "xai-oauth"),
    ("xai-",          "xai-oauth"),
    ("xai-oauth",     "xai-oauth"),
    ("xiaomi-",       "xiaomi"),
    ("mimo-",         "xiaomi"),
    ("xiaomi",        "xiaomi"),
    ("fugu-",         "sakana"),
    ("sakana-",       "sakana"),
]



def detect_provider(model: str) -> str:
    """Return provider name for a model string.
    Supports 'provider/model' explicit format, or auto-detect by prefix."""
    if "/" in model:
        p = model.split("/", 1)[0]
        if p in PROVIDERS:
            return p
    for prefix, pname in _PREFIXES:
        if model.lower().startswith(prefix):
            return pname
    return "openai"   # fallback




XAI_OAUTH_CLIENT_ID = "b1a00492-073a-47ea-816f-4c329264a828"
XAI_OAUTH_SCOPE = "openid profile email offline_access grok-cli:access api:access"
XAI_OAUTH_ISSUER = "https://auth.x.ai"
XAI_OAUTH_DISCOVERY_URL = f"{XAI_OAUTH_ISSUER}/.well-known/openid-configuration"
XAI_OAUTH_REDIRECT_HOST = "127.0.0.1"
XAI_OAUTH_REDIRECT_PORT = 56121
XAI_OAUTH_REDIRECT_PATH = "/callback"
XAI_OAUTH_BASE_URL = "https://api.x.ai/v1"


def _load_grok_build_session_token() -> str | None:
    """Load the real session token from the official Grok Build TUI (this Grok you're talking to).
    Stored in ~/.grok/auth.json after `grok login` (or when you launched this session).
    This is THE auth the official `grok` / Grok Build CLI uses — same OAuth client_id,
    same backend. Using it directly is 100x more reliable than Playwright harvest.
    "grok build y ya".
    """
    import json
    import os
    import pathlib

    grok_home = os.environ.get("GROK_HOME") or str(pathlib.Path.home() / ".grok")
    auth_path = pathlib.Path(grok_home) / "auth.json"

    if not auth_path.exists():
        return None

    try:
        data = json.loads(auth_path.read_text())
        # The key is usually "https://auth.x.ai::b1a00492-073a-47ea-816f-4c329264a828"
        # (exact same client_id we had hardcoded for the harvest flow)
        for entry in data.values():
            if isinstance(entry, dict):
                key = entry.get("key")
                if isinstance(key, str) and len(key) > 40:
                    # Looks like a JWT / access token
                    return key
        # Fallback: sometimes it might be under 'token' or top level
        if isinstance(data.get("key"), str) and len(data["key"]) > 40:
            return data["key"]
    except Exception:
        pass
    return None


# NOTE: The old Playwright browser harvest for Grok (/login grok with _xai_oauth_login)
# has been completely removed for cleanliness.
# The only supported ways to authenticate Grok models are now:
#   - Official Grok Build TUI (`grok login`) → ~/.grok/auth.json (preferred)
#   - Direct XAI_API_KEY
#
# If you want the old harvest behavior you can keep a copy of the previous providers.py
# or implement it yourself. We no longer maintain it here.


# ── Native OAuth 2.0 + PKCE login (no browser automation, no `grok` binary) ──
# Implements the exact Authorization-Code + PKCE flow the official Grok CLI uses
# against auth.x.ai. Opens the system browser, captures the code on the loopback
# redirect (127.0.0.1:56121/callback), exchanges it for tokens, and persists them
# with refresh support. This lets `/login grok` work WITHOUT the `grok` binary.

def _xai_oauth_store_path():
    import os, pathlib
    home = pathlib.Path(os.environ.get("DULUS_HOME") or (pathlib.Path.home() / ".dulus"))
    home.mkdir(parents=True, exist_ok=True)
    return home / "xai_oauth.json"


def _xai_oauth_load_store() -> dict:
    import json
    p = _xai_oauth_store_path()
    if not p.exists():
        return {}
    try:
        return json.loads(p.read_text(encoding="utf-8"))
    except Exception:
        return {}


def _xai_oauth_save_store(data: dict) -> None:
    import json
    try:
        _xai_oauth_store_path().write_text(
            json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
    except Exception:
        pass


def _xai_pkce_pair():
    """Return (code_verifier, code_challenge) for PKCE S256."""
    import base64, hashlib, secrets
    verifier = base64.urlsafe_b64encode(secrets.token_bytes(40)).decode("ascii").rstrip("=")
    digest = hashlib.sha256(verifier.encode("ascii")).digest()
    challenge = base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")
    return verifier, challenge


def _xai_oauth_endpoints() -> dict:
    """Fetch OIDC discovery; fall back to known endpoints if the network call fails."""
    import requests as _req
    try:
        r = _req.get(XAI_OAUTH_DISCOVERY_URL, timeout=15)
        if r.status_code == 200:
            d = r.json()
            return {
                "authorization_endpoint": d["authorization_endpoint"],
                "token_endpoint": d["token_endpoint"],
            }
    except Exception:
        pass
    return {
        "authorization_endpoint": f"{XAI_OAUTH_ISSUER}/oauth2/authorize",
        "token_endpoint": f"{XAI_OAUTH_ISSUER}/oauth2/token",
    }


def _xai_oauth_refresh(refresh_token: str) -> dict | None:
    """Exchange a refresh_token for a fresh access_token. Returns the new store dict or None."""
    import time
    import requests as _req
    eps = _xai_oauth_endpoints()
    try:
        r = _req.post(eps["token_endpoint"], data={
            "grant_type": "refresh_token",
            "refresh_token": refresh_token,
            "client_id": XAI_OAUTH_CLIENT_ID,
        }, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=30)
        if r.status_code != 200:
            return None
        tok = r.json()
    except Exception:
        return None
    store = {
        "access_token": tok.get("access_token", ""),
        # Some servers rotate the refresh token; keep the new one if present.
        "refresh_token": tok.get("refresh_token") or refresh_token,
        "token_type": tok.get("token_type", "Bearer"),
        "obtained_at": time.time(),
        "expires_at": time.time() + int(tok.get("expires_in", 3600)) - 60,
    }
    _xai_oauth_save_store(store)
    return store


def _xai_oauth_login(config: dict, notify=print) -> str | None:
    """Run the native Authorization-Code + PKCE flow. Opens the browser, captures
    the loopback callback, exchanges the code, persists tokens. Returns the access
    token on success, None on failure."""
    import secrets, threading, time, webbrowser
    from urllib.parse import urlencode, urlparse, parse_qs
    from http.server import BaseHTTPRequestHandler, HTTPServer
    import requests as _req

    eps = _xai_oauth_endpoints()
    verifier, challenge = _xai_pkce_pair()
    state = secrets.token_urlsafe(24)
    redirect_uri = f"http://{XAI_OAUTH_REDIRECT_HOST}:{XAI_OAUTH_REDIRECT_PORT}{XAI_OAUTH_REDIRECT_PATH}"

    captured: dict = {}

    class _Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            qs = parse_qs(urlparse(self.path).query)
            captured["code"] = (qs.get("code") or [None])[0]
            captured["state"] = (qs.get("state") or [None])[0]
            captured["error"] = (qs.get("error") or [None])[0]
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.end_headers()
            ok = captured.get("code") and not captured.get("error")
            msg = ("<h2>&#129413; Dulus &mdash; Grok login complete</h2>"
                   "<p>You can close this tab and return to the terminal.</p>") if ok else \
                  ("<h2>Dulus &mdash; login failed</h2><p>%s</p>" % (captured.get("error") or "no code"))
            self.wfile.write(b"<html><body style='font-family:sans-serif;background:#111;color:#eee;"
                             b"text-align:center;padding-top:60px'>" + msg.encode("utf-8") + b"</body></html>")

        def log_message(self, *_a):
            pass  # silence the default stderr logging

    try:
        server = HTTPServer((XAI_OAUTH_REDIRECT_HOST, XAI_OAUTH_REDIRECT_PORT), _Handler)
    except OSError as e:
        notify(f"[xai/grok] Cannot bind {redirect_uri} ({e}). Is another login in progress?")
        return None
    server.timeout = 1

    auth_url = eps["authorization_endpoint"] + "?" + urlencode({
        "response_type": "code",
        "client_id": XAI_OAUTH_CLIENT_ID,
        "redirect_uri": redirect_uri,
        "scope": XAI_OAUTH_SCOPE,
        "state": state,
        "code_challenge": challenge,
        "code_challenge_method": "S256",
    })

    notify("[xai/grok] Opening your browser to log in to X / Grok…")
    notify(f"[xai/grok] If it doesn't open, paste this URL manually:\n{auth_url}")
    try:
        webbrowser.open(auth_url)
    except Exception:
        pass

    # Wait for the callback (up to ~3 minutes), staying responsive.
    deadline = time.time() + 180
    while "code" not in captured and "error" not in captured and time.time() < deadline:
        server.handle_request()
    try:
        server.server_close()
    except Exception:
        pass

    if captured.get("error"):
        notify(f"[xai/grok] Login denied: {captured['error']}")
        return None
    if not captured.get("code"):
        notify("[xai/grok] Login timed out (no callback received).")
        return None
    if captured.get("state") != state:
        notify("[xai/grok] State mismatch — aborting for safety (possible CSRF).")
        return None

    # Exchange the authorization code for tokens.
    try:
        r = _req.post(eps["token_endpoint"], data={
            "grant_type": "authorization_code",
            "code": captured["code"],
            "redirect_uri": redirect_uri,
            "client_id": XAI_OAUTH_CLIENT_ID,
            "code_verifier": verifier,
        }, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=30)
    except Exception as e:
        notify(f"[xai/grok] Token exchange failed: {e}")
        return None
    if r.status_code != 200:
        notify(f"[xai/grok] Token endpoint HTTP {r.status_code}: {(r.text or '')[:300]}")
        return None

    tok = r.json()
    store = {
        "access_token": tok.get("access_token", ""),
        "refresh_token": tok.get("refresh_token", ""),
        "token_type": tok.get("token_type", "Bearer"),
        "obtained_at": time.time(),
        "expires_at": time.time() + int(tok.get("expires_in", 3600)) - 60,
    }
    _xai_oauth_save_store(store)
    notify("[xai/grok] ✅ Logged in. Token saved — Grok models are ready.")
    return store["access_token"] or None


def _xai_oauth_get_token(config: dict) -> str:
    """Get access token for xAI / Grok, in priority order:
    1. Official Grok Build TUI session from ~/.grok/auth.json (if `grok login` was used).
    2. Dulus-native OAuth store from /login grok (auto-refreshes when expired).
    3. XAI_API_KEY (env or config) as direct fallback.
    """
    import os, time

    # 1. Official Grok Build TUI session (best path if the user has the binary)
    grok_token = _load_grok_build_session_token()
    if grok_token:
        return grok_token

    # 2. Dulus-native OAuth store (from /login grok). Refresh on expiry.
    store = _xai_oauth_load_store()
    if store.get("access_token"):
        if not _is_token_expired(store):
            return store["access_token"]
        if store.get("refresh_token"):
            refreshed = _xai_oauth_refresh(store["refresh_token"])
            if refreshed and refreshed.get("access_token"):
                return refreshed["access_token"]
        # Expired and refresh failed — fall through (caller may re-login).

    # 3. Direct API key fallback
    return os.environ.get("XAI_API_KEY") or config.get("xai_api_key") or ""


def _is_token_expired(auth_data: dict) -> bool:
    """True if the stored token is past (or within the 60s buffer of) expiry."""
    import time
    exp = auth_data.get("expires_at")
    if not exp:
        return False  # unknown lifetime → assume valid, let a 401 trigger refresh
    try:
        return time.time() >= float(exp)
    except Exception:
        return False


# ── Grok / xAI remaining usage (Grok Build `/usage` path) ─────────────────────
# Source: open-source xai-org/grok-build (`extensions/billing.rs`).
# The official TUI `/usage` command hits the CLI chat proxy with the same OAuth
# token from `grok login` / `/login grok` — NOT the Management API key.
#
#   GET https://cli-chat-proxy.grok.com/v1/billing?format=credits
#   GET https://cli-chat-proxy.grok.com/v1/billing   (legacy cents)
#
# Toolbar must NEVER block on network: peek_* returns cache only and refreshes
# in a background thread.

GROK_CLI_CHAT_PROXY_BASE = "https://cli-chat-proxy.grok.com/v1"
_GROK_USAGE_CACHE: dict[str, Any] = {
    "snapshot": None,       # dict | None
    "fetched_at": 0.0,      # monotonic-ish wall time
    "error": None,          # str | None
    "refreshing": False,
}
_GROK_USAGE_LOCK = None  # lazy threading.Lock
_GROK_USAGE_TTL_OK_S = 120.0
_GROK_USAGE_TTL_ERR_S = 60.0


def _grok_usage_lock():
    global _GROK_USAGE_LOCK
    if _GROK_USAGE_LOCK is None:
        import threading
        _GROK_USAGE_LOCK = threading.Lock()
    return _GROK_USAGE_LOCK


def is_grok_model(model: str | None) -> bool:
    """True when the active model routes through xAI / Grok OAuth."""
    if not model:
        return False
    m = str(model).strip().lower()
    if not m:
        return False
    # Explicit provider prefix
    if m.startswith(("grok-", "xai-", "xai/", "xai-oauth/", "grok/")):
        return True
    # Bare model ids used in the wild
    if m.startswith("grok"):
        return True
    # provider/model form
    if "/" in m:
        head, _tail = m.split("/", 1)
        if head in ("xai", "xai-oauth", "grok", "grok-build"):
            return True
    try:
        return detect_provider(model) == "xai-oauth"
    except Exception:
        return False


def _cent_val(obj: Any) -> int | None:
    """Extract USD-cents integer from Grok billing Cent shape ``{"val": N}``."""
    if obj is None:
        return None
    if isinstance(obj, (int, float)):
        return int(obj)
    if isinstance(obj, dict):
        if "val" in obj:
            try:
                return int(obj.get("val") or 0)
            except (TypeError, ValueError):
                return 0
    return None


def _parse_grok_billing_payloads(
    credits: dict | None,
    legacy: dict | None = None,
) -> dict[str, Any]:
    """Normalize Grok Build billing JSON into a stable snapshot dict."""
    credits = credits if isinstance(credits, dict) else {}
    legacy = legacy if isinstance(legacy, dict) else {}
    # NOTE: bind to a local before isinstance() so type checkers can narrow it.
    # Calling .get() twice reads as two unrelated expressions and leaves the
    # result Optional, which makes every downstream cfg.get(...) an error.
    _cfg_raw = credits.get("config")
    cfg: dict[str, Any] = _cfg_raw if isinstance(_cfg_raw, dict) else {}
    _leg_cfg_raw = legacy.get("config")
    leg_cfg: dict[str, Any] = _leg_cfg_raw if isinstance(_leg_cfg_raw, dict) else {}

    usage_pct = cfg.get("creditUsagePercent")
    if usage_pct is None:
        usage_pct = cfg.get("credit_usage_percent")
    try:
        usage_pct_f = float(usage_pct) if usage_pct is not None else None
    except (TypeError, ValueError):
        usage_pct_f = None

    remaining_pct = None
    if usage_pct_f is not None:
        remaining_pct = max(0.0, min(100.0, 100.0 - usage_pct_f))

    period = cfg.get("currentPeriod") or cfg.get("current_period") or {}
    period_type = None
    period_start = None
    period_end = None
    if isinstance(period, dict):
        period_type = period.get("type")
        period_start = period.get("start")
        period_end = period.get("end")
    if not period_start:
        period_start = cfg.get("billingPeriodStart") or cfg.get("billing_period_start")
        period_start = period_start or leg_cfg.get("billingPeriodStart")
    if not period_end:
        period_end = cfg.get("billingPeriodEnd") or cfg.get("billing_period_end")
        period_end = period_end or leg_cfg.get("billingPeriodEnd")

    prepaid_cents = _cent_val(cfg.get("prepaidBalance") or cfg.get("prepaid_balance"))
    on_demand_cap = _cent_val(cfg.get("onDemandCap") or cfg.get("on_demand_cap")
                              or leg_cfg.get("onDemandCap"))
    on_demand_used = _cent_val(cfg.get("onDemandUsed") or cfg.get("on_demand_used")
                               or leg_cfg.get("onDemandUsed"))

    # Legacy monthly cents (still useful for dollar remaining display)
    monthly_limit_cents = _cent_val(leg_cfg.get("monthlyLimit") or leg_cfg.get("monthly_limit")
                                    or cfg.get("monthlyLimit") or cfg.get("monthly_limit"))
    used_cents = _cent_val(leg_cfg.get("used") or cfg.get("used"))

    remaining_usd = None
    used_usd = None
    limit_usd = None
    if monthly_limit_cents is not None and used_cents is not None:
        limit_usd = monthly_limit_cents / 100.0
        used_usd = used_cents / 100.0
        remaining_usd = max(0.0, (monthly_limit_cents - used_cents) / 100.0)
    prepaid_usd = (prepaid_cents / 100.0) if prepaid_cents is not None else None

    # Status buckets for toolbar color
    status = "unknown"
    if remaining_pct is not None:
        if remaining_pct <= 0.5:
            status = "exhausted"
        elif remaining_pct < 20.0:
            status = "low"
        else:
            status = "ok"
    elif remaining_usd is not None and limit_usd is not None:
        if remaining_usd <= 0:
            status = "exhausted"
        elif limit_usd > 0 and (remaining_usd / limit_usd) < 0.20:
            status = "low"
        else:
            status = "ok"

    product_usage = cfg.get("productUsage") or cfg.get("product_usage") or []

    return {
        "provider": "grok",
        "usage_percent": usage_pct_f,
        "remaining_percent": remaining_pct,
        "remaining_usd": remaining_usd,
        "used_usd": used_usd,
        "limit_usd": limit_usd,
        "prepaid_usd": prepaid_usd,
        "on_demand_cap_usd": (on_demand_cap / 100.0) if on_demand_cap is not None else None,
        "on_demand_used_usd": (on_demand_used / 100.0) if on_demand_used is not None else None,
        "period_type": period_type,
        "period_start": period_start,
        "period_end": period_end,
        "is_unified_billing": cfg.get("isUnifiedBillingUser", cfg.get("is_unified_billing_user")),
        "product_usage": product_usage if isinstance(product_usage, list) else [],
        "status": status,
        "raw_credits": credits or None,
        "raw_legacy": legacy or None,
        "checked_at": time.time(),
    }


def fetch_grok_billing(config: dict | None = None, *, timeout: float = 12.0) -> dict[str, Any]:
    """Live-fetch Grok Build billing/usage via OAuth (same path as `/usage`).

    Raises on hard failures (no token / HTTP error). Prefer
    ``get_grok_usage_snapshot`` for UI paths that must stay non-blocking.
    """
    config = config or {}
    token = _xai_oauth_get_token(config) or ""
    if not token:
        raise RuntimeError("No Grok/xAI OAuth token. Run /login grok first.")

    base = GROK_CLI_CHAT_PROXY_BASE.rstrip("/")
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json",
        "User-Agent": "dulus-grok-usage/1.0",
        "x-grok-client-mode": "cli",
    }

    def _get(url: str) -> dict:
        resp = requests.get(url, headers=headers, timeout=timeout)
        if resp.status_code == 401:
            # One refresh attempt via native store, then retry once.
            store = _xai_oauth_load_store()
            if store.get("refresh_token"):
                refreshed = _xai_oauth_refresh(store["refresh_token"])
                if refreshed and refreshed.get("access_token"):
                    headers["Authorization"] = f"Bearer {refreshed['access_token']}"
                    resp = requests.get(url, headers=headers, timeout=timeout)
        if resp.status_code >= 400:
            body = (resp.text or "")[:240]
            raise RuntimeError(f"Grok billing HTTP {resp.status_code}: {body}")
        try:
            data = resp.json()
        except Exception as e:
            raise RuntimeError(f"Grok billing: invalid JSON ({e})") from e
        return data if isinstance(data, dict) else {}

    credits = _get(f"{base}/billing?format=credits")
    legacy: dict = {}
    try:
        legacy = _get(f"{base}/billing")
    except Exception:
        # Credits-only is enough for the toolbar; legacy is enrichment.
        legacy = {}
    snap = _parse_grok_billing_payloads(credits, legacy)
    # Keep the toolbar cache warm whenever anyone does a live fetch.
    try:
        with _grok_usage_lock():
            _GROK_USAGE_CACHE["snapshot"] = snap
            _GROK_USAGE_CACHE["fetched_at"] = time.time()
            _GROK_USAGE_CACHE["error"] = None
            _GROK_USAGE_CACHE["refreshing"] = False
    except Exception:
        pass
    return snap


def get_grok_usage_snapshot(
    config: dict | None = None,
    *,
    force: bool = False,
    wait: bool = False,
    timeout: float = 12.0,
) -> dict[str, Any] | None:
    """Return cached Grok usage snapshot; refresh in background when stale.

    - ``wait=False`` (default): never blocks — returns last good snapshot or None
      and kicks a background refresh if needed. Safe for the TUI toolbar.
    - ``wait=True`` / ``force=True``: fetch synchronously (for /quota, tests).
    """
    config = config or {}
    now = time.time()
    lock = _grok_usage_lock()

    with lock:
        snap = _GROK_USAGE_CACHE.get("snapshot")
        fetched_at = float(_GROK_USAGE_CACHE.get("fetched_at") or 0.0)
        err = _GROK_USAGE_CACHE.get("error")
        refreshing = bool(_GROK_USAGE_CACHE.get("refreshing"))
        ttl = _GROK_USAGE_TTL_OK_S if snap is not None else _GROK_USAGE_TTL_ERR_S
        age = now - fetched_at if fetched_at else 1e18
        stale = force or age >= ttl

    if not stale and snap is not None:
        return snap

    if wait or force:
        try:
            fresh = fetch_grok_billing(config, timeout=timeout)
            with lock:
                _GROK_USAGE_CACHE["snapshot"] = fresh
                _GROK_USAGE_CACHE["fetched_at"] = time.time()
                _GROK_USAGE_CACHE["error"] = None
                _GROK_USAGE_CACHE["refreshing"] = False
            return fresh
        except Exception as e:
            with lock:
                _GROK_USAGE_CACHE["error"] = str(e)[:240]
                _GROK_USAGE_CACHE["fetched_at"] = time.time()
                _GROK_USAGE_CACHE["refreshing"] = False
            return snap  # last good, if any

    # Non-blocking path: spawn refresh if not already running
    if stale and not refreshing:
        import threading

        def _bg():
            try:
                fresh = fetch_grok_billing(config, timeout=timeout)
                with lock:
                    _GROK_USAGE_CACHE["snapshot"] = fresh
                    _GROK_USAGE_CACHE["fetched_at"] = time.time()
                    _GROK_USAGE_CACHE["error"] = None
            except Exception as e:
                with lock:
                    _GROK_USAGE_CACHE["error"] = str(e)[:240]
                    _GROK_USAGE_CACHE["fetched_at"] = time.time()
            finally:
                with lock:
                    _GROK_USAGE_CACHE["refreshing"] = False

        with lock:
            if _GROK_USAGE_CACHE.get("refreshing"):
                return snap
            _GROK_USAGE_CACHE["refreshing"] = True
        t = threading.Thread(target=_bg, name="grok-usage-refresh", daemon=True)
        t.start()

    return snap


def format_grok_usage_toolbar(snapshot: dict[str, Any] | None) -> str:
    """Short plain label for the TUI toolbar (no ANSI). Empty if unknown."""
    if not snapshot:
        return ""
    # Prefer percent remaining (Grok Build primary signal)
    rem_pct = snapshot.get("remaining_percent")
    if rem_pct is not None:
        try:
            return f"💳 {int(rem_pct)}% left"
        except (TypeError, ValueError):
            pass
    rem_usd = snapshot.get("remaining_usd")
    if rem_usd is not None:
        try:
            v = float(rem_usd)
            if v >= 100:
                return f"💳 ${v:.0f} left"
            return f"💳 ${v:.2f} left"
        except (TypeError, ValueError):
            pass
    prepaid = snapshot.get("prepaid_usd")
    if prepaid is not None:
        try:
            v = float(prepaid)
            if v > 0:
                return f"💳 ${v:.2f} prepaid"
        except (TypeError, ValueError):
            pass
    return ""


def peek_grok_usage_toolbar(config: dict | None = None, model: str | None = None) -> str:
    """Toolbar-safe Grok remaining-usage label.

    Returns "" unless the active model is Grok and we already have (or can
    background-fetch) a snapshot. Never raises, never blocks on network.
    """
    try:
        cfg = config or {}
        m = model if model is not None else cfg.get("model", "")
        if not is_grok_model(m):
            return ""
        snap = get_grok_usage_snapshot(cfg, force=False, wait=False)
        return format_grok_usage_toolbar(snap)
    except Exception:
        return ""


def reset_grok_usage_cache() -> None:
    """Test helper — clear the in-process Grok usage cache."""
    with _grok_usage_lock():
        _GROK_USAGE_CACHE["snapshot"] = None
        _GROK_USAGE_CACHE["fetched_at"] = 0.0
        _GROK_USAGE_CACHE["error"] = None
        _GROK_USAGE_CACHE["refreshing"] = False


# ── Anthropic / Claude OAuth (Claude Pro/Max subscription login — NO API key) ──
# Mirrors the Grok OAuth flow above, for Anthropic's subscription auth. The user
# logs in at claude.ai (the browser part), the authorization code is shown on the
# console.anthropic.com callback page as  CODE#STATE , pasted back into the
# terminal, and exchanged for an OAuth access/refresh token. claude-* models then
# run on the user's Claude subscription via `Authorization: Bearer` + the oauth
# beta header — never an API key. This is the replacement for the cookie webbridge.
ANTHROPIC_OAUTH_CLIENT_ID     = "9d1c250a-e61b-44d9-88ed-5944d1962f5e"
ANTHROPIC_OAUTH_AUTHORIZE_URL = "https://claude.ai/oauth/authorize"
ANTHROPIC_OAUTH_TOKEN_URL     = "https://console.anthropic.com/v1/oauth/token"
ANTHROPIC_OAUTH_REDIRECT_URI  = "https://console.anthropic.com/oauth/code/callback"
ANTHROPIC_OAUTH_SCOPE         = "org:create_api_key user:profile user:inference"
ANTHROPIC_OAUTH_BETA          = "oauth-2025-04-20"
# OAuth subscription tokens are only authorized for the Claude Code client, so the
# FIRST system block must carry this exact identity or the API returns 403.
ANTHROPIC_OAUTH_IDENTITY      = "You are Claude Code, Anthropic's official CLI for Claude."


def _anthropic_oauth_store_path():
    import os, pathlib
    home = pathlib.Path(os.environ.get("DULUS_HOME") or (pathlib.Path.home() / ".dulus"))
    home.mkdir(parents=True, exist_ok=True)
    return home / "anthropic_oauth.json"


def _anthropic_oauth_load_store() -> dict:
    import json
    p = _anthropic_oauth_store_path()
    if not p.exists():
        return {}
    try:
        return json.loads(p.read_text(encoding="utf-8"))
    except Exception:
        return {}


def _anthropic_oauth_save_store(data: dict) -> None:
    import json
    try:
        _anthropic_oauth_store_path().write_text(
            json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
    except Exception:
        pass


def _anthropic_oauth_refresh(refresh_token: str) -> dict | None:
    """Exchange a refresh_token for a fresh access_token. Returns the new store dict
    or None. Body is JSON (Anthropic's token endpoint expects JSON, unlike xAI)."""
    import time
    import requests as _req
    try:
        r = _req.post(ANTHROPIC_OAUTH_TOKEN_URL, json={
            "grant_type": "refresh_token",
            "refresh_token": refresh_token,
            "client_id": ANTHROPIC_OAUTH_CLIENT_ID,
        }, headers={"Content-Type": "application/json"}, timeout=30)
        if r.status_code != 200:
            return None
        tok = r.json()
    except Exception:
        return None
    store = {
        "access_token": tok.get("access_token", ""),
        # Anthropic rotates refresh tokens; keep the new one if present.
        "refresh_token": tok.get("refresh_token") or refresh_token,
        "token_type": tok.get("token_type", "Bearer"),
        "scope": tok.get("scope", ANTHROPIC_OAUTH_SCOPE),
        "obtained_at": time.time(),
        "expires_at": time.time() + int(tok.get("expires_in", 3600)) - 60,
    }
    _anthropic_oauth_save_store(store)
    return store


def _anthropic_oauth_login(config: dict, notify=print, get_code=None) -> str | None:
    """Native Authorization-Code + PKCE login for Claude Pro/Max. Opens the browser
    to claude.ai, the user approves and copies the code shown on the callback page,
    pastes it back here; we exchange it for OAuth tokens. Returns the access token on
    success, None on failure. NO API key is involved anywhere in this flow."""
    import secrets, time, webbrowser
    from urllib.parse import urlencode, urlparse, parse_qs
    import requests as _req

    verifier, challenge = _xai_pkce_pair()  # generic PKCE S256 helper (reused)
    state = secrets.token_urlsafe(24)

    auth_url = ANTHROPIC_OAUTH_AUTHORIZE_URL + "?" + urlencode({
        "code": "true",
        "client_id": ANTHROPIC_OAUTH_CLIENT_ID,
        "response_type": "code",
        "redirect_uri": ANTHROPIC_OAUTH_REDIRECT_URI,
        "scope": ANTHROPIC_OAUTH_SCOPE,
        "code_challenge": challenge,
        "code_challenge_method": "S256",
        "state": state,
    })

    notify("[claude] Opening your browser to log in to your Claude account…")
    notify(f"[claude] If it doesn't open, paste this URL manually:\n{auth_url}")
    try:
        webbrowser.open(auth_url)
    except Exception:
        pass

    # The callback page shows a code like  CODE#STATE  — the user pastes it back.
    raw = ""
    try:
        if get_code is not None:
            raw = (get_code() or "").strip()
        else:
            raw = input("[claude] Paste the code from the browser here: ").strip()
    except (EOFError, KeyboardInterrupt):
        notify("[claude] Login cancelled.")
        return None

    if not raw:
        notify("[claude] No code provided.")
        return None

    # Accept a bare 'CODE#STATE', a bare CODE, or the full redirected callback URL.
    code = raw
    st = state
    if "code=" in raw:
        qs = parse_qs(urlparse(raw).query)
        code = (qs.get("code") or [raw])[0]
        st = (qs.get("state") or [state])[0]
    if "#" in code:
        code, _, st = code.partition("#")

    try:
        r = _req.post(ANTHROPIC_OAUTH_TOKEN_URL, json={
            "grant_type": "authorization_code",
            "client_id": ANTHROPIC_OAUTH_CLIENT_ID,
            "code": code,
            "state": st,
            "redirect_uri": ANTHROPIC_OAUTH_REDIRECT_URI,
            "code_verifier": verifier,
        }, headers={"Content-Type": "application/json"}, timeout=30)
    except Exception as e:
        notify(f"[claude] Token exchange failed: {e}")
        return None

    if r.status_code != 200:
        notify(f"[claude] Token exchange HTTP {r.status_code}: {(r.text or '')[:300]}")
        return None

    tok = r.json()
    store = {
        "access_token": tok.get("access_token", ""),
        "refresh_token": tok.get("refresh_token", ""),
        "token_type": tok.get("token_type", "Bearer"),
        "scope": tok.get("scope", ANTHROPIC_OAUTH_SCOPE),
        "obtained_at": time.time(),
        "expires_at": time.time() + int(tok.get("expires_in", 3600)) - 60,
    }
    if not store["access_token"]:
        notify("[claude] No access_token in token response.")
        return None
    _anthropic_oauth_save_store(store)
    return store["access_token"]


def _anthropic_oauth_get_token(config: dict) -> str:
    """Return a valid Claude OAuth access token (from /login claude), refreshing on
    expiry. Returns "" when no OAuth session exists — callers then fall back to the
    ANTHROPIC_API_KEY path."""
    store = _anthropic_oauth_load_store()
    if not store.get("access_token"):
        return ""
    if not _is_token_expired(store):
        return store["access_token"]
    if store.get("refresh_token"):
        refreshed = _anthropic_oauth_refresh(store["refresh_token"])
        if refreshed and refreshed.get("access_token"):
            return refreshed["access_token"]
    return ""  # expired and refresh failed — caller should prompt /login claude


def _anthropic_oauth_system_blocks(system, cc_marker: dict | None = None):
    """Ensure the Claude Code identity is the FIRST system block (mandatory for OAuth
    subscription tokens, else the API 403s). Returns a list of content blocks, keeping
    the user's real system prompt as a cached second block."""
    cc_marker = cc_marker or {"type": "ephemeral"}
    identity = {"type": "text", "text": ANTHROPIC_OAUTH_IDENTITY}
    if isinstance(system, str):
        rest = ([{"type": "text", "text": system,
                  "cache_control": cc_marker}] if system else [])
        return [identity] + rest
    if isinstance(system, list):
        if (system and isinstance(system[0], dict)
                and str(system[0].get("text", "")).startswith("You are Claude Code")):
            return system  # identity already present
        return [identity] + system
    return [identity]


# ── Generic OAuth token-store helpers (shared by chatgpt-oauth) ──────────
# Ported from the private build's refactor so chatgpt-oauth can reuse one
# store implementation. The per-provider _xai_*/_anthropic_* stores are left
# untouched; these only back the ChatGPT/Codex store below.
def _oauth_store_path(filename: str):
    """Return a path inside the private Dulus auth directory.

    OAuth stores contain bearer and refresh tokens.  Keeping the directory
    creation and file handling in one place prevents the xAI and Anthropic
    implementations from drifting apart.
    """
    import os
    import pathlib

    home = pathlib.Path(
        os.environ.get("DULUS_HOME") or (pathlib.Path.home() / ".dulus")
    )
    home.mkdir(parents=True, exist_ok=True)
    return home / filename


def _oauth_load_store(path) -> dict:
    """Load an OAuth JSON object, treating corrupt/non-object data as absent."""
    import json

    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError):
        return {}
    return data if isinstance(data, dict) else {}


def _oauth_save_store(path, data: dict) -> None:
    """Atomically save an OAuth token store with owner-only permissions.

    A process interruption during ``write_text`` used to leave a truncated
    auth file, effectively logging the user out.  Writing beside the target
    and replacing it atomically avoids that failure mode.  ``chmod`` is best
    effort on Windows and enforces 0600 on POSIX.
    """
    import json
    import os

    tmp = path.with_name(path.name + ".tmp")
    try:
        tmp.write_text(
            json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8"
        )
        try:
            os.chmod(tmp, 0o600)
        except OSError:
            pass
        os.replace(tmp, path)
    except (OSError, TypeError, ValueError):
        try:
            tmp.unlink(missing_ok=True)
        except OSError:
            pass


def _oauth_expires_at(token_payload: dict, default_seconds: int = 3600) -> float:
    """Calculate a buffered expiry even when a provider returns odd metadata."""
    import time

    try:
        lifetime = max(0, int(float(token_payload.get("expires_in", default_seconds))))
    except (TypeError, ValueError):
        lifetime = default_seconds
    return time.time() + lifetime - 60


# ── ChatGPT / Codex OAuth (ChatGPT Plus/Pro/Team subscription — NO API key) ──
# Same Authorization-Code + PKCE flow the official Codex CLI uses against
# auth.openai.com (client_id app_EMoamEEZ73f0CkXaXp7hrann). Tokens land in
# ~/.dulus/chatgpt_oauth.json; we ALSO reuse ~/.codex/auth.json when present
# (so a prior `codex login` Just Works). Inference goes to the Codex backend
# Responses API (chatgpt.com/backend-api/codex/responses) — the OAuth token is
# rejected by the regular platform /v1/chat/completions endpoint.
#
# Refs (reverse-engineered from Codex CLI + OpenCode + codex-openai-proxy):
#   https://developers.openai.com/codex/auth
#   client_id / endpoints documented in openai/codex + anomalyco/opencode
CHATGPT_OAUTH_CLIENT_ID     = "app_EMoamEEZ73f0CkXaXp7hrann"
CHATGPT_OAUTH_AUTHORIZE_URL = "https://auth.openai.com/oauth/authorize"
CHATGPT_OAUTH_TOKEN_URL     = "https://auth.openai.com/oauth/token"
CHATGPT_OAUTH_REDIRECT_URI  = "http://localhost:1455/auth/callback"
CHATGPT_OAUTH_SCOPE         = "openid profile email offline_access"
CHATGPT_OAUTH_BASE_URL      = "https://chatgpt.com/backend-api/codex"
CHATGPT_OAUTH_ORIGINATOR    = "codex_cli_rs"
CHATGPT_OAUTH_CLIENT_VER    = "0.144.0"  # client version for the current model set
CHATGPT_OAUTH_USER_AGENT    = f"codex_cli_rs/{CHATGPT_OAUTH_CLIENT_VER}"
# Local loopback capture (Codex CLI default port). If 1455 is busy we fall
# back to paste-the-callback-URL mode, same as Claude/Z.ai.
CHATGPT_OAUTH_REDIRECT_PORT = 1455


def _chatgpt_oauth_store_path():
    return _oauth_store_path("chatgpt_oauth.json")


def _chatgpt_oauth_load_store() -> dict:
    return _oauth_load_store(_chatgpt_oauth_store_path())


def _chatgpt_oauth_save_store(data: dict) -> None:
    _oauth_save_store(_chatgpt_oauth_store_path(), data)


def _chatgpt_jwt_payload(token: str | None) -> dict:
    """Decode a JWT payload without verifying the signature (we only need claims)."""
    if not token or not isinstance(token, str) or token.count(".") < 2:
        return {}
    try:
        import base64
        payload = token.split(".")[1]
        payload += "=" * (-len(payload) % 4)
        data = json.loads(base64.urlsafe_b64decode(payload.encode("ascii")))
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _chatgpt_extract_account_id(
    id_token: str | None = None,
    access_token: str | None = None,
    fallback: str | None = None,
) -> str:
    """Pull chatgpt_account_id from id_token → access_token → explicit fallback."""
    for tok in (id_token, access_token):
        pl = _chatgpt_jwt_payload(tok)
        if not pl:
            continue
        auth = pl.get("https://api.openai.com/auth") or {}
        if isinstance(auth, dict):
            aid = auth.get("chatgpt_account_id") or auth.get("account_id")
            if aid:
                return str(aid)
        aid = pl.get("chatgpt_account_id") or pl.get("account_id")
        if aid:
            return str(aid)
    return str(fallback or "")


def _chatgpt_codex_auth_path():
    """Path to the official Codex CLI auth cache (~/.codex/auth.json)."""
    import os
    import pathlib
    home = os.environ.get("CODEX_HOME") or str(pathlib.Path.home() / ".codex")
    override = os.environ.get("CODEX_AUTH_FILE")
    return pathlib.Path(override) if override else pathlib.Path(home) / "auth.json"


def _chatgpt_load_codex_cli_auth() -> dict:
    """Read ~/.codex/auth.json (written by `codex login`) into our store shape.

    Returns {} when missing/unreadable. Does NOT refresh — caller handles that.
    """
    import pathlib
    path = _chatgpt_codex_auth_path()
    try:
        if not pathlib.Path(path).is_file():
            return {}
        raw = json.loads(pathlib.Path(path).read_text(encoding="utf-8"))
    except Exception:
        return {}
    if not isinstance(raw, dict):
        return {}
    _tk = raw.get("tokens")
    tokens = _tk if isinstance(_tk, dict) else {}
    access = tokens.get("access_token") or ""
    refresh = tokens.get("refresh_token") or ""
    if not access and not refresh:
        return {}
    id_token = tokens.get("id_token") or ""
    account_id = (
        tokens.get("account_id")
        or _chatgpt_extract_account_id(id_token, access)
        or ""
    )
    # Prefer exp claim from the access JWT; fall back to last_refresh + 1h.
    expires_at = None
    pl = _chatgpt_jwt_payload(access)
    if pl.get("exp") is not None:
        try:
            expires_at = float(pl["exp"])
        except (TypeError, ValueError):
            expires_at = None
    if expires_at is None:
        # Codex stores last_refresh as ISO; give a conservative 50min window.
        try:
            from datetime import datetime, timezone
            lr = raw.get("last_refresh") or ""
            if lr:
                dt = datetime.fromisoformat(str(lr).replace("Z", "+00:00"))
                expires_at = dt.timestamp() + 3000.0
        except Exception:
            expires_at = None
    return {
        "access_token": access,
        "refresh_token": refresh,
        "id_token": id_token,
        "account_id": account_id,
        "token_type": "Bearer",
        "scope": CHATGPT_OAUTH_SCOPE,
        "obtained_at": time.time(),
        "expires_at": expires_at,
        "source": "codex-cli",
    }


def _chatgpt_oauth_apply_token_response(tok: dict, prev: dict | None = None) -> dict:
    """Normalize a token-endpoint JSON body into our on-disk store shape."""
    prev = prev or {}
    access = tok.get("access_token") or ""
    refresh = tok.get("refresh_token") or prev.get("refresh_token") or ""
    id_token = tok.get("id_token") or prev.get("id_token") or ""
    account_id = (
        tok.get("account_id")
        or _chatgpt_extract_account_id(id_token, access, prev.get("account_id"))
        or prev.get("account_id")
        or ""
    )
    store = {
        "access_token": access,
        "refresh_token": refresh,
        "id_token": id_token,
        "account_id": account_id,
        "token_type": tok.get("token_type", "Bearer"),
        "scope": tok.get("scope", CHATGPT_OAUTH_SCOPE),
        "obtained_at": time.time(),
        "expires_at": _oauth_expires_at(tok),
        "source": prev.get("source") or "dulus",
    }
    # If the token response didn't give expires_in, fall back to JWT exp.
    if not store.get("expires_at"):
        pl = _chatgpt_jwt_payload(access)
        if pl.get("exp") is not None:
            try:
                store["expires_at"] = float(pl["exp"])
            except (TypeError, ValueError):
                pass
    return store


def _chatgpt_oauth_refresh(refresh_token: str, prev: dict | None = None) -> dict | None:
    """Refresh a ChatGPT/Codex OAuth access token. Returns updated store or None."""
    import requests as _req
    if not refresh_token:
        return None
    try:
        r = _req.post(
            CHATGPT_OAUTH_TOKEN_URL,
            data={
                "grant_type": "refresh_token",
                "refresh_token": refresh_token,
                "client_id": CHATGPT_OAUTH_CLIENT_ID,
            },
            headers={
                "Content-Type": "application/x-www-form-urlencoded",
                "Accept": "application/json",
            },
            timeout=30,
        )
    except Exception:
        return None
    if r.status_code != 200:
        return None
    try:
        tok = r.json()
    except Exception:
        return None
    if not isinstance(tok, dict) or not tok.get("access_token"):
        return None
    store = _chatgpt_oauth_apply_token_response(tok, prev)
    _chatgpt_oauth_save_store(store)
    return store


def _chatgpt_oauth_login(config: dict, notify=print, get_code=None) -> str | None:
    """Native Authorization-Code + PKCE login for ChatGPT/Codex.

    Opens the browser to auth.openai.com (Codex CLI client). Prefers a local
    loopback capture on :1455 (Codex's redirect_uri); falls back to paste-the-
    callback-URL if the port is busy or get_code is supplied. Returns the
    access token on success, None on failure.
    """
    import secrets
    import socket
    import webbrowser
    from http.server import BaseHTTPRequestHandler, HTTPServer
    from urllib.parse import parse_qs, urlencode, urlparse
    import requests as _req

    verifier, challenge = _xai_pkce_pair()  # generic S256 helper
    state = secrets.token_urlsafe(24)

    auth_params = {
        "response_type": "code",
        "client_id": CHATGPT_OAUTH_CLIENT_ID,
        "redirect_uri": CHATGPT_OAUTH_REDIRECT_URI,
        "scope": CHATGPT_OAUTH_SCOPE,
        "code_challenge": challenge,
        "code_challenge_method": "S256",
        "state": state,
        "id_token_add_organizations": "true",
        "codex_cli_simplified_flow": "true",
        "originator": CHATGPT_OAUTH_ORIGINATOR,
    }
    auth_url = CHATGPT_OAUTH_AUTHORIZE_URL + "?" + urlencode(auth_params)

    captured: dict = {"code": "", "state": "", "error": ""}

    def _parse_callback(raw: str) -> None:
        raw = (raw or "").strip()
        if not raw:
            return
        # Accept full callback URL, bare code, or code#state.
        if "code=" in raw or raw.startswith("http"):
            qs = parse_qs(urlparse(raw).query)
            captured["code"] = (qs.get("code") or [""])[0]
            captured["state"] = (qs.get("state") or [state])[0]
            captured["error"] = (qs.get("error") or [""])[0]
            return
        if "#" in raw:
            code, _, st = raw.partition("#")
            captured["code"] = code
            captured["state"] = st or state
            return
        captured["code"] = raw
        captured["state"] = state

    # Try loopback capture on Codex's default port first.
    server = None
    port_free = False
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            s.bind(("127.0.0.1", CHATGPT_OAUTH_REDIRECT_PORT))
            port_free = True
    except OSError:
        port_free = False

    if port_free and get_code is None:
        class _Handler(BaseHTTPRequestHandler):
            def do_GET(self):  # noqa: N802
                qs = parse_qs(urlparse(self.path).query)
                captured["code"] = (qs.get("code") or [""])[0]
                captured["state"] = (qs.get("state") or [""])[0]
                captured["error"] = (qs.get("error") or [""])[0]
                ok = bool(captured["code"]) and not captured["error"]
                html = (
                    "<html><body style='font-family:sans-serif;padding:2rem'>"
                    "<h2>Dulus &mdash; ChatGPT login OK</h2>"
                    "<p>You can close this tab and return to the terminal.</p>"
                    "</body></html>"
                    if ok else
                    "<html><body style='font-family:sans-serif;padding:2rem'>"
                    "<h2>Dulus &mdash; ChatGPT login failed</h2>"
                    "<p>No authorization code received. Close this tab and retry.</p>"
                    "</body></html>"
                )
                body = html.encode("utf-8")
                self.send_response(200 if ok else 400)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)

            def log_message(self, fmt, *args):  # silence
                return

        try:
            server = HTTPServer(("127.0.0.1", CHATGPT_OAUTH_REDIRECT_PORT), _Handler)
            server.timeout = 1.0
        except OSError:
            server = None

    notify("[chatgpt] Opening your browser to log in with ChatGPT (Codex OAuth)…")
    notify(f"[chatgpt] If it doesn't open, paste this URL manually:\n{auth_url}")
    try:
        webbrowser.open(auth_url)
    except Exception:
        pass

    if server is not None:
        notify(
            f"[chatgpt] Waiting for browser callback on "
            f"http://localhost:{CHATGPT_OAUTH_REDIRECT_PORT}/auth/callback …"
        )
        notify("[chatgpt] (Ctrl+C to cancel)")
        try:
            deadline = time.time() + 300.0
            while time.time() < deadline and not captured["code"] and not captured["error"]:
                server.handle_request()
        except KeyboardInterrupt:
            notify("[chatgpt] Login cancelled.")
            try:
                server.server_close()
            except Exception:
                pass
            return None
        try:
            server.server_close()
        except Exception:
            pass
    else:
        # Port busy or get_code supplied — paste mode.
        notify(
            "[chatgpt] Paste the full redirect URL from the browser "
            "(looks like http://localhost:1455/auth/callback?code=…&state=…):"
        )
        try:
            if get_code is not None:
                raw = (get_code() or "").strip()
            else:
                raw = input("[chatgpt] Callback URL or code: ").strip()
        except (EOFError, KeyboardInterrupt):
            notify("[chatgpt] Login cancelled.")
            return None
        _parse_callback(raw)

    if captured.get("error"):
        notify(f"[chatgpt] Auth error from OpenAI: {captured['error']}")
        return None
    code = (captured.get("code") or "").strip()
    if not code:
        notify("[chatgpt] No authorization code received.")
        return None
    st = (captured.get("state") or "").strip()
    if st and st != state:
        notify("[chatgpt] State mismatch — aborting (possible CSRF). Re-run /login chatgpt.")
        return None

    try:
        r = _req.post(
            CHATGPT_OAUTH_TOKEN_URL,
            data={
                "grant_type": "authorization_code",
                "client_id": CHATGPT_OAUTH_CLIENT_ID,
                "code": code,
                "redirect_uri": CHATGPT_OAUTH_REDIRECT_URI,
                "code_verifier": verifier,
            },
            headers={
                "Content-Type": "application/x-www-form-urlencoded",
                "Accept": "application/json",
            },
            timeout=30,
        )
    except Exception as e:
        notify(f"[chatgpt] Token exchange failed: {e}")
        return None

    if r.status_code != 200:
        notify(f"[chatgpt] Token exchange HTTP {r.status_code}: {(r.text or '')[:300]}")
        return None
    try:
        tok = r.json()
    except Exception as e:
        notify(f"[chatgpt] Invalid token response: {e}")
        return None
    if not isinstance(tok, dict) or not tok.get("access_token"):
        notify("[chatgpt] No access_token in token response.")
        return None

    store = _chatgpt_oauth_apply_token_response(tok)
    store["source"] = "dulus"
    _chatgpt_oauth_save_store(store)
    notify(
        f"[chatgpt] Logged in"
        + (f" (account {(store.get('account_id') or '')[:8]}…)" if store.get("account_id") else "")
        + ". chatgpt/* models now use your ChatGPT subscription."
    )
    return store["access_token"]


def _chatgpt_oauth_get_session(config: dict | None = None) -> dict:
    """Return a valid ChatGPT OAuth session dict {access_token, account_id, ...}.

    Priority:
      1. Dulus-native store (~/.dulus/chatgpt_oauth.json) from `/login chatgpt`
      2. Official Codex CLI cache (~/.codex/auth.json) from `codex login`
    Auto-refreshes when expired. Returns {} when nothing usable is available.
    """
    config = config or {}

    # 1. Dulus-native store
    store = _chatgpt_oauth_load_store()
    if store.get("access_token") or store.get("refresh_token"):
        if store.get("access_token") and not _is_token_expired(store):
            return store
        if store.get("refresh_token"):
            refreshed = _chatgpt_oauth_refresh(store["refresh_token"], store)
            if refreshed and refreshed.get("access_token"):
                return refreshed

    # 2. Official Codex CLI cache
    cli = _chatgpt_load_codex_cli_auth()
    if cli.get("access_token") or cli.get("refresh_token"):
        if cli.get("access_token") and not _is_token_expired(cli):
            # Mirror into Dulus store so subsequent calls are fast + refreshable.
            try:
                _chatgpt_oauth_save_store({**cli, "source": "codex-cli"})
            except Exception:
                pass
            return cli
        if cli.get("refresh_token"):
            refreshed = _chatgpt_oauth_refresh(cli["refresh_token"], cli)
            if refreshed and refreshed.get("access_token"):
                return refreshed

    return {}


def _chatgpt_oauth_get_token(config: dict | None = None) -> str:
    """Convenience: just the bearer access token, or ''."""
    return (_chatgpt_oauth_get_session(config) or {}).get("access_token") or ""


def _chatgpt_oauth_headers(session: dict) -> dict:
    """Build the headers Codex CLI sends to chatgpt.com/backend-api/codex."""
    headers = {
        "Authorization": f"Bearer {session.get('access_token', '')}",
        "Content-Type": "application/json",
        "Accept": "text/event-stream",
        "OpenAI-Beta": "responses=experimental",
        "originator": CHATGPT_OAUTH_ORIGINATOR,
        "User-Agent": CHATGPT_OAUTH_USER_AGENT,
        "version": CHATGPT_OAUTH_CLIENT_VER,
    }
    account_id = session.get("account_id") or _chatgpt_extract_account_id(
        session.get("id_token"), session.get("access_token")
    )
    if account_id:
        headers["ChatGPT-Account-Id"] = account_id
    return headers


def _chatgpt_parse_reasoning_suffix(model: str) -> tuple[str, str | None]:
    """Strip optional reasoning-effort suffixes from a model id.

    Examples:
      gpt-5.4-high     → (gpt-5.4, high)
      gpt-5.5-xhigh    → (gpt-5.5, xhigh)
      chatgpt/gpt-5.4  → (gpt-5.4, None)
    """
    m = (model or "").strip()
    for prefix in ("chatgpt/", "chatgpt-", "codex/", "codex-"):
        if m.lower().startswith(prefix):
            m = m[len(prefix):]
            break
    effort = None
    for suffix in ("-ultra", "-xhigh", "-max", "-high", "-medium", "-low", "-minimal"):
        if m.lower().endswith(suffix):
            effort = suffix.lstrip("-")
            m = m[: -len(suffix)]
            break
    return m, effort


def _chatgpt_messages_to_responses_input(
    system: str,
    messages: list,
) -> tuple[str, list]:
    """Translate Dulus neutral messages → Codex Responses API (instructions, input)."""
    instruction_parts: list[str] = []
    if system:
        instruction_parts.append(str(system))

    # Sanitize orphans the same way chat-completions does, then convert.
    openai_msgs = messages_to_openai(messages)

    input_items: list[dict] = []
    for m in openai_msgs:
        role = m.get("role")
        if role in ("system", "developer"):
            content = m.get("content") or ""
            if isinstance(content, list):
                content = "".join(
                    (p.get("text") or "") if isinstance(p, dict) else str(p)
                    for p in content
                )
            if content:
                instruction_parts.append(str(content))
            continue

        if role == "user":
            content = m.get("content") or ""
            if isinstance(content, list):
                parts = []
                for p in content:
                    if isinstance(p, dict):
                        txt = p.get("text") or p.get("input_text") or ""
                        if txt:
                            parts.append({"type": "input_text", "text": str(txt)})
                        # Best-effort image passthrough
                        elif p.get("type") in ("image_url", "input_image"):
                            url = (
                                (p.get("image_url") or {}).get("url")
                                if isinstance(p.get("image_url"), dict)
                                else p.get("image_url") or p.get("url")
                            )
                            if url:
                                parts.append({"type": "input_image", "image_url": url})
                    elif p:
                        parts.append({"type": "input_text", "text": str(p)})
                if not parts:
                    parts = [{"type": "input_text", "text": ""}]
            else:
                parts = [{"type": "input_text", "text": str(content)}]
            input_items.append({"type": "message", "role": "user", "content": parts})
            continue

        if role == "assistant":
            content = m.get("content") or ""
            if isinstance(content, list):
                text = "".join(
                    (p.get("text") or "") if isinstance(p, dict) else str(p)
                    for p in content
                )
            else:
                text = str(content) if content is not None else ""
            if text:
                input_items.append({
                    "type": "message",
                    "role": "assistant",
                    "content": [{"type": "output_text", "text": text}],
                })
            for tc in (m.get("tool_calls") or []):
                fn = tc.get("function") or {}
                input_items.append({
                    "type": "function_call",
                    "call_id": tc.get("id") or "",
                    "name": fn.get("name") or tc.get("name") or "",
                    "arguments": fn.get("arguments") or tc.get("arguments") or "{}",
                })
            continue

        if role == "tool":
            input_items.append({
                "type": "function_call_output",
                "call_id": m.get("tool_call_id") or "",
                "output": str(m.get("content") or ""),
            })
            continue

        # Unknown role → treat as user text
        input_items.append({
            "type": "message",
            "role": "user",
            "content": [{"type": "input_text", "text": str(m.get("content") or "")}],
        })

    instructions = "\n\n".join(p for p in instruction_parts if p)
    return instructions, input_items


def _chatgpt_tools_to_responses(tool_schemas: list) -> list:
    """Convert Anthropic-style tool schemas → Responses API function tools."""
    out = []
    for t in tools_to_openai(tool_schemas or []):
        fn = t.get("function") or {}
        name = fn.get("name") or t.get("name")
        if not name:
            continue
        out.append({
            "type": "function",
            "name": name,
            "description": fn.get("description") or t.get("description") or "",
            "parameters": fn.get("parameters") or {"type": "object", "properties": {}},
        })
    return out


def stream_chatgpt_oauth(
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
) -> Generator:
    """Stream from ChatGPT/Codex subscription via the Responses API backend.

    Requires a prior `/login chatgpt` (or an existing `codex login` session in
    ~/.codex/auth.json). Yields TextChunk / ThinkingChunk / AssistantTurn.
    """
    import requests as _req

    session = _chatgpt_oauth_get_session(config)
    if not session.get("access_token"):
        yield TextChunk(
            "[chatgpt] No ChatGPT/Codex OAuth session. Run `/login chatgpt` "
            "(or `codex login`) to sign in with your ChatGPT account — no API key."
        )
        yield AssistantTurn(
            "[chatgpt] No ChatGPT/Codex OAuth session. Run `/login chatgpt`.",
            [], 0, 0, error=True,
        )
        return

    wire_model, effort_from_suffix = _chatgpt_parse_reasoning_suffix(model)
    instructions, input_items = _chatgpt_messages_to_responses_input(system, messages)

    body: dict[str, Any] = {
        "model": wire_model,
        "instructions": instructions,
        "input": input_items,
        "stream": True,
        "store": False,
    }

    # Prompt caching: OpenAI auto-caches the prompt prefix (>1024 tokens), but
    # routes cache lookups by `prompt_cache_key`. Without a stable key, turns of
    # the same conversation can hit different cache nodes and miss. Key off the
    # system prompt (stable across a session) + account so every turn of this
    # chat lands on the same cached prefix. Cheap, and turns cache hits from
    # "sometimes" into "reliably".
    try:
        import hashlib as _hl
        _seed = (instructions or "") + "|" + (session.get("account_id") or "")
        body["prompt_cache_key"] = "dulus-" + _hl.sha256(_seed.encode("utf-8")).hexdigest()[:32]
    except Exception:
        pass

    # Reasoning effort: config override → model suffix → omit (server default).
    effort = (
        (config or {}).get("chatgpt_reasoning_effort")
        or (config or {}).get("reasoning_effort")
        or effort_from_suffix
    )
    if effort:
        body["reasoning"] = {"effort": str(effort)}

    tools = _chatgpt_tools_to_responses(tool_schemas)
    if tools:
        body["tools"] = tools
        if not (config or {}).get("disable_tool_choice"):
            body["tool_choice"] = "auto"

    url = CHATGPT_OAUTH_BASE_URL.rstrip("/") + "/responses"
    headers = _chatgpt_oauth_headers(session)

    text = ""
    thinking = ""
    in_tok = 0
    out_tok = 0
    cached_tok = 0  # prompt-cached input tokens reported by the Responses API
    # tool_buf keyed by call_id (or synthetic index when id arrives late)
    tool_buf: dict[str, dict] = {}
    tool_order: list[str] = []
    current_call_id: str | None = None
    stream_interrupted = False
    auth_retried = False
    resp = None

    def _ensure_tool(call_id: str | None, name: str | None = None) -> str:
        nonlocal current_call_id
        cid = call_id or current_call_id or f"call_{len(tool_order)}"
        if cid not in tool_buf:
            tool_buf[cid] = {"id": cid, "name": name or "", "args": ""}
            tool_order.append(cid)
        elif name and not tool_buf[cid].get("name"):
            tool_buf[cid]["name"] = name
        current_call_id = cid
        return cid

    def _do_request(hdrs: dict):
        return _req.post(url, headers=hdrs, json=body, stream=True, timeout=300)

    try:
        resp = _do_request(headers)
        # One silent refresh+retry on 401
        if resp.status_code == 401 and session.get("refresh_token") and not auth_retried:
            auth_retried = True
            refreshed = _chatgpt_oauth_refresh(session["refresh_token"], session)
            if refreshed and refreshed.get("access_token"):
                session = refreshed
                headers = _chatgpt_oauth_headers(session)
                try:
                    resp.close()
                except Exception:
                    pass
                resp = _do_request(headers)

        if resp.status_code == 401:
            yield TextChunk(
                "[chatgpt] Auth error 401. Run `/login chatgpt force` to re-authenticate."
            )
            yield AssistantTurn(
                "[chatgpt] Auth error 401. Run `/login chatgpt force`.",
                [], 0, 0, error=True,
            )
            return
        if resp.status_code == 402:
            body_preview = (resp.text or "")[:300]
            yield TextChunk(
                "[chatgpt] HTTP 402 — ChatGPT workspace inactive / plan required. "
                f"Check your ChatGPT subscription. Detail: {body_preview}"
            )
            yield AssistantTurn(
                "[chatgpt] HTTP 402 — ChatGPT workspace inactive / plan required.",
                [], 0, 0, error=True,
            )
            return
        if resp.status_code >= 400:
            body_preview = (resp.text or "")[:400]
            yield TextChunk(f"[chatgpt] HTTP {resp.status_code}: {body_preview}")
            yield AssistantTurn(
                f"[chatgpt] HTTP {resp.status_code}: {body_preview}",
                [], 0, 0, error=True,
            )
            return

        event_type = ""
        for raw_line in resp.iter_lines(decode_unicode=True):
            if raw_line is None:
                continue
            line = raw_line.strip() if isinstance(raw_line, str) else raw_line.decode("utf-8", "replace").strip()
            if not line:
                continue
            if line.startswith(":"):
                continue
            if line.startswith("event:"):
                event_type = line[6:].strip()
                continue
            if not line.startswith("data:"):
                continue
            data = line[5:].strip()
            if data == "[DONE]":
                break
            try:
                event = json.loads(data)
            except Exception:
                continue
            if not isinstance(event, dict):
                continue
            et = event.get("type") or event_type or ""

            if et in ("response.output_text.delta", "response.refusal.delta"):
                delta = event.get("delta") or ""
                if delta:
                    text += delta
                    yield TextChunk(delta)
                continue

            if et in (
                "response.reasoning_summary_text.delta",
                "response.reasoning_text.delta",
            ):
                delta = event.get("delta") or ""
                if delta:
                    thinking += delta
                    yield ThinkingChunk(delta)
                continue

            if et == "response.output_item.added":
                item = event.get("item") or {}
                if isinstance(item, dict) and item.get("type") == "function_call":
                    _ensure_tool(item.get("call_id") or item.get("id"), item.get("name"))
                continue

            if et == "response.function_call_arguments.delta":
                cid = event.get("call_id") or current_call_id
                key = _ensure_tool(cid)
                delta_args = event.get("delta") or ""
                if delta_args:
                    tool_buf[key]["args"] += delta_args
                continue

            if et == "response.function_call_arguments.done":
                cid = event.get("call_id") or current_call_id
                key = _ensure_tool(cid)
                final_args = event.get("arguments")
                if final_args is not None:
                    tool_buf[key]["args"] = final_args
                continue

            if et == "response.output_item.done":
                item = event.get("item") or {}
                if isinstance(item, dict) and item.get("type") == "function_call":
                    key = _ensure_tool(item.get("call_id") or item.get("id"), item.get("name"))
                    if item.get("name"):
                        tool_buf[key]["name"] = item["name"]
                    if item.get("arguments") is not None:
                        tool_buf[key]["args"] = item["arguments"]
                    if item.get("call_id"):
                        tool_buf[key]["id"] = item["call_id"]
                continue

            if et in ("response.completed", "response.incomplete"):
                resp_obj = event.get("response") or {}
                usage = resp_obj.get("usage") or event.get("usage") or {}
                if isinstance(usage, dict):
                    in_tok = int(usage.get("input_tokens") or usage.get("prompt_tokens") or in_tok or 0)
                    out_tok = int(usage.get("output_tokens") or usage.get("completion_tokens") or out_tok or 0)
                    # Responses API reports the cached slice under
                    # input_tokens_details.cached_tokens (prompt_tokens_details on
                    # the Chat-Completions shape). Surfacing it lets the toolbar
                    # credit cache hits instead of billing full input every turn.
                    _det = usage.get("input_tokens_details") or usage.get("prompt_tokens_details") or {}
                    if isinstance(_det, dict):
                        cached_tok = int(_det.get("cached_tokens") or cached_tok or 0)
                continue

            if et in ("response.failed", "error"):
                err = event.get("error") or event.get("response", {}).get("error") or event
                msg = ""
                if isinstance(err, dict):
                    msg = err.get("message") or err.get("code") or json.dumps(err)[:300]
                else:
                    msg = str(err)[:300]
                note = f"\n\n_[chatgpt error: {msg}]_"
                text += note
                yield TextChunk(note)
                continue

    except _req.exceptions.ChunkedEncodingError:
        stream_interrupted = True
    except _req.exceptions.ConnectionError:
        stream_interrupted = True
    except Exception as e:
        msg = f"[chatgpt] Error: {e}"
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return
    finally:
        if resp is not None:
            try:
                resp.close()
            except Exception:
                pass

    if stream_interrupted and not tool_buf:
        note = "\n\n_[chatgpt: el streaming se cortó a media respuesta — esto es parcial, reintenta]_"
        text += note
        yield TextChunk(note)

    # Build final tool_calls list in arrival order
    ordered_buf = {i: tool_buf[k] for i, k in enumerate(tool_order)} if tool_order else tool_buf
    final_tool_calls = _finalize_tool_calls(ordered_buf)

    yield AssistantTurn(text, final_tool_calls, in_tok, out_tok, thinking=thinking,
                        cache_read_tokens=cached_tok)


# ── Meta AI (Muse Spark) — API key, Responses-API-style SSE ────────────────
META_API_BASE_URL = "https://api.meta.ai/v1"


def stream_meta(
    api_key: str,
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
) -> Generator:
    """Meta AI Muse Spark via api.meta.ai/v1/responses (SSE, OpenAI Responses
    wire format). Key: /config meta_api_key=... or META_API_KEY.
    Yields TextChunk / ThinkingChunk / AssistantTurn."""
    import requests as _req

    if not api_key:
        yield TextChunk(
            "[meta] No API key. Set one with `/config meta_api_key=...` "
            "or export META_API_KEY."
        )
        yield AssistantTurn(
            "[meta] No API key. Set META_API_KEY or /config meta_api_key=...",
            [], 0, 0, error=True,
        )
        return

    instructions, input_items = _chatgpt_messages_to_responses_input(system, messages)

    body: dict[str, Any] = {
        "model": model,
        "input": input_items,
        "stream": True,
        "temperature": 1,
        "max_output_tokens": 32000,
        "top_p": 1,
    }
    if instructions:
        body["instructions"] = instructions

    tools = _chatgpt_tools_to_responses(tool_schemas)
    if tools:
        body["tools"] = tools
        if not (config or {}).get("disable_tool_choice"):
            body["tool_choice"] = "auto"

    base = (PROVIDERS.get("meta", {}).get("base_url") or META_API_BASE_URL).rstrip("/")
    url = base + "/responses"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
        "Accept": "text/event-stream",
    }

    text = ""
    thinking = ""
    in_tok = 0
    out_tok = 0
    cached_tok = 0
    tool_buf: dict[str, dict] = {}
    tool_order: list[str] = []
    current_call_id: str | None = None
    stream_interrupted = False
    resp = None

    def _ensure_tool(call_id: str | None, name: str | None = None) -> str:
        nonlocal current_call_id
        cid = call_id or current_call_id or f"call_{len(tool_order)}"
        if cid not in tool_buf:
            tool_buf[cid] = {"id": cid, "name": name or "", "args": ""}
            tool_order.append(cid)
        elif name and not tool_buf[cid].get("name"):
            tool_buf[cid]["name"] = name
        current_call_id = cid
        return cid

    try:
        resp = _req.post(url, headers=headers, json=body, stream=True, timeout=300)

        if resp.status_code in (401, 403):
            body_preview = (resp.text or "")[:300]
            yield TextChunk(
                f"[meta] Auth error {resp.status_code}. Check your key: "
                f"/config meta_api_key=... Detail: {body_preview}"
            )
            yield AssistantTurn(
                f"[meta] Auth error {resp.status_code}. Check meta_api_key.",
                [], 0, 0, error=True,
            )
            return
        if resp.status_code >= 400:
            body_preview = (resp.text or "")[:400]
            yield TextChunk(f"[meta] HTTP {resp.status_code}: {body_preview}")
            yield AssistantTurn(
                f"[meta] HTTP {resp.status_code}: {body_preview}",
                [], 0, 0, error=True,
            )
            return

        event_type = ""
        # NOTE: iterate RAW BYTES (no decode_unicode). requests defaults to
        # ISO-8859-1 for text/event-stream when the server omits charset, which
        # mangled UTF-8 accents (á→á, ñ→ñ) in Spanish output. Decoding each
        # line explicitly as UTF-8 fixes the mojibake.
        for raw_line in resp.iter_lines():
            if raw_line is None:
                continue
            line = raw_line.decode("utf-8", "replace").strip() if isinstance(raw_line, (bytes, bytearray)) else raw_line.strip()
            if not line:
                continue
            if line.startswith(":"):
                continue
            if line.startswith("event:"):
                event_type = line[6:].strip()
                continue
            if not line.startswith("data:"):
                continue
            data = line[5:].strip()
            if data == "[DONE]":
                break
            try:
                event = json.loads(data)
            except Exception:
                continue
            if not isinstance(event, dict):
                continue
            et = event.get("type") or event_type or ""

            if et in ("response.output_text.delta", "response.refusal.delta"):
                delta = event.get("delta") or ""
                if delta:
                    text += delta
                    yield TextChunk(delta)
                continue

            if et in (
                "response.reasoning_summary_text.delta",
                "response.reasoning_text.delta",
            ):
                delta = event.get("delta") or ""
                if delta:
                    thinking += delta
                    yield ThinkingChunk(delta)
                continue

            if et == "response.output_item.added":
                item = event.get("item") or {}
                if isinstance(item, dict) and item.get("type") == "function_call":
                    _ensure_tool(item.get("call_id") or item.get("id"), item.get("name"))
                continue

            if et == "response.function_call_arguments.delta":
                cid = event.get("call_id") or current_call_id
                key = _ensure_tool(cid)
                delta_args = event.get("delta") or ""
                if delta_args:
                    tool_buf[key]["args"] += delta_args
                continue

            if et == "response.function_call_arguments.done":
                cid = event.get("call_id") or current_call_id
                key = _ensure_tool(cid)
                final_args = event.get("arguments")
                if final_args is not None:
                    tool_buf[key]["args"] = final_args
                continue

            if et == "response.output_item.done":
                item = event.get("item") or {}
                if isinstance(item, dict) and item.get("type") == "function_call":
                    key = _ensure_tool(item.get("call_id") or item.get("id"), item.get("name"))
                    if item.get("name"):
                        tool_buf[key]["name"] = item["name"]
                    if item.get("arguments") is not None:
                        tool_buf[key]["args"] = item["arguments"]
                    if item.get("call_id"):
                        tool_buf[key]["id"] = item["call_id"]
                continue

            if et in ("response.completed", "response.incomplete"):
                resp_obj = event.get("response") or {}
                usage = resp_obj.get("usage") or event.get("usage") or {}
                if isinstance(usage, dict):
                    in_tok = int(usage.get("input_tokens") or usage.get("prompt_tokens") or in_tok or 0)
                    out_tok = int(usage.get("output_tokens") or usage.get("completion_tokens") or out_tok or 0)
                    _det = usage.get("input_tokens_details") or usage.get("prompt_tokens_details") or {}
                    if isinstance(_det, dict):
                        cached_tok = int(_det.get("cached_tokens") or cached_tok or 0)
                continue

            if et in ("response.failed", "error"):
                err = event.get("error") or event.get("response", {}).get("error") or event
                if isinstance(err, dict):
                    msg = err.get("message") or err.get("code") or json.dumps(err)[:300]
                else:
                    msg = str(err)[:300]
                note = f"\n\n_[meta error: {msg}]_"
                text += note
                yield TextChunk(note)
                continue

    except _req.exceptions.ChunkedEncodingError:
        stream_interrupted = True
    except _req.exceptions.ConnectionError:
        stream_interrupted = True
    except Exception as e:
        msg = f"[meta] Error: {e}"
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return
    finally:
        if resp is not None:
            try:
                resp.close()
            except Exception:
                pass

    if stream_interrupted and not tool_buf:
        note = "\n\n_[meta: el streaming se cortó a media respuesta — esto es parcial, reintenta]_"
        text += note
        yield TextChunk(note)

    ordered_buf = {i: tool_buf[k] for i, k in enumerate(tool_order)} if tool_order else tool_buf
    final_tool_calls = _finalize_tool_calls(ordered_buf)

    yield AssistantTurn(text, final_tool_calls, in_tok, out_tok, thinking=thinking,
                        cache_read_tokens=cached_tok)


def stream_xai_oauth(
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
    _auth_path: str | None = None,  # ignored: we only use official Grok TUI (~/.grok/auth.json) or XAI_API_KEY
) -> Generator:
    """Stream from xAI / Grok using the official Grok Build TUI session (preferred)
    or a direct XAI_API_KEY.

    The old Playwright browser harvest for Grok has been removed for cleanliness.
    We source the token exclusively via _load_grok_build_session_token() + XAI_API_KEY.
    """
    import requests as _req

    # Now that Dulus uses the official Grok Build TUI session (grok login),
    # we get the raw token (JWT / access token) from ~/.grok/auth.json.
    # XAI_API_KEY (env or config) is the fallback and is also expected raw.
    # We normalize to always send a proper "Bearer <token>" header.
    token = _xai_oauth_get_token(config) or ""

    if not token:
        yield TextChunk("[xai/grok] No official Grok Build TUI session found and no XAI_API_KEY.\n"
                        "Run `grok login` (recommended) or set XAI_API_KEY.")
        return

    # Clean session + headers for official Grok Build TUI session token
    # or direct XAI_API_KEY fallback.
    # The old Playwright harvest flow for xAI/Grok has been removed.
    session = _req.Session()

    # Present as the official Grok CLI client. We're reusing the Grok Build TUI
    # session token (scope grok-cli:access) even when the binary isn't running,
    # so the headers must match what xAI's backend expects from its first-party
    # client — otherwise some accounts get throttled/blocked. (OpenClaw learned
    # this the hard way: spoof the real client headers and the API behaves.)
    headers = {
        "Content-Type": "application/json",
        "Accept": "text/event-stream",
        "User-Agent": "grok-cli/1.0.0",
        "x-client-name": "grok-cli",
        "x-client-platform": "cli",
        "Accept-Language": "en-US,en;q=0.9",
    }

    # Always produce a single proper "Bearer <token>" (handles raw tokens from
    # the TUI auth.json or XAI_API_KEY, and tolerates if someone put "Bearer ..."
    # in the env var by accident).
    t = str(token).strip()
    if not t.lower().startswith("bearer "):
        t = f"Bearer {t}"
    headers["Authorization"] = t

    # --- Dulus environment awareness (stronger for native Grok) ---
    # We send real structured tools=, but we also inject explicit context so the model
    # strongly understands it is inside Dulus and that tools are executed for real by the harness.
    # This is the native-API equivalent of the raw text manifest injection you used for web models.
    effective_system = system or ""
    if tool_schemas and not config.get("no_tools"):
        dulus_note = (
            "\n\n[ YOU ARE RUNNING INSIDE DULUS AGENT HARNESS ]\n"
            "You are the LLM backend for Dulus, a powerful autonomous agent.\n"
            "The Dulus runtime will actually execute any tool calls you make using the provided function definitions.\n"
            "Output tool calls in the standard function calling format.\n"
            "Tool results will be returned to you in subsequent messages with role \"tool\" (with the tool_call_id).\n"
            "Use tools aggressively when they help. After tools return, continue until you can give a complete final answer.\n"
            "Never simulate tool results yourself — always call the tool."
        )
        effective_system = (system or "") + dulus_note

    oai_messages = [{"role": "system", "content": effective_system}] + messages_to_openai(messages)
    payload = {
        "model": model,
        "messages": oai_messages,
        "stream": True,
    }
    if tool_schemas and not config.get("no_tools"):
        payload["tools"] = tools_to_openai(tool_schemas)

    # reasoning_effort: ONLY the grok-3-mini family accepts it, and xAI only
    # supports two values here ("low" | "high") — there is no "medium". So we
    # fold the harness-wide 0-4 thinking level into those two buckets:
    #   level >= 3  -> "high"   (max/raw and the True default both land here)
    #   level 1-2   -> "low"
    #   level 0     -> "low"    (mini always reasons a little; can't fully disable)
    # grok-4 reasons by default and 400s if you send this param, so we skip it there.
    if "mini" in model.lower():
        _lvl = _thinking_level_from(config.get("thinking", 0))
        payload["reasoning_effort"] = "high" if _lvl >= 3 else "low"

    # Ask for usage in the stream (helps /cost and visibility)
    payload["stream_options"] = {"include_usage": True}

    url = f"{XAI_OAUTH_BASE_URL}/chat/completions"
    try:
        # timeout=(connect, read): read is the gap allowed *between* streamed chunks.
        # grok-4 reasons internally and can pause well past 120s, which would raise
        # ReadTimeout mid-stream and truncate the reply — give it a generous window.
        resp = session.post(url, headers=headers, json=payload, stream=True, timeout=(15, 180))
        # On 401, the native-OAuth access token likely expired mid-session. Try a
        # silent refresh with the stored refresh_token and retry the request once.
        if resp.status_code == 401:
            _store = _xai_oauth_load_store()
            if _store.get("refresh_token"):
                _refreshed = _xai_oauth_refresh(_store["refresh_token"])
                if _refreshed and _refreshed.get("access_token"):
                    headers["Authorization"] = f"Bearer {_refreshed['access_token']}"
                    resp = session.post(url, headers=headers, json=payload, stream=True, timeout=(15, 180))
        if resp.status_code != 200:
            body_preview = ""
            try:
                body_preview = (resp.text or "")[:500]
            except Exception:
                pass
            if resp.status_code in (401, 403):
                yield TextChunk(f"[xai/grok] Auth error {resp.status_code}. Run `/login grok` to re-authenticate "
                                f"(native OAuth), or `grok login` for the official CLI. Body: {body_preview}")
            else:
                yield TextChunk(f"[xai-oauth] HTTP {resp.status_code}: {body_preview}")
            return
    except Exception as e:
        yield TextChunk(f"[xai-oauth] Error: {e}")
        return

    # Proper native OpenAI tool_calls handling (critical for grok-* via real API).
    # Previously this only did WebToolParser on content deltas → when Grok emitted
    # structured tool_calls the stream would finish with no tool_calls returned,
    # making the agent think the model didn't want to use tools → "no termina de responder".
    text = ""
    thinking = ""
    tool_buf = {}
    in_tok = out_tok = 0
    parser = WebToolParser()  # fallback: catch any <tool_call> XML the model might emit in content

    stream_interrupted = False

    def _iter_stream(r):
        # Guard iter_lines so a mid-stream drop (read timeout, connection reset,
        # chunked-encoding error) ends the loop cleanly instead of bubbling up and
        # killing the whole turn — otherwise everything Grok already streamed (text
        # AND buffered tool_calls) would be lost and the agent gets an empty reply.
        nonlocal stream_interrupted
        try:
            for _ln in r.iter_lines():
                yield _ln
        except Exception:
            stream_interrupted = True

    for line in _iter_stream(resp):
        if not line:
            continue
        line = line.decode("utf-8") if isinstance(line, bytes) else line
        if line.startswith("data: "):
            data_str = line[6:]
            if data_str.strip() == "[DONE]":
                break
            try:
                data = json.loads(data_str)
                delta = data.get("choices", [{}])[0].get("delta", {})

                # Normal text content
                content = delta.get("content") or ""
                if content:
                    text += content
                    display = parser.parse_chunk(content)
                    if display:
                        yield TextChunk(display)

                # Native tool_calls (OpenAI format) — this is what real Grok returns
                # when we send "tools": [...] in the payload.
                for tc in (delta.get("tool_calls") or []):
                    idx = tc.get("index", 0)
                    if idx not in tool_buf:
                        tool_buf[idx] = {"id": "", "name": "", "args": ""}
                    if tc.get("id"):
                        tool_buf[idx]["id"] = tc["id"]
                    fn = tc.get("function", {}) or {}
                    if fn.get("name"):
                        tool_buf[idx]["name"] += fn["name"]
                    if fn.get("arguments"):
                        tool_buf[idx]["args"] += fn["arguments"]

                # Some Grok variants stream reasoning/thinking (grok-3-mini family
                # returns reasoning_content; grok-4 reasons internally and hides it).
                # Show it when it arrives — every other provider here yields it too.
                reasoning = delta.get("reasoning_content") or delta.get("reasoning")
                if reasoning:
                    thinking += reasoning
                    yield ThinkingChunk(reasoning)

                # Usage (comes especially when we set stream_options include_usage)
                usage = data.get("usage") or {}
                if usage:
                    in_tok = usage.get("prompt_tokens", in_tok) or in_tok
                    out_tok = usage.get("completion_tokens", out_tok) or out_tok

            except Exception:
                continue

    # Flush any leftover XML-style tool calls that might have been in content
    remaining = parser.flush()
    if remaining:
        text += remaining
        yield TextChunk(remaining)

    # If the stream dropped mid-response and the model was not in the middle of a
    # tool call, mark the reply as partial so a cut answer isn't mistaken for a
    # complete one. We still yield everything received below.
    if stream_interrupted and not tool_buf:
        note = "\n\n_[grok: el streaming se cortó a media respuesta — esto es parcial, reintenta]_"
        text += note
        yield TextChunk(note)

    final_tool_calls = _finalize_tool_calls(tool_buf)

    # Belt-and-suspenders: merge anything the XML parser caught too
    if getattr(parser, "tool_calls", None):
        final_tool_calls.extend(parser.tool_calls)

    yield AssistantTurn(text, final_tool_calls, in_tok, out_tok, thinking=thinking)


















def _yield_web_parsed(parser: WebToolParser, raw_content: str):
    """Parse accumulated raw content and yield any resulting text chunk."""
    if raw_content:
        text = parser.parse_chunk(raw_content)
        text += parser.flush()
        if text:
            yield TextChunk(text)


def _yield_final_turn(text: str, parser: WebToolParser, update_text: bool = False):
    """Flush parser, yield remaining text, and emit final AssistantTurn."""
    remaining = parser.flush()
    if remaining:
        if update_text:
            text += remaining
        yield TextChunk(remaining)
    yield AssistantTurn(text, parser.tool_calls, 0, 0)


def _iter_jsonl(path):
    """Yield parsed JSON objects from a JSONL file, skipping invalid lines."""
    try:
        with open(path, "r", encoding="utf-8", errors="ignore") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    yield json.loads(line)
                except Exception:
                    continue
    except Exception:
        return
















def bare_model(model: str) -> str:
    """Strip 'provider/' prefix if present."""
    return model.split("/", 1)[1] if "/" in model else model


def get_api_key(provider_name: str, config: dict) -> str:
    # moonshot is an alias for kimi (same API backend and credentials)
    if provider_name == "moonshot":
        provider_name = "kimi"
    prov = PROVIDERS.get(provider_name, {})
    # 1. Check config dict (e.g. config["kimi_api_key"])
    cfg_key = config.get(f"{provider_name}_api_key", "")
    if cfg_key:
        return cfg_key
    
    # Alias fallback: moonshot <-> kimi
    if provider_name == "moonshot":
        cfg_key = config.get("kimi_api_key", "")
        if cfg_key: return cfg_key
    elif provider_name == "kimi":
        cfg_key = config.get("moonshot_api_key", "")
        if cfg_key: return cfg_key
    elif provider_name == "kimi-code":
        # Try multiple API keys in order of preference (fallback chain)
        for key_name in ("kimi_code_api_key", "kimi_code2_api_key", "kimi_code3_api_key"):
            cfg_key = config.get(key_name, "")
            if cfg_key: return cfg_key

    # 2. Check env var
    env_var = prov.get("api_key_env")
    if env_var:
        import os
        return os.environ.get(env_var, "")
    # 3. Hardcoded (for local providers)
    return prov.get("api_key", "")


def calc_cost(model: str, in_tok: int, out_tok: int) -> float:
    ic, oc = COSTS.get(bare_model(model), (0.0, 0.0))
    return (in_tok * ic + out_tok * oc) / 1_000_000


# ── Native tool-call format interceptors ──────────────────────────────────
# Some models (Gemma 3/4, Mistral, ...) emit their NATIVE tool-call format
# inside `delta.content` even when the API has been told to use OpenAI-style
# tool schemas. Without interception the user sees raw markers like
# `<|tool_call>call:Foo{"x":1}<tool_call|>` streamed as text, and the
# intended tool call never fires — and on Ollama Cloud / vLLM the broken
# format can also trip a 502 from the upstream proxy. The helpers below let
# stream_ollama / stream_openai_compat detect these markers, switch into
# buffer mode, and parse the buffered tail into proper tool_calls.
_NATIVE_TOOL_OPENERS = (
    "<|tool_call|>",   # Gemma official
    "<|tool_call>",    # Gemma 4 asymmetric variant seen in the wild
    "<tool_call>",     # Hermes / Qwen
    "[TOOL_CALLS]",    # Mistral
)

_GEMMA_QUOTE_TOKEN_FIXES = (
    ("<|\"|>", '"'),
    ("<|'|>", "'"),
)

_NATIVE_FMT_V2 = re.compile(
    r"<\|?tool_call\|?>\s*call:\s*(\w+)\s*(\{.*?\})\s*<\|?(?:end_)?(?:/)?tool_call\|?>",
    re.DOTALL,
)
_NATIVE_FMT_V1 = re.compile(
    r"<\|?tool_call\|?>\s*(\{.*?\})\s*<\|?(?:end_)?(?:/)?tool_call\|?>",
    re.DOTALL,
)
_NATIVE_FMT_MISTRAL = re.compile(r"\[TOOL_CALLS\]\s*(\[.*?\])", re.DOTALL)


def _find_native_tool_marker(text: str) -> "int | None":
    earliest = None
    for opener in _NATIVE_TOOL_OPENERS:
        idx = text.find(opener)
        if idx != -1 and (earliest is None or idx < earliest):
            earliest = idx
    return earliest


def _extract_native_tool_calls(buf: str) -> list:
    """Parse buffered native-format tool calls. Returns [] on any failure."""
    if not buf:
        return []
    for tok, repl in _GEMMA_QUOTE_TOKEN_FIXES:
        buf = buf.replace(tok, repl)

    out: list = []

    # Format 2 first (more specific — explicit `call:NAME` outside the JSON)
    for m in _NATIVE_FMT_V2.finditer(buf):
        name, body = m.group(1), m.group(2)
        try:
            args = json.loads(body)
            if not isinstance(args, dict):
                args = {"_raw": body}
        except json.JSONDecodeError:
            args = {"_raw": body}
        out.append({"id": f"native_call_{len(out)}", "name": name, "input": args})

    # Format 1: JSON envelope with `name` + `arguments`
    if not out:
        for m in _NATIVE_FMT_V1.finditer(buf):
            try:
                parsed = json.loads(m.group(1))
                if isinstance(parsed, dict):
                    name = parsed.get("name") or parsed.get("function") or ""
                    args = parsed.get("arguments") or parsed.get("args") or {}
                    if name:
                        if not isinstance(args, dict):
                            args = {"_raw": str(args)}
                        out.append({
                            "id": f"native_call_{len(out)}",
                            "name": name, "input": args,
                        })
            except json.JSONDecodeError:
                continue

    # Mistral [TOOL_CALLS] [{...}, {...}]
    if not out:
        for m in _NATIVE_FMT_MISTRAL.finditer(buf):
            try:
                arr = json.loads(m.group(1))
                if isinstance(arr, list):
                    for item in arr:
                        if not isinstance(item, dict):
                            continue
                        name = item.get("name") or (item.get("function") or {}).get("name") or ""
                        args = item.get("arguments") or (item.get("function") or {}).get("arguments") or {}
                        if name:
                            if not isinstance(args, dict):
                                args = {"_raw": str(args)}
                            out.append({
                                "id": f"native_call_{len(out)}",
                                "name": name, "input": args,
                            })
            except json.JSONDecodeError:
                continue

    return out


def estimate_tokens_kimi(api_key: str, model: str, messages: list) -> int | None:
    """Estimate token count using Kimi's native API endpoint.
    
    Args:
        api_key: Moonshot API key
        model: Model name (e.g., "kimi-k2.5")
        messages: List of message dicts with "role" and "content"
    Returns:
        Estimated token count, or None if the request fails
    """
    if not api_key:
        return None
    
    url = "https://api.moonshot.ai/v1/tokenizers/estimate-token-count"
    
    # Convert messages to Kimi format (similar to OpenAI format)
    kimi_messages = []
    for m in messages:
        role = m.get("role", "user")
        content = m.get("content", "")
        if isinstance(content, str):
            kimi_messages.append({"role": role, "content": content})
        elif isinstance(content, list):
            # Multimodal content - extract text parts
            text_parts = []
            for part in content:
                if isinstance(part, dict) and part.get("type") == "text":
                    text_parts.append(part.get("text", ""))
            if text_parts:
                kimi_messages.append({"role": role, "content": " ".join(text_parts)})
    
    payload = {
        "model": model,
        "messages": kimi_messages
    }
    
    try:
        req = urllib.request.Request(
            url,
            data=json.dumps(payload).encode("utf-8"),
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {api_key}"
            }
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            data = json.loads(resp.read().decode("utf-8"))
            # Response: {"data": {"total_tokens": 123}}
            if "data" in data and "total_tokens" in data["data"]:
                return data["data"]["total_tokens"]
            return None
    except Exception:
        # Silently fail - caller will fall back to character-based estimation
        return None


# ── Tool schema conversion ─────────────────────────────────────────────────

def scrub_any_type(obj: Any) -> Any:
    """Recursively remove 'type': 'any' from schema dictionaries as it's not valid JSON Schema."""
    if isinstance(obj, dict):
        new_obj = {}
        for k, v in obj.items():
            if k == "type" and v == "any":
                continue
            new_obj[k] = scrub_any_type(v)
        return new_obj
    elif isinstance(obj, list):
        return [scrub_any_type(item) for item in obj]
    return obj


def coerce_type_arrays(obj: Any) -> Any:
    """Coerce JSON-Schema ``"type": [...]`` arrays to a single string type.

    Optional/Union Python hints adapt to schemas like ``"type": ["string", "null"]``.
    Anthropic/OpenAI tolerate that, but Moonshot/Kimi rejects type arrays outright
    ("invalid type in type array"), which 400s every request once such a tool is
    loaded. We pick the first non-"null" concrete type (falling back to "string")
    so an auto-adapted plugin can never brick the whole session.
    """
    if isinstance(obj, dict):
        new_obj = {}
        for k, v in obj.items():
            if k == "type" and isinstance(v, list):
                concrete = [t for t in v if isinstance(t, str) and t != "null"]
                new_obj[k] = concrete[0] if concrete else "string"
            else:
                new_obj[k] = coerce_type_arrays(v)
        return new_obj
    elif isinstance(obj, list):
        return [coerce_type_arrays(item) for item in obj]
    return obj


def tools_to_openai(tool_schemas: list) -> list:
    """Convert Anthropic-style tool schemas to OpenAI function-calling format."""
    out = []
    for t in tool_schemas:
        if not isinstance(t, dict) or "name" not in t:
            continue
        
        # Handle different schema names (Anthropic input_schema vs OpenAI parameters)
        params = t.get("input_schema") or t.get("parameters")
        if params is None:
            # Fallback to empty object if missing, better than crashing
            params = {"type": "object", "properties": {}}
        elif isinstance(params, str):
            # Auto-adapted plugins / external skills sometimes serialize the
            # JSON-Schema parameters as a JSON string. OpenAI/Anthropic tolerate
            # it in places, but Perplexity (and other strict OpenAI-compatible
            # backends) reject it with "Tool parameters must be a JSON object".
            try:
                params = json.loads(params)
            except Exception:
                params = {"type": "object", "properties": {}}
        
        # Scrub invalid 'any' types that some models hallucinate
        params = scrub_any_type(params)
        # Coerce ["string","null"]-style type arrays to a single type — Moonshot/Kimi
        # rejects type arrays and would 400 every request once such a tool loads.
        params = coerce_type_arrays(params)

        out.append({
            "type": "function",
            "function": {
                "name":        t["name"],
                "description": t.get("description", ""),
                "parameters":  params,
            },
        })
    return out


# ── Message format conversion ──────────────────────────────────────────────
#
# Internal "neutral" message format:
#   {"role": "user",      "content": "text"}
#   {"role": "assistant", "content": "text", "tool_calls": [
#       {"id": "...", "name": "...", "input": {...}}
#   ]}
#   {"role": "tool", "tool_call_id": "...", "name": "...", "content": "..."}

def messages_to_anthropic(messages: list) -> list:
    """Convert neutral messages → Anthropic API format.

    Also sanitizes orphan tool_calls — if an assistant message has tool_calls
    but the matching tool responses are missing (e.g. user interrupted mid-call),
    the tool_calls are stripped. Anthropic requires every tool_use block to have
    a corresponding tool_result block in the IMMEDIATELY following message, or
    it 400s the whole request.
    """
    # ── Sanitize orphan tool_calls (position-aware) ───────────────────────
    # Anthropic is stricter than OpenAI: each tool_use must be answered by a
    # tool_result in the IMMEDIATELY next message. A tool result that exists
    # later in history (e.g. a user message got interleaved) doesn't count.
    sanitized = []
    for idx, m in enumerate(messages):
        if m.get("role") == "assistant" and m.get("tool_calls"):
            # Collect ids answered by the consecutive tool messages right after
            answered_here = set()
            j = idx + 1
            while j < len(messages) and messages[j].get("role") == "tool":
                answered_here.add(messages[j].get("tool_call_id"))
                j += 1
            valid_tcs = [tc for tc in m["tool_calls"] if tc.get("id") in answered_here]
            if valid_tcs:
                sanitized.append({**m, "tool_calls": valid_tcs})
            else:
                # All tool_calls are orphans — strip them, keep text content only
                sanitized.append({k: v for k, v in m.items() if k != "tool_calls"}
                                 | {"content": m.get("content") or "(interrupted)"})
        elif m.get("role") == "tool":
            # Drop tool results whose call was stripped (orphan results also 400)
            k = idx - 1
            while k >= 0 and messages[k].get("role") == "tool":
                k -= 1
            owner = messages[k] if k >= 0 else {}
            owner_ids = {tc.get("id") for tc in (owner.get("tool_calls") or [])} if owner.get("role") == "assistant" else set()
            if m.get("tool_call_id") in owner_ids:
                sanitized.append(m)
            # else: orphan tool_result — skip it
        else:
            sanitized.append(m)
    messages = sanitized

    result = []
    # Thinking blocks are only REQUIRED on the last assistant turn (tool-use
    # replay). Older ones are ignored server-side but still travel in the
    # payload — with a 16K thinking budget that bloats every request. Replay
    # thinking ONLY for the final assistant message.
    _last_assistant_idx = -1
    for _j in range(len(messages) - 1, -1, -1):
        if messages[_j].get("role") == "assistant":
            _last_assistant_idx = _j
            break
    i = 0
    while i < len(messages):
        m = messages[i]
        role = m["role"]

        if role == "user":
            imgs = m.get("images")
            if imgs:
                # Anthropic multipart vision: image blocks FIRST, then text
                # (per Anthropic docs recommendation). Without this branch,
                # /image attachments were silently dropped for claude-* models
                # while Ollama/OpenAI paths attached them fine.
                blocks = []
                for img_b64 in imgs:
                    blocks.append({
                        "type": "image",
                        "source": {
                            "type":       "base64",
                            "media_type": "image/png",
                            "data":       img_b64,
                        },
                    })
                blocks.append({"type": "text", "text": m["content"]})
                result.append({"role": "user", "content": blocks})
            else:
                result.append({"role": "user", "content": m["content"]})
            i += 1

        elif role == "assistant":
            blocks = []
            thinking = m.get("thinking", "")
            thinking_sig = m.get("thinking_signature", "")
            # A thinking block can only be replayed WITH its signature — the API
            # rejects one missing the field. On newer Claude models thinking is
            # returned with display="omitted", so the thinking text is empty but
            # the signature still must be passed back. Replay the block whenever
            # we have the signature, using an empty string for the thinking text
            # when the model chose not to return it.
            if thinking_sig and i == _last_assistant_idx:
                blocks.append({
                    "type": "thinking",
                    "thinking": thinking or "",
                    "signature": thinking_sig,
                })
            
            text = m.get("content", "")
            if text:
                blocks.append({"type": "text", "text": text})
            for tc in m.get("tool_calls", []):
                blocks.append({
                    "type":  "tool_use",
                    "id":    tc["id"],
                    "name":  tc["name"],
                    "input": tc["input"],
                })
            result.append({"role": "assistant", "content": blocks})
            i += 1

        elif role == "tool":
            # Collect consecutive tool results into one user message
            tool_blocks = []
            while i < len(messages) and messages[i]["role"] == "tool":
                t = messages[i]
                tool_blocks.append({
                    "type":        "tool_result",
                    "tool_use_id": t["tool_call_id"],
                    "content":     t["content"],
                })
                i += 1
            result.append({"role": "user", "content": tool_blocks})

        else:
            i += 1

    return result


def messages_to_openai(messages: list, ollama_native_images: bool = False, model: str = "") -> list:
    """Convert neutral messages → OpenAI API format.

    Also sanitizes orphan tool_calls — if an assistant message has tool_calls
    but the matching tool responses are missing (e.g. user interrupted mid-call),
    the tool_calls are stripped to avoid API rejection.
    """
    # ── Sanitize orphan tool_calls ────────────────────────────────────────
    # Collect all tool_call_ids that have a matching tool response
    answered_ids = {m.get("tool_call_id") for m in messages if m.get("role") == "tool"}
    sanitized = []
    for m in messages:
        if m.get("role") == "assistant" and m.get("tool_calls"):
            # Keep only tool_calls that have a matching response
            valid_tcs = [tc for tc in m["tool_calls"] if tc.get("id") in answered_ids]
            if valid_tcs:
                sanitized.append({**m, "tool_calls": valid_tcs})
            else:
                # All tool_calls are orphans — strip them, keep text content only
                sanitized.append({"role": "assistant", "content": m.get("content") or "(interrupted)"})
        else:
            sanitized.append(m)
    messages = sanitized

    # Kimi K2.5 / K2.6 accept image_url AND video_url multimodal content, no
    # matter which provider serves them (native Moonshot, Azure deployment, etc.).
    _ml = (model or "").lower()
    _is_kimi_mm = "kimi-k2.5" in _ml or "kimi-k2.6" in _ml

    result = []
    for m in messages:
        role = m["role"]

        if role == "user":
            content = m["content"]
            imgs = m.get("images")
            vids = m.get("videos")
            if ollama_native_images and imgs:
                # Ollama /api/chat native: bare base64 list on the message
                msg_out = {"role": "user", "content": content, "images": imgs}
            elif imgs or (vids and _is_kimi_mm):
                # OpenAI / Gemini / Kimi multipart vision format
                parts = [{"type": "text", "text": content}]
                for img_b64 in (imgs or []):
                    parts.append({
                        "type": "image_url",
                        "image_url": {"url": f"data:image/png;base64,{img_b64}"},
                    })
                # Kimi K2.5/K2.6 also accept video_url as a data URL. Other models
                # would reject it, so videos are attached ONLY for those models.
                if _is_kimi_mm:
                    for vid in (vids or []):
                        if isinstance(vid, dict):
                            v_b64, v_mime = vid.get("data", ""), vid.get("mime", "video/mp4")
                        else:
                            v_b64, v_mime = vid, "video/mp4"
                        if v_b64:
                            parts.append({
                                "type": "video_url",
                                "video_url": {"url": f"data:{v_mime};base64,{v_b64}"},
                            })
                msg_out = {"role": "user", "content": parts}
            else:
                msg_out = {"role": "user", "content": content}
            result.append(msg_out)

        elif role == "assistant":
            msg: dict = {"role": "assistant", "content": m.get("content") or None}
            if "thinking" in m and m["thinking"]:
                msg["reasoning_content"] = m["thinking"]
            
            tcs = m.get("tool_calls", [])
            if tcs:
                msg["tool_calls"] = []
                for tc in tcs:
                    # Prefer the exact bytes the model originally streamed. Re-serializing
                    # the parsed dict can reorder keys / change separators and invalidate
                    # the provider's prefix cache for every subsequent turn.
                    raw = tc.get("arguments_raw")
                    if not isinstance(raw, str) or not raw:
                        raw = json.dumps(tc.get("input") or {}, ensure_ascii=False)
                    tc_msg = {
                        "id":   tc["id"],
                        "type": "function",
                        "function": {
                            "name":      tc["name"],
                            "arguments": raw,
                        },
                    }
                    # Pass through provider-specific fields (e.g. Gemini thought_signature)
                    if tc.get("extra_content"):
                        tc_msg["extra_content"] = tc["extra_content"]
                    msg["tool_calls"].append(tc_msg)
            result.append(msg)

        elif role == "tool":
            result.append({
                "role":         "tool",
                "tool_call_id": m["tool_call_id"],
                "content":      m["content"],
            })

    return result


# ── Streaming adapters ─────────────────────────────────────────────────────

class TextChunk:
    def __init__(self, text): self.text = text

class ThinkingChunk:
    def __init__(self, text): self.text = text

class AssistantTurn:
    """Completed assistant turn with text + tool_calls + thinking."""
    def __init__(self, text, tool_calls, in_tokens, out_tokens, thinking="", error=False,
                 cache_creation_tokens=0, cache_read_tokens=0, thinking_signature="",
                 finish_reason=""):
        self.text        = text
        self.tool_calls  = tool_calls   # list of {id, name, input}
        self.in_tokens   = in_tokens
        self.out_tokens  = out_tokens
        self.thinking    = thinking
        # Anthropic extended-thinking signature (empty for other providers). Must
        # be replayed verbatim alongside the thinking text or the API 400s.
        self.thinking_signature = thinking_signature
        self.error       = error
        # Why the turn ended: "stop" | "tool_calls" | "length" | "". Lets the
        # agent loop tell a real completion from a max_tokens truncation.
        self.finish_reason = finish_reason
        # Anthropic explicit caching + OpenAI prompt-cached tokens.
        # 0 when the provider doesn't report it.
        self.cache_creation_tokens = cache_creation_tokens
        self.cache_read_tokens     = cache_read_tokens


def friendly_api_error(exc: Exception) -> str:
    """Map common API exceptions to short, actionable hints for the user.

    Returns a single-line string suitable for streaming back to the REPL.
    Falls back to the raw exception message when no pattern matches.
    """
    s = str(exc).lower()
    etype = type(exc).__name__

    # Kimi Code quota. Checked before the generic auth/permission branches: the
    # payload carries a 403 and the word "access", which would otherwise be read
    # as a bad key. Quota is billed per ACCOUNT, so a new key changes nothing.
    if "access_terminated_error" in s:
        return ("Kimi membership quota is spent for this billing cycle. It is billed per "
                "account, so a new API key won't help — switch model with /model, or wait "
                "for the cycle to reset.")
    # Auth / key problems
    if "authentication" in s or "invalid_api_key" in s or "401" in s or etype == "AuthenticationError":
        return "API key is missing or invalid. Run /config <provider>_api_key=... or set the env var."
    # Quota / free-tier exhaustion (NVIDIA "ResourceExhausted", Gemini quota, …)
    if ("resourceexhausted" in s or "resource_exhausted" in s or "resource exhausted" in s
            or "quota" in s or "request limit reached" in s):
        return ("Provider quota exhausted (free-tier or account limit reached). "
                "Wait for the quota to reset, or switch provider/model with /model.")
    # Rate limit
    if "rate limit" in s or "rate_limit" in s or "429" in s or etype == "RateLimitError":
        return "Rate limit hit. Wait a bit and retry, or switch model with /model."
    # Overload / capacity
    if "overloaded" in s or "capacity" in s or "503" in s or "502" in s:
        return "Provider is overloaded right now. Retry in a few seconds or switch model."
    # Context / token limit
    if "context_length" in s or "maximum context" in s or "too many tokens" in s or "context_window" in s:
        return "Context window exceeded. Try /compact to shrink history or /clear to reset."
    # Bad request / tool schema
    if "invalid_request" in s or "400" in s or etype == "BadRequestError":
        return f"API rejected the request: {exc}. Check tool schemas, message format, or model name."
    # Network / DNS
    if "connection" in s or "timeout" in s or "dns" in s or etype in ("APIConnectionError", "ConnectTimeout"):
        return "Network problem reaching the API. Check connection, VPN, or provider status."
    # Permission / model access
    if "permission" in s or "model_not_found" in s or "404" in s:
        return "Model not found or not enabled for your account. Check model name or billing."
    return f"API error: {exc}"


def _thinking_level_from(value) -> int:
    """Coerce legacy bool/int thinking config into an int 0-3.

    Level 4 is clamped to 3: the largest budgets push providers into
    RESOURCE_EXHAUSTED. 3 is the effective max everywhere, whatever the UI
    toggle says.
    """
    if value is True:  return 3
    if value is False or value is None: return 0
    try:
        lvl = int(value)
    except (TypeError, ValueError):
        return 0
    return max(0, min(3, lvl))


_ANTHROPIC_TOOL_ALLOWED_KEYS = {
    "name", "description", "input_schema", "cache_control", "type",
}


def _sanitize_tool_for_anthropic(t: dict) -> dict | None:
    """Coerce any tool schema (OpenAI / mixed / Anthropic) into Anthropic shape.

    Handles three common mis-formats seen from plugins/MCP/auto-adapter:
      (a) OpenAI wrapped:      {"type":"function","function":{"name","description","parameters"}}
      (b) OpenAI custom:       {"type":"custom","custom":{"name","description","parameters"}}
      (c) Loose:               {"name","description","parameters":{...}}   (parameters → input_schema)
    Plus the correct Anthropic shape (passes through).

    Returns None when the schema is irrecoverable (no name).
    """
    if not isinstance(t, dict):
        return None

    inner = t
    if "function" in t and isinstance(t["function"], dict):
        inner = {**t.get("function", {})}
        inner.setdefault("type", "custom")
    elif "custom" in t and isinstance(t["custom"], dict):
        inner = {**t.get("custom", {})}
        inner.setdefault("type", "custom")

    out: dict = {}
    name = inner.get("name") or t.get("name")
    if not name:
        return None
    out["name"] = name

    desc = inner.get("description") or t.get("description") or ""
    if desc:
        out["description"] = desc

    schema = (inner.get("input_schema")
              or inner.get("parameters")
              or t.get("input_schema")
              or t.get("parameters")
              or {"type": "object", "properties": {}})
    out["input_schema"] = schema

    cc = inner.get("cache_control") or t.get("cache_control")
    if cc:
        out["cache_control"] = cc

    return out


def stream_anthropic(
    api_key: str,
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
) -> Generator:
    """Stream from Anthropic API. Yields TextChunk/ThinkingChunk, then AssistantTurn.

    Prompt caching: marks up to 3 cache breakpoints — system prompt, tools
    block, and the latest user message. Anthropic caches everything BEFORE
    each breakpoint, so the conversation history up to the latest user turn
    rides the same cache as long as it's appended (not edited). 4-breakpoint
    cap is the API limit; 3 is the practical sweet spot for an agent loop.
    """
    import anthropic as _ant

    # Cache TTL: default "ephemeral" = 5-minute TTL. With a large tool registry
    # + system prompt the cached prefix can be 30-60K tokens; any pause > 5 min
    # between turns expires it and the FULL prefix is re-written at 1.25x base
    # input price. `/config cache_ttl=1h` switches to the 1-hour TTL (2x write
    # cost once, but survives think-pauses) — much cheaper for interactive
    # sessions on Opus/Sonnet. Requires the extended-cache-ttl beta header.
    _cache_ttl = str(config.get("cache_ttl", "")).strip().lower()
    _use_1h = _cache_ttl in ("1h", "1hr", "1hour", "3600", "3600s")
    if _use_1h:
        _cc_marker = {"type": "ephemeral", "ttl": "1h"}
    else:
        _cc_marker = {"type": "ephemeral"}
    _extra_betas = ["extended-cache-ttl-2025-04-11"] if _use_1h else []

    # Prefer the Dulus-native Claude subscription OAuth token (from /login claude)
    # over an API key — this is the "no API key, no cookie webbridge" path. Passing
    # auth_token (an explicit credential) makes the SDK skip ANTHROPIC_API_KEY env
    # resolution, so it sends ONLY `Authorization: Bearer` (no x-api-key).
    _oauth_tok = _anthropic_oauth_get_token(config)
    _is_oauth = bool(_oauth_tok)
    if _is_oauth:
        _betas = ",".join([ANTHROPIC_OAUTH_BETA] + _extra_betas)
        client = _ant.Anthropic(
            auth_token=_oauth_tok,
            default_headers={"anthropic-beta": _betas},
        )
    elif _extra_betas:
        client = _ant.Anthropic(
            api_key=api_key,
            default_headers={"anthropic-beta": ",".join(_extra_betas)},
        )
    else:
        client = _ant.Anthropic(api_key=api_key)

    # 1) System prompt as a single text block with cache_control. Under OAuth the
    #    Claude Code identity must be the first system block (else the API 403s).
    if _is_oauth:
        system_blocks = _anthropic_oauth_system_blocks(system, _cc_marker)
    elif isinstance(system, str) and system:
        system_blocks = [{
            "type": "text",
            "text": system,
            "cache_control": _cc_marker,
        }]
    else:
        system_blocks = system  # already structured, leave as-is

    # 2) Tools: sanitize → some plugins/MCP/auto-adapter register schemas
    #    using OpenAI's wire format ({type:"function", function:{...}} or
    #    {custom:{parameters:...}}). Anthropic rejects anything outside
    #    {name, description, input_schema, cache_control, type}. We coerce
    #    each schema into the Anthropic shape and silently drop unknown
    #    top-level keys so one bad plugin doesn't 400 the whole turn.
    cached_tools = [_sanitize_tool_for_anthropic(t) for t in (tool_schemas or [])]
    cached_tools = [t for t in cached_tools if t]
    if cached_tools:
        last_tool = dict(cached_tools[-1])
        last_tool["cache_control"] = _cc_marker
        cached_tools[-1] = last_tool

    # 3) Latest user message: marker on the last content block. Caches the
    #    full prior conversation so multi-turn sessions hit the cache.
    ant_messages = messages_to_anthropic(messages)
    for i in range(len(ant_messages) - 1, -1, -1):
        m = ant_messages[i]
        if m.get("role") != "user":
            continue
        c = m.get("content")
        if isinstance(c, str):
            m["content"] = [{
                "type": "text",
                "text": c,
                "cache_control": _cc_marker,
            }]
        elif isinstance(c, list) and c:
            last = c[-1]
            if isinstance(last, dict) and "cache_control" not in last:
                # Copy-before-mark: mutating the shared block in-place would
                # persist the marker into state.messages, accumulating extra
                # breakpoints turn after turn (>4 markers → API 400).
                c = list(c)
                c[-1] = {**last, "cache_control": _cc_marker}
                m["content"] = c
        break

    kwargs = {
        "model":      model,
        "max_tokens": config.get("max_tokens", 128000),
        "system":     system_blocks,
        "messages":   ant_messages,
        "tools":      cached_tools,
    }
    _thk_raw = config.get("thinking", 0)
    _thk_level = _thinking_level_from(_thk_raw)
    if _thk_level > 0:
        # Budget scales with level: 1=low, 2=medium, 3=high, 4=normal (mid). Explicit
        # thinking_budget in config still wins when provided.
        _level_budgets = {1: 2048, 2: 6000, 3: 16000, 4: 8192}
        budget = config.get("thinking_budget") or _level_budgets[_thk_level]
        # Anthropic needs max_tokens > budget_tokens, and the ANSWER must fit in
        # whatever's left after thinking. Cap the budget so text always has room —
        # otherwise Claude "thinks" up to the limit and stops with no answer.
        _mx = int(kwargs.get("max_tokens") or 0)
        _answer_room = 4096
        if _mx and budget >= _mx - _answer_room:
            budget = max(1024, _mx - _answer_room)
        kwargs["thinking"] = {
            "type":          "enabled",
            "budget_tokens": budget,
        }

    tool_calls = []
    text       = ""
    thinking   = ""
    thinking_signature = ""

    try:
        with client.messages.stream(**kwargs) as stream:
            for event in stream:
                etype = getattr(event, "type", None)
                if etype == "content_block_delta":
                    delta: Any = getattr(event, "delta", None)
                    dtype = getattr(delta, "type", None)
                    if dtype == "text_delta":
                        text += delta.text
                        yield TextChunk(delta.text)
                    elif dtype == "thinking_delta":
                        thinking += delta.thinking
                        yield ThinkingChunk(delta.thinking)
                    elif dtype == "signature_delta":
                        # Signature for the current thinking block. It is bound to
                        # the thinking text and is REQUIRED verbatim on replay when
                        # extended thinking is combined with tool use — capture it.
                        thinking_signature += getattr(delta, "signature", "") or ""

            final = stream.get_final_message()
            _final_text_parts = []
            for block in final.content:
                if block.type == "tool_use":
                    tool_calls.append({
                        "id":    block.id,
                        "name":  block.name,
                        "input": block.input,
                    })
                elif block.type == "thinking":
                    # The final message carries the authoritative signature — prefer it.
                    _sig = getattr(block, "signature", "") or ""
                    if _sig:
                        thinking_signature = _sig
                elif block.type == "text":
                    _final_text_parts.append(getattr(block, "text", "") or "")

            # Recover the answer from the final message if the streamed deltas
            # missed it (seen under the Claude OAuth/beta path, where the text can
            # surface only on the final message). Without this the turn looks like
            # a "thinking-only" stall even though Claude did answer.
            if not (text or "").strip() and _final_text_parts:
                text = "".join(_final_text_parts)
                if text:
                    yield TextChunk(text)

            # Map Anthropic stop_reason → finish_reason so the agent loop can tell
            # a real end from a max_tokens truncation.
            _sr = getattr(final, "stop_reason", "") or ""
            _finish = {"max_tokens": "length", "tool_use": "tool_calls",
                       "end_turn": "stop", "stop_sequence": "stop"}.get(_sr, _sr or "stop")

            _cc = getattr(final.usage, "cache_creation_input_tokens", 0) or 0
            _cr = getattr(final.usage, "cache_read_input_tokens", 0) or 0
            yield AssistantTurn(
                text, tool_calls,
                final.usage.input_tokens,
                final.usage.output_tokens,
                thinking=thinking,
                thinking_signature=thinking_signature,
                cache_creation_tokens=_cc,
                cache_read_tokens=_cr,
                finish_reason=_finish,
            )
    except KeyboardInterrupt:
        # User Ctrl+C'd mid-stream. Yield whatever text we already got and
        # bail cleanly — do NOT route through friendly_api_error, which sees
        # cleanup-side httpx noise ("401", "authentication") and mislabels
        # the cancel as a "wrong API key" error.
        yield AssistantTurn(text, tool_calls, 0, 0, thinking=thinking,
                            thinking_signature=thinking_signature, error=False)
        return
    except Exception as _e:
        # Filter out cancellation/cleanup exceptions that masquerade as auth
        # errors. httpx ReadError / RemoteProtocolError / CancelledError raised
        # during stream teardown after a Ctrl+C are NOT auth failures.
        _etype = type(_e).__name__
        if _etype in ("CancelledError", "ReadError", "RemoteProtocolError", "CloseError"):
            yield AssistantTurn(text, tool_calls, 0, 0, thinking=thinking,
                                thinking_signature=thinking_signature, error=False)
            return
        msg = friendly_api_error(_e)
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return


def stream_kimi(
    api_key: str,
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
) -> Generator:
    """Stream from Kimi API using native HTTP requests. Yields TextChunk, then AssistantTurn.
    
    This is a native implementation using urllib.request instead of the OpenAI SDK,
    allowing direct comparison with the OpenAI-compatible version.
    
    Token estimation:
    1. Input tokens: Estimados ANTES usando estimate_tokens_kimi() (endpoint nativo de Kimi)
    2. Output tokens: Capturados del campo usage de la respuesta streaming
    """
    url = "https://api.moonshot.ai/v1/chat/completions"

    # Build messages
    kimi_messages = [{"role": "system", "content": system}] + messages_to_openai(messages, model=model)

    # Kimi rejects assistant messages with null/empty content and no tool_calls
    # (happens when a prior turn was thinking-only or interrupted).
    # Replace empty content with a placeholder so the conversation chain stays valid.
    for _m in kimi_messages:
        if _m.get("role") == "assistant" and not _m.get("tool_calls"):
            if not _m.get("content"):
                _m["content"] = "..."

    # === CONTADOR DE TOKENS ===
    # Input: Estimación por caracteres (fallback simple y confiable)
    # Output: Capturado del usage del stream
    in_tok = 0
    
    # Build request payload
    payload: dict = {
        "model": model,
        "messages": kimi_messages,
        "stream": True,
        "stream_options": {"include_usage": True},  # ensure token usage in stream
    }

    # Kimi thinking control
    thinking_mode = "enabled" if config.get("thinking", False) else "disabled"
    payload["thinking"] = {"type": thinking_mode}

    # Tools
    if tool_schemas and not config.get("no_tools"):
        payload["tools"] = tools_to_openai(tool_schemas)
        if not config.get("disable_tool_choice"):
            payload["tool_choice"] = "auto"

    # Max tokens (Kimi prefers max_completion_tokens like OpenAI new API)
    if config.get("max_tokens"):
        prov_cap = PROVIDERS.get("kimi", {}).get("max_completion_tokens")
        mt = config["max_tokens"]
        payload["max_completion_tokens"] = min(mt, prov_cap) if prov_cap else mt
    
    # Extra options
    if config.get("temperature") is not None:
        payload["temperature"] = config["temperature"]
    if config.get("top_p") is not None:
        payload["top_p"] = config["top_p"]
    
    # Make request
    req = urllib.request.Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {api_key}"
        },
        method="POST"
    )
    
    text = ""
    thinking = ""
    tool_buf: dict = {}
    out_tok = 0
    cached_tok = 0
    
    # Estimación simple de tokens de entrada (caracteres / 4)
    # Esto es aproximado pero confiable
    total_chars = len(system) + sum(len(str(m.get("content", ""))) for m in messages)
    in_tok = max(1, total_chars // 4)
    
    try:
        resp = urllib.request.urlopen(req, timeout=300)
    except urllib.error.HTTPError as e:
        try:
            err_body = e.read().decode("utf-8")
            err_data = json.loads(err_body)
            err_msg = err_data.get("error", {}).get("message", str(e))
        except:
            err_msg = str(e)
        msg = f"Error: Kimi API error: {err_msg}"
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return
    except Exception as e:
        msg = f"Error: Failed to connect to Kimi API: {e}"
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return
    
    # Parse SSE stream
    for line in resp:
        line = line.decode("utf-8").strip()
        if not line or not line.startswith("data: "):
            continue
        
        data_str = line[6:]  # Remove "data: " prefix
        if data_str == "[DONE]":
            break
        
        try:
            data = json.loads(data_str)
        except json.JSONDecodeError:
            continue
        
        # Extract usage if present
        if "usage" in data and data["usage"]:
            u = data["usage"]
            in_tok = u.get("prompt_tokens", 0) or in_tok
            out_tok = u.get("completion_tokens", 0) or out_tok
            # Kimi exposes cached prompt tokens at top-level usage.cached_tokens
            # (some accounts also report prompt_tokens_details.cached_tokens).
            cached_tok = (
                u.get("cached_tokens", 0)
                or (u.get("prompt_tokens_details") or {}).get("cached_tokens", 0)
                or cached_tok
            )
        
        # Extract choices
        choices = data.get("choices", [])
        if not choices:
            continue
        
        delta = choices[0].get("delta", {})
        
        # Content
        content = delta.get("content")
        if content:
            text += content
            yield TextChunk(content)
        
        # Reasoning content
        reasoning = delta.get("reasoning_content") or delta.get("reasoning")
        if reasoning:
            thinking += reasoning
            yield ThinkingChunk(reasoning)
        
        # Tool calls
        tool_calls = delta.get("tool_calls", [])
        for tc in tool_calls:
            idx = tc.get("index", 0)
            if idx not in tool_buf:
                tool_buf[idx] = {"id": "", "name": "", "args": ""}
            if tc.get("id"):
                tool_buf[idx]["id"] = tc["id"]
            fn = tc.get("function", {})
            if fn.get("name"):
                tool_buf[idx]["name"] += fn["name"]
            if fn.get("arguments"):
                tool_buf[idx]["args"] += fn["arguments"]
    
    final_tool_calls = _finalize_tool_calls(tool_buf)

    yield AssistantTurn(text, final_tool_calls, in_tok, out_tok, thinking=thinking,
                        cache_read_tokens=cached_tok)


def _oai_uses_completion_tokens(model: str) -> bool:
    """Newer OpenAI models (gpt-5.x and the o1/o3/o4 reasoning family) reject the
    legacy `max_tokens` parameter and require `max_completion_tokens` instead."""
    m = (model or "").lower().rsplit("/", 1)[-1]
    return m.startswith(("o1", "o3", "o4")) or m.startswith("gpt-5")


def _finalize_tool_calls(tool_buf: dict, include_extra: bool = False) -> list:
    """Convert buffered OpenAI-style tool deltas into final tool_call dicts.

    Also keeps ``arguments_raw`` — the exact JSON string the model streamed.
    Re-dumping ``input`` via json.dumps() on replay can change key order /
    whitespace and bust provider prompt caches on the next turn (xAI/OpenAI
    prefix cache is byte-sensitive on earlier messages).
    """
    tool_calls = []
    for idx in sorted(tool_buf):
        v = tool_buf[idx]
        raw_args = v.get("args") or ""
        try:
            inp = json.loads(raw_args) if raw_args else {}
        except json.JSONDecodeError:
            inp = {"_raw": raw_args}
        tc_entry = {
            "id": v["id"] or f"call_{idx}",
            "name": v["name"],
            "input": inp,
            "arguments_raw": raw_args,
        }
        if include_extra and v.get("extra_content"):
            tc_entry["extra_content"] = v["extra_content"]
        tool_calls.append(tc_entry)
    return tool_calls


def stream_litellm(
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
) -> Generator:
    """Stream via LiteLLM — one entry point routing to 100+ backends.

    `model` arrives WITHOUT the leading "litellm/" prefix (bare_model strips
    one level). So for `litellm/openrouter/anthropic/claude-3-5-sonnet`,
    `model` here is `openrouter/anthropic/claude-3-5-sonnet` — exactly the
    format LiteLLM's `completion()` expects.

    LiteLLM auto-resolves the underlying API key from env (OPENROUTER_API_KEY,
    GROQ_API_KEY, TOGETHER_API_KEY, …). For Dulus we also mirror the relevant
    config-stored key into the env before dispatching, so users who put their
    keys under `/config openrouter_api_key=...` Just Work without `export`.
    """
    # Robust detection: ImportError covers "not installed at all", but a
    # phantom `litellm` namespace (stale stub, name collision with a local
    # litellm.py, partial pip install) can let the import succeed while
    # `.completion` is missing — that surfaces as a confusing AttributeError
    # mid-call. Check both, and treat missing-completion the same as
    # missing-package so the user gets the same friendly install hint.
    try:
        import litellm  # type: ignore
    except ImportError:
        litellm = None  # type: ignore
    if litellm is None or not hasattr(litellm, "completion"):
        where = ""
        try:
            if litellm is not None and getattr(litellm, "__file__", None):
                where = f"\n  (found a partial 'litellm' module at: {litellm.__file__})"
        except Exception:
            pass
        msg = (
            "[litellm] Not installed (or installed module is missing .completion). Run:\n"
            "  pip install 'dulus[litellm]'\n"
            "or directly:\n"
            "  pip install -U litellm" + where
        )
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return

    # Mirror config-stored API keys into the env so LiteLLM finds them.
    # LiteLLM reads ~30 env vars depending on the backend prefix; we surface
    # the most common ones from config.
    import os as _os
    _env_map = {
        "openrouter":   "OPENROUTER_API_KEY",
        "groq":         "GROQ_API_KEY",
        "together_ai":  "TOGETHER_API_KEY",
        "perplexity":   "PERPLEXITYAI_API_KEY",
        "cohere":       "COHERE_API_KEY",
        "mistral":      "MISTRAL_API_KEY",
        "fireworks_ai": "FIREWORKS_API_KEY",
        "xai":          "XAI_API_KEY",
        "xai-oauth":    None,  # X login (SuperGrok/Premium+) – black magic, no separate key
        "anyscale":     "ANYSCALE_API_KEY",
        "deepinfra":    "DEEPINFRA_API_KEY",
        "replicate":    "REPLICATE_API_KEY",
        "azure":        "AZURE_API_KEY",
        "bedrock":      "AWS_ACCESS_KEY_ID",     # bedrock needs AWS creds, not single key
        "vertex_ai":    "GOOGLE_APPLICATION_CREDENTIALS",
        "openai":       "OPENAI_API_KEY",
        "anthropic":    "ANTHROPIC_API_KEY",
        "gemini":       "GEMINI_API_KEY",
    }
    backend = model.split("/", 1)[0] if "/" in model else ""
    env_name = _env_map.get(backend)
    if env_name:
        cfg_key = (
            config.get(f"{backend}_api_key", "")
            or config.get(f"litellm_{backend}_api_key", "")
            or config.get("litellm_api_key", "")
        )
        if cfg_key and not _os.environ.get(env_name):
            _os.environ[env_name] = cfg_key

    oai_messages = [{"role": "system", "content": system}] + messages_to_openai(messages)

    kwargs: dict = {
        "model":    model,
        "messages": oai_messages,
        "stream":   True,
    }
    if tool_schemas and not config.get("no_tools"):
        kwargs["tools"] = tools_to_openai(tool_schemas)
        if not config.get("disable_tool_choice"):
            kwargs["tool_choice"] = "auto"
        if backend == "perplexity":
            # Perplexity has two APIs reachable through LiteLLM:
            #   1. Sonar Chat Completions (perplexity/sonar*): does NOT accept
            #      OpenAI function tools at all; sending them returns
            #      "Tool parameters must be a JSON object".
            #   2. Agent/Responses API third-party wrappers (perplexity/openai/*,
            #      perplexity/anthropic/*, presets, etc.): do support tools.
            # We keep tools only for the second group.
            _perplexity_tail = model.split("/", 1)[1] if "/" in model else ""
            _perplexity_native_prefixes = ("sonar", "r1-1776", "preset")
            _is_perplexity_native = any(
                _perplexity_tail.startswith(p) for p in _perplexity_native_prefixes
            )
            if _is_perplexity_native:
                kwargs.pop("tools", None)
                kwargs.pop("tool_choice", None)
                if not getattr(stream_litellm, "_perplexity_tool_warned", False):
                    stream_litellm._perplexity_tool_warned = True  # type: ignore[attr-defined]
                    yield TextChunk(
                        "[litellm/perplexity] Sonar chat completions do not support "
                        "function tools; tools omitted for this call."
                    )
            else:
                kwargs["allowed_openai_params"] = ["tools", "tool_choice"]
    _litellm_tokens = min(
        int(config.get("litellm_tokens") or config.get("max_tokens") or -1),
        -1,
    )
    if _litellm_tokens:
        if _oai_uses_completion_tokens(model):
            kwargs["max_completion_tokens"] = _litellm_tokens
        else:
            kwargs["max_tokens"] = _litellm_tokens

    text     = ""
    thinking = ""
    tool_buf: dict = {}
    in_tok = out_tok = -1

    try:
        response = litellm.completion(**kwargs)
    except Exception as e:
        msg = f"[litellm] {type(e).__name__}: {e}"
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return

    try:
        for chunk in response:
            # LiteLLM normalises every backend into OpenAI ChunkChoice format.
            choices = getattr(chunk, "choices", None) or []
            if not choices:
                # Usage block sometimes arrives in a separate trailing chunk.
                u = getattr(chunk, "usage", None)
                if u is not None:
                    in_tok = getattr(u, "prompt_tokens", 0) or 0
                    out_tok = getattr(u, "completion_tokens", 0) or 0
                continue
            delta = getattr(choices[0], "delta", None)
            if delta is None:
                continue
            piece = getattr(delta, "content", None) or ""
            if piece:
                text += piece
                yield TextChunk(piece)
            tc_list = getattr(delta, "tool_calls", None) or []
            for tc in tc_list:
                idx = getattr(tc, "index", 0) or 0
                slot = tool_buf.setdefault(idx, {"id": "", "name": "", "args": ""})
                if getattr(tc, "id", None):
                    slot["id"] = tc.id
                fn = getattr(tc, "function", None)
                if fn is not None:
                    if getattr(fn, "name", None):
                        slot["name"] = fn.name
                    args_piece = getattr(fn, "arguments", "") or ""
                    if args_piece:
                        slot["args"] += args_piece
    except Exception as e:
        msg = f"[litellm stream] {type(e).__name__}: {e}"
        yield TextChunk(msg)
        yield AssistantTurn(text or msg, [], in_tok, out_tok, error=True)
        return

    # Flatten tool_buf into AssistantTurn tool_calls.
    tool_calls = []
    for idx in sorted(tool_buf.keys()):
        slot = tool_buf[idx]
        if not slot["name"]:
            continue
        try:
            args = json.loads(slot["args"] or "{}")
        except Exception:
            args = {}
        tool_calls.append({"id": slot["id"] or f"call_{idx}", "name": slot["name"], "input": args})

    yield AssistantTurn(text, tool_calls, in_tok, out_tok, thinking=thinking)


def _get_nvidia_fallback_chain(config: dict) -> list[str]:
    chain = config.get("nvidia_fallback_chain")
    if chain:
        return chain
    import json as _json, os as _os
    p = _os.path.join(_os.path.expanduser("~"), ".dulus", "nvidia-providers.json")
    if _os.path.exists(p):
        try:
            return _json.loads(open(p, encoding="utf-8").read()).get("fallback_models", [])
        except Exception:
            pass
    return [
        "deepseek-ai/deepseek-v4-flash",
        "z-ai/glm-4.7",
        "z-ai/glm-5.1",
        "minimaxai/minimax-m2.7",
        "moonshotai/kimi-k2-instruct",
        "mistralai/mistral-nemotron",
        "bytedance/seed-oss-36b-instruct",
        "upstage/solar-10.7b-instruct",
        "stepfun-ai/step-3.5-flash",
        "meta/llama-3.3-70b-instruct",
        "moonshotai/kimi-k2.5",
        "deepseek-ai/deepseek-r1",
        "nvidia/llama-3.1-nemotron-70b-instruct",
        "qwen/qwen2.5-72b-instruct",
    ]


def _get_modelstudio_fallback_chain(config: dict) -> list[str]:
    """Ordered model fallback chain for Alibaba Model Studio (Singapore).

    Override via `/config modelstudio_fallback_chain=qwen-max,qwen-plus,...`
    (comma string or list); otherwise defaults to the provider's model list.
    Lets Dulus auto-switch models when one fails (e.g. 403 quota exhausted).
    """
    chain = config.get("modelstudio_fallback_chain")
    if isinstance(chain, str):
        chain = [m.strip() for m in chain.split(",") if m.strip()]
    if not chain:
        chain = PROVIDERS.get("modelstudio", {}).get("models", [])
    return list(chain or [])


def stream_openai_compat(
    api_key: str,
    base_url: str,
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
) -> Generator:
    """Stream from any OpenAI-compatible API. Yields TextChunk, then AssistantTurn."""
    from openai import OpenAI
    # Detect kimi-code by base_url, NOT by model name. Reason: when invoked as
    # `kimi-code/kimi-k2.5` (or k2.6, kimi-latest, etc.), `model` arrives here
    # already stripped to the bare name, and detect_provider("kimi-k2.5") falls
    # through to the generic "kimi" prefix → header omitted → 403.
    # The /coding/v1 endpoint is unique to kimi-code regardless of model.
    # api.kimi.ai and api.kimi.com are the same Kimi Code backend (same models,
    # same quota errors, both behind Cloudflare), so accept either host — the
    # endpoint 403s any client it can't recognise as a known coding agent.
    _kimi_base = (base_url or "").lower()
    _is_kimi_code = (
        "api.kimi.com/coding" in _kimi_base
        or "api.kimi.ai/coding" in _kimi_base
        or detect_provider(model) in ("kimi-code", "kimi-code2", "kimi-code3")
    )
    client_kwargs: dict = {"api_key": api_key or "dummy", "base_url": base_url}
    if _is_kimi_code:
        # Kimi Code API whitelists only known Coding Agents by User-Agent.
        # Without this header the API returns 403.
        client_kwargs["default_headers"] = {"User-Agent": "KimiCLI/1.30.0"}
    client = OpenAI(**client_kwargs)

    # Grok-style agentic nudge: parallel_tool_calls only ALLOWS batching — the
    # model still decides. Grok feels fast because xAI also tells it to use tools
    # aggressively and keep going. Mirror that so Kimi/DeepSeek/etc. actually
    # batch independent calls. Byte-stable (cacheable); only when tools are on
    # and parallel isn't disabled.
    _sys = system or ""
    if tool_schemas and not config.get("no_tools") and not config.get("disable_parallel_tools"):
        _sys = _sys + (
            "\n\n[Tool use] Tools are executed for real by the Dulus harness. "
            "CHAIN AGGRESSIVELY — every tool call is a round-trip that burns "
            "tokens and time: (1) INDEPENDENT actions (multiple reads, searches, "
            "fetches) → emit them ALL as parallel tool calls in ONE response, "
            "never drip-feed one per turn. (2) DEPENDENT steps (scan → filter → "
            "drill down) → run them INSIDE a single Python() or Bash() call, "
            "not across turns. (3) Hunting through many files? Do the whole "
            "hunt in the Python kernel (os.walk + regex over contents) — one "
            "Python call beats 10 Grep/Glob round-trips. Keep calling tools "
            "until you can give a complete final answer. Never simulate tool results."
        )
    oai_messages = [{"role": "system", "content": _sys}] + messages_to_openai(messages, model=model)

    _is_nvidia = detect_provider(model) == "nvidia-web"

    # Alibaba Model Studio (Singapore): detect by endpoint host so it stays true
    # even though the bare model name (qwen-*) would route to plain DashScope.
    _is_modelstudio = "maas.aliyuncs.com" in (base_url or "")
    _ms_remaining = None
    if _is_modelstudio:
        _ms_remaining = config.get("_modelstudio_remaining")
        if _ms_remaining is None:
            _ms_remaining = [m for m in _get_modelstudio_fallback_chain(config) if m != model]

    kwargs: dict = {
        "model":    model,
        "messages": oai_messages,
        "stream":   True,
        "stream_options": {"include_usage": True},
    }

    # Pass num_ctx for known Ollama/LM Studio ports only — avoids matching other local servers (e.g. vLLM on :8000)
    _is_local_ollama = "11434" in base_url
    _is_lmstudio     = "1234" in base_url and ("lmstudio" in base_url or "localhost" in base_url or "127.0.0.1" in base_url)
    if _is_local_ollama or _is_lmstudio:
        # Same single source of truth as stream_ollama: config.json first,
        # provider default only when the user hasn't set anything.
        prov = detect_provider(model)
        fallback = PROVIDERS.get(prov if prov in ("ollama", "lmstudio") else "ollama", {}).get("context_limit", 128000)
        # Context window (num_ctx) is its OWN knob, decoupled from max_tokens
        # (which is the output cap). context_limit first, provider default otherwise.
        ctx_limit = int(config.get("context_limit") or fallback)
        _ctx_body: dict = {"options": {"num_ctx": ctx_limit}}  # bare dict: extra_body mixes int/str shapes
        kwargs["extra_body"] = _ctx_body

    # Kimi thinking control (v1.0.1.20+)
    # Gate by the REAL endpoint host, not just the model-name prefix: Kimi models
    # are also hosted on Azure/other OpenAI-compatible gateways that reject the
    # non-standard `thinking` arg (400 unrecognized_request_argument).
    _is_native_kimi_host = any(
        h in (base_url or "")
        for h in ("api.moonshot.ai", "api.kimi.com")
    )
    if _is_native_kimi_host:
        # Gate on the host (reliable) — by here `model` is the bare wire name.
        # Two families per Kimi's docs (kimi.com/code/docs → Model Configuration):
        #   k3 / k3-256k     → reasoning_effort: low | high | max  (default high)
        #   kimi-for-coding  → thinking {type: enabled|disabled}
        #   moonshot-v1-*    → thinking (same as before)
        _wire = str(model or "").lower()
        if _wire.startswith("k3"):
            _eff = str(config.get("reasoning_effort", "high")).strip().lower()
            kwargs["reasoning_effort"] = _eff if _eff in ("low", "high", "max") else "high"
        else:
            if not kwargs.get("extra_body"): kwargs["extra_body"] = {}
            # Kimi expects an object: {"type": "enabled" | "disabled"}
            mode = "enabled" if config.get("thinking", False) else "disabled"
            kwargs["extra_body"]["thinking"] = {"type": mode}

    # DeepSeek reasoning control (reasoning_effort for thinking models)
    if detect_provider(model) == "deepseek":
        if config.get("thinking", False):
            # Map thinking mode to reasoning_effort
            kwargs["reasoning_effort"] = "medium"  # default
        else:
            kwargs["reasoning_effort"] = "none"

    # NVIDIA NIM thinking control (chat_template_kwargs)
    if _is_nvidia and config.get("thinking", False):
        if not kwargs.get("extra_body"):
            kwargs["extra_body"] = {}
        kwargs["extra_body"]["chat_template_kwargs"] = {
            "thinking": True,
            "reasoning_effort": "high",
        }

    if tool_schemas and not config.get("no_tools"):
        kwargs["tools"] = tools_to_openai(tool_schemas)
        # "auto" requires vLLM --enable-auto-tool-choice; omit if server doesn't support it
        if not config.get("disable_tool_choice"):
            kwargs["tool_choice"] = "auto"
            # Let the model batch INDEPENDENT tool calls into ONE response. The
            # agent loop already executes every tool_call in a turn, so without
            # this many OpenAI-compatible servers (Kimi included) default parallel
            # OFF and the model emits one tool per API round-trip — each round-trip
            # resends the growing context and re-primes the cache. Batching = fewer
            # round-trips. Opt out with /config disable_parallel_tools=1.
            if not config.get("disable_parallel_tools"):
                kwargs["parallel_tool_calls"] = True
    if config.get("max_tokens"):
        prov_cap = PROVIDERS.get(detect_provider(model), {}).get("max_completion_tokens")
        if _is_modelstudio:
            prov_cap = PROVIDERS.get("modelstudio", {}).get("max_completion_tokens") or prov_cap
        mt = config["max_tokens"]
        mt = min(mt, prov_cap) if prov_cap else mt
        if _oai_uses_completion_tokens(model):
            kwargs["max_completion_tokens"] = mt
        else:
            kwargs["max_tokens"] = mt

    text          = ""
    thinking      = ""
    tool_buf: dict = {}   # index → {id, name, args_str}
    in_tok = out_tok = 0
    cached_tok = 0  # OpenAI-compat prefix-cached prompt tokens (when reported)

    try:
        from openai import AuthenticationError, RateLimitError, APIConnectionError, APIStatusError
        stream = client.chat.completions.create(**kwargs)
    except (AuthenticationError, RateLimitError, APIConnectionError, APIStatusError) as e:
        import sys;
        if _is_nvidia:
            print(f"[nvidia-web RAW ERROR] {type(e).__name__}: {e}", file=sys.stderr, flush=True)
            if not config.get("_nvidia_fallback_active"):
                chain = _get_nvidia_fallback_chain(config)
                bare = model.split("/", 1)[-1] if "/" in model else model
                try:
                    idx = chain.index(bare)
                    remaining = chain[idx + 1:]
                except ValueError:
                    remaining = chain
                for next_model in remaining:
                    full = f"nvidia-web/{next_model}"
                    yield TextChunk(f"\n⚡ NVIDIA rate limit — switching to {next_model}...\n")
                    fallback_config = {**config, "_nvidia_fallback_active": True}
                    yield from stream_openai_compat(api_key, base_url, full, system, messages, tool_schemas, fallback_config)
                    return
        if _is_modelstudio and _ms_remaining:
            _nm = _ms_remaining[0]
            yield TextChunk(f"\n⚡ Model Studio: '{model}' failed ({type(e).__name__}) — switching to {_nm}…\n")
            yield from stream_openai_compat(
                api_key, base_url, _nm, system, messages, tool_schemas,
                {**config, "_modelstudio_remaining": _ms_remaining[1:]},
            )
            return
        msg = friendly_api_error(e)
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return
    except Exception as e:
        if _is_nvidia:
            import sys; print(f"[nvidia-web RAW ERROR] {type(e).__name__}: {e}", file=sys.stderr, flush=True)
        if _is_modelstudio and _ms_remaining:
            _nm = _ms_remaining[0]
            yield TextChunk(f"\n⚡ Model Studio: '{model}' failed ({type(e).__name__}) — switching to {_nm}…\n")
            yield from stream_openai_compat(
                api_key, base_url, _nm, system, messages, tool_schemas,
                {**config, "_modelstudio_remaining": _ms_remaining[1:]},
            )
            return
        msg = friendly_api_error(e)
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return

    in_thought = False
    def _extract_cached(u) -> int:
        # Cached prompt tokens come in different shapes depending on provider:
        #   OpenAI:    usage.prompt_tokens_details.cached_tokens
        #   Kimi/code: usage.cached_tokens (top-level) or same as OpenAI
        #   DeepSeek:  usage.prompt_cache_hit_tokens
        #   Anthropic-style proxy: usage.cache_read_input_tokens
        c = 0
        details = getattr(u, "prompt_tokens_details", None)
        if details:
            c = (
                getattr(details, "cached_tokens", 0)
                or (details.get("cached_tokens", 0) if isinstance(details, dict) else 0)
                or 0
            )
        return (
            c
            or getattr(u, "cached_tokens", 0)
            or getattr(u, "prompt_cache_hit_tokens", 0)
            or getattr(u, "cache_read_input_tokens", 0)
            or 0
        )

    # Iterate manually: quota errors (e.g. NVIDIA free-tier "ResourceExhausted")
    # surface on the first READ of the SSE stream, after create() has already
    # returned — so the try/except around create() above never sees them.
    # Catching them here converts the crash into a normal error turn (issue #18).
    _stream_iter = iter(stream)
    while True:
        try:
            chunk = next(_stream_iter)
        except StopIteration:
            break
        except Exception as e:
            _partial = bool(text or thinking or tool_buf)
            # Transient mid-stream failures (connection reset, 5xx) keep the
            # old behavior: propagate so _ProviderRetry can retry the request.
            if _ProviderRetry.is_retryable(e):
                raise
            if _is_nvidia:
                import sys
                print(f"[nvidia-web RAW ERROR] {type(e).__name__}: {e}", file=sys.stderr, flush=True)
                # Fall back only when nothing streamed yet — otherwise the
                # rerun would duplicate the partial answer.
                if not _partial and not config.get("_nvidia_fallback_active"):
                    chain = _get_nvidia_fallback_chain(config)
                    bare = model.split("/", 1)[-1] if "/" in model else model
                    try:
                        idx = chain.index(bare)
                        remaining = chain[idx + 1:]
                    except ValueError:
                        remaining = chain
                    for next_model in remaining:
                        full = f"nvidia-web/{next_model}"
                        yield TextChunk(f"\n⚡ NVIDIA quota/rate limit — switching to {next_model}...\n")
                        fallback_config = {**config, "_nvidia_fallback_active": True}
                        yield from stream_openai_compat(api_key, base_url, full, system, messages, tool_schemas, fallback_config)
                        return
            if _is_modelstudio and _ms_remaining and not _partial:
                _nm = _ms_remaining[0]
                yield TextChunk(f"\n⚡ Model Studio: '{model}' failed mid-stream ({type(e).__name__}) — switching to {_nm}…\n")
                yield from stream_openai_compat(
                    api_key, base_url, _nm, system, messages, tool_schemas,
                    {**config, "_modelstudio_remaining": _ms_remaining[1:]},
                )
                return
            msg = friendly_api_error(e)
            yield TextChunk(("\n" if _partial else "") + msg)
            yield AssistantTurn(msg, [], in_tok, out_tok, error=True)
            return
        if not chunk.choices:
            # usage-only chunk (some providers send this last)
            if hasattr(chunk, "usage") and chunk.usage:
                u = chunk.usage
                in_tok  = getattr(u, "prompt_tokens", 0) or in_tok
                out_tok = getattr(u, "completion_tokens", 0) or out_tok
                cached_tok = _extract_cached(u) or cached_tok
            continue

        choice = chunk.choices[0]
        delta  = choice.delta

        content = delta.content
        if content:
            # Heuristic: detect reasoning tags in the content stream
            lower_c = content.lower()
            if "<thought" in lower_c or "<reasoning" in lower_c:
                in_thought = True
            
            # If we are inside a thought block, check for closing tags
            if in_thought:
                if "</thought" in lower_c or "</reasoning" in lower_c:
                    # Closing tag found: yield current chunk as thinking, then flip
                    yield ThinkingChunk(content)
                    in_thought = False
                else:
                    yield ThinkingChunk(content)
            else:
                text += content
                yield TextChunk(content)

        # Capture native reasoning content (DeepSeek/Gemini/OpenAI/Custom)
        reasoning = (
            getattr(delta, "reasoning_content", None)
            or getattr(delta, "reasoning", None)
            or getattr(delta, "thought", None)
        )
        if reasoning:
            thinking += reasoning
            yield ThinkingChunk(reasoning)

        if delta.tool_calls:
            for tc in delta.tool_calls:
                idx = tc.index
                if idx not in tool_buf:
                    tool_buf[idx] = {"id": "", "name": "", "args": "", "extra_content": None}
                if tc.id:
                    tool_buf[idx]["id"] = tc.id
                if tc.function:
                    if tc.function.name:
                        tool_buf[idx]["name"] += tc.function.name
                    if tc.function.arguments:
                        tool_buf[idx]["args"] += tc.function.arguments
                # Capture extra_content (e.g. Gemini thought_signature)
                extra = getattr(tc, "extra_content", None)
                if extra:
                    tool_buf[idx]["extra_content"] = extra

        # Some providers include usage in the last chunk
        if hasattr(chunk, "usage") and chunk.usage:
            u = chunk.usage
            in_tok  = (getattr(u, "prompt_tokens", 0) or getattr(u, "prompt_token_count", 0) or in_tok)
            out_tok = (getattr(u, "completion_tokens", 0) or getattr(u, "candidate_token_count", 0) or out_tok)
            cached_tok = _extract_cached(u) or cached_tok
        elif hasattr(chunk, "x_groq") and chunk.x_groq and "usage" in chunk.x_groq:
            # Groq-specific usage
            u = chunk.x_groq["usage"]
            in_tok = u.get("prompt_tokens", 0) or in_tok
            out_tok = u.get("completion_tokens", 0) or out_tok
        elif hasattr(chunk, "model_extra") and chunk.model_extra and "usage" in chunk.model_extra:
            # Pydantic v2 / Gemini proxy fallback
            u = chunk.model_extra["usage"]
            if u:
                if isinstance(u, dict):
                    in_tok = u.get("prompt_tokens", 0) or in_tok
                    out_tok = u.get("completion_tokens", 0) or out_tok
                else:
                    in_tok = getattr(u, "prompt_tokens", 0) or in_tok
                    out_tok = getattr(u, "completion_tokens", 0) or out_tok

    tool_calls = _finalize_tool_calls(tool_buf, include_extra=True)

    yield AssistantTurn(
        text, tool_calls, in_tok, out_tok,
        thinking=thinking,
        cache_read_tokens=cached_tok,
    )


def _flatten_tool_messages(messages: list) -> list:
    """Convert tool-call history to plain text for models without native tool support.

    Transforms:
      - assistant messages with tool_calls → text + inline <tool_call> representation
      - role:tool messages → role:user with [Tool Result] prefix
    This lets the model see the full conversation without needing the tools API.
    """
    out = []
    for m in messages:
        role = m.get("role", "")

        if role == "assistant":
            text = m.get("content") or ""
            tcs = m.get("tool_calls", [])
            if tcs:
                # Append inline <tool_call> tags so the model sees what it called
                parts = [text] if text else []
                for tc in tcs:
                    fn = tc.get("function", {})
                    name = fn.get("name", tc.get("name", ""))
                    args = fn.get("arguments", tc.get("input", {}))
                    if isinstance(args, str):
                        try:
                            args = json.loads(args)
                        except Exception:
                            pass
                    parts.append(
                        f'<tool_call>{json.dumps({"name": name, "input": args}, ensure_ascii=False)}</tool_call>'
                    )
                out.append({"role": "assistant", "content": "\n".join(parts)})
            else:
                out.append({"role": "assistant", "content": text})

        elif role == "tool":
            # Convert tool result to a user message the model can read
            name = m.get("name", m.get("tool_call_id", "unknown"))
            content = m.get("content", "")
            # Make format more explicit for DeepSeek-R1
            tool_result_msg = f"[Tool Result: {name}]\n{content}\n\n[INSTRUCTION: Use this data to respond. Do not ask what to do next.]"
            out.append({
                "role": "user",
                "content": tool_result_msg,
            })

        else:
            # system / user — pass through as-is
            out.append(m)

    return out


def _build_prompt_tool_manifest(tool_schemas: list) -> str:
    """Build the text block injected into the system prompt for prompt-based tool calling."""
    oai_tools = tools_to_openai(tool_schemas)
    tool_lines = []
    for t in oai_tools:
        fn = t.get("function", t)
        name = fn.get("name", "")
        desc = fn.get("description", "")
        params = json.dumps(fn.get("parameters", {}))
        tool_lines.append(f"  - {name}: {desc}\n    Parameters: {params}")

    return (
        "\n\n[TOOL USE]\nYou have access to these tools. "
        "When you need to use a tool, you MUST output EXACTLY this format (no extra text):\n"
        '<tool_call>{"name": "tool_name", "input": {"param": "value"}}</tool_call>\n\n'
        "EXAMPLE - If you need to search, output ONLY this exact line:\n"
        '<tool_call>{"name": "web_search", "input": {"query": "search term"}}</tool_call>\n\n'
        "CRITICAL RULES:\n"
        "1. ALWAYS wrap the JSON in <tool_call>...</tool_call> tags\n"
        "2. The <tool_call> tag must be on its own line with NO extra text before or after\n"
        "3. Use ONLY the exact JSON format: {\"name\": \"tool_name\", \"input\": {...}}\n"
        "4. Output the tool call IMMEDIATELY - do not explain first\n"
        "5. Wait for the tool result before continuing\n"
        "6. After your thinking is done, output ONLY the <tool_call> line\n"
        "7. AFTER calling a tool, WAIT for [Tool Result] and READ it before responding\n"
        "8. DO NOT call multiple tools at once - wait for each result\n\n"
        "WORKFLOW EXAMPLE:\n"
        "User: 'List files'\n"
        "Assistant: <tool_call>{\"name\": \"Glob\", \"input\": {\"pattern\": \"*\"}}</tool_call>\n"
        "[Tool Result: Glob]\n"
        "file1.txt\nfile2.txt\n"
        "Assistant: 'Found files: file1.txt, file2.txt'\n\n"
        "Available tools:\n" + "\n".join(tool_lines)
    )


def _get_gcloud_token() -> str:
    """Obtain OAuth2 access token from gcloud CLI."""
    use_shell = platform.system() == "Windows"
    result = subprocess.run(
        "gcloud auth print-access-token",
        capture_output=True,
        text=True,
        check=True,
        shell=use_shell,
    )
    return result.stdout.strip()


def _openai_messages_to_vertex_contents(messages: list) -> list:
    """Convert OpenAI-format messages to Vertex AI generateContent 'contents'."""
    contents = []
    for m in messages:
        role = m.get("role", "user")
        content = m.get("content", "")

        if role == "system":
            continue  # handled separately as systemInstruction

        if role == "user":
            parts = []
            if content:
                parts.append({"text": content})
            contents.append({"role": "user", "parts": parts})

        elif role == "assistant":
            parts = []
            if content:
                parts.append({"text": content})
            # Native tool calls from OpenAI format
            for tc in m.get("tool_calls", []):
                fn = tc.get("function", {})
                name = fn.get("name", tc.get("name", ""))
                args = fn.get("arguments", tc.get("input", {}))
                if isinstance(args, str):
                    try:
                        args = json.loads(args)
                    except Exception:
                        args = {}
                if name:
                    parts.append({
                        "functionCall": {"name": name, "args": args}
                    })
            contents.append({"role": "model", "parts": parts})

        elif role == "tool":
            name = m.get("name", m.get("tool_call_id", "unknown"))
            parts = [{
                "functionResponse": {
                    "name": name,
                    "response": {"result": content},
                }
            }]
            contents.append({"role": "user", "parts": parts})

    return contents


def _openai_tools_to_vertex_tools(tool_schemas: list) -> list:
    """Convert OpenAI-format tools to Vertex AI functionDeclarations."""
    declarations = []
    for t in tool_schemas:
        if not isinstance(t, dict):
            continue
        fn = t.get("function", t)
        name = fn.get("name", t.get("name", ""))
        if not name:
            continue
        declarations.append({
            "name": name,
            "description": fn.get("description", t.get("description", "")),
            "parameters": fn.get("parameters", t.get("input_schema", {"type": "object", "properties": {}})),
        })
    return [{"functionDeclarations": declarations}] if declarations else []


def stream_gcloud(
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
) -> Generator:
    """Stream from Google Cloud Vertex AI using gcloud OAuth2 authentication.

    Uses the generateContent REST API directly with Bearer tokens from
    `gcloud auth print-access-token`. Supports native function calling.
    """
    # ── Auth ────────────────────────────────────────────────────────────────
    try:
        token = _get_gcloud_token()
    except Exception as e:
        msg = f"[gcloud] Failed to get gcloud token: {e}. Run `gcloud auth login`."
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return

    # ── Configurable project/location (fallback to hardcoded) ─────────────
    project_id = config.get("gcloud_project_id", "gen-lang-client-0108363942")
    location   = config.get("gcloud_location", "us-west1")
    bare = model.split("/")[-1] if "/" in model else model

    url = (
        f"https://{location}-aiplatform.googleapis.com/v1/"
        f"projects/{project_id}/locations/{location}/"
        f"publishers/google/models/{bare}:generateContent"
    )

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }

    # ── Convert messages ────────────────────────────────────────────────────
    oai_messages = messages_to_openai(messages)
    contents = _openai_messages_to_vertex_contents(oai_messages)

    # Gemini 2.5 models "think" before answering, and that reasoning is billed
    # against maxOutputTokens. With a small cap the model can burn the whole
    # budget thinking and return no answer at all ("no responde"). Give it a
    # generous output budget so thinking never starves the answer — general fix,
    # works no matter what thinking/reasoning toggle the user has on.
    prov_cap = PROVIDERS.get("gcloud", {}).get("max_completion_tokens", 65536) or 65536
    req_max = config.get("max_tokens") or 0
    safe_max = min(prov_cap, max(req_max, 32768))

    payload: dict = {
        "contents": contents,
        "generationConfig": {
            "temperature": config.get("temperature", 0.7),
            "maxOutputTokens": safe_max,
            # Surface the model's reasoning as thought parts; depth stays dynamic
            # (the model decides) so this behaves for every model and toggle.
            "thinkingConfig": {"includeThoughts": True},
        },
    }

    if system:
        payload["systemInstruction"] = {"parts": [{"text": system}]}

    # ── Tools ───────────────────────────────────────────────────────────────
    if tool_schemas and not config.get("no_tools"):
        vertex_tools = _openai_tools_to_vertex_tools(tools_to_openai(tool_schemas))
        if vertex_tools:
            payload["tools"] = vertex_tools
            payload["toolConfig"] = {
                "functionCallingConfig": {"mode": "AUTO"}
            }

    # ── Request ─────────────────────────────────────────────────────────────
    text = ""
    thinking = ""
    tool_calls: list = []
    in_tok = out_tok = 0

    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=120)
        if resp.status_code != 200:
            msg = f"[gcloud] HTTP {resp.status_code}: {resp.text[:400]}"
            yield TextChunk(msg)
            yield AssistantTurn(msg, [], 0, 0, error=True)
            return
        data = resp.json()
    except Exception as e:
        msg = f"[gcloud] Request error: {e}"
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return

    # ── Parse response ──────────────────────────────────────────────────────
    candidates = data.get("candidates", [])
    if not candidates:
        msg = "[gcloud] No candidates in response."
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return

    candidate = candidates[0]
    parts = candidate.get("content", {}).get("parts", [])

    for part in parts:
        if "text" in part:
            # Gemini flags reasoning parts with "thought": true — route those to
            # the thinking channel; everything else is the visible answer.
            if part.get("thought"):
                chunk_think = part["text"]
                thinking += chunk_think
                yield ThinkingChunk(chunk_think)
            else:
                chunk_text = part["text"]
                text += chunk_text
                yield TextChunk(chunk_text)

        if "functionCall" in part:
            fc = part["functionCall"]
            tool_calls.append({
                "id": f"call_gc_{len(tool_calls)}",
                "name": fc.get("name", ""),
                "input": fc.get("args", {}),
            })

    # Token usage (Vertex AI sometimes includes usageMetadata)
    usage = data.get("usageMetadata", {})
    in_tok = usage.get("promptTokenCount", 0)
    out_tok = usage.get("candidatesTokenCount", 0)

    # Safety net: if the model produced no answer and no tool call (e.g. it hit
    # the token cap mid-thought), say so instead of returning dead silence.
    if not text and not tool_calls:
        fr = candidate.get("finishReason", "")
        note = f"[gcloud] No answer returned (finishReason={fr or 'unknown'}). Try again."
        yield TextChunk(note)
        yield AssistantTurn(note, [], in_tok, out_tok, thinking=thinking, error=True)
        return

    yield AssistantTurn(text, tool_calls, in_tok, out_tok, thinking=thinking)


def _ollama_pull_stream(base_url: str, model: str, timeout: int = 1800) -> Generator:
    """Pull a model into Ollama, yielding human-readable progress lines.

    Mirrors what `ollama run <model>` does implicitly: if the model isn't on
    disk, download it first instead of dying with a 404. This is what makes
    the HuggingFace "Use this model -> Dulus" snippet
    (`dulus --model ollama/hf.co/<repo>:<quant>`) work verbatim on a machine
    that has never seen that GGUF before.

    Yields str progress lines. Final yield is True on success, False on failure.
    """
    import time as _t
    req = urllib.request.Request(
        f"{base_url.rstrip('/')}/api/pull",
        data=json.dumps({"model": model, "stream": True}).encode("utf-8"),
        headers={"Content-Type": "application/json"},
    )
    last_emit = 0.0
    last_pct = -1
    ok = False
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            for raw in resp:
                if not raw:
                    continue
                try:
                    ev = json.loads(raw.decode("utf-8", "replace").strip())
                except Exception:
                    continue
                if ev.get("error"):
                    yield f"\n  [ollama] pull failed: {ev['error']}\n"
                    return
                status = ev.get("status", "")
                total = ev.get("total") or 0
                done = ev.get("completed") or 0
                now = _t.time()
                if total and done:
                    pct = int(done * 100 / total)
                    # Throttle: only redraw every 2% or every 1.5s.
                    if pct != last_pct and (pct - last_pct >= 2 or now - last_emit > 1.5):
                        mb_d, mb_t = done / 1048576, total / 1048576
                        bar_n = int(pct / 5)
                        bar = "#" * bar_n + "-" * (20 - bar_n)
                        yield f"\r  [{bar}] {pct:3d}%  {mb_d:.0f}/{mb_t:.0f} MB  {status}"
                        last_pct, last_emit = pct, now
                elif status and now - last_emit > 0.5:
                    yield f"\r  [ollama] {status}...".ljust(60)
                    last_emit = now
                if status == "success":
                    ok = True
    except urllib.error.HTTPError as he:
        body = ""
        try:
            body = he.read().decode("utf-8", "replace")[:200]
        except Exception:
            pass
        yield f"\n  [ollama] pull rejected ({he.code}): {body or he.reason}\n"
        return
    except Exception as e:
        yield f"\n  [ollama] pull error: {type(e).__name__}: {e}\n"
        return
    yield ok


def stream_ollama(
    base_url: str,
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
) -> Generator:
    # pass_images=True: Ollama /api/chat accepts base64 images natively in the message
    oai_messages: list = [{"role": "system", "content": system}] + messages_to_openai(messages, ollama_native_images=True)
    
    # Ollama requires tool arguments as dict objects, not strings. OpenAI uses strings.
    for m in oai_messages:
        if m.get("content") is None:
            m["content"] = ""
        if "tool_calls" in m and m["tool_calls"]:
            for tc in m["tool_calls"]:
                fn = tc.get("function", {})
                if isinstance(fn.get("arguments"), str):
                    try:
                        fn["arguments"] = json.loads(fn["arguments"])
                    except Exception:
                        pass

    # ── DeepSeek-R1 Specific Fix ─────────────────────────────────────────
    # Simplified instructions for smaller models
    is_deepseek_r1 = "deepseek-r1" in model.lower()
    if is_deepseek_r1:
        deepseek_fix = (
            '\n\nRules: Reply directly. Use tools ONLY when needed. Format: <tool_call>{"name": "...", "input": {}}</tool_call>\n'
        )
        for msg in oai_messages:
            if msg.get("role") == "system":
                msg["content"] += deepseek_fix
                break

    # ── Check if a previous turn already detected no native tool support ──
    # Use model-specific key to persist across sessions
    _no_native_tools_key = f"_no_native_tools_{model}"
    _prompt_tool_mode = False
    
    # Check both the old generic flag and the new model-specific flag
    if (config.get("_prompt_tool_mode") or config.get(_no_native_tools_key)) and tool_schemas and not config.get("no_tools"):
        _prompt_tool_mode = True
        # Flatten tool messages in history so the model can read them as plain text
        oai_messages = _flatten_tool_messages(oai_messages)
        # Inject tool manifest into system prompt
        tool_manifest = _build_prompt_tool_manifest(tool_schemas)
        for msg in oai_messages:
            if msg.get("role") == "system":
                msg["content"] += tool_manifest
                break

    payload = {
        "model": model,
        "messages": oai_messages,
        "stream": True,
        "options": {
            # Context window and output length are independent knobs:
            #   context_limit -> num_ctx (how much the model can READ)
            #   max_tokens    -> num_predict (how much it can WRITE)
            # 32k is the num_ctx fallback only when context_limit is unset.
            "num_ctx": int(config.get("context_limit") or 32768),
            "num_predict": int(config.get("max_tokens") or -1),
        }
    }

    # Honor `thinking` — tell Ollama-hosted reasoning models (Qwen3, etc.)
    # to skip the thinking phase instead of streaming <think> blocks.
    # Cap at level 3: level 4 stalls Ollama after the first message.
    _thk = _thinking_level_from(config.get("thinking", 0))
    if _thk > 3:
        _thk = 3
        config = {**config, "thinking": 3}
    if _thk == 0:
        payload["think"] = False
        payload["options"]["enable_thinking"] = False

    if tool_schemas and not config.get("no_tools") and not _prompt_tool_mode:
        payload["tools"] = tools_to_openai(tool_schemas)
    
    def _make_request(p):
        return urllib.request.Request(
            f"{base_url.rstrip('/')}/api/chat",
            data=json.dumps(p).encode("utf-8"),
            headers={"Content-Type": "application/json"}
        )

    req = _make_request(payload)

    text = ""
    thinking = ""
    tool_buf: dict = {}

    # Native tool-call interceptor state. When the model emits its native
    # `<|tool_call|>...` envelope inside `content` (Gemma 3/4 in particular
    # do this even when given OpenAI-style tool schemas), we stop yielding
    # text and accumulate everything from the marker onward. At end-of-stream
    # we parse the buffer into proper tool_calls. Without this the user sees
    # `<|tool_call>call:Foo{...}<tool_call|>` as raw text, the tool never
    # fires, and on Ollama Cloud the malformed exchange can trip a 502.
    _native_buf = ""           # text accumulated after a native marker
    _native_intercept = False  # True once we've seen any native marker

    # State for prompt-based tool call parsing across streamed chunks
    use_deep_tools = config.get("deep_tools", False) if config else False
    _auto_wrap_json = is_deepseek_r1 and use_deep_tools
    parser = WebToolParser(auto_wrap_json=_auto_wrap_json)

    # Cloud-routed Ollama models (e.g. minimax-m2.7:cloud) need a moment before
    # the proxy starts streaming real content — without this, the first response
    # can come back empty.
    import time as _time
    if ":cloud" in model.lower():
        _time.sleep(1)

    try:
        resp_cm = urllib.request.urlopen(req)
    except urllib.error.HTTPError as he:
        # Ollama (esp. Cloud) returns transient 5xx/429 under load. This urlopen
        # happens BEFORE any streaming, so retrying re-issues the request cleanly
        # with no duplicate output. Retry transient codes a few times, then fail
        # SOFT with a friendly turn instead of letting the HTTPError bubble all
        # the way up and crash the app (Sentry: HTTP 500 in stream_ollama).
        resp_cm = None

        # 404 "model not found" = the GGUF simply isn't pulled yet. `ollama run`
        # downloads it transparently; the HTTP API does not. Match `ollama run`
        # so the HuggingFace snippet `dulus --model ollama/hf.co/<repo>:<quant>`
        # works verbatim on a fresh machine instead of dying with a bare 404.
        if he.code == 404 and not (config or {}).get("no_auto_pull"):
            _body = ""
            try:
                _body = he.read().decode("utf-8", "replace")
            except Exception:
                pass
            if "not found" in (_body or "").lower() or "not found" in str(getattr(he, "reason", "")).lower():
                yield TextChunk(f"\n  [ollama] Model '{model}' not found locally — downloading...\n")
                _pulled = False
                for _ev in _ollama_pull_stream(base_url, model):
                    if isinstance(_ev, bool):
                        _pulled = _ev
                    else:
                        yield TextChunk(_ev)
                if _pulled:
                    yield TextChunk("\n  [ollama] Download complete. Starting...\n")
                    try:
                        resp_cm = urllib.request.urlopen(_make_request(payload))
                    except urllib.error.HTTPError as _he2:
                        # Fall through to the 400/transient handling below — a
                        # freshly pulled GGUF very often has no tool template,
                        # which surfaces as 400 "does not support tools".
                        he = _he2
                    except Exception as _e2:
                        _m = (f"[ollama] Model '{model}' downloaded but the request still "
                              f"failed: {_e2}. Try again or switch with /model.")
                        yield TextChunk(_m)
                        yield AssistantTurn(_m, [], 0, 0, error=True)
                        return
                else:
                    _m = (f"[ollama] Could not download '{model}' from {base_url}. "
                          "Check the model name (for GGUF use "
                          "`ollama/hf.co/<user>/<repo>:<QUANT>`) and your connection.")
                    yield TextChunk(_m)
                    yield AssistantTurn(_m, [], 0, 0, error=True)
                    return

        # 400 "<model> does not support tools" — many GGUF chat models (SmolLM2,
        # base Llama/Phi builds, most HuggingFace quants) have no tool-calling
        # template, so Ollama rejects the request outright. Dulus already has a
        # prompt-based tool mode for exactly this case, but nothing ever flipped
        # the switch. Detect it here, remember it for this model, and replay the
        # turn with the prompt-tool manifest instead of native `tools`.
        if resp_cm is None and he.code == 400:
            _body400 = ""
            try:
                _body400 = he.read().decode("utf-8", "replace")
            except Exception:
                pass
            if "does not support tools" in _body400.lower():
                if config is not None:
                    config[_no_native_tools_key] = True
                if not _prompt_tool_mode:
                    yield from stream_ollama(base_url, model, system, messages,
                                             tool_schemas, config)
                    return
                _m = (f"[ollama] Model '{model}' does not support tools and the "
                      "prompt-based fallback also failed. Use a tool-capable model "
                      "or run with --no-tools.")
                yield TextChunk(_m)
                yield AssistantTurn(_m, [], 0, 0, error=True)
                return

        _transient = resp_cm is None and he.code in (408, 425, 429, 500, 502, 503, 504)
        if _transient:
            for _attempt in range(3):
                _time.sleep(min(1.5 * (2 ** _attempt), 8.0))
                try:
                    resp_cm = urllib.request.urlopen(req)
                    break
                except urllib.error.HTTPError as he2:
                    he = he2
                    continue
                except Exception:
                    break
        if resp_cm is None:
            code = getattr(he, "code", "?")
            if _transient:
                msg = (f"[ollama] Server error {code} from {base_url} after retries. "
                       "The Ollama backend is overloaded or unavailable — try again in a "
                       "moment, or switch provider with /model.")
            else:
                msg = (f"[ollama] Request rejected ({code}) by {base_url}: {he.reason}. "
                       "Check the model name and your Ollama config, or switch with /model.")
            yield TextChunk(msg)
            yield AssistantTurn(msg, [], 0, 0, error=True)
            return
    except (urllib.error.URLError, ConnectionError, OSError) as e:
        # Ollama not reachable. NEVER auto-start `ollama serve` from here:
        # spawning it detached with default env stomps the user's own server
        # config (and the retry loop hammered the port for ~12s). Fail fast
        # with a clear message instead.
        import shutil as _sh
        if _sh.which("ollama"):
            msg = (f"[ollama] Could not connect to Ollama at {base_url} ({e}). "
                   "Start it yourself with `ollama serve`, or switch provider with /model.")
        else:
            msg = (f"[ollama] Could not connect to Ollama at {base_url} ({e}). "
                   "Ollama isn't installed — get it at https://ollama.com/download, "
                   "or switch to a cloud provider with /model. Run /help for options.")
        yield TextChunk(msg)
        yield AssistantTurn(msg, [], 0, 0, error=True)
        return

    # Buffer for accumulating thinking content to reduce word-by-word chunks
    _thinking_buffer = ""
    
    with resp_cm as resp:
        for line in resp:
            if not line.strip(): continue
            try:
                data = json.loads(line)
            except Exception: continue

            msg = data.get("message", {})
            reasoning = None
            for r_key in ["thinking", "reasoning", "thought", "reasoning_content"]:
                if r_key in msg and msg[r_key]:
                    reasoning = msg[r_key]
                    break

            if reasoning:
                thinking += reasoning
                _thinking_buffer += reasoning
                if len(_thinking_buffer) >= 20 or (_thinking_buffer and reasoning and reasoning[0] in " \n\t.,;:!?"):
                    yield ThinkingChunk(_thinking_buffer)
                    _thinking_buffer = ""

            content = msg.get("content", "") if "content" in msg else ""
            if content:
                # Flush thinking buffer before content
                if _thinking_buffer:
                    yield ThinkingChunk(_thinking_buffer)
                    _thinking_buffer = ""
                
                if _prompt_tool_mode:
                    display = parser.parse_chunk(content)
                    if display:
                        text += display
                        yield TextChunk(display)
                elif _native_intercept:
                    # Already inside a native tool-call envelope — buffer silently.
                    _native_buf += content
                else:
                    marker = _find_native_tool_marker(content)
                    if marker is not None:
                        # Yield clean prefix, then start buffering from the marker.
                        prefix = content[:marker]
                        if prefix:
                            text += prefix
                            yield TextChunk(prefix)
                        _native_buf += content[marker:]
                        _native_intercept = True
                    else:
                        text += content
                        yield TextChunk(content)

            # Handle native ollama tools format
            for tc in msg.get("tool_calls", []):
                fn = tc.get("function", {})
                idx = len(tool_buf) # Ollama sends complete tool calls, not delta
                tool_buf[idx] = {
                    "id": "call_ollama" + str(idx),
                    "name": fn.get("name", ""),
                    "input": fn.get("arguments", {})
                }

    # Flush any remaining thinking buffer at end of stream
    if _thinking_buffer:
        yield ThinkingChunk(_thinking_buffer)

    if _prompt_tool_mode:
        remaining = parser.flush()
        if remaining:
            text += remaining
            yield TextChunk(remaining)

    tool_calls = []
    # Merge native Ollama tools
    for idx in sorted(tool_buf):
        v = tool_buf[idx]
        tool_calls.append({"id": v["id"], "name": v["name"], "input": v["input"]})

    # Merge native-format tool calls intercepted from `content` (Gemma 3/4 etc.)
    if _native_intercept:
        intercepted = _extract_native_tool_calls(_native_buf)
        if intercepted:
            tool_calls.extend(intercepted)
        else:
            # Parser couldn't make sense of it — surface the raw buffer so the
            # user sees something instead of a silent stall.
            text += _native_buf
            yield TextChunk(_native_buf)

    # Merge prompt-based tools from parser
    if _prompt_tool_mode:
        for tc in parser.tool_calls:
            tool_calls.append(tc)

    # NOTE: Sanitizer temporarily disabled due to space issues
    # if is_deepseek_r1:
    #     text = _sanitize_deepseek_output(text)
    #     thinking = _sanitize_deepseek_output(thinking)

    # Ollama doesn't return exact token counts via livestream easily until "done",
    # but we can do a rough estimate or 0, dulus handles zero gracefully

    # For cloud-routed models: if text is empty (timing issue), retry once with longer wait
    if not text and not tool_calls and ":cloud" in model.lower():
        _time.sleep(2)
        try:
            req2 = _make_request(payload)
            text2 = ""
            thinking2 = ""
            with urllib.request.urlopen(req2) as resp2:
                for line in resp2:
                    if not line.strip(): continue
                    try:
                        data2 = json.loads(line)
                    except Exception: continue
                    msg2 = data2.get("message", {})
                    c2 = msg2.get("content", "") if "content" in msg2 else ""
                    if c2:
                        text2 += c2
                        yield TextChunk(c2)
                    for r_key in ["thinking", "reasoning", "thought", "reasoning_content"]:
                        if r_key in msg2 and msg2[r_key]:
                            thinking2 += msg2[r_key]
                            break
            yield AssistantTurn(text2 or "[ollama-cloud: no response]", [], 0, 0, thinking=thinking2)
        except Exception as _e:
            msg_err = f"[ollama-cloud] Retry failed: {_e}"
            yield TextChunk(msg_err)
            yield AssistantTurn(msg_err, [], 0, 0, error=True)
        return

    yield AssistantTurn(text, tool_calls, 0, 0, thinking=thinking)


# ── Kimi Code OAuth (Kimi membership login — NO API key) ──────────────────
# Device-Authorization flow ported from the official open-source Kimi Code CLI
# (MoonshotAI/kimi-code, packages/oauth — TypeScript). Three POSTs, all
# form-encoded against auth.kimi.com:
#   1. /api/oauth/device_authorization      → user_code + device_code + verify URL
#   2. /api/oauth/token (grant=device_code) → poll until the user approves
#   3. /api/oauth/token (grant=refresh_token) → refresh on expiry
# The access token authenticates the SAME api.kimi.com/coding/v1 endpoint the
# kimi-code API-key provider uses — usage bills against the user's Kimi
# membership, not platform API credits. Reuses the official CLI's credential
# file (~/.kimi-code/credentials/kimi-code.json) when present — one login, both.
KIMI_OAUTH_HOST           = "https://auth.kimi.com"
KIMI_OAUTH_CLIENT_ID      = "17e5f671-d194-4dfb-9706-5516cb48c098"  # official Kimi Code CLI client
KIMI_OAUTH_DEVICE_URL     = f"{KIMI_OAUTH_HOST}/api/oauth/device_authorization"
KIMI_OAUTH_TOKEN_URL      = f"{KIMI_OAUTH_HOST}/api/oauth/token"
KIMI_OAUTH_DEVICE_GRANT   = "urn:ietf:params:oauth:grant-type:device_code"
KIMI_OAUTH_API_BASE_URL   = "https://api.kimi.com/coding/v1"
KIMI_OAUTH_UA             = "KimiCLI/1.30.0"  # same UA the kimi-code API whitelists


def _kimi_oauth_store_path():
    return _oauth_store_path("kimi_oauth.json")


def _kimi_oauth_load_store() -> dict:
    return _oauth_load_store(_kimi_oauth_store_path())


def _kimi_oauth_save_store(data: dict) -> None:
    _oauth_save_store(_kimi_oauth_store_path(), data)


def _kimi_oauth_clear_store() -> list[str]:
    """Delete the Kimi OAuth session *and* the device identity.

    A plain re-login is not enough to switch Kimi accounts: the refresh token in
    ``kimi_oauth.json`` silently revives the previous session, and the stable
    ``kimi_device_id`` keeps the host tied to the old device. Both files have to
    go before the device flow can bind a different account.

    Returns the names of the files actually removed (for user feedback).
    """
    removed: list[str] = []
    for path in (_kimi_oauth_store_path(), _oauth_store_path("kimi_device_id")):
        try:
            path.unlink()
            removed.append(path.name)
        except FileNotFoundError:
            pass
        except OSError:
            pass
    return removed


def _kimi_device_headers() -> dict:
    """X-Msh-* device identity headers the official CLI sends on OAuth calls.

    The device id is a random UUID persisted next to the Dulus auth stores so
    the host sees a stable device across logins/refreshes."""
    import platform
    import uuid

    id_path = _oauth_store_path("kimi_device_id")
    device_id = ""
    try:
        device_id = id_path.read_text(encoding="utf-8").strip()
    except OSError:
        pass
    if not device_id:
        device_id = str(uuid.uuid4())
        try:
            id_path.write_text(device_id, encoding="utf-8")
        except OSError:
            pass
    model = f"{platform.system()} {platform.release()} {platform.machine()}".strip()
    return {
        "User-Agent": KIMI_OAUTH_UA,
        "X-Msh-Platform": "kimi_code_cli",
        "X-Msh-Version": KIMI_OAUTH_UA.split("/")[-1],
        "X-Msh-Device-Name": platform.node() or "unknown",
        "X-Msh-Device-Model": model or "unknown",
        "X-Msh-Os-Version": platform.release() or "unknown",
        "X-Msh-Device-Id": device_id,
    }


def _kimi_oauth_load_cli_credentials() -> dict:
    """Load the official Kimi Code CLI credential file, if present.

    Same snake_case wire format the TypeScript CLI writes to
    ~/.kimi-code/credentials/kimi-code.json — so logging in via the official
    CLI (or via Dulus) is a single sign-on for both."""
    import json
    import pathlib

    path = pathlib.Path.home() / ".kimi-code" / "credentials" / "kimi-code.json"
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError):
        return {}
    return data if isinstance(data, dict) else {}


def _kimi_oauth_refresh(refresh_token: str) -> dict | None:
    """Exchange a refresh_token for a fresh access token (official CLI's
    /api/oauth/token grant=refresh_token). Returns the updated store or None.
    401/403/invalid_grant (revoked) and network errors all land on None — the
    caller then prompts a fresh /login kimi."""
    import time
    import requests as _req

    try:
        r = _req.post(KIMI_OAUTH_TOKEN_URL, data={
            "client_id": KIMI_OAUTH_CLIENT_ID,
            "grant_type": "refresh_token",
            "refresh_token": refresh_token,
        }, headers={**_kimi_device_headers(),
                    "Content-Type": "application/x-www-form-urlencoded",
                    "Accept": "application/json"}, timeout=30)
        body = r.json() if r.content else {}
    except Exception:
        return None
    if r.status_code != 200 or not isinstance(body, dict) or not body.get("access_token"):
        return None
    store = _kimi_oauth_load_store()
    store.update({
        "access_token": body["access_token"],
        "refresh_token": body.get("refresh_token") or refresh_token,
        "obtained_at": time.time(),
        "expires_at": _oauth_expires_at(body),
    })
    _kimi_oauth_save_store(store)
    return store


def _kimi_oauth_login(config: dict, notify=print) -> str | None:
    """Kimi membership OAuth login (device flow — no API key). Opens the
    verification URL in the browser; the user approves the shown device code
    and we poll auth.kimi.com until approved. Returns the access token or None."""
    import time
    import webbrowser
    import requests as _req

    headers = {**_kimi_device_headers(),
               "Content-Type": "application/x-www-form-urlencoded",
               "Accept": "application/json"}
    try:
        r = _req.post(KIMI_OAUTH_DEVICE_URL,
                      data={"client_id": KIMI_OAUTH_CLIENT_ID},
                      headers=headers, timeout=30)
        data = r.json() if r.content else {}
    except Exception as e:
        notify(f"[kimi] Device authorization request failed: {e}")
        return None
    if r.status_code != 200 or not isinstance(data, dict):
        notify(f"[kimi] Device authorization failed (HTTP {r.status_code}).")
        return None

    user_code = (data.get("user_code") or "").strip()
    device_code = (data.get("device_code") or "").strip()
    verify_url = (data.get("verification_uri_complete")
                  or data.get("verification_uri") or "").strip()
    if not user_code or not device_code or not verify_url:
        notify("[kimi] Device authorization response incomplete (missing code/URL).")
        return None
    try:
        interval = max(1, int(float(data.get("interval") or 5)))
        expires_in = max(30, int(float(data.get("expires_in") or 600)))
    except (TypeError, ValueError):
        interval, expires_in = 5, 600

    notify("[kimi] Opening your browser to log in to your Kimi membership…")
    notify(f"[kimi] If it doesn't open, go to:\n{verify_url}")
    notify(f"[kimi] Your device code: {user_code}")
    try:
        webbrowser.open(verify_url)
    except Exception:
        pass
    notify("[kimi] Waiting for approval… (Ctrl+C to cancel)")

    deadline = time.time() + expires_in
    try:
        while time.time() < deadline:
            time.sleep(interval)
            try:
                r = _req.post(KIMI_OAUTH_TOKEN_URL, data={
                    "client_id": KIMI_OAUTH_CLIENT_ID,
                    "device_code": device_code,
                    "grant_type": KIMI_OAUTH_DEVICE_GRANT,
                }, headers=headers, timeout=30)
                body = r.json() if r.content else {}
            except Exception:
                continue  # transient network hiccup — keep polling until deadline
            if r.status_code == 200 and isinstance(body, dict) and body.get("access_token"):
                store = {
                    "access_token": body["access_token"],
                    "refresh_token": body.get("refresh_token") or "",
                    "scope": body.get("scope") or "",
                    "token_type": body.get("token_type") or "Bearer",
                    "obtained_at": time.time(),
                    "expires_at": _oauth_expires_at(body),
                }
                _kimi_oauth_save_store(store)
                notify("[kimi] Logged in. kimi-oauth/* models ready (Kimi membership, no API key).")
                return store["access_token"]
            err = (body.get("error") or "") if isinstance(body, dict) else ""
            if err in ("authorization_pending", "slow_down"):
                if err == "slow_down":
                    interval += 5  # server asked us to back off
                continue
            if err == "expired_token":
                notify("[kimi] Device code expired. Re-run /login kimi.")
                return None
            if err == "access_denied":
                notify("[kimi] Authorization denied.")
                return None
            if r.status_code >= 500:
                continue  # server-side wobble — keep polling until deadline
            notify(f"[kimi] Polling failed: {err or f'HTTP {r.status_code}'}")
            return None
    except KeyboardInterrupt:
        notify("[kimi] Login cancelled.")
        return None
    notify("[kimi] Login timed out waiting for approval. Re-run /login kimi.")
    return None


def _kimi_oauth_get_token(config: dict) -> str:
    """Return a valid Kimi OAuth access token, refreshing on expiry.

    Sources, in order:
      1. Dulus-native store (~/.dulus/kimi_oauth.json) from /login kimi.
      2. Official Kimi Code CLI credentials (~/.kimi-code/credentials/kimi-code.json).
    Returns "" when no session exists — the caller should prompt /login kimi."""
    store = _kimi_oauth_load_store()
    if store.get("access_token"):
        if not _is_token_expired(store):
            return store["access_token"]
        if store.get("refresh_token"):
            refreshed = _kimi_oauth_refresh(store["refresh_token"])
            if refreshed and refreshed.get("access_token"):
                return refreshed["access_token"]

    cli = _kimi_oauth_load_cli_credentials()
    if cli.get("access_token"):
        if not _is_token_expired(cli):
            _kimi_oauth_save_store({**cli, "source": "kimi-code-cli"})
            return cli["access_token"]
        if cli.get("refresh_token"):
            refreshed = _kimi_oauth_refresh(cli["refresh_token"])
            if refreshed and refreshed.get("access_token"):
                return refreshed["access_token"]
    return ""  # no session anywhere — needs /login kimi


def stream_kimi_oauth(
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
) -> Generator:
    """Stream Kimi from api.kimi.com/coding/v1 using the membership OAuth token
    (/login kimi). Same OpenAI-compatible endpoint as the kimi-code API-key
    provider — usage bills against the Kimi membership, not API credits."""
    token = _kimi_oauth_get_token(config)
    if not token:
        yield TextChunk("[kimi] No Kimi OAuth session (or it expired). Run `/login kimi` "
                        "to sign in with your Kimi membership — no API key needed.")
        return
    yield from stream_openai_compat(token, KIMI_OAUTH_API_BASE_URL, model, system,
                                    messages, tool_schemas, config)


def stream_dulus_account(
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
) -> Generator:
    """Stream Fuel-metered dulus-* models from the Dulus control plane.

    The credential comes from `/login dulus` (OAuth access token) or a
    ``dulus_sk_*`` key (`/login dulus key`, or DULUS_API_KEY in the
    environment). The control plane owns the upstream provider keys and
    settles Fuel per token — the client only ever holds its own account
    credential.
    """
    try:
        import dulus_account
        token = dulus_account.access_token(notify=lambda _m: None)
    except Exception:
        token = ""
    if not token:
        yield TextChunk("[dulus] No Dulus account session. Run `/login dulus` to sign in "
                        "(OAuth — no API key), or `/login dulus key` to mint a "
                        "dulus_sk_* key for CI/servers.")
        return
    base_url = (config.get("dulus_base_url")
                or _os_tls.environ.get("DULUS_BASE_URL", "")).rstrip("/") \
        or PROVIDERS["dulus"]["base_url"]
    yield from stream_openai_compat(token, base_url, model, system,
                                    messages, tool_schemas, config)


def stream(
    model: str,
    system: str,
    messages: list,
    tool_schemas: list,
    config: dict,
) -> Generator:
    """
    Unified streaming entry point.
    Auto-detects provider from model string.
    Yields: TextChunk | ThinkingChunk | AssistantTurn

    All provider calls are wrapped with automatic retry on transient
    failures (timeouts, 429 rate-limit, 5xx server errors).
    """
    provider_name = detect_provider(model)
    model_name    = bare_model(model)
    prov          = PROVIDERS.get(provider_name, PROVIDERS["openai"])
    api_key       = get_api_key(provider_name, config)

    def _inner_stream() -> Generator:
        nonlocal api_key  # modelstudio branch may reassign (DashScope fallback)
        if prov["type"] == "kimi-oauth":
            # Kimi membership OAuth (`/login kimi` or ~/.kimi-code credentials)
            # → api.kimi.com/coding/v1 (same endpoint as kimi-code API key).
            yield from stream_kimi_oauth(model_name, system, messages, tool_schemas, config)
        elif prov["type"] == "dulus-oauth":
            # Dulus account (`/login dulus`) → control plane, Fuel-metered.
            yield from stream_dulus_account(model_name, system, messages, tool_schemas, config)
        elif prov["type"] == "xai-oauth":
            # Official Grok Build TUI session (`grok login` → ~/.grok/auth.json) or XAI_API_KEY.
            # No separate harvest file is used (old Playwright harvest for Grok was removed).
            yield from stream_xai_oauth(model_name, system, messages, tool_schemas, config)
        elif prov["type"] == "chatgpt-oauth":
            # ChatGPT/Codex subscription OAuth (`/login chatgpt` or ~/.codex/auth.json)
            # → chatgpt.com/backend-api/codex/responses (Responses API, not chat/completions)
            yield from stream_chatgpt_oauth(model_name, system, messages, tool_schemas, config)
        elif prov["type"] == "meta":
            # Meta AI (Muse Spark) via api.meta.ai/v1/responses — API key (META_API_KEY).
            yield from stream_meta(api_key, model_name, system, messages, tool_schemas, config)
        elif prov["type"] == "gcloud":
            yield from stream_gcloud(model_name, system, messages, tool_schemas, config)
        elif prov["type"] == "litellm":
            yield from stream_litellm(model_name, system, messages, tool_schemas, config)
        elif prov["type"] == "anthropic":
            yield from stream_anthropic(api_key, model_name, system, messages, tool_schemas, config)
        elif prov["type"] == "ollama":
            base_url = prov.get("base_url", "http://localhost:11434")
            yield from stream_ollama(base_url, model_name, system, messages, tool_schemas, config)
        elif provider_name in ("kimi", "moonshot"):
            # Use native Kimi HTTP implementation for testing/comparison
            yield from stream_kimi(api_key, model_name, system, messages, tool_schemas, config)
        else:
            import os as _os
            if provider_name == "custom":
                base_url = (config.get("custom_base_url")
                            or _os.environ.get("CUSTOM_BASE_URL", ""))
                if not base_url:
                    raise ValueError(
                        "custom provider requires a base_url. "
                        "Set CUSTOM_BASE_URL env var or run: /config custom_base_url=http://..."
                    )
            elif provider_name == "azure":
                # Azure OpenAI endpoint is per-resource — read it from the user's
                # env/config instead of hardcoding one. Accept either the bare
                # resource URL or one already ending in /openai/v1/.
                base_url = (config.get("azure_base_url")
                            or _os.environ.get("AZURE_OPENAI_ENDPOINT", "")).rstrip("/")
                if not base_url:
                    raise ValueError(
                        "azure provider requires an endpoint. Set AZURE_OPENAI_ENDPOINT "
                        "env var or run: /config azure_base_url=https://<resource>.cognitiveservices.azure.com"
                    )
                if not base_url.endswith("/openai/v1"):
                    base_url = base_url + "/openai/v1"
                base_url = base_url + "/"
            elif provider_name == "modelstudio":
                # Alibaba Cloud Model Studio (Singapore). Allow overriding the
                # workspace without editing the registry: a full base_url wins,
                # otherwise just the workspace ID is spliced into the template.
                base_url = (config.get("modelstudio_base_url")
                            or _os.environ.get("MODELSTUDIO_BASE_URL", "")).rstrip("/")
                if not base_url:
                    ws = (config.get("modelstudio_workspace_id")
                          or _os.environ.get("MODELSTUDIO_WORKSPACE_ID", "")).strip()
                    if ws:
                        base_url = f"https://{ws}.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1"
                    else:
                        base_url = prov.get("base_url", "")
                # Model Studio shares Alibaba's DashScope key if a dedicated one
                # isn't set.
                if not api_key:
                    api_key = _os.environ.get("DASHSCOPE_API_KEY", "") or config.get("dashscope_api_key", "")
            elif provider_name == "amd":
                # AMD Developer Cloud: a vLLM/TGI server you launched on an AMD
                # GPU. The instance IP changes each session, so read it from
                # env/config rather than hardcoding.
                base_url = (config.get("amd_base_url")
                            or _os.environ.get("AMD_BASE_URL", "")).rstrip("/")
                if not base_url:
                    raise ValueError(
                        "amd provider requires your AMD Dev Cloud endpoint. "
                        "Set AMD_BASE_URL env var or run: "
                        "/config amd_base_url=http://<instance-ip>:8000/v1"
                    )
                # vLLM accepts any key; send a placeholder if none configured.
                if not api_key:
                    api_key = "amd-dev-cloud"
            elif provider_name == "edge":
                # Edge / local models over an OpenAI-compatible server (llama.cpp
                # `llama-server`, Ollama, or the on-device Android APK). Port is
                # configurable so it can dodge conflicts on Termux / point at a
                # phone on the LAN without editing the registry.
                base_url = (config.get("edge_base_url")
                            or _os.environ.get("DULUS_EDGE_BASE_URL", "")).rstrip("/") \
                           or prov.get("base_url", "http://127.0.0.1:8080/v1")
                if not api_key:
                    api_key = "dulus-edge"
            else:
                base_url = prov.get("base_url", "https://api.openai.com/v1")
            yield from stream_openai_compat(
                api_key, base_url, model_name, system, messages, tool_schemas, config
            )

    # Wrap with retry on transient failures
    yield from _ProviderRetry.wrap_generator(_inner_stream)


def list_ollama_models(base_url: str) -> list[str]:
    """Fetch locally available model tags from Ollama server."""
    try:
        url = f"{base_url.rstrip('/')}/api/tags"
        with urllib.request.urlopen(url, timeout=3) as resp:
            data = json.loads(resp.read().decode("utf-8"))
            # Ollama returns {"models": [{"name": "llama3:latest", ...}, ...]}
            return [m["name"] for m in data.get("models", [])]
    except Exception:
        return []