client.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# providers/openai_compat/client.py
 
"""
Generic OpenAI-compatible streaming client with retry.
 
Purpose
-------
:class:`OpenAICompatClient` is a :class:`BaseClient` for any lane that
speaks the OpenAI chat-completions dialect. It is written FRESH on the
shape the DeepSeek bundle proved (plan 215 P2) — this module imports
nothing from ``providers.deepseek`` — and parameterises the two things
that differ per lane: the base URL and the extra request headers three
of the lanes need to pass their WAF or their attribution requirements.
 
Responsibilities
----------------
- :class:`OpenAICompatClient` — ``stream_chat`` / ``one_shot`` /
  ``list_models`` against one lane's endpoint.
- :func:`_is_retryable` — auth errors fail fast; transient HTTP and
  network errors retry.
- :func:`_plain` — recursively coerce SDK/pydantic objects to plain
  dicts so no model object leaks past this client.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Up to 3 attempts on transient errors, delays 1/2/4 seconds.
- Authentication failures are NEVER retried — a wrong key does not
  become right, and retrying burns a free lane's daily request budget.
- ``stream_chat`` yields ``("reasoning", str)``, ``("content", str)``,
  ``("tool_call", dict)``, ``("usage", dict)`` in arrival order, then a
  terminal ``("finish_reason", str)`` — the CONTRACTS.md §5 tuple
  contract, identical to the DeepSeek client's.
- The SDK is constructed with ``max_retries=0`` (we own the backoff) and
  an httpx client with ``trust_env=False``, so the transport-trust env
  family (HTTP(S)_PROXY / ALL_PROXY / SSL_CERT_* / REQUESTS_CA_BUNDLE)
  cannot MITM or redirect the provider transcript (M8.0 WU-4c, F72/F79).
  This mirrors the DeepSeek client deliberately: a new egress path that
  silently re-trusted the environment would be a security regression.
- ``extra_headers`` carries lane metadata ONLY (User-Agent, attribution).
  The API key travels via the SDK's own auth handling and is never
  placed in a header dict here.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- openai (chat-completions client and streaming)
- httpx (lazy; transport with trust_env disabled)
- option_c_agent_contracts (lazy; SupportsCancellation type of the stream_chat cancellation token)
"""
from __future__ import annotations
 
import time
from collections import OrderedDict
from collections.abc import Iterator, Mapping
from typing import Any, TYPE_CHECKING
 
from oc_diagnostics import _dbg
 
from openai import OpenAI, AuthenticationError, APIStatusError
 
if TYPE_CHECKING:
    from option_c_agent_contracts import SupportsCancellation
 
_RETRY_DELAYS = (1, 2, 4)  # seconds between attempts (3 attempts total)
_RETRYABLE_STATUS = {429, 500, 502, 503, 524}
 
 
def _is_retryable(exc: Exception) -> bool:
    """Return True if *exc* is a transient error worth retrying."""
    if isinstance(exc, AuthenticationError):
        return False  # wrong key — retrying won't help
    if isinstance(exc, APIStatusError):
        return exc.status_code in _RETRYABLE_STATUS
    return True  # network errors, timeouts, etc.
 
 
def _plain(value: Any) -> Any:
    """Recursively convert pydantic / arbitrary objects to plain dicts and lists."""
    func = "_plain"
    if isinstance(value, dict):
        return {k: _plain(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_plain(v) for v in value]
    if value is None or isinstance(value, (str, int, float, bool, bytes)):
        return value
    if hasattr(value, "model_dump"):
        return _plain(value.model_dump())
    try:
        return _plain(value.__dict__)
    except AttributeError as exc:
        # VALIDATION_ERROR — slots-only or built-in object; coerce via str()
        _dbg("OC_CHAT.CLIENT", func,
             f"VALIDATION_ERROR: {type(value).__name__}: {type(exc).__name__}: {exc}", 3)
        return str(value)
 
 
class OpenAICompatClient:
    """Streaming chat client for one OpenAI-compatible lane."""
 
    def __init__(
        self,
        api_key: str,
        base_url: str,
        *,
        extra_headers: Mapping[str, str] | None = None,
        lane_name: str = "openai_compat",
    ) -> None:
        func = "__init__"
        import httpx
 
        self.base_url = base_url
        self.lane_name = lane_name
        self._headers = dict(extra_headers or {})
        # M8.0 WU-4c (F72/F79): trust_env=False — see the module Contracts block.
        self._client = OpenAI(
            api_key=api_key,
            base_url=base_url,
            max_retries=0,
            default_headers=self._headers or None,
            http_client=httpx.Client(trust_env=False),
        )
        _dbg("OC_CHAT.CLIENT", func,
             f"OpenAICompatClient lane={lane_name} base_url={base_url} "
             f"extra_headers={sorted(self._headers)}", 3)
 
    # ── Model listing ────────────────────────────────────────────────────────
 
    def list_models(self) -> list[str]:
        """Return the lane's live model ids (the account's own catalogue)."""
        return [m.id for m in self._client.models.list()]
 
    # ── Streaming chat ───────────────────────────────────────────────────────
 
    def stream_chat(
        self,
        messages: list[dict],
        model: str,
        effort: str | None = None,
        extra_body: dict | None = None,
        tools: list[dict] | None = None,
        max_tokens: int | None = None,
        *,
        cancellation: "SupportsCancellation | None" = None,
    ) -> Iterator[tuple[str, Any]]:
        """Stream a chat completion, yielding typed ``(chunk_type, data)`` tuples.
 
        See the module Contracts block for the tuple contract. *effort* is
        forwarded as ``reasoning_effort`` only when the caller supplies it;
        no lane in the shipped catalogue is configured for reasoning, so the
        common path sends the model's as-shipped defaults.
 
        If *cancellation* is provided the token is polled before yielding
        each chunk; a cancelled token raises and the stream is abandoned.
        """
        func = "stream_chat"
        kwargs: "OrderedDict[str, Any]" = OrderedDict()
        kwargs["model"] = model
        if tools:
            kwargs["tools"] = tools
        kwargs["stream"] = True
        kwargs["messages"] = messages
        if effort is not None:
            kwargs["reasoning_effort"] = effort
        if extra_body:
            kwargs["extra_body"] = extra_body
        if max_tokens is not None:
            kwargs["max_tokens"] = max_tokens
 
        stream = self._create_with_retry(kwargs)
 
        tool_calls_acc: dict[int, dict] = {}
        finish_reason: str | None = None
 
        for chunk in stream:
            if cancellation is not None:
                cancellation.raise_if_cancelled()
 
            usage = getattr(chunk, "usage", None)
            if usage is not None:
                yield ("usage", _plain(usage))
 
            choices = getattr(chunk, "choices", None)
            if not choices:
                continue
 
            choice = choices[0]
            # Terminal-chunk only; keep the last non-empty value. This is the
            # sole signal separating "the model finished" from "it ran out of
            # budget", so a truncated tool_call is diagnosable downstream.
            if getattr(choice, "finish_reason", None):
                finish_reason = choice.finish_reason
 
            delta = getattr(choice, "delta", None)
            if delta is None:
                continue
 
            # Several lanes surface chain-of-thought on this non-standard
            # field (the DeepSeek spelling); absent everywhere else.
            reasoning = getattr(delta, "reasoning_content", None)
            if reasoning:
                yield ("reasoning", reasoning)
 
            if getattr(delta, "content", None):
                yield ("content", delta.content)
 
            for tc in (getattr(delta, "tool_calls", None) or []):
                idx = getattr(tc, "index", 0) or 0
                acc = tool_calls_acc.setdefault(idx, {"id": "", "name": "", "arguments": ""})
                if getattr(tc, "id", None):
                    acc["id"] = tc.id
                fn = getattr(tc, "function", None)
                if fn is not None:
                    if getattr(fn, "name", None):
                        acc["name"] = fn.name
                    if getattr(fn, "arguments", None):
                        acc["arguments"] += fn.arguments
 
        for idx in sorted(tool_calls_acc):
            yield ("tool_call", tool_calls_acc[idx])
 
        # Last, so a consumer that reads it has the whole stream in hand.
        if finish_reason is not None:
            _dbg("OC_CHAT.CLIENT", func,
                 f"lane={self.lane_name} finish_reason={finish_reason}", 4)
            yield ("finish_reason", finish_reason)
 
    # ── One-shot (non-streaming) ─────────────────────────────────────────────
 
    def one_shot(
        self,
        messages: list[dict],
        model: str,
        *,
        effort: str | None = None,
        extra_body: dict | None = None,
        max_tokens: int | None = None,
    ) -> str:
        """Non-streaming single call with retry (context-compression summaries)."""
        kwargs: "OrderedDict[str, Any]" = OrderedDict()
        kwargs["model"] = model
        kwargs["stream"] = False
        kwargs["messages"] = messages
        if effort is not None:
            kwargs["reasoning_effort"] = effort
        if extra_body:
            kwargs["extra_body"] = extra_body
        if max_tokens is not None:
            kwargs["max_tokens"] = max_tokens
 
        resp = self._create_with_retry(kwargs)
        return resp.choices[0].message.content or ""
 
    # ── Retry ────────────────────────────────────────────────────────────────
 
    def _create_with_retry(self, kwargs: "OrderedDict[str, Any]") -> Any:
        """Call ``chat.completions.create`` with the backoff schedule.
 
        Shared by both call paths so the "auth fails fast, transient
        retries" rule cannot drift between streaming and one-shot.
        """
        func = "_create_with_retry"
        last_exc: Exception | None = None
        for delay in (*_RETRY_DELAYS, None):
            try:
                return self._client.chat.completions.create(**kwargs)
            except Exception as exc:
                last_exc = exc
                if not _is_retryable(exc) or delay is None:
                    raise
                # EXTERNAL_ERROR — transient upstream failure; back off and retry.
                _dbg("OC_CHAT.CLIENT", func,
                     f"EXTERNAL_ERROR: lane={self.lane_name} retry in {delay}s: "
                     f"{type(exc).__name__}: {exc}", 2)
                print(f"\n[Retrying in {delay}s - {type(exc).__name__}]", flush=True)
                time.sleep(delay)
        raise last_exc  # unreachable: the delay=None arm re-raises
 
 
__all__ = ["OpenAICompatClient"]