Record Real Sessions.
Replay Them Deterministically.

Start with typed, validated schemas. This keeps fixtures forward-compatible and makes replay bugs obvious.

from __future__ import annotations
from dataclasses import dataclass, field, asdict
from enum import Enum
import json, time


class StepType(str, Enum):
    """Discriminates the kind of recorded interaction."""
    TOOL_CALL = "tool_call"
    LLM = "llm"
    RETRIEVAL = "retrieval"


@dataclass(frozen=True)
class Step:
    """A single, replayable interaction within a session.

    Args:
        type: The category of the step.
        name: Logical name (e.g. the tool or model called).
        args: Inputs used to produce ``output``.
        output: The recorded result returned to the agent.
        latency_ms: Observed latency, retained for offline analysis.
    """
    type: StepType
    name: str
    args: dict
    output: object
    latency_ms: int = 0


@dataclass
class Fixture:
    """A complete, serializable record of one agent session."""
    session_id: str
    query: str
    steps: list[Step] = field(default_factory=list)
    final_output: str | None = None
    expected: dict = field(default_factory=dict)
    schema_version: str = "1.2"

    def to_json(self) -> str:
        return json.dumps(asdict(self), indent=2, default=str)

    @classmethod
    def from_json(cls, raw: str) -> "Fixture":
        data = json.loads(raw)
        data["steps"] = [Step(**s) for s in data.get("steps", [])]
        return cls(**data)

A recorder wraps your real tool/LLM clients, forwarding calls while logging every request and response.

import time
from .models import Fixture, Step, StepType


class Recorder:
    """Transparent proxy that logs tool and LLM calls into a Fixture.

    The recorder forwards each call to the real implementation and stores
    the result, so a live run produces a replayable snapshot as a side effect.
    """

    def __init__(self, session_id: str, query: str):
        self.fixture = Fixture(session_id=session_id, query=query)

    def tool(self, name: str, fn):
        """Wrap a callable so its inputs and outputs are recorded.

        Args:
            name: Logical tool name stored in the fixture.
            fn: The real tool implementation to invoke.

        Returns:
            A wrapped callable with identical behavior.
        """
        def wrapped(**kwargs):
            start = time.perf_counter()
            output = fn(**kwargs)
            elapsed = int((time.perf_counter() - start) * 1000)
            self.fixture.steps.append(
                Step(StepType.TOOL_CALL, name, kwargs, output, elapsed)
            )
            return output
        return wrapped

    def save(self, path: str) -> None:
        """Persist the recorded session to disk as JSON."""
        with open(path, "w", encoding="utf-8") as fh:
            fh.write(self.fixture.to_json())

On replay, the mock layer returns recorded outputs in order — matched by tool name and arguments — so the agent runs its real logic with zero network calls.

from typing import Callable
from .models import Fixture, StepType


class ReplayError(RuntimeError):
    """Raised when the agent requests a call not present in the fixture."""


class MockLayer:
    """Serves recorded outputs deterministically during replay."""

    def __init__(self, fixture: Fixture):
        self._steps = list(fixture.steps)
        self._cursor = 0

    def tool(self, name: str):
        """Return a mock callable that yields the next recorded output.

        Raises:
            ReplayError: If the requested tool does not match the recording.
        """
        def wrapped(**kwargs):
            if self._cursor >= len(self._steps):
                raise ReplayError(f"No recorded step left for tool '{name}'")
            step = self._steps[self._cursor]
            if step.name != name:
                raise ReplayError(
                    f"Expected '{step.name}' but agent called '{name}'"
                )
            self._cursor += 1
            return step.output
        return wrapped


class ReplayHarness:
    """Loads fixtures, replays them against agent logic, and scores output."""

    def __init__(self, fixtures_dir: str = "fixtures"):
        self.fixtures_dir = fixtures_dir

    def load(self, session_id: str) -> Fixture:
        with open(f"{self.fixtures_dir}/{session_id}.json") as fh:
            return Fixture.from_json(fh.read())

    def replay(
        self,
        session_id: str,
        agent_factory: Callable[[MockLayer], Callable[[str], str]],
        evaluators: dict[str, Callable] | None = None,
    ) -> dict:
        """Replay a fixture and evaluate the deterministic output.

        Args:
            session_id: Fixture identifier to load.
            agent_factory: Builds an agent bound to the mock layer.
            evaluators: Named scoring functions ``fn(output, fixture) -> value``.

        Returns:
            A mapping of evaluator name to score, plus the raw output.
        """
        fixture = self.load(session_id)
        agent = agent_factory(MockLayer(fixture))
        output = agent(fixture.query)
        scores = {
            name: fn(output, fixture)
            for name, fn in (evaluators or {}).items()
        }
        return {"output": output, "scores": scores}

Evaluators turn output into numbers. Wire them into pytest so every commit re-runs your fixtures offline.

import pytest
from harness.replay import ReplayHarness


def exact_match(output: str, fixture) -> float:
    """Return 1.0 when the output matches the recorded final output."""
    return float(output.strip() == (fixture.final_output or "").strip())


def mentions_refund(output: str, fixture) -> float:
    """Heuristic: reward answers that address the refund intent."""
    return float("refund" in output.lower())


@pytest.mark.parametrize("session_id", ["support_12345", "support_67890"])
def test_support_agent_regression(session_id):
    harness = ReplayHarness()
    result = harness.replay(
        session_id,
        agent_factory=build_support_agent,   # binds tools to the mock layer
        evaluators={
            "exact_match": exact_match,
            "mentions_refund": mentions_refund,
        },
    )
    assert result["scores"]["mentions_refund"] == 1.0
    assert result["scores"]["exact_match"] >= 0.0