Replay-Based Regression: Deterministic Re-Runs of Agent Workflows

LLMs are non-deterministic. Tools return different data. Networks fail. Replay-based regression testing solves this by recording an agent’s execution—every LLM call, tool response, and state change—then replaying it deterministically to catch regressions.

The Problem with Live Agent Testing

Running your agent against live APIs in CI is:

• Flaky: Network failures cause false negatives
• Slow: Waiting for LLM responses takes minutes per test
• Expensive: Every CI run burns API credits
• Non-reproducible: Same test gives different results each time

Replay testing eliminates all of these by recording once and replaying forever.

Architecture

┌──────────────┐     ┌──────────────────┐     ┌──────────────────┐
│  Record Mode │     │  Replay Mode     │     │  Assertion Layer │
│              │     │                  │     │                  │
│  - LLM calls │     │  - Mock LLM      │     │  - State checks  │
│  - Tool resp  │────▶│  - Mock tools    │────▶│  - Path checks   │
│  - State     │     │  - Deterministic │     │  - Output checks │
└──────────────┘     └──────────────────┘     └──────────────────┘

Implementation with Microsoft Agent Framework

import json
import hashlib
from pathlib import Path
from typing import Any
from dataclasses import dataclass, field

@dataclass
class RecordedEvent:
    event_type: str  # "llm_call", "tool_call", "tool_response", "state_change"
    timestamp: float
    input_hash: str
    output: Any
    metadata: dict = field(default_factory=dict)

class ReplayRecorder:
    def __init__(self, session_dir: str):
        self.session_dir = Path(session_dir)
        self.session_dir.mkdir(parents=True, exist_ok=True)
        self.events: list[RecordedEvent] = []
        self.session_id = hashlib.md5(str(time.time()).encode()).hexdigest()[:8]
    
    def record_llm_call(self, messages: list, model: str, response: str, usage: dict):
        input_hash = hashlib.sha256(json.dumps(messages, sort_keys=True).encode()).hexdigest()
        self.events.append(RecordedEvent(
            event_type="llm_call",
            timestamp=time.time(),
            input_hash=input_hash,
            output=response,
            metadata={"model": model, "usage": usage}
        ))
    
    def record_tool_call(self, tool_name: str, args: dict, response: Any):
        input_hash = hashlib.sha256(json.dumps({"tool": tool_name, "args": args}, sort_keys=True).encode()).hexdigest()
        self.events.append(RecordedEvent(
            event_type="tool_response",
            timestamp=time.time(),
            input_hash=input_hash,
            output=response,
            metadata={"tool_name": tool_name}
        ))
    
    def save(self):
        session_file = self.session_dir / f"session_{self.session_id}.json"
        with open(session_file, "w") as f:
            json.dump([vars(e) for e in self.events], f, indent=2, default=str)

class ReplayPlayer:
    def __init__(self, session_file: str):
        with open(session_file) as f:
            raw_events = json.load(f)
        self.events = {e["input_hash"]: e for e in raw_events}
    
    def replay_llm_call(self, messages: list, model: str) -> str:
        input_hash = hashlib.sha256(json.dumps(messages, sort_keys=True).encode()).hexdigest()
        if input_hash in self.events:
            return self.events[input_hash]["output"]
        raise ValueError(f"No recorded response for hash {input_hash}")
    
    def replay_tool_call(self, tool_name: str, args: dict) -> Any:
        input_hash = hashlib.sha256(json.dumps({"tool": tool_name, "args": args}, sort_keys=True).encode()).hexdigest()
        if input_hash in self.events:
            return self.events[input_hash]["output"]
        raise ValueError(f"No recorded tool response for {tool_name} with hash {input_hash}")

Decorator-Based Recording

import functools
import json
import hashlib

def recordable(recorder: ReplayRecorder):
    """Decorator that records function calls for replay."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Create deterministic input hash
            input_data = {"func": func.__name__, "args": args, "kwargs": kwargs}
            input_hash = hashlib.sha256(
                json.dumps(input_data, sort_keys=True, default=str).encode()
            ).hexdigest()
            
            # Check if we have a recorded response
            if hasattr(recorder, 'has_event') and recorder.has_event(input_hash):
                return recorder.get_event(input_hash)["output"]
            
            # Execute and record
            result = func(*args, **kwargs)
            recorder.record_llm_call(
                messages=args[0] if args else kwargs.get("messages", []),
                model=kwargs.get("model", "unknown"),
                response=result,
                usage={}
            )
            return result
        return wrapper
    return decorator

# Usage
@recordable(my_recorder)
def call_llm(messages, model="gpt-4o"):
    # Your actual LLM call
    return openai.chat.completions.create(model=model, messages=messages)

Assertion-Based Regression

@dataclass
class AgentAssertion:
    name: str
    check_fn: callable
    description: str

class RegressionSuite:
    def __init__(self):
        self.assertions: list[AgentAssertion] = []
    
    def add_assertion(self, name: str, check_fn, description: str):
        self.assertions.append(AgentAssertion(name, check_fn, description))
    
    def run(self, replay_session: str) -> dict:
        player = ReplayPlayer(replay_session)
        results = []
        
        for assertion in self.assertions:
            try:
                passed = assertion.check_fn(player)
                results.append({
                    "name": assertion.name,
                    "passed": passed,
                    "description": assertion.description
                })
            except Exception as e:
                results.append({
                    "name": assertion.name,
                    "passed": False,
                    "error": str(e),
                    "description": assertion.description
                })
        
        return {
            "total": len(results),
            "passed": sum(1 for r in results if r["passed"]),
            "failed": sum(1 for r in results if not r["passed"]),
            "results": results
        }

# Define regression assertions
suite = RegressionSuite()

suite.add_assertion(
    "tool_selection",
    lambda player: "get_stock_price" in str(player.events),
    "Agent should use get_stock_price for price queries"
)

suite.add_assertion(
    "no_hallucination",
    lambda player: all(
        "I don't have access" not in e["output"]
        for e in player.events.values()
        if e["event_type"] == "llm_call"
    ),
    "Agent should not claim lack of access when tools are available"
)

suite.add_assertion(
    "max_turns",
    lambda player: len(player.events) <= 15,
    "Agent should complete task in under 15 steps"
)

CI Integration

# .github/workflows/agent-regression.yml
name: Agent Regression Tests
on:
  pull_request:
    paths:
      - 'src/agent/**'
      - 'src/tools/**'

jobs:
  replay-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Run replay tests
        run: |
          python -m pytest tests/replay/ \
            --replay-dir=tests/fixtures/sessions/ \
            --junitxml=results.xml
      
      - name: Upload results
        uses: actions/upload-artifact@v4
        with:
          name: regression-results
          path: results.xml

Recording Production Sessions

import logging
from contextlib import contextmanager

@contextmanager
def record_session(agent, session_dir: str):
    """Context manager that records agent sessions for replay."""
    recorder = ReplayRecorder(session_dir)
    
    original_llm = agent.llm
    original_tools = agent.tools
    
    # Wrap LLM with recorder
    agent.llm = RecordedLLM(original_llm, recorder)
    
    # Wrap tools with recorder
    agent.tools = {name: RecordedTool(tool, recorder) for name, tool in original_tools.items()}
    
    try:
        yield agent
    finally:
        recorder.save()
        agent.llm = original_llm
        agent.tools = original_tools

# Record a production session for later replay
with record_session(agent, "tests/fixtures/sessions/") as recorded_agent:
    result = recorded_agent.run("What's the weather in Tokyo?")

Managing Session Fixes

class SessionFixtures:
    def __init__(self, fixtures_dir: str):
        self.fixtures_dir = Path(fixtures_dir)
    
    def add_fixtures_for_task(self, task: str, fixtures: dict):
        """Add recorded fixtures for a specific task."""
        task_hash = hashlib.md5(task.encode()).hexdigest()[:8]
        fixture_file = self.fixtures_dir / f"task_{task_hash}.json"
        
        with open(fixture_file, "w") as f:
            json.dump({"task": task, "fixtures": fixtures}, f, indent=2)
    
    def get_fixtures(self, task: str) -> dict:
        task_hash = hashlib.md5(task.encode()).hexdigest()[:8]
        fixture_file = self.fixtures_dir / f"task_{task_hash}.json"
        
        if fixture_file.exists():
            with open(fixture_file) as f:
                return json.load(f)
        return {}

Replay-based regression testing gives you the speed and determinism of unit tests with the coverage of integration tests. Record sessions from production, replay them in CI, and sleep soundly knowing your agent hasn’t regressed.