orchestrata_ultimate.llm.client

Reference › orchestrata_ultimate › llm › client

client is the async facade every volk uses to reach a language model; the base agent (base) and several code drivers import it. Volks call call_llm, call_reasoner or call_with_tools and never see which provider answers. Provider and model are resolved at call time from config by volk alias, and the request is routed through a process-wide ProviderManager from the providers package. Specification 028 is the main source of requirements. The curated narrative is llm.

• call_llm builds a system and user message pair, resolves the model with get_volk_model unless one is passed, and sends it through _execute_with_retry. A graph_context string is prepended to the user message, and the system prompt is extended to require a closing "Entities used" section. The reply is then checked by a graph-context validator, and the call is retried with a stricter prompt up to a bounded number of times if cited entities are missing.
• call_reasoner and call_with_tools follow the same path for reasoning-tier calls and tool-calling turns; the latter takes a message list and tool schemas.
• _get_provider_manager builds the singleton once. PROVIDER_CONFIGS is a table of _ProviderSpec entries, one per backend (Claude CLI, Anthropic, DeepSeek, OpenAI, OpenRouter, Ollama), each with a builder; _register_all_providers registers those whose credentials are present in the environment, and the Claude CLI and Ollama need no key. Environment values come from the process environment and from .env files parsed by _load_dotenv.
• _execute_with_retry retries with exponential backoff. _should_stop_retrying stops on a non-transient error (_is_transient), at MAX_RETRIES attempts, or when a total time budget is spent; LLM_RETRY_TOTAL_BUDGET_SECONDS sets the budget (default 900) and LLM_RETRY_DELAY_SCALE scales the delays. _compute_retry_delay picks the delay, aware of rate-limit errors. After the retries it raises LLMClientError from exceptions.
• When LLM_CASSETTE_DIR is set, _execute_with_retry becomes a record and replay boundary instead of calling a provider, using _cassette: a hit is replayed, LLM_CASSETTE_SYNTH=1 returns a synthetic reply, LLM_CASSETTE_RECORD=1 records a real reply, and otherwise it raises.
• _capture_exchange appends each request and response as a JSON line to the file named by ORCHESTRATA_CONVERSATION_LOG when that variable is set, and does nothing otherwise; it never raises into the caller.

At a glance

Class diagram

Dependency diagram

Truncated at 25 nodes: showing 13 of 14 imports and 11 of 11 importers.

API

Constants

Classes

_ProviderSpec

Bases: NamedTuple · Line 336

One data-driven entry in the provider-registration table.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/llm/client.py

# mypy: ignore-errors
"""Async LLM client for orchestrata volk brains.

Provides call_llm() and call_reasoner() — the primary interface
all volks use to invoke LLM reasoning. Routes requests through
the production ProviderManager with per-volk provider/model
assignment from VolkConfig.

Usage:
    from orchestrata_ultimate.llm.client import call_llm

    response = await call_llm(
        "You are a requirements analyst.",
        "Analyze this intent: ...",
        volk="U:BA",
    )
"""

from __future__ import annotations

import asyncio
import logging
import threading
import time
from collections.abc import Callable, Mapping
from typing import TYPE_CHECKING, Any, NamedTuple

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    Message,
    ProviderRequest,
    ProviderResponse,
)
from orchestrata_ultimate.llm.config import (
    get_volk_model,
)
from orchestrata_ultimate.llm.exceptions import (
    LLMClientError,
    LLMProviderUnavailableError,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.providers.provider_manager import ProviderManager

_logger = logging.getLogger(__name__)

# Retry configuration for transient errors. Delays must be generous enough to
# clear a provider's per-minute rate-limit window (HTTP 429): under a sustained
# implement/review loop, deepseek-v4-pro returns 429 in bursts, and a short
# backoff (the old 1+2+4+8 = 15s effective) exhausted retries and crashed the
# volk. Effective cumulative backoff here is 5+10+20+40+60 = 135s.
MAX_RETRIES = 6
RETRY_DELAYS = [5, 10, 20, 40, 60, 60]

# Canned, broadly-viable text returned by the synthetic-replay cassette mode
# (LLM_CASSETTE_SYNTH=1). Deliberately contains prose AND a fenced code block so
# it satisfies the common shape assertions (non-empty content, "contains code",
# "mentions a function") of LLM-routed tests without any real provider call.
# Policy 2026-08-04: "do not call the llm in the tests, just return any viable
# text and save the time."
# The text also carries the Architect volk's design-section markers
# (ARCHITECTURE:/REQUIREMENTS:/FILE_TREE:/API_SPEC:/CLASS_DIAGRAM:, parsed by
# architect_project._parse_design) with non-blank placeholder bodies, so a
# synth-served design response never trips the all-core-sections-blank guard.
# It must NOT contain "REQUIREMENTS_COMPLETE" — the Concierge intake volk
# branches on that token.
_SYNTHETIC_LLM_TEXT = (
    "Here is a viable response.\n\n"
    "```python\n"
    "def solution():\n"
    '    """Placeholder implementation generated for offline test replay."""\n'
    "    return True\n"
    "```\n\n"
    "ARCHITECTURE:\n"
    "Single-module placeholder architecture: one entry point delegating to a\n"
    "pure function; no external services.\n\n"
    "REQUIREMENTS:\n"
    "- The response must be viable, deterministic, and produced offline.\n\n"
    "FILE_TREE:\n"
    "src/\n"
    "  solution.py\n\n"
    "API_SPEC:\n"
    "solution.py: def solution() -> bool — returns True.\n\n"
    "CLASS_DIAGRAM:\n"
    "classDiagram\n"
    "    class Solution\n\n"
    "This placeholder satisfies the request."
)

_TRANSIENT_MARKERS = frozenset(
    {
        "500",
        "502",
        "503",
        "429",
        "Timeout",
        "timed out",
        "Connection refused",
        "ConnectionReset",
        "Connection reset",
        "RemoteDisconnected",
        "BrokenPipe",
    }
)

_RATE_LIMIT_MARKERS = frozenset(
    {
        "RPM exhausted",
        "TPM exhausted",
        "rate limit exhausted",
        "Too Many Requests",
    }
)

# Lazy-initialized ProviderManager singleton
_provider_manager = None
_provider_manager_lock = threading.Lock()


def _parse_dotenv_file(path: str) -> dict[str, str]:
    """Parse a .env file and return missing environment variables.

    Args:
        path: Path to the .env file.

    Returns:
        Mapping of variable names to values that are not already
        present in the process environment.
    """
    import os

    loaded: dict[str, str] = {}
    with open(path) as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, v = line.split("=", 1)
            v = v.strip().strip('"').strip("'")
            key = k.strip()
            if v and key not in os.environ:
                loaded[key] = v
    return loaded


def _load_dotenv() -> dict[str, str]:
    """Load environment variables from .env files.

    Returns:
        Effective environment mapping: the current process environment
        merged with any variables loaded from .env files. Values already
        present in ``os.environ`` take precedence over .env entries.
    """
    import os

    merged: dict[str, str] = dict(os.environ)
    for env_path in [".env", "/opt/orchestrata/.env"]:
        if not os.path.exists(env_path):
            continue
        try:
            merged.update(_parse_dotenv_file(env_path))
        except Exception as exc:
            _logger.debug("dotenv load: %s", exc)
    return merged


def _build_claude_cli(env: Mapping[str, str]) -> tuple[Any, Any] | None:
    """Build the Claude CLI adapter (OAuth, no API key needed).

    Primary adapter for container deployments where the Claude CLI
    binary is available on PATH. Returns ``None`` (no registration,
    no log) when the CLI binary isn't found on this host.
    """
    from orchestrata_ultimate.core.protocols import ModelPricing, ProviderConfig
    from orchestrata_ultimate.providers.adapters.claude_cli_adapter import (
        ClaudeCLIAdapter,
    )

    _free = ModelPricing()
    cfg = ProviderConfig(
        models=["sonnet", "opus", "haiku"],
        rate_limit_rpm=20,
        timeout_seconds=600.0,
        pricing={"sonnet": _free, "opus": _free, "haiku": _free},
    )
    cli_adapter = ClaudeCLIAdapter(timeout_seconds=cfg.timeout_seconds, config=cfg)
    if not cli_adapter._claude_bin:
        return None
    return cli_adapter, cfg


def _build_anthropic(env: Mapping[str, str]) -> tuple[Any, Any] | None:
    """Build the Anthropic API adapter (key already confirmed present)."""
    from orchestrata_ultimate.core.protocols import ModelPricing, ProviderConfig
    from orchestrata_ultimate.providers.adapters.anthropic_adapter import (
        AnthropicAdapter,
    )

    anthropic_key = env.get("ANTHROPIC_API_KEY", "")
    _p = ModelPricing()
    _haiku_pricing = ModelPricing(input_per_1m=1.00, output_per_1m=5.00)
    cfg = ProviderConfig(
        api_key=anthropic_key,
        models=["claude-sonnet-4-6", "claude-opus-4-6", "claude-haiku-4-5"],
        rate_limit_rpm=50,
        timeout_seconds=120.0,
        pricing={
            "claude-sonnet-4-6": _p,
            "claude-opus-4-6": _p,
            "claude-haiku-4-5": _haiku_pricing,
        },
    )
    return AnthropicAdapter(cfg), cfg


def _build_deepseek(env: Mapping[str, str]) -> tuple[Any, Any] | None:
    """Build the DeepSeek adapter (key already confirmed present)."""
    from orchestrata_ultimate.core.protocols import ModelPricing, ProviderConfig
    from orchestrata_ultimate.providers.adapters.deepseek_adapter import (
        DeepSeekAdapter,
    )

    deepseek_key = env.get("DEEPSEEK_API_KEY", "")
    _p = ModelPricing()
    cfg = ProviderConfig(
        api_key=deepseek_key,
        base_url="https://api.deepseek.com/v1",
        models=["deepseek-v4-pro", "deepseek-v4-flash"],
        rate_limit_rpm=30,
        timeout_seconds=180.0,
        pricing={
            "deepseek-v4-pro": _p,
            "deepseek-v4-flash": _p,
        },
    )
    return DeepSeekAdapter(cfg), cfg


def _build_openai(env: Mapping[str, str]) -> tuple[Any, Any] | None:
    """Build the OpenAI adapter (key already confirmed present)."""
    from orchestrata_ultimate.core.protocols import ModelPricing, ProviderConfig
    from orchestrata_ultimate.providers.adapters.openai_adapter import OpenAIAdapter

    openai_key = env.get("OPENAI_API_KEY", "")
    _p = ModelPricing()
    models = ["gpt-4o", "gpt-4o-mini", "o1", "o3-mini"]
    cfg = ProviderConfig(
        api_key=openai_key,
        base_url="https://api.openai.com/v1",
        models=models,
        rate_limit_rpm=500,
        timeout_seconds=60.0,
        pricing={m: _p for m in models},
    )
    return OpenAIAdapter(cfg), cfg


def _build_openrouter(env: Mapping[str, str]) -> tuple[Any, Any] | None:
    """Build the OpenRouter adapter (key already confirmed present).

    OpenRouter routes a single API key to 339+ upstream models including
    Anthropic (Claude Opus 4.8, Sonnet 4.6), Qwen (3.7-plus, 3.7-max, coder),
    Kimi (k2.7-code, k2-thinking), DeepSeek, Meta, and Gemini 3.1.
    """
    from orchestrata_ultimate.core.protocols import ModelPricing, ProviderConfig
    from orchestrata_ultimate.providers.adapters.openrouter_adapter import (
        OpenRouterAdapter,
    )

    openrouter_key = env.get("OPENROUTER_API_KEY", "")
    _p = ModelPricing()
    models = [
        "anthropic/claude-sonnet-4-6",
        "anthropic/claude-haiku-4-5-20251001",
        "deepseek/deepseek-chat",
        "deepseek/deepseek-reasoner",
        "qwen/qwen3.7-plus",
        "qwen/qwen3.7-max",
        "qwen/qwen3-coder-plus",
        "qwen/qwen3-coder",
        "moonshotai/kimi-k2.7-code",
        "moonshotai/kimi-k2-thinking",
        "meta-llama/llama-3.3-70b-instruct",
        "google/gemini-3.1-pro-preview",
        "minimax/minimax-m3",
        # openrouter-cheap profile tier (2026-07-13): real pricing below so
        # the cost tracker reports true spend for the cheap-model spike.
        "deepseek/deepseek-v4-flash",
        "qwen/qwen3-coder-next",
    ]
    pricing = {m: _p for m in models}
    pricing["deepseek/deepseek-v4-flash"] = ModelPricing(
        input_per_1m=0.077, output_per_1m=0.154, context_window=1_048_576
    )
    pricing["qwen/qwen3-coder-next"] = ModelPricing(
        input_per_1m=0.11, output_per_1m=0.80, context_window=262_144
    )
    # openrouter-minimax orchestration tier (user-selected 2026-07-13 after
    # the v4-flash reliability failures in the LRU-rebuild demo runs 3/4):
    # MiniMax M3, catalog price 0.30/1.20 per 1M, 1M context.
    pricing["minimax/minimax-m3"] = ModelPricing(
        input_per_1m=0.30, output_per_1m=1.20, context_window=1_048_576
    )
    cfg = ProviderConfig(
        api_key=openrouter_key,
        base_url="https://openrouter.ai/api/v1",
        models=models,
        rate_limit_rpm=60,
        timeout_seconds=180.0,
        pricing=pricing,
    )
    return OpenRouterAdapter(cfg), cfg


def _build_ollama(env: Mapping[str, str]) -> tuple[Any, Any] | None:
    """Build the Ollama adapter for local inference (no key needed).

    Raises ``KeyError`` (caught by the caller) when ``OLLAMA_BASE_URL``
    isn't set, matching the original unconditional-attempt behavior.
    """
    from orchestrata_ultimate.core.protocols import ModelPricing, ProviderConfig
    from orchestrata_ultimate.providers.adapters.ollama_adapter import OllamaAdapter

    ollama_url = env["OLLAMA_BASE_URL"]
    _p = ModelPricing()
    cfg = ProviderConfig(
        base_url=ollama_url,
        models=["qwen3:1.7b"],
        rate_limit_rpm=120,
        timeout_seconds=120.0,
        pricing={"qwen3:1.7b": _p},
    )
    return OllamaAdapter(cfg), cfg


class _ProviderSpec(NamedTuple):
    """One data-driven entry in the provider-registration table.

    Attributes:
        key: Registry name passed to ``manager.register_provider``.
        display_name: Human-readable name used in log messages.
        env_key: Env var whose presence gates registration, or ``None``
            when the provider is always attempted (its ``build`` may
            still raise/return ``None`` to skip).
        build: Builds ``(adapter, config)`` from the effective env, or
            returns ``None`` to skip registration silently.
        quiet_failure: When True, a build exception is logged at DEBUG
            ("... skipped") instead of WARNING ("... failed") — matches
            the original claude_cli/ollama "best-effort, no key required"
            adapters vs. the API-key-gated ones.
    """

    key: str
    display_name: str
    env_key: str | None
    build: Callable[[Mapping[str, str]], tuple[Any, Any] | None]
    quiet_failure: bool = False


PROVIDER_CONFIGS: tuple[_ProviderSpec, ...] = (
    _ProviderSpec("claude", "Claude CLI", None, _build_claude_cli, quiet_failure=True),
    _ProviderSpec("anthropic", "Anthropic", "ANTHROPIC_API_KEY", _build_anthropic),
    _ProviderSpec("deepseek", "DeepSeek", "DEEPSEEK_API_KEY", _build_deepseek),
    _ProviderSpec("openai", "OpenAI", "OPENAI_API_KEY", _build_openai),
    _ProviderSpec("openrouter", "OpenRouter", "OPENROUTER_API_KEY", _build_openrouter),
    _ProviderSpec("ollama", "Ollama", None, _build_ollama, quiet_failure=True),
)


def _register_one(
    manager: ProviderManager,
    spec: _ProviderSpec,
    env: Mapping[str, str] | None = None,
) -> None:
    """Register a single provider per its table entry.

    Args:
        manager: The ProviderManager to register the adapter with.
        spec: The provider's table entry (build fn, env gate, logging).
        env: Optional explicit environment mapping (defaults to
            ``os.environ``).
    """
    import os

    # NOTE: falls back to os.environ on any falsy env (None *or* {}) to
    # preserve the original per-provider ``(env or os.environ)`` behavior.
    effective_env = env or os.environ
    if spec.env_key is not None and not effective_env.get(spec.env_key, ""):
        return
    try:
        result = spec.build(effective_env)
    except Exception as exc:
        log = _logger.debug if spec.quiet_failure else _logger.warning
        verb = "skipped" if spec.quiet_failure else "failed"
        log("%s adapter %s: %s", spec.display_name, verb, exc)
        return
    if result is None:
        return
    adapter, cfg = result
    manager.register_provider(spec.key, adapter, cfg)
    _logger.info("Registered %s adapter", spec.display_name)


def _register_claude_cli(
    manager: ProviderManager, env: Mapping[str, str] | None = None
) -> None:
    """Register Claude CLI adapter (OAuth, no API key needed)."""
    _register_one(manager, PROVIDER_CONFIGS[0], env)


def _register_anthropic(
    manager: ProviderManager, env: Mapping[str, str] | None = None
) -> None:
    """Register Anthropic API adapter if ANTHROPIC_API_KEY is set."""
    _register_one(manager, PROVIDER_CONFIGS[1], env)


def _register_deepseek(
    manager: ProviderManager, env: Mapping[str, str] | None = None
) -> None:
    """Register DeepSeek adapter if DEEPSEEK_API_KEY is set."""
    _register_one(manager, PROVIDER_CONFIGS[2], env)


def _register_openai(
    manager: ProviderManager, env: Mapping[str, str] | None = None
) -> None:
    """Register OpenAI adapter if OPENAI_API_KEY is set."""
    _register_one(manager, PROVIDER_CONFIGS[3], env)


def _register_openrouter(
    manager: ProviderManager, env: Mapping[str, str] | None = None
) -> None:
    """Register OpenRouter adapter with latest Qwen 3.7, Kimi K2, Claude Opus 4.8."""
    _register_one(manager, PROVIDER_CONFIGS[4], env)


def _register_ollama(
    manager: ProviderManager, env: Mapping[str, str] | None = None
) -> None:
    """Register Ollama adapter for local inference (no key needed)."""
    _register_one(manager, PROVIDER_CONFIGS[5], env)


def _register_all_providers(
    manager: ProviderManager,
    env: Mapping[str, str] | None = None,
) -> None:
    """Register every provider in ``PROVIDER_CONFIGS`` against ``manager``.

    Args:
        manager: The ProviderManager to register adapters with.
        env: The effective environment mapping (already resolved by the
            caller — see :func:`_register_adapters`).
    """
    for spec in PROVIDER_CONFIGS:
        _register_one(manager, spec, env)


def _register_adapters(
    manager: ProviderManager,
    env: Mapping[str, str] | None = None,
) -> None:
    """Register all available LLM provider adapters.

    Loads environment variables from .env files, then registers
    each adapter that has credentials available.

    Args:
        manager: The ProviderManager to register adapters with.
        env: Optional explicit environment mapping. When provided, it is
            forwarded directly to each registration helper and no .env file
            is loaded. When omitted, variables are loaded from .env files.
    """
    effective_env = env if env is not None else _load_dotenv()
    _register_all_providers(manager, effective_env)


@traces(spec="017", reqs=["REQ-017-014"])
@traces(spec="028", reqs=["REQ-028-050", "REQ-028-051", "REQ-028-052", "REQ-028-053"])
def _get_provider_manager() -> ProviderManager:
    """Get or create the singleton ProviderManager.

    Registers all available adapters on first call.

    Returns:
        The ProviderManager instance.

    Raises:
        LLMProviderUnavailableError: If provider package unavailable.
    """
    global _provider_manager
    if _provider_manager is not None:
        return _provider_manager
    with _provider_manager_lock:
        if _provider_manager is not None:
            return _provider_manager
        try:
            from orchestrata_ultimate.providers.provider_manager import (
                ProviderManager,
            )

            manager = ProviderManager()
            _register_adapters(manager)
            _provider_manager = manager
            return _provider_manager
        except ImportError as exc:
            raise LLMProviderUnavailableError(
                "orchestrata-provider package not installed"
            ) from exc


@traces(spec="017", reqs=["REQ-017-015"])
@traces(
    spec="028",
    reqs=["REQ-028-029", "REQ-028-030", "REQ-028-031", "REQ-028-032", "REQ-028-033"],
)
def _is_transient(exc: Exception) -> bool:
    """Check if an exception represents a transient error.

    Args:
        exc: The exception to check.

    Returns:
        True if the error is transient and should be retried.
    """
    err_str = str(exc)
    if err_str.startswith("No eligible provider found") and (
        "unknown model/capabilities" in err_str
    ):
        # This exact rejection reason (providers/router.py
        # get_rejection_reasons) means NO registered provider serves the
        # requested model/capabilities — a CONFIGURATION verdict retrying
        # cannot change. Defensive pin only: this message matches no
        # _TRANSIENT_MARKERS/_RATE_LIMIT_MARKERS entry today, so without
        # this branch it is already non-retryable; the branch exists so a
        # future marker addition cannot silently make a config verdict
        # retryable. It is NOT the retry driver of the 2026-08-05 CXP
        # accept-queue wedge (backlog 129/128) — that driver is still
        # unidentified (backlog). Other "No eligible provider" rejections
        # (RPM/TPM exhausted, breaker cooldown) stay retryable — their
        # causes clear with time.
        return False
    return any(
        marker in err_str for marker in _TRANSIENT_MARKERS
    ) or _is_rate_limit_error(exc)


@traces(spec="017", reqs=["REQ-017-015"])
@traces(
    spec="028",
    reqs=["REQ-028-034", "REQ-028-035", "REQ-028-036", "REQ-028-037", "REQ-028-038"],
)
def _is_rate_limit_error(exc: Exception) -> bool:
    """Check if an exception is a rate-limit exhaustion signal.

    Detects both local router rejection (RPM/TPM exhausted) and upstream
    HTTP 429 responses so retries can back off past the rate window.
    """
    err_str = str(exc)
    return any(marker in err_str for marker in _RATE_LIMIT_MARKERS)


@traces(spec="017", reqs=["REQ-017-016"])
@traces(
    spec="028",
    reqs=["REQ-028-034", "REQ-028-035", "REQ-028-036", "REQ-028-037", "REQ-028-038"],
)
async def _execute_with_retry(
    request: ProviderRequest,
    label: str,
    env: Mapping[str, str] | None = None,
) -> ProviderResponse:
    """Execute a provider request with exponential backoff retry.

    When ``LLM_CASSETTE_DIR`` is set (env or ``os.environ``), this is a
    record/replay cassette boundary instead of a real provider call: a
    cassette hit returns the captured ``ProviderResponse`` with no network
    call and no retries; a miss either records a real response
    (``LLM_CASSETTE_RECORD=1``) or fails loudly (replay-only — the test
    suite's default mode, so a missing cassette is never silently skipped).
    Empty/unset ``LLM_CASSETTE_DIR`` leaves production behavior untouched.

    Args:
        request: The ProviderRequest to execute.
        label: Label for logging retry attempts.
        env: Optional environment mapping (defaults to ``os.environ``).

    Returns:
        The ProviderResponse.

    Raises:
        LLMClientError: After all retries exhausted, or (cassette mode) when
            no cassette exists for the request and recording is not enabled.
    """
    import os  # noqa: PLC0415

    from orchestrata_ultimate.llm._cassette import CassetteStore  # noqa: PLC0415

    effective_env = env or os.environ
    cassette_dir = effective_env.get("LLM_CASSETTE_DIR", "")
    if cassette_dir:
        import dataclasses  # noqa: PLC0415

        store = CassetteStore(cassette_dir)
        cassette_key = store.key(request)
        hit = store.load(cassette_key)
        if hit is None:
            # Provider-agnostic fallback (policy 2026-08-04): replay the Claude
            # recording of THIS prompt for whatever model it routed to, presented
            # under the requesting model's identity — "copy Claude's answer as if
            # it were the other LLM's." Exact model-specific key wins when
            # present; this only fires on an exact miss, so a genuine per-model
            # cassette is never shadowed.
            generic = store.load(store.content_key(request))
            if generic is not None:
                hit = dataclasses.replace(generic, model_used=request.model)
        if hit is not None:
            _capture_exchange(request, hit, label, 0, env=env)
            return hit
        if effective_env.get("LLM_CASSETTE_SYNTH") == "1":
            # Synthetic replay (policy 2026-08-04: "do not call the llm in the
            # tests, just return any viable text"). On a cassette miss, return a
            # canned, broadly-viable ProviderResponse under the requesting
            # model's identity instead of placing a real call OR raising — so an
            # LLM-routed test completes instantly with plausible text and never
            # touches a provider. Opt-in via env (the test fixtures set it); the
            # strict default lane leaves it unset and still fails loud on a miss.
            synthetic = ProviderResponse(
                content=_SYNTHETIC_LLM_TEXT,
                model_used=request.model or "synthetic",
                provider_used="synthetic",
            )
            _capture_exchange(request, synthetic, label, 0, env=env)
            return synthetic
        if effective_env.get("LLM_CASSETTE_RECORD") == "1":
            manager = _get_provider_manager()
            result = await manager.execute(request)
            _capture_exchange(request, result, label, 0, env=env)
            # Save the exact (model-specific) cassette AND a provider-agnostic
            # copy, so a single Claude recording serves every model that later
            # sends the same prompt.
            store.save(cassette_key, result)
            store.save(store.content_key(request), result)
            return result
        raise LLMClientError(
            f"no LLM cassette for key {cassette_key} (model={request.model}); "
            "re-record locally with LLM_CASSETTE_RECORD=1"
        )

    manager = _get_provider_manager()
    last_exc = None
    # Retry backoff is real wall-clock. LLM_RETRY_DELAY_SCALE compresses it
    # (default 1.0 = production timing per RETRY_DELAYS) so retry-path tests
    # exercise the FULL retry/backoff logic — same retry count, same branches,
    # same sleep call — without sleeping real seconds. Production is untouched.
    retry_delay_scale = float(effective_env.get("LLM_RETRY_DELAY_SCALE", "1"))
    # Total-elapsed budget across ALL attempts (2026-08-05 kernel wedge root
    # cause): claude_cli_adapter's per-attempt failure "Claude CLI timed out
    # after 600s" contains the transient marker "timed out", so the attempt
    # counter alone allowed 6 × 600s + 135s backoff ≈ 62 min of one pinned
    # kernel handler per volk call — the handler pool exhausted and the CXP
    # accept queue wedged at backlog 129/128. Fast transients (connection
    # refused, 5xx) still get all MAX_RETRIES attempts — they fit inside the
    # budget in seconds; only slow-failure modes lose their multiplier.
    retry_budget = float(effective_env.get("LLM_RETRY_TOTAL_BUDGET_SECONDS", "900"))
    started = time.monotonic()
    attempts_made = 0

    for attempt in range(MAX_RETRIES):
        attempts_made = attempt + 1
        try:
            result = await manager.execute(request)
            _capture_exchange(request, result, label, attempt, env=env)
            return result
        except Exception as exc:
            last_exc = exc
            elapsed = time.monotonic() - started
            if _should_stop_retrying(exc, attempt, elapsed, retry_budget):
                break
            delay = _compute_retry_delay(manager, request, exc, attempt)
            _logger.warning(
                "%s attempt %d failed: %s — retrying in %ds",
                label,
                attempt + 1,
                exc,
                delay,
            )
            await asyncio.sleep(delay * retry_delay_scale)

    raise LLMClientError(
        f"{label} failed after {attempts_made} attempt(s): {last_exc}"
    ) from last_exc


def _should_stop_retrying(
    exc: Exception, attempt: int, elapsed: float, budget: float
) -> bool:
    """Decide whether the retry loop stops after a failed attempt.

    Stops on: a non-transient error (retrying cannot change the outcome),
    the attempt counter reaching ``MAX_RETRIES``, or total elapsed time
    reaching ``budget`` — the last one bounds handler occupancy when each
    attempt is itself slow (the 62-minute wedge scenario above).
    """
    return not _is_transient(exc) or attempt >= MAX_RETRIES - 1 or elapsed >= budget


def _compute_retry_delay(
    manager: ProviderManager,
    request: ProviderRequest,
    exc: Exception,
    attempt: int,
) -> int:
    """Choose the retry delay for a failed attempt.

    For rate-limit failures this waits long enough to clear the local
    sliding-window remainder rather than burning retries inside the same
    minute. Falls back to the default exponential backoff for other
    transient errors.
    """
    if not _is_rate_limit_error(exc):
        return RETRY_DELAYS[attempt]
    window_remainder = manager.get_rate_limit_backoff(request)
    if window_remainder <= 0.0:
        # Upstream 429 with no local window metadata — wait a full window.
        return 60
    # Wait past the remainder; a 1s buffer is enough to clear the window.
    return max(int(window_remainder) + 1, 1)


def _capture_exchange(
    request: ProviderRequest,
    result: ProviderResponse,
    label: str,
    attempt: int,
    env: Mapping[str, str] | None = None,
) -> None:
    """Append a full LLM request/response to ``ORCHESTRATA_CONVERSATION_LOG``.

    Real-time provenance for volk conversations: every provider round-trip
    (including validation retries) is captured as one flushed JSONL line when
    the env var is set, and is a no-op otherwise (so production never logs
    prompt/response content by default — SPEC-100 FR-009). Never raises into
    the caller: capture must never break a real LLM call.
    """
    import os  # noqa: PLC0415

    path = (env or os.environ).get("ORCHESTRATA_CONVERSATION_LOG", "")
    if not path:
        return
    try:
        import json  # noqa: PLC0415
        from datetime import UTC, datetime  # noqa: PLC0415

        try:
            from orchestrata_ultimate.core.logging import (  # noqa: PLC0415
                correlation_id,
                session_id,
                volk_id,
            )

            cid, sid, vid = (
                correlation_id.get(""),
                session_id.get(""),
                volk_id.get(""),
            )
        except Exception:
            cid = sid = vid = ""

        entry = {
            "ts": datetime.now(UTC).isoformat(),
            "label": label,
            "attempt": attempt + 1,
            "correlation_id": cid,
            "session_id": sid,
            "volk": getattr(request, "volk_id", None) or vid,
            "model": getattr(request, "model", None),
            "messages": [
                {"role": m.role, "content": m.content}
                for m in getattr(request, "messages", [])
            ],
            "response": getattr(result, "content", None),
            "usage": str(getattr(result, "usage", None)),
            # Diagnostic provenance (run-3 forensics 2026-07-13: a mid-
            # sentence 227-char architect answer was undiagnosable without
            # finish_reason/upstream — the adapter returns both in metadata
            # but they were dropped here). tool_calls as a count: the raw
            # entries repeat the arguments already visible in the next
            # request's messages.
            "finish_reason": (getattr(result, "metadata", None) or {}).get(
                "finish_reason", ""
            ),
            "upstream_model": (getattr(result, "metadata", None) or {}).get(
                "upstream_model", ""
            ),
            "tool_call_count": len(getattr(result, "tool_calls", None) or []),
        }
        with open(path, "a", encoding="utf-8") as fh:
            fh.write(json.dumps(entry, default=str) + "\n")
            fh.flush()
    except Exception:
        pass


@traces(
    spec="028",
    reqs=[
        "REQ-028-003",
        "REQ-028-005",
        "REQ-028-006",
        "REQ-028-040",
        "REQ-028-042",
        "REQ-028-043",
        "REQ-028-049",
    ],
)
async def call_llm(
    system_prompt: str,
    user_message: str,
    model: str | None = None,
    temperature: float = 0.0,
    volk: str | None = None,
    graph_context: str | None = None,
    env: Mapping[str, str] | None = None,
) -> str:
    """Call the configured LLM provider and return the text response.

    Routes through ProviderManager with per-volk provider/model
    assignment. Retries transient errors with exponential backoff.

    If graph_context is provided, it is prepended to the user message
    so the LLM processes it as part of the prompt — not optional, not
    a tool the LLM can choose to ignore. The data is in the tokens.

    Args:
        system_prompt: System prompt for the LLM.
        user_message: User message / query.
        model: Explicit model override (optional).
        temperature: Sampling temperature (0.0 = deterministic).
        volk: Volk alias for provider/model routing (e.g. "U:Cx").
        graph_context: Pre-formatted knowledge graph context to inject
            into the prompt. When set, prepended to user_message.
        env: Optional environment mapping (defaults to ``os.environ``).

    Returns:
        The LLM response text.

    Raises:
        LLMClientError: If the call fails after all retries.
        LLMProviderUnavailableError: If no provider is available.
    """
    volk_model = model
    if not model:
        # Resolve from per-volk config. NOTE: get_volk_model() (and the
        # underlying VolkConfig) has no "default" fallback for an
        # unconfigured/None volk — it raises LLMConfigError. Callers that
        # omit both `model` and `volk` will see that LLMConfigError instead
        # of the LLMClientError/LLMProviderUnavailableError documented above;
        # always pass a `volk` registered in configs/spike-profiles.yaml
        # (see SPEC-028 REQ-028-021/028 for the intended "default" sentinel,
        # which is not yet implemented in llm/config.py).
        volk_model = get_volk_model(volk)

    # Inject graph context directly into the prompt tokens.
    # The LLM has no choice — it processes this data because it's
    # in the message, not in a side-channel it can ignore.
    effective_message = user_message
    if graph_context:
        effective_message = graph_context + "\n\n---\n\n" + user_message
        # Phase 0: graph-context responses must cite the entities they used.
        # Volk calls already get this via compose_persona(); add it here for
        # direct call_llm() callers so validation has something to enforce.
        if "## Entities used" not in system_prompt:
            system_prompt = (
                f"{system_prompt}\n\n"
                "You MUST end your response with a '## Entities used' section "
                "listing every class, function, requirement, decision, and "
                "artifact you acted on from the provided context."
            )

    request = ProviderRequest(
        messages=[
            Message(role="system", content=system_prompt),
            Message(role="user", content=effective_message),
        ],
        model=volk_model,
        parameters={"temperature": temperature},
        volk_id=volk,
    )

    label = f"call_llm(volk={volk})"
    start = time.monotonic()
    response = await _execute_with_retry(request, label, env=env)
    elapsed_ms = (time.monotonic() - start) * 1000

    _logger.debug(
        "%s completed in %.0fms (model=%s, tokens=%s)",
        label,
        elapsed_ms,
        response.model_used if hasattr(response, "model_used") else "?",
        getattr(response, "usage", None),
    )

    # Post-response validation: enforce graph context compliance.
    # If graph_context was injected and contains entities, validate
    # that the response references them. Retry if it doesn't.
    if graph_context:
        try:
            from orchestrata_ultimate.knowledge_graph.graph_context_validator import (
                MAX_VALIDATION_RETRIES,
                GraphContextValidator,
            )

            validator = GraphContextValidator(graph_context)
            if validator.has_entities:
                result = validator.validate(response.content)
                if not result.passed:
                    _logger.warning(
                        "%s graph context validation FAILED: "
                        "%d/%d public entities referenced (%.0f%%, "
                        "required: 100%%). Retrying.",
                        label,
                        result.public_referenced,
                        result.public_count,
                        result.compliance_ratio * 100,
                    )
                    # Retry with forced entity list. Seed with the original
                    # response so the "retries exhausted" return below is safe even
                    # when MAX_VALIDATION_RETRIES == 0 (loop body never runs).
                    retry_response = response
                    for retry in range(MAX_VALIDATION_RETRIES):
                        retry_prompt = validator.build_retry_prompt(
                            effective_message, result
                        )
                        retry_request = ProviderRequest(
                            messages=[
                                Message(role="system", content=system_prompt),
                                Message(role="user", content=retry_prompt),
                            ],
                            model=volk_model,
                            parameters={"temperature": temperature},
                            volk_id=volk,
                        )
                        retry_response = await _execute_with_retry(
                            retry_request, f"{label}_retry{retry + 1}", env=env
                        )
                        result = validator.validate(retry_response.content)
                        if result.passed:
                            _logger.info(
                                "%s graph context validation PASSED on retry %d",
                                label,
                                retry + 1,
                            )
                            return retry_response.content
                        _logger.warning(
                            "%s retry %d still failed: %d/%d public entities (%.0f%%)",
                            label,
                            retry + 1,
                            result.public_referenced,
                            result.public_count,
                            result.compliance_ratio * 100,
                        )
                    # All retries exhausted — return last response with warning
                    _logger.error(
                        "%s graph context validation FAILED after %d retries. "
                        "Proceeding with non-compliant response.",
                        label,
                        MAX_VALIDATION_RETRIES,
                    )
                    return retry_response.content
        except ImportError:
            pass  # knowledge package not installed — skip validation

    return response.content


@traces(spec="017", reqs=["REQ-017-018"])
@traces(
    spec="028",
    reqs=["REQ-028-011", "REQ-028-012", "REQ-028-013", "REQ-028-014", "REQ-028-015"],
)
async def call_reasoner(
    user_message: str,
    model: str | None = None,
    temperature: float = 1.0,
    volk: str | None = None,
    graph_context: str | None = None,
    env: Mapping[str, str] | None = None,
) -> str:
    """Call a reasoning-tier LLM model for deep analysis.

    Used by the Architect volk (U:Ar) for architectural reasoning.
    Routes to the reasoning model variant for the configured provider.

    Args:
        user_message: The reasoning prompt.
        model: Explicit model override (optional).
        temperature: Sampling temperature (default 1.0 for creativity).
        volk: Volk alias (defaults to reasoning tier).
        graph_context: Pre-formatted knowledge graph context to inject.
        env: Optional environment mapping (defaults to ``os.environ``).

    Returns:
        The reasoning model response text.

    Raises:
        LLMClientError: If the call fails after all retries.
    """
    system_prompt = "You are a deep reasoning architect. Think step by step."
    reasoning_model = model
    if not reasoning_model:
        if volk:
            reasoning_model = get_volk_model(volk)
        else:
            reasoning_model = get_volk_model("U:Ar")

    return await call_llm(
        system_prompt,
        user_message,
        model=reasoning_model,
        temperature=temperature,
        volk=volk or "U:Ar",
        graph_context=graph_context,
        env=env,
    )


async def call_with_tools(
    messages: list[Message],
    tools: list[dict],
    model: str | None = None,
    temperature: float = 0.2,
    volk: str | None = None,
    tool_choice: str = "auto",
    env: Mapping[str, str] | None = None,
) -> ProviderResponse:
    """Call the LLM with function/tool calling (SPEC-103 FR-005).

    Unlike :func:`call_llm`, this takes a full message list (so the caller can
    carry assistant tool_calls and tool-result turns) and returns the whole
    :class:`ProviderResponse` so the caller can read ``.tool_calls`` as well as
    ``.content``. ``tools`` and ``tool_choice`` are placed in the request
    parameters so the adapter forwards them to the provider via ``**parameters``.

    Args:
        messages: Conversation so far (system/user/assistant/tool messages).
        tools: OpenAI/DeepSeek-compatible function schemas.
        model: Explicit model override; defaults to the volk's model.
        temperature: Sampling temperature (low for deterministic refactors).
        volk: Volk alias for provider/model routing.
        tool_choice: "auto" (default), "none", or a forced tool.
        env: Optional environment mapping (defaults to ``os.environ``).

    Returns:
        The ProviderResponse, carrying ``content`` and ``tool_calls``.

    Raises:
        LLMClientError: If the call fails after all retries.
    """
    resolved = model
    if not resolved:
        resolved = get_volk_model(volk)
    request = ProviderRequest(
        messages=messages,
        model=resolved,
        parameters={
            "temperature": temperature,
            "tools": tools,
            "tool_choice": tool_choice,
        },
        tools=tools,
        tool_choice=tool_choice,
        volk_id=volk,
    )
    return await _execute_with_retry(request, f"call_with_tools(volk={volk})", env=env)