provider.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# providers/openai_compat/provider.py
 
"""
OpenAICompatProvider — one class, one configured instance per lane.
 
Purpose
-------
Concrete :class:`ProviderPlugin` for any OpenAI-compatible lane. Unlike
the single-provider bundles, this class is NOT configured by class
attributes: each instance is built from a :class:`Lane` catalogue row, so
seven rows produce seven independently-named providers backed by one
implementation (plan 215 P1). The registry keys on ``name`` and only
rejects a DIFFERENT class re-using an existing name, so registering one
class under seven names is exactly what its contract allows.
 
Responsibilities
----------------
- :class:`OpenAICompatProvider` — implement the abstract surface from a
  lane row.
- :meth:`OpenAICompatProvider.register_lane` — build and register one
  instance for a row.
- :func:`register_all` — validate the catalogue, then register every row.
- :func:`registered_lanes` — the lane rows this bundle registered, in
  catalogue declaration order.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Identity (``name`` / ``api_key_env_var`` / ``base_url_env_var`` /
  ``default_base_url`` / ``default_model``) is set per INSTANCE from the
  row, shadowing the ABC's class-level defaults. Two instances never
  share mutable state.
- :meth:`build_extra_body` returns ``None`` always: no catalogue lane is
  configured for reasoning (the OCON as-shipped condition), so there is
  no provider-specific payload to add. A lane that later needs one adds
  it here keyed on its own name — never a blanket default.
- :meth:`transform_outgoing_messages` is a true pass-through: no lane has
  a prefix-cache trick to serve.
- :meth:`parse_usage` delegates to :func:`normalise_usage`, which is
  total over malformed input.
- :func:`register_all` is idempotent and validates FIRST: an invalid row
  raises :exc:`CatalogueError` before ANY lane is registered, so a
  half-registered bundle is unreachable (plan 215 P5).
- Nothing here reads an API key. The harness resolves the key from the
  env-var NAME this provider advertises.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- option_c_agent_contracts (ProviderPlugin ABC, ModelCaps, UsageStats, registry)
- option_c_agent_contracts.provider_plugin (OAuthCapability return type of the oauth property)
- providers.openai_compat.catalogue (LANES, Lane rows and validate_catalogue for register_all)
- providers.openai_compat.client (OpenAICompatClient returned by build_client)
- providers.openai_compat.usage (normalise_usage behind parse_usage)
- providers.openai_compat.oauth (lazy; OpenRouterOAuth for the openrouter lane's oauth property)
"""
from __future__ import annotations
 
from typing import Sequence
 
from oc_diagnostics import _dbg
 
from option_c_agent_contracts import (
    ProviderPlugin,
    ModelCaps,
    UsageStats,
    BaseClient,
)
 
from option_c_agent_contracts.provider_plugin import OAuthCapability
 
from .catalogue import LANES, Lane, validate_catalogue
from .client import OpenAICompatClient
from .usage import normalise_usage
 
 
#: Rows registered by :func:`register_all`, in catalogue declaration order.
#: The opening menu discovers lanes through the registry rather than through
#: this list; it exists so a caller can ask what the bundle contributed.
_REGISTERED: list[Lane] = []
 
 
class OpenAICompatProvider(ProviderPlugin):
    """A :class:`ProviderPlugin` configured from one :class:`Lane` row."""
 
    def __init__(self, lane: Lane) -> None:
        func = "__init__"
        self.lane = lane
        # Per-instance identity shadows the ABC's class attributes — this is
        # what lets one class serve seven distinctly-named lanes.
        self.name = lane.name
        self.api_key_env_var = lane.api_key_env_var
        self.base_url_env_var = lane.base_url_env_var
        self.default_base_url = lane.default_base_url
        self.default_model = lane.default_model
        self._models = lane.model_caps()
        _dbg("OC_CHAT.CLIENT", func,
             f"lane={lane.name} models={len(self._models)} free={lane.free} "
             f"verification={lane.verification}", 3)
 
    @property
    def oauth(self) -> OAuthCapability | None:
        """Expose OAuth only for the OpenRouter lane; all other rows opt out."""
        if self.name != 'openrouter':
            return None
        from .oauth import OpenRouterOAuth
        return OpenRouterOAuth()
 
    # ── Client construction ──────────────────────────────────────────────────
 
    def build_client(self, api_key: str, base_url: str) -> BaseClient:
        func = "build_client"
        _dbg("OC_CHAT.CLIENT", func, f"lane={self.name} base_url={base_url}", 2)
        return OpenAICompatClient(
            api_key=api_key,
            base_url=base_url,
            extra_headers=self.lane.extra_headers,
            lane_name=self.name,
        )
 
    # ── Model registry ───────────────────────────────────────────────────────
 
    def models(self) -> dict[str, ModelCaps]:
        return self._models
 
    def fetch_live_models(self, client: BaseClient) -> list[str]:
        return client.list_models()
 
    def default_caps(self) -> ModelCaps:
        """Caps for a live-discovered model absent from the curated set.
 
        Deliberately the same honest shape the catalogue uses: tools
        attemptable, nothing claimed about thinking/reasoning, and the
        contract's conservative context default rather than a guess.
        """
        return ModelCaps(
            supports_thinking=False,
            supports_reasoning=False,
            supports_tools=True,
            max_context_tokens=65_536,
            tool_schema_dialect="openai_chat",
        )
 
    # ── Tier mapping ─────────────────────────────────────────────────────────
 
    def tier(self, name: str) -> str | None:
        return self.lane.tiers.get(name)
 
    # ── Per-call shaping ─────────────────────────────────────────────────────
 
    def build_extra_body(
        self,
        model: str,
        *,
        enable_thinking: bool,
        effort: str | None,
    ) -> dict | None:
        """Return ``None`` — no catalogue lane carries a thinking payload.
 
        DeepSeek returns ``{"thinking": {"type": "enabled"}}`` here. Every
        lane in this bundle runs as-shipped: no ``ModelCaps`` in the
        catalogue advertises ``supports_thinking``, so honouring
        *enable_thinking* would mean inventing a payload shape that was
        never proven against the endpoint.
        """
        return None
 
    def transform_outgoing_messages(
        self,
        messages: list[dict],
        tool_schemas: list[dict] | None,
    ) -> list[dict]:
        """Return *messages* unchanged — no lane has a prefix-cache trick."""
        return messages
 
    # ── Response parsing ─────────────────────────────────────────────────────
 
    def parse_usage(self, usage: dict) -> UsageStats:
        return normalise_usage(usage)
 
    # ── Registration ─────────────────────────────────────────────────────────
 
    @classmethod
    def register_lane(cls, lane: Lane) -> "OpenAICompatProvider":
        """Build and register one provider instance for *lane*."""
        from option_c_agent_contracts import registry
 
        instance = cls(lane)
        registry.register(instance)
        return instance
 
 
def register_all(lanes: Sequence[Lane] = LANES) -> list["OpenAICompatProvider"]:
    """Validate *lanes*, then register one provider per row.
 
    Validation runs to completion BEFORE the first registration, so an
    invalid row aborts loudly with its name and leaves the registry
    untouched (plan 215 P5/W1).
    """
    func = "register_all"
    validated = validate_catalogue(lanes)
    instances = [OpenAICompatProvider.register_lane(lane) for lane in validated]
    _REGISTERED.clear()
    _REGISTERED.extend(validated)
    _dbg("OC_CHAT.CLIENT", func,
         f"registered {len(instances)} lane(s): "
         f"{', '.join(i.name for i in instances)}", 2)
    return instances
 
 
def registered_lanes() -> tuple[Lane, ...]:
    """The lane rows this bundle registered, in catalogue declaration order."""
    return tuple(_REGISTERED)
 
 
__all__ = [
    "OpenAICompatProvider",
    "register_all",
    "registered_lanes",
]