codec_llm.py

"""CODEC LLM call helper — the single canonical OpenAI-style chat/completions caller.

A-12 (PR-3E): before this, ~45 sites hand-rolled the same `chat/completions`
POST — build headers (`Authorization: Bearer …`, `Content-Type`), assemble the
payload (`model`/`messages`/`max_tokens`/`temperature`/
`chat_template_kwargs.enable_thinking=False`), parse `choices[0].message`
(content, with a `reasoning` fallback), and strip `<think>…</think>`. A model
upgrade or API-shape fix then meant editing 20+ places.

This module centralizes the **non-streaming** call. It is intentionally
config-agnostic — each caller passes its own `base_url` / `model` / `api_key`
/ tuning — so it's a pure "build payload → POST → parse" helper with no import
cycle into codec_config. (Streaming SSE + the remaining call sites are migrated
in later A-12 tranches; this PR covers the call() API + codec.py + codec_session.)

NOTE: `codec_llm_proxy` is a *priority queue* (semaphore), not an HTTP proxy —
orthogonal to this module. Callers that want prioritization still wrap the call
in `llm_queue_sync(...)`; behavior parity for the migrated sites means we do NOT
add queue acquisition here (none of them used it).
"""
from __future__ import annotations

import logging
import re
import time
from typing import Any, AsyncIterator, Dict, Iterator, List, Optional

log = logging.getLogger("codec.llm")


class LLMError(Exception):
    """Raised by ``call(raise_on_error=True)`` on any non-success outcome —
    non-200 (after retries), a request exception (after retries), or a 200 with
    empty/unparseable content. The default ``raise_on_error=False`` keeps the
    never-raise → "" contract that the streaming/best-effort callers rely on.
    Fail-loud callers (agent_plan/runner, textassist, the regen script) opt in
    and map this onto their own error handling."""


_THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)

# Sentinel yielded by stream(keepalive=True) on empty "thinking" chunks so an SSE
# caller (e.g. the dashboard) can emit a transport keepalive to hold its tunnel
# open. Never yielded when keepalive=False (the default) — content-only callers
# (codec_session.qwen_stream) are unaffected.
KEEPALIVE = object()

# Yielded by stream(usage_sentinel=True) when the server reports token counts in
# its final chunk. Carries data, so unlike the bare object() sentinels it is a
# dict subclass — callers check isinstance(item, StreamUsage) BEFORE treating an
# item as text, otherwise it would be rendered into the reply.
class StreamUsage(dict):
    """Token accounting for one streamed reply (prompt/completion/total)."""

# Sentinels yielded by stream(error_sentinel=True) — 2026-07 chat-visibility
# fix. stream() never raises by contract, so before these existed a mid-reply
# connection drop / non-200 / read timeout was indistinguishable from a clean
# finish: the dashboard rendered an empty or silently-truncated bubble.
#   STREAM_ERROR   — the stream died abnormally (connect/HTTP/read error).
#   FINISH_LENGTH  — the model stopped at the max_tokens cap
#                    (finish_reason == "length"), i.e. the reply is truncated.
# Only yielded when error_sentinel=True; all existing callers are unaffected.
STREAM_ERROR = object()
FINISH_LENGTH = object()


def strip_think(text: str) -> str:
    """Remove <think>…</think> reasoning blocks and surrounding whitespace."""
    if not text:
        return ""
    return _THINK_RE.sub("", text).strip()


def extract_content(response_json: Dict[str, Any]) -> str:
    """Pull the assistant text from an OpenAI-style response: prefer
    `choices[0].message.content`, fall back to `.reasoning` (some local
    servers put the answer there when content is empty). `<think>` stripped.
    Returns "" on any shape mismatch."""
    try:
        msg = response_json["choices"][0]["message"]
    except (KeyError, IndexError, TypeError):
        return ""
    content = (msg.get("content") or "").strip()
    if content:
        return strip_think(content)
    reasoning = (msg.get("reasoning") or "").strip()
    if reasoning:
        return strip_think(reasoning)
    return ""


# The OpenAI-style chat API accepts only these roles. CODEC's memory stores
# other roles internally (notably "fact" from fact_extract / memory_save), and
# more than one caller has replayed those straight into the messages array — a
# hard 422 that surfaces to the user as "Qwen busy". This is the last line of
# defence: coerce anything non-standard to a safe role rather than 422.
_VALID_LLM_ROLES = {"user", "assistant", "system", "developer", "tool"}


def _sanitize_roles(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Return messages with only API-valid roles. A message whose role isn't
    recognised (e.g. 'fact') is relabelled 'user' — valid at ANY position, so it
    can't trigger a 422 (bad role) or a 500 ('system message must be first' when a
    stray fact lands mid-conversation). Its content is preserved so it still
    informs the model."""
    out: List[Dict[str, Any]] = []
    for m in messages:
        if m.get("role") in _VALID_LLM_ROLES:
            out.append(m)
        else:
            out.append({**m, "role": "user"})
    return out


def _build_request(
    messages: List[Dict[str, Any]],
    *,
    model: str,
    api_key: str,
    max_tokens: int,
    temperature: float,
    enable_thinking: bool,
    extra_kwargs: Optional[Dict[str, Any]],
    stream: bool = False,
) -> tuple[Dict[str, str], Dict[str, Any]]:
    """Build the (headers, payload) for an OpenAI-style chat/completions request.
    Shared by call() and stream() so headers/auth/payload shape never drift.
    The `stream` flag is applied LAST so extra_kwargs can't clobber it."""
    headers = {"Content-Type": "application/json"}
    if api_key:
        headers["Authorization"] = "Bearer " + api_key
    payload: Dict[str, Any] = {
        "model": model,
        "messages": _sanitize_roles(messages),
        "max_tokens": max_tokens,
        "temperature": temperature,
        "chat_template_kwargs": {"enable_thinking": enable_thinking},
    }
    if extra_kwargs:
        payload.update(extra_kwargs)
    if stream:
        payload["stream"] = True
    return headers, payload


def _cloud_blocked_msg(base_url: str) -> Optional[str]:
    """License gate for the 'cloud_proxy' feature (paid edition only).

    Returns a user-facing message if this is a CLOUD call (non-localhost
    base_url) that the current license doesn't permit; otherwise None.

    Local calls (localhost / 127.0.0.1 / 0.0.0.0) are NEVER gated. OSS/dev
    builds always return None (feature_allowed → True). Fail-open: any
    licensing fault returns None so transport is never broken by licensing.
    """
    try:
        bl = (base_url or "").lower()
        if "localhost" in bl or "127.0.0.1" in bl or "0.0.0.0" in bl:
            return None  # local model — always allowed
        import codec_license
        if codec_license.feature_allowed("cloud_proxy"):
            return None
        st = codec_license.license_state()
        return (f"\U0001F512 Cloud models require an active CODEC license — "
                f"{st.reason}. Activate in Settings, or switch to the local model.")
    except Exception:
        return None  # fail-open — licensing must never break the LLM transport


def _cloud_route(base_url: str, model: str):
    """The registered cloud model (codec_cloud_models) served at `base_url`, or
    None. Local URLs return before the registry is read.

    Fail-open like the licence gate: if the registry cannot be read, the request
    goes out unchanged, and without the entry's key a cloud model rejects it, so
    nothing is spent unmetered."""
    try:
        import codec_cloud_models
        return codec_cloud_models.route(base_url, model)
    except Exception as e:
        log.warning("cloud model check failed: %s", e)
        return None


class _CloudReply:
    """The parts of a `requests` / `httpx` response that call() and acall() read."""

    def __init__(self, status_code: int, text: str):
        self.status_code, self.text = status_code, text

    def json(self) -> Any:
        import json as _json
        return _json.loads(self.text)


class _StreamedReply:
    """Rebuilds one chat.completion from a streamed reply."""

    def __init__(self, timeout: float, first_reply_s: float):
        self.timeout = timeout
        self.deadline = time.monotonic() + timeout
        self.first_reply_s = min(timeout, first_reply_s)
        self.first_by = time.monotonic() + self.first_reply_s
        self.content: List[str] = []
        self.reasoning: List[str] = []
        self.usage: Optional[Dict[str, Any]] = None
        self.done = False

    def feed(self, line: Any) -> None:
        """Take one SSE line. Raises TimeoutError past the deadline."""
        import json as _json
        now = time.monotonic()
        if now > self.deadline:
            raise TimeoutError(f"no complete reply within {self.timeout:.0f}s")
        if not self.content and now > self.first_by:
            raise TimeoutError(f"no reply started within {self.first_reply_s:.0f}s")
        if isinstance(line, (bytes, bytearray)):
            line = line.decode("utf-8", "replace")
        if not line or not line.startswith("data: "):
            return
        data = line[6:]
        if data.strip() == "[DONE]":
            self.done = True
            return
        try:
            obj = _json.loads(data)
        except ValueError:
            return
        if not isinstance(obj, dict):
            return
        if obj.get("usage"):
            self.usage = obj["usage"]
        for ch in obj.get("choices") or []:
            d = (ch or {}).get("delta") or {}
            if d.get("content"):
                self.content.append(d["content"])
            r = d.get("reasoning_content") or d.get("reasoning")
            if r:
                self.reasoning.append(r)

    def reply(self) -> _CloudReply:
        import json as _json
        msg: Dict[str, Any] = {"role": "assistant", "content": "".join(self.content)}
        if self.reasoning:
            msg["reasoning"] = "".join(self.reasoning)
        body: Dict[str, Any] = {"choices": [{"message": msg}]}
        if self.usage:
            body["usage"] = self.usage
        return _CloudReply(200, _json.dumps(body))


def _streamed(payload: Dict[str, Any]) -> Dict[str, Any]:
    return dict(payload, stream=True, stream_options={"include_usage": True})


def _cloud_post(requests: Any, url: str, payload: Dict[str, Any],
                headers: Dict[str, str], timeout: float,
                first_reply_s: float) -> _CloudReply:
    """One reply from a cloud model, asked for as a stream, with a wall-clock
    deadline of `timeout` seconds.

    Measured on MiMo (2026-09-27): streamed requests answered in 1-3 s every
    time, while non-streamed ones took 1.7 s to over 45 s, once 489 s. While a
    non-streamed request waits, MiMo sends keep-alive newlines, so the per-read
    timeout never fires. A stream, rebuilt into the single reply the caller
    expects, avoids both."""
    acc = _StreamedReply(timeout, first_reply_s)
    with requests.post(url, json=_streamed(payload), headers=headers,
                       timeout=timeout, stream=True) as r:
        if r.status_code != 200:
            return _CloudReply(r.status_code, getattr(r, "text", "") or "")
        for line in r.iter_lines():
            acc.feed(line)
            if acc.done:
                break
    return acc.reply()


async def _cloud_apost(client: Any, url: str, payload: Dict[str, Any],
                       headers: Dict[str, str], timeout: float,
                       first_reply_s: float) -> _CloudReply:
    """Async twin of _cloud_post for acall()."""
    import asyncio

    async def _run() -> _CloudReply:
        acc = _StreamedReply(timeout, first_reply_s)
        async with client.stream("POST", url, json=_streamed(payload),
                                 headers=headers) as r:
            if getattr(r, "status_code", 200) != 200:
                return _CloudReply(r.status_code, "")
            async for line in r.aiter_lines():
                acc.feed(line)
                if acc.done:
                    break
        return acc.reply()

    try:
        return await asyncio.wait_for(_run(), timeout)
    except asyncio.TimeoutError:
        raise TimeoutError(f"no complete reply within {timeout:.0f}s") from None


def _preflight(base_url: str, model: str):
    """(block message or None, cloud route or None): the licence gate, then the
    cloud-model check (key, spend cap)."""
    blocked = _cloud_blocked_msg(base_url)
    if blocked is not None:
        return blocked, None
    route = _cloud_route(base_url, model)
    if route is not None and route.blocked:
        return route.blocked, None
    return None, route


def call(
    messages: List[Dict[str, Any]],
    *,
    base_url: str,
    model: str,
    api_key: str = "",
    max_tokens: int = 500,
    temperature: float = 0.7,
    timeout: float = 120.0,
    retries: int = 1,
    enable_thinking: bool = False,
    extra_kwargs: Optional[Dict[str, Any]] = None,
    raise_on_error: bool = False,
    usage_out: Optional[Dict[str, Any]] = None,
) -> str:
    """POST `messages` to `<base_url>/chat/completions` and return the parsed,
    `<think>`-stripped assistant text.

    `usage_out`, when given, receives the server's `usage` block (token counts)
    of the successful reply, if it sent one (UI P3.10: model compare).

    `retries` includes the first attempt (retries=3 → up to 3 tries with
    exponential 2**n backoff between them, matching codec_session.qwen_call).

    Error contract:
    - `raise_on_error=False` (default): never raises — network/parse errors and
      empty/unparseable 200s are logged and yield "".
    - `raise_on_error=True`: raises `LLMError` on EVERY non-success outcome
      (non-200 after retries, request exception after retries, or a 200 with
      empty/unparseable content). For fail-loud callers that must not silently
      proceed on an empty answer.
    """
    import requests
    _blocked, _route = _preflight(base_url, model)
    if _blocked is not None:
        if raise_on_error:
            raise LLMError(_blocked)
        return _blocked
    if _route is not None:
        model, api_key, extra_kwargs = _route.model, _route.api_key, _route.kwargs(extra_kwargs)
    headers, payload = _build_request(
        messages, model=model, api_key=api_key, max_tokens=max_tokens,
        temperature=temperature, enable_thinking=enable_thinking,
        extra_kwargs=extra_kwargs,
    )
    if _route is not None:
        _route.shape(payload, stream=False)

    attempts = max(1, retries)
    url = base_url.rstrip("/") + "/chat/completions"
    last_error: Optional[Exception] = None
    for attempt in range(attempts):
        try:
            if _route is None:
                r = requests.post(url, json=payload, headers=headers, timeout=timeout)
            else:
                r = _cloud_post(requests, url, payload, headers, timeout,
                                _route.first_reply_s)
            if r.status_code == 200:
                _body = r.json()
                resp = extract_content(_body)
                if _route is not None and isinstance(_body, dict):
                    _route.record(_body.get("usage"), messages, resp)
                if usage_out is not None and isinstance(_body, dict) and isinstance(_body.get("usage"), dict):
                    usage_out.update(_body["usage"])
                if resp:
                    return resp
                # 200 but empty/odd shape — nothing more to get; don't retry.
                if raise_on_error:
                    raise LLMError("LLM returned empty or unparseable content")
                return ""
            last_error = LLMError(f"LLM call returned {r.status_code}: {r.text[:200]}")
            log.warning("LLM call %s returned %s: %s", url, r.status_code, r.text[:200])
        except LLMError:
            raise  # empty-200 in raise mode — propagate, don't swallow as a retry
        except Exception as e:
            last_error = e
            log.warning("LLM call attempt %d/%d failed: %s", attempt + 1, attempts, e)
            if attempt < attempts - 1:
                time.sleep(2 ** attempt)
    if raise_on_error:
        raise LLMError(f"LLM call failed after {attempts} attempt(s): {last_error}")
    return ""


def stream(
    messages: List[Dict[str, Any]],
    *,
    base_url: str,
    model: str,
    api_key: str = "",
    max_tokens: int = 500,
    temperature: float = 0.7,
    timeout: float = 120.0,
    enable_thinking: bool = False,
    extra_kwargs: Optional[Dict[str, Any]] = None,
    keepalive: bool = False,
    error_sentinel: bool = False,
    inline_reasoning: bool = False,
    usage_sentinel: bool = False,
) -> Iterator[Any]:
    """POST with `stream=True` and yield the RAW assistant content deltas in
    order. Centralizes the SSE plumbing: header/payload build (shared with
    call()), `data: ` framing, the `[DONE]` sentinel, `choices[0].delta.content`
    extraction, and per-chunk parse tolerance.

    Think-stripping is intentionally NOT done here — callers that show tokens
    live (e.g. codec_session.qwen_stream) strip `<think>` on the accumulated
    result, and the dashboard owns its own cross-chunk tag machine. Never
    raises: on connect/HTTP/parse error it logs and stops yielding, so the
    caller sees a short/empty stream and applies its own fallback.

    `keepalive=True` (default off): on an empty "thinking" chunk, yield the
    `KEEPALIVE` sentinel every 10th empty (1st, 11th, …) so an SSE caller can
    emit a transport keepalive. Content-only callers leave it off and only ever
    see `str` deltas.

    `error_sentinel=True` (default off): yield `STREAM_ERROR` when the stream
    dies abnormally (non-200, connect/read exception) and `FINISH_LENGTH` when
    the model stops at the max_tokens cap — so a UI caller can tell the user
    the reply was interrupted / truncated instead of rendering an empty or
    silently-cut bubble. Callers that leave it off see the old behavior
    (stream just ends).
    """
    import json as _json
    import requests
    _blocked, _route = _preflight(base_url, model)
    if _blocked is not None:
        yield _blocked
        return
    if _route is not None:
        model, api_key, extra_kwargs = _route.model, _route.api_key, _route.kwargs(extra_kwargs)
    headers, payload = _build_request(
        messages, model=model, api_key=api_key, max_tokens=max_tokens,
        temperature=temperature, enable_thinking=enable_thinking,
        extra_kwargs=extra_kwargs, stream=True,
    )
    if usage_sentinel:
        # OpenAI-style: usage is omitted from streamed responses unless asked
        # for. mlx_vlm.server honours this and emits a final choices-empty chunk
        # carrying prompt/completion tokens.
        payload["stream_options"] = {"include_usage": True}
    if _route is not None:
        _route.shape(payload, stream=True)
    url = base_url.rstrip("/") + "/chat/completions"
    _empty = 0  # empty "thinking" chunks seen (drives keepalive)
    # Cloud metering: charged once per reply, from the usage chunk, or from an
    # estimate if the stream ended without one.
    _billable = _charged = False
    _out: List[str] = []
    _first_by = (time.monotonic() + min(timeout, _route.first_reply_s)
                 if _route is not None else None)
    try:
        with requests.post(url, json=payload, headers=headers,
                           timeout=timeout, stream=True) as r:
            if r.status_code != 200:
                log.warning("LLM stream %s returned %s: %s",
                            url, r.status_code, getattr(r, "text", "")[:200])
                if error_sentinel:
                    yield STREAM_ERROR
                return
            _billable = _route is not None
            reasoning_open = False  # inline_reasoning: are we inside a synth <think>?
            for line in r.iter_lines():
                if _first_by is not None and not _out and time.monotonic() > _first_by:
                    # Keep-alive newlines (see _cloud_post) would otherwise
                    # hold a reply that never starts open for minutes.
                    log.warning("cloud stream %s sent no content in time", url)
                    if error_sentinel:
                        yield STREAM_ERROR
                    return
                if not line:
                    continue
                if isinstance(line, (bytes, bytearray)):
                    line = line.decode("utf-8", "replace")
                if not line.startswith("data: "):
                    continue
                data = line[6:]
                if data.strip() == "[DONE]":
                    break
                try:
                    _obj = _json.loads(data)
                except Exception as e:
                    log.warning("LLM stream chunk parse failed: %s", e)
                    continue
                # The usage chunk carries an EMPTY choices list, so it must be
                # handled before the choices[0] access below — that access used
                # to raise IndexError and the counts were swallowed as a parse
                # failure.
                if _obj.get("usage"):
                    if _billable and not _charged:
                        _route.record(_obj["usage"])
                        _charged = True
                    if usage_sentinel:
                        # Carry `timings` alongside the token counts when the
                        # server provides it. predicted_per_second is the TRUE
                        # generation rate; completion_tokens/wall-time is not —
                        # it charges prompt processing to the reply and reads an
                        # order of magnitude too slow after a long prompt.
                        _u = dict(_obj["usage"])
                        if isinstance(_obj.get("timings"), dict):
                            _u["timings"] = _obj["timings"]
                        yield StreamUsage(_u)
                    if not _obj.get("choices"):
                        continue
                try:
                    choice = _obj.get("choices", [{}])[0]
                    _d = choice.get("delta", {})
                    delta = _d.get("content", "")
                except Exception as e:
                    log.warning("LLM stream chunk parse failed: %s", e)
                    continue
                # inline_reasoning: some models (Qwen3 etc.) stream their thinking
                # in a separate `reasoning_content`/`reasoning` delta rather than
                # inline <think>…</think>. Wrap it back into synthetic <think> tags
                # so the caller's existing <think> machine can surface it as
                # train-of-thought. Off by default → callers see content only.
                if inline_reasoning:
                    rdelta = _d.get("reasoning_content") or _d.get("reasoning") or ""
                    if rdelta:
                        if not reasoning_open:
                            yield "<think>"
                            reasoning_open = True
                        yield rdelta
                    if delta and reasoning_open:
                        yield "</think>"
                        reasoning_open = False
                if delta:
                    if _billable:
                        _out.append(delta)
                    yield delta
                elif keepalive and not (inline_reasoning and reasoning_open):
                    _empty += 1
                    if _empty % 10 == 1:   # 1st, 11th, 21st … (matches dashboard)
                        yield KEEPALIVE
                if error_sentinel and choice.get("finish_reason") == "length":
                    # Model hit the max_tokens cap — reply is truncated.
                    yield FINISH_LENGTH
            if reasoning_open:
                yield "</think>"   # close a think block if content never arrived
    except Exception as e:
        log.warning("LLM stream call failed: %s", e)
        if error_sentinel:
            yield STREAM_ERROR
        return
    finally:
        if _billable and not _charged:
            _route.record(None, messages, "".join(_out))


async def acall(
    messages: List[Dict[str, Any]],
    *,
    base_url: str,
    model: str,
    api_key: str = "",
    max_tokens: int = 500,
    temperature: float = 0.7,
    timeout: float = 120.0,
    enable_thinking: bool = False,
    extra_kwargs: Optional[Dict[str, Any]] = None,
    http: Optional[Any] = None,
    raise_on_error: bool = False,
) -> str:
    """Async sibling of call() — a single non-streaming POST via an httpx
    AsyncClient. Reuses the caller's client when `http` is given (e.g. agents'
    module `_async_http`), else makes + closes its own. When a client is
    injected we do NOT pass a per-request timeout — the client's configured
    timeout applies (exact parity with the inline sites). `raise_on_error`
    mirrors call(): raise `LLMError` on non-200 / exception / empty, else "".
    The queue (codec_llm_proxy) stays at the call site — never owned here.
    """
    import httpx
    _blocked, _route = _preflight(base_url, model)
    if _blocked is not None:
        if raise_on_error:
            raise LLMError(_blocked)
        return _blocked
    if _route is not None:
        model, api_key, extra_kwargs = _route.model, _route.api_key, _route.kwargs(extra_kwargs)
    headers, payload = _build_request(
        messages, model=model, api_key=api_key, max_tokens=max_tokens,
        temperature=temperature, enable_thinking=enable_thinking,
        extra_kwargs=extra_kwargs,
    )
    if _route is not None:
        _route.shape(payload, stream=False)
    url = base_url.rstrip("/") + "/chat/completions"
    own_client = http is None
    client = http or httpx.AsyncClient(timeout=timeout)
    try:
        try:
            if _route is None:
                r = await client.post(url, json=payload, headers=headers)
            else:  # streamed, with a wall-clock deadline (see _cloud_post)
                r = await _cloud_apost(client, url, payload, headers, timeout,
                                       _route.first_reply_s)
            if r.status_code == 200:
                _body = r.json()
                resp = extract_content(_body)
                if _route is not None and isinstance(_body, dict):
                    _route.record(_body.get("usage"), messages, resp)
                if resp:
                    return resp
                if raise_on_error:
                    raise LLMError("LLM returned empty or unparseable content")
                return ""
            if raise_on_error:
                raise LLMError(f"async LLM call returned {r.status_code}")
            log.warning("async LLM call %s returned %s", url, r.status_code)
            return ""
        except LLMError:
            raise
        except Exception as e:
            if raise_on_error:
                raise LLMError(f"async LLM call failed: {e}") from e
            log.warning("async LLM call failed: %s", e)
            return ""
    finally:
        if own_client:
            await client.aclose()


async def astream(
    messages: List[Dict[str, Any]],
    *,
    base_url: str,
    model: str,
    api_key: str = "",
    max_tokens: int = 500,
    temperature: float = 0.7,
    timeout: float = 120.0,
    enable_thinking: bool = False,
    extra_kwargs: Optional[Dict[str, Any]] = None,
    http: Optional[Any] = None,
    keepalive: bool = False,
) -> AsyncIterator[Any]:
    """Async sibling of stream() — yields the RAW assistant content deltas (and
    the `KEEPALIVE` sentinel on empty chunks when `keepalive=True`) over httpx
    streaming. Reuses the caller's client when `http` is given (e.g. voice's
    `self._http`), else makes + closes its own.

    Contract difference vs sync stream(): astream **propagates** exceptions —
    it does NOT swallow connect/stream errors — because its consumer
    (codec_voice._stream_qwen) wraps the loop in try/except to speak a failure
    and a silent stream would be a UX regression. The queue stays at the call
    site. `<think>` stripping is the caller's job (voice strips per-token).
    """
    import json as _json
    import httpx
    _blocked, _route = _preflight(base_url, model)
    if _blocked is not None:
        yield _blocked
        return
    if _route is not None:
        model, api_key, extra_kwargs = _route.model, _route.api_key, _route.kwargs(extra_kwargs)
    headers, payload = _build_request(
        messages, model=model, api_key=api_key, max_tokens=max_tokens,
        temperature=temperature, enable_thinking=enable_thinking,
        extra_kwargs=extra_kwargs, stream=True,
    )
    if _route is not None:
        _route.shape(payload, stream=True)
    url = base_url.rstrip("/") + "/chat/completions"
    own_client = http is None
    client = http or httpx.AsyncClient(timeout=timeout)
    _empty = 0
    # Cloud metering, as in stream(): once per reply, usage chunk or estimate.
    _billable = _charged = False
    _out: List[str] = []
    _first_by = (time.monotonic() + min(timeout, _route.first_reply_s)
                 if _route is not None else None)
    try:
        async with client.stream("POST", url, json=payload, headers=headers) as resp:
            _billable = _route is not None and getattr(resp, "status_code", 200) == 200
            async for line in resp.aiter_lines():
                if _first_by is not None and not _out and time.monotonic() > _first_by:
                    raise LLMError("cloud model sent no reply in time")
                if not line or not line.startswith("data: "):
                    continue
                data = line[6:]
                if data.strip() == "[DONE]":
                    return
                try:
                    _obj = _json.loads(data)
                    if _billable and not _charged and _obj.get("usage"):
                        _route.record(_obj["usage"])
                        _charged = True
                    delta = (_obj.get("choices", [{}])[0]
                             .get("delta", {}).get("content", ""))
                except (ValueError, KeyError, IndexError, TypeError, AttributeError):
                    continue
                if delta:
                    if _billable:
                        _out.append(delta)
                    yield delta
                elif keepalive:
                    _empty += 1
                    if _empty % 10 == 1:
                        yield KEEPALIVE
    finally:
        if _billable and not _charged:
            _route.record(None, messages, "".join(_out))
        if own_client:
            await client.aclose()