orchestrata_ultimate.llm._cassette

Reference › orchestrata_ultimate › llm › _cassette

_cassette is the record and replay store that lets tests and offline runs reuse captured model exchanges without any network call. client uses it for chat completions and the semantic-graph embedders use it for embeddings.

CassetteStore is a directory of captured chat exchanges: key computes a deterministic key from a provider request, content_key computes a model-independent key from the prompt alone, and load and save read and write a real ProviderResponse. PayloadCassetteStore does the same for arbitrary JSON exchanges under a kind label, and its save never raises into the caller. cassette_exchange applies four modes to a non-chat exchange: with no cassette directory (EMBEDDING_CASSETTE_DIR, falling back to LLM_CASSETTE_DIR) it runs the live call; a cassette hit is replayed; EMBEDDING_CASSETTE_SYNTH or LLM_CASSETTE_SYNTH equal to 1 returns the synthetic value; LLM_CASSETTE_RECORD=1 runs live once and records. Otherwise it raises LLMCassetteMiss, so a missing recording is a gap and not a silent billed call. synthetic_embedding and synthetic_embeddings produce deterministic unit vectors to stand in for real embeddings.

At a glance

Class diagram

Dependency diagram

API

Classes

CassetteStore

Line 37

Directory-backed store of captured LLM request/response pairs.

PayloadCassetteStore

Line 191

Directory-backed record/replay of arbitrary JSON provider exchanges.

LLMCassetteMiss

Bases: RuntimeError · Line 262

Replay-only mode was asked for an exchange nothing has recorded.

Functions

Source

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

"""LLM record/replay cassette store.

Not mocking: a cassette is the exact ``ProviderResponse`` a real provider
returned, captured once and replayed byte-for-byte forever after. This lets
the test suite exercise the full ``call_llm``/``call_with_tools`` path
without ever placing a network call to a real LLM provider.

Wired into ``llm/client.py::_execute_with_retry`` (env-gated via
``LLM_CASSETTE_DIR``/``LLM_CASSETTE_RECORD``): a cassette hit short-circuits
``manager.execute()`` entirely; a miss either records a fresh real response
(``LLM_CASSETTE_RECORD=1``) or raises loudly (replay-only, the default/CI
mode) so a missing cassette is never silently papered over.
"""

from __future__ import annotations

import hashlib
import json
import logging
import os
import struct
from collections.abc import Awaitable, Callable, Mapping, Sequence
from dataclasses import asdict
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.protocols import (
    CostRecord,
    ProviderRequest,
    ProviderResponse,
    TokenUsage,
)

_logger = logging.getLogger(__name__)


class CassetteStore:
    """Directory-backed store of captured LLM request/response pairs.

    One JSON file per cassette, named ``<key>.json`` under ``directory``,
    where ``key`` is a deterministic hash of the request's model/temperature/
    messages (see :meth:`key`). Mirrors ``_capture_exchange``'s safety
    contract: a broken cassette directory or a corrupt file must never break
    a real LLM call, so ``save`` never raises into the caller.
    """

    def __init__(self, directory: str | Path) -> None:
        self.directory = Path(directory)

    def key(self, request: ProviderRequest) -> str:
        """Compute the deterministic cassette key for a request.

        sha256 hex digest over canonical JSON of ``{model, temperature,
        messages}`` — temperature is included only when the request's
        ``parameters`` carries one, so requests that never set it hash
        identically regardless of internal default plumbing.

        Args:
            request: The ProviderRequest to key.

        Returns:
            A 64-character hex digest, stable for identical prompts.
        """
        payload: dict[str, Any] = {
            "model": request.model,
            "messages": [
                {"role": m.role, "content": m.content} for m in request.messages
            ],
        }
        if request.parameters and "temperature" in request.parameters:
            payload["temperature"] = request.parameters["temperature"]
        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    def content_key(self, request: ProviderRequest) -> str:
        """Provider-agnostic cassette key — hashes the prompt but NOT the model.

        Same canonical payload as :meth:`key` minus ``model``: the messages
        (and temperature, when set). One recording therefore serves EVERY
        provider that sends the same prompt.

        Policy (Mohamed, 2026-08-04): the suite exercises Claude only. Each
        prompt's answer is recorded ONCE against Claude and replayed for any
        model the prompt would otherwise route to — "copy Claude's answer as
        if it were the other LLM's." A non-Claude-routed test never needs a
        real non-Claude key or network call: its exact (model-specific) key
        misses and this content key serves the Claude recording (retiring the
        provider-key blocker on non-Claude cassettes).

        Args:
            request: The ProviderRequest to key.

        Returns:
            A 64-character hex digest, stable for identical prompts across
            every model.
        """
        payload: dict[str, Any] = {
            "messages": [
                {"role": m.role, "content": m.content} for m in request.messages
            ],
        }
        if request.parameters and "temperature" in request.parameters:
            payload["temperature"] = request.parameters["temperature"]
        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    def _path(self, key: str) -> Path:
        return self.directory / f"{key}.json"

    def load(self, key: str) -> ProviderResponse | None:
        """Load a cassette by key, reconstructing a real ProviderResponse.

        Args:
            key: The cassette key from :meth:`key`.

        Returns:
            The reconstructed ProviderResponse, or None if no cassette
            exists for this key (or the cassette file is unreadable/corrupt
            — treated as a miss rather than a hard failure, matching
            ``_capture_exchange``'s never-raise-into-the-caller contract).
        """
        path = self._path(key)
        if not path.exists():
            return None
        try:
            with open(path, encoding="utf-8") as fh:
                data = json.load(fh)
            usage_data = data.get("usage") or {}
            cost_data = data.get("cost") or {}
            return ProviderResponse(
                content=data.get("content", ""),
                model_used=data.get("model_used", ""),
                provider_used=data.get("provider_used", ""),
                usage=TokenUsage(
                    input_tokens=usage_data.get("input_tokens", 0),
                    output_tokens=usage_data.get("output_tokens", 0),
                    total_tokens=usage_data.get("total_tokens", 0),
                ),
                cost=CostRecord(
                    input_cost=cost_data.get("input_cost", 0.0),
                    output_cost=cost_data.get("output_cost", 0.0),
                    total_cost=cost_data.get("total_cost", 0.0),
                    currency=cost_data.get("currency", "USD"),
                ),
                latency_ms=data.get("latency_ms", 0.0),
                metadata=data.get("metadata") or {},
                tool_calls=data.get("tool_calls") or [],
                reasoning_content=data.get("reasoning_content"),
            )
        except Exception:
            _logger.warning("cassette %s is unreadable/corrupt — treating as miss", key)
            return None

    def save(self, key: str, response: ProviderResponse) -> None:
        """Persist a real ProviderResponse as a cassette.

        Pretty JSON with stable (declaration-order) key order. Never raises
        into the caller: a failure to write the cassette must not break the
        real LLM call that produced ``response``.

        Args:
            key: The cassette key from :meth:`key`.
            response: The real ProviderResponse to persist.
        """
        try:
            self.directory.mkdir(parents=True, exist_ok=True)
            data = asdict(response)
            with open(self._path(key), "w", encoding="utf-8") as fh:
                json.dump(data, fh, indent=2, sort_keys=False)
                fh.write("\n")
        except Exception:
            _logger.warning("failed to save cassette %s", key)


# ---------------------------------------------------------------------------
# Provider exchanges that are not chat completions.
#
# CassetteStore above is typed to ProviderRequest/ProviderResponse, which is
# the right shape for a chat completion and the wrong one for an embedding
# batch or a rerank. Those paths do not go through llm/client.py at all — they
# hold their own httpx client — so for as long as the cassette lived only in
# the client, every embedding the suite computed was a real billed request to
# OpenRouter. That is what it was: the suite went red the day the key hit its
# credit cap, on tests whose stated subject was the MISSING-key path.
#
# Same directory, same env switch, same four modes, so there is one cassette
# concept to reason about and not two.
# ---------------------------------------------------------------------------


class PayloadCassetteStore:
    """Directory-backed record/replay of arbitrary JSON provider exchanges.

    Keyed by a sha256 over canonical JSON of ``{kind, **payload}``, so an
    embedding request and a rerank request that happened to carry identical
    fields still land in different cassettes.
    """

    def __init__(self, directory: str | Path, *, kind: str) -> None:
        self.directory = Path(directory)
        self.kind = kind

    def key(self, payload: Mapping[str, Any]) -> str:
        """Deterministic cassette key for *payload* under this store's kind."""
        canonical = json.dumps(
            {"kind": self.kind, **dict(payload)},
            sort_keys=True,
            separators=(",", ":"),
        )
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    def _path(self, key: str) -> Path:
        return self.directory / f"{self.kind}-{key}.json"

    def load(self, key: str) -> Any | None:
        """The recorded value, or None on a miss/unreadable cassette."""
        path = self._path(key)
        if not path.exists():
            return None
        try:
            with open(path, encoding="utf-8") as fh:
                return json.load(fh)["value"]
        except Exception:
            _logger.warning("cassette %s is unreadable/corrupt — treating as miss", key)
            return None

    def save(self, key: str, value: Any) -> None:
        """Persist *value*. Never raises into the caller — a cassette that
        cannot be written must not break the real call that produced it."""
        try:
            self.directory.mkdir(parents=True, exist_ok=True)
            with open(self._path(key), "w", encoding="utf-8") as fh:
                json.dump({"kind": self.kind, "value": value}, fh, indent=2)
                fh.write("\n")
        except Exception:
            _logger.warning("failed to save cassette %s", key)


def synthetic_embedding(text: str, dimension: int) -> list[float]:
    """A deterministic unit vector standing in for a real embedding.

    Derived from the sha256 of *text*, so it is stable across runs and across
    processes, and two different texts get two different vectors — enough for
    a similarity assertion to mean something without a provider. Unit-normed
    because every consumer here compares by cosine.
    """
    raw = bytearray()
    counter = 0
    while len(raw) < dimension * 4:
        raw += hashlib.sha256(f"{counter}:{text}".encode()).digest()
        counter += 1
    values = [
        struct.unpack_from("<i", bytes(raw), offset=i * 4)[0] / 2_147_483_648.0
        for i in range(dimension)
    ]
    norm = sum(value * value for value in values) ** 0.5
    if norm == 0.0:  # every-byte-zero is unreachable from sha256, but cheap to hold
        return [0.0] * dimension
    return [value / norm for value in values]


class LLMCassetteMiss(RuntimeError):
    """Replay-only mode was asked for an exchange nothing has recorded."""


async def cassette_exchange(
    *,
    kind: str,
    payload: Mapping[str, Any],
    synth: Callable[[], Any],
    live: Callable[[], Awaitable[Any]],
    env: Mapping[str, str] | None = None,
) -> Any:
    """Replay, synthesize, record, or refuse — for a non-chat exchange.

    The same four modes ``llm/client.py::_execute_with_retry`` applies to a
    chat completion, in the same order and off the same environment keys:

    1. ``LLM_CASSETTE_DIR`` unset — *live* runs. This is a real script talking
       to a real provider; the suite never leaves it unset for these paths.
    2. A cassette exists — replayed byte-for-byte, no network.
    3. ``LLM_CASSETTE_SYNTH=1`` — *synth* answers, no network. What the suite
       uses, so a test that merely needs A vector never spends a credit.
    4. ``LLM_CASSETTE_RECORD=1`` — *live* runs once and is recorded.

    Otherwise it raises, because a missing cassette in replay-only mode is a
    gap to fix and not a reason to quietly place a billed call.
    """
    effective_env = env if env is not None else os.environ
    # Own key pair, falling back to the chat one. The suite cannot set
    # LLM_CASSETTE_DIR globally — that boundary raises "no cassette" in front
    # of 14 offline _execute_with_retry unit tests that drive the retry logic
    # with an injected provider (see tests/conftest.py). No such test drives
    # THIS function, so EMBEDDING_CASSETTE_DIR can be a suite-wide default and
    # make "no real provider call in the suite" a property rather than a habit.
    cassette_dir = effective_env.get("EMBEDDING_CASSETTE_DIR") or effective_env.get(
        "LLM_CASSETTE_DIR", ""
    )
    if not cassette_dir:
        return await live()

    store = PayloadCassetteStore(cassette_dir, kind=kind)
    key = store.key(payload)
    hit = store.load(key)
    if hit is not None:
        return hit
    synth_on = (
        effective_env.get("EMBEDDING_CASSETTE_SYNTH")
        or effective_env.get("LLM_CASSETTE_SYNTH", "")
    ) == "1"
    if synth_on:
        return synth()
    if effective_env.get("LLM_CASSETTE_RECORD") == "1":
        result = await live()
        store.save(key, result)
        return result
    raise LLMCassetteMiss(
        f"no {kind} cassette for key {key}; re-record locally with "
        "LLM_CASSETTE_RECORD=1"
    )


def synthetic_embeddings(texts: Sequence[str], dimension: int) -> list[list[float]]:
    """:func:`synthetic_embedding` over a batch, preserving input order."""
    return [synthetic_embedding(text, dimension) for text in texts]