provider.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# providers/deepseek/provider.py
 
"""
DeepSeek provider implementation.
 
Purpose
-------
Concrete :class:`oc_chat.core.provider_plugin.ProviderPlugin` for the
DeepSeek API. Encapsulates everything provider-specific: client
construction, the model/tier registry, the ``extra_body`` shape for
thinking, the prefix-cache sentinel embedding, and prompt-cache
hit/miss extraction.
 
Importing the parent package (``import providers.deepseek``) triggers
self-registration via ``DeepSeekProvider.register()``. The rest of
the app retrieves it via
``oc_chat.core.provider_plugin.registry.get("deepseek")``.
 
Responsibilities
----------------
- :class:`DeepSeekProvider` — implement every abstract method on
  :class:`ProviderPlugin` for DeepSeek.
- :meth:`build_client` — construct a :class:`DeepSeekClient` with
  the supplied API key and base URL.
- :meth:`build_extra_body` — produce the
  ``{"thinking": {"type": "enabled"}}`` shape when the target model
  supports thinking.
- :meth:`transform_outgoing_messages` — deprecated no-op since plan 135
  (the dump is gone; ``render_request`` is the seam now).
- :meth:`parse_usage` — convert raw usage dicts to :class:`UsageStats`
  via the shared cache-hit/miss helper.
- :meth:`tier` / :meth:`models` — expose the static registries.
- :meth:`fetch_live_models` — list live models, dropping retired ids.
- :meth:`migrate_model_id` — map a retired alias to its V4 replacement
  for saved-session migration (plan 135 §4.6).
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- ``transform_outgoing_messages`` returns its input unchanged (plan 135
  neutered the dump; it must never re-embed).
- ``build_extra_body`` returns ``None`` for models whose ``ModelCaps``
  do not advertise ``supports_thinking``.
- ``tier`` returns ``None`` (not a model id) when the requested tier
  is not in :data:`TIER_MAP`.
- ``migrate_model_id`` returns ``None`` for live ids; for a retired alias
  it returns ``(replacement, notice)`` with a distinct per-alias target.
- All abstract surface lives here; the bundle's other modules are
  pure data / SDK plumbing.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- option_c_agent_contracts (ProviderPlugin base class and the ModelCaps, UsageStats and BaseClient types)
- providers.deepseek.client (DeepSeekClient returned by build_client)
- providers.deepseek.cache_usage (prompt-cache hit and miss extraction for parse_usage)
- providers.deepseek.models (MODELS, TIER_MAP and RETIRED_ALIASES registries)
"""
from __future__ import annotations
 
from oc_diagnostics import _dbg
 
from option_c_agent_contracts import (
    ProviderPlugin,
    ModelCaps,
    UsageStats,
    BaseClient,
)
 
from .client import DeepSeekClient
from .cache_usage import usage_prompt_cache_hit_miss
from .models import MODELS, TIER_MAP, RETIRED_ALIASES
 
 
# DeepSeek API base URL. Override by setting ``DEEPSEEK_BASE_URL`` in the
# environment — ClientInterface picks the env value over this default.
_DEFAULT_BASE_URL = "https://api.deepseek.com"
 
 
# DeepSeek prefix-cache sentinel — duplicated here from
# oc_chat/core/session.py during the C.2 transition. session.py is still
# the authority for the current session-mutating helper; once C.3 routes
# message transformation through the provider, this becomes the canonical
# location and session.py imports from here.
TOOL_SCHEMA_SENTINEL = "<!--dc:embedded-tool-schemas-->"
 
 
class DeepSeekProvider(ProviderPlugin):
    name = "deepseek"
    # OCT-LINT: disable=no_hardcoded_secrets reason="Env var NAME to read the key from, not a credential"
    api_key_env_var = "DEEPSEEK_API_KEY"
    base_url_env_var = "DEEPSEEK_BASE_URL"
    default_base_url = _DEFAULT_BASE_URL
    # Plan 135 §4.6 (operator-confirmed): the default is the reasoning tier — a
    # deliberate capability↑/cost↑/latency↑ change from the old deepseek-chat.
    default_model = "deepseek-v4-pro"
 
    # ── Client construction ──────────────────────────────────────────────────
 
    def build_client(self, api_key: str, base_url: str) -> BaseClient:
        func = "build_client"
        _dbg("OC_CHAT.CLIENT", func, f"DeepSeekProvider base_url={base_url}", 2)
        return DeepSeekClient(api_key=api_key, base_url=base_url)
 
    # ── Model registry ───────────────────────────────────────────────────────
 
    def models(self) -> dict[str, ModelCaps]:
        return MODELS
 
    def fetch_live_models(self, client: BaseClient) -> list[str]:
        # Plan 135 §4.6: drop retired ids so live discovery cannot re-surface
        # deepseek-chat / deepseek-reasoner in /models after their removal.
        return [m for m in client.list_models() if m not in RETIRED_ALIASES]
 
    def default_caps(self) -> ModelCaps:
        # Permissive default for newly-discovered DeepSeek models: tools yes,
        # 64K context to be safe. Specific known models override via MODELS.
        return ModelCaps(
            supports_thinking=False,
            supports_reasoning=False,
            supports_tools=True,
            max_context_tokens=65_536,
        )
 
    def migrate_model_id(self, model_id: str) -> tuple[str, str] | None:
        func = "migrate_model_id"
        replacement = RETIRED_ALIASES.get(model_id)
        if replacement is None:
            return None
        # Distinct per-alias notice (plan 135 §4.6) — never collapse to one target.
        tier_desc = "reasoning tier" if replacement == "deepseek-v4-pro" else "cheap tier"
        notice = (
            f"[Model '{model_id}' is retired — migrated this session to "
            f"'{replacement}' ({tier_desc}).]"
        )
        _dbg("OC_CHAT.CLIENT", func, f"{model_id} -> {replacement}", 2)
        return replacement, notice
 
    # ── Tier mapping ─────────────────────────────────────────────────────────
 
    def tier(self, name: str) -> str | None:
        return TIER_MAP.get(name)
 
    # ── Per-call shaping ─────────────────────────────────────────────────────
 
    def build_extra_body(
        self,
        model: str,
        *,
        enable_thinking: bool,
        effort: str | None,  # noqa: ARG002 — DeepSeek puts effort in `reasoning_effort` kwarg, not extra_body
    ) -> dict | None:
        """Return the ``extra_body`` payload that toggles thinking on
        DeepSeek's thinking-capable models (e.g. deepseek-v4-pro).
 
        Effort is *not* part of ``extra_body`` for DeepSeek — it's a
        top-level ``reasoning_effort`` SDK kwarg, handled by the client
        directly. This method only owns the thinking flag.
        """
        if not enable_thinking:
            return None
        caps = MODELS.get(model)
        if caps is None or not caps.supports_thinking:
            return None
        return {"thinking": {"type": "enabled"}}
 
    def transform_outgoing_messages(
        self,
        messages: list[dict],
        tool_schemas: list[dict] | None,
    ) -> list[dict]:
        """Return *messages* unchanged (plan 135 §4.6 — neutered).
 
        Deprecated: this hook used to embed the full tool-schema JSON dump into
        ``messages[0]`` for DeepSeek's disk prefix cache. Plan 135 removed the
        dump everywhere — ``session.messages[0]`` already carries the slim
        policy + compact tool map (the full schema travels via the API ``tools``
        parameter), and :meth:`render_request` is the provider seam now. This is
        kept as a no-op for interface stability; it must never re-embed.
        """
        return messages
 
    # ── Optional extension hooks ─────────────────────────────────────────────
 
    _warned_opts: set = set()
 
    def translate_generation_options(self, opts: "GenerationOptions") -> dict | None:  # type: ignore[override]
        from option_c_agent_contracts import GenerationOptions
        func = "translate_generation_options"
        _SUPPORTED = {"temperature", "top_p", "seed"}
        d: dict = {}
        for fname, val in opts.model_dump(exclude_none=True).items():
            if fname in _SUPPORTED:
                d[fname] = val
            elif fname not in self._warned_opts:
                self._warned_opts.add(fname)
                _dbg("OC_CHAT.CLIENT", func,
                     f"DeepSeek ignores option {fname!r}", 2)
        return d if d else None
 
    # ── Response parsing ─────────────────────────────────────────────────────
 
    def parse_usage(self, usage: dict) -> UsageStats:
        if not isinstance(usage, dict):
            return UsageStats()
        hit, miss = usage_prompt_cache_hit_miss(usage)
        details = usage.get("completion_tokens_details") or {}
        return UsageStats(
            prompt_tokens=int(usage.get("prompt_tokens") or 0),
            completion_tokens=int(usage.get("completion_tokens") or 0),
            total_tokens=int(usage.get("total_tokens") or 0),
            cache_hit_tokens=hit,
            cache_miss_tokens=miss,
            # Plan 135 §3.9: separately-billed thinking tokens when reported.
            reasoning_tokens=int((details.get("reasoning_tokens") if isinstance(details, dict) else 0) or 0),
        )