---
title: Record-Replay Testing System
description: Understanding how Llama Stack captures and replays API interactions for testing
sidebar_label: Testing Record-Replay
sidebar_position: 4
---

import Tabs from '@theme/Tabs';
import TabItem from '@theme/TabItem';

Understanding how Llama Stack captures and replays API interactions for reliable, cost-effective testing.

## Overview

The record-replay system solves a fundamental challenge in AI testing: **How do you test against expensive, non-deterministic APIs without breaking the bank or dealing with flaky tests?**

**The solution:** Intercept API calls, store real responses, and replay them later. This gives you real API behavior without the cost or variability.

## System Architecture

### Request Hashing

Every API request gets converted to a deterministic hash for lookup:

```python
def normalize_request(method: str, url: str, headers: dict, body: dict) -> str:
    normalized = {
        "method": method.upper(),
        "endpoint": urlparse(url).path,  # Just the path, not full URL
        "body": body,  # Request parameters
    }
    return hashlib.sha256(json.dumps(normalized, sort_keys=True).encode()).hexdigest()
```

:::warning[Precise Hashing]
The hashing is intentionally precise. Different whitespace, float precision, or parameter order produces different hashes. This prevents subtle bugs from false cache hits.

```python
# These produce DIFFERENT hashes:
{"content": "Hello world"}
{"content": "Hello   world\n"}
{"temperature": 0.7}
{"temperature": 0.7000001}
```
:::

### Client Interception

The system patches OpenAI and Ollama client methods to intercept calls before they leave your application. This happens transparently - your test code doesn't change.

### Storage Architecture

Recordings are stored as JSON files in the recording directory, looked up by their request hash:

```
recordings/
└── responses/
    ├── abc123def456.json  # Individual response files
    └── def789ghi012.json
```

**JSON files** store complete request/response pairs in human-readable format for debugging.

## Recording Modes

The system supports three distinct modes for different testing scenarios:

<Tabs>
<TabItem value="live" label="LIVE Mode">

**Direct API calls** with no recording or replay:

```python
with inference_recording(mode=InferenceMode.LIVE):
    response = await client.chat.completions.create(...)
```

**Use for:**
- Initial development and debugging
- Testing against real APIs
- Validating new functionality

</TabItem>
<TabItem value="record" label="RECORD Mode">

**Captures API interactions** while passing through real responses:

```python
with inference_recording(mode=InferenceMode.RECORD, storage_dir="./recordings"):
    response = await client.chat.completions.create(...)
    # Real API call made, response captured AND returned
```

**The recording process:**
1. Request intercepted and hashed
2. Real API call executed
3. Response captured and serialized
4. Recording stored to disk
5. Original response returned to caller

</TabItem>
<TabItem value="replay" label="REPLAY Mode">

**Returns stored responses** instead of making API calls:

```python
with inference_recording(mode=InferenceMode.REPLAY, storage_dir="./recordings"):
    response = await client.chat.completions.create(...)
    # No API call made, cached response returned instantly
```

**The replay process:**
1. Request intercepted and hashed
2. Hash looked up in SQLite index
3. Response loaded from JSON file
4. Response deserialized and returned
5. Error if no recording found

</TabItem>
</Tabs>

## Streaming Support

Streaming APIs present a unique challenge: how do you capture an async generator?

### The Challenge

```python
# How do you record this?
async for chunk in client.chat.completions.create(stream=True):
    process(chunk)
```

### The Solution

The system captures all chunks immediately before yielding any:

<Tabs>
<TabItem value="capture" label="Stream Capture">

```python
async def handle_streaming_record(response):
    # Capture complete stream first
    chunks = []
    async for chunk in response:
        chunks.append(chunk)

    # Store complete recording
    storage.store_recording(
        request_hash,
        request_data,
        {"body": chunks, "is_streaming": True}
    )

    # Return generator that replays captured chunks
    async def replay_stream():
        for chunk in chunks:
            yield chunk

    return replay_stream()
```

</TabItem>
<TabItem value="benefits" label="Benefits">

This approach ensures:

- **Complete capture** - The entire stream is saved atomically
- **Interface preservation** - The returned object behaves like the original API
- **Deterministic replay** - Same chunks in the same order every time
- **No API changes** - Your streaming code works unchanged

</TabItem>
</Tabs>

## Serialization

API responses contain complex Pydantic objects that need careful serialization:

```python
def _serialize_response(response):
    if hasattr(response, "model_dump"):
        # Preserve type information for proper deserialization
        return {
            "__type__": f"{response.__class__.__module__}.{response.__class__.__qualname__}",
            "__data__": response.model_dump(mode="json"),
        }
    return response
```

This preserves **type safety** - when replayed, you get the same Pydantic objects with all their validation and methods.

## Usage in Testing

### Environment Variables

Control recording behavior globally:

<Tabs>
<TabItem value="env-vars" label="Environment Variables">

```bash
# Set recording mode (default: replay)
export LLAMA_STACK_TEST_INFERENCE_MODE=replay

# Set recording directory (default: tests/integration/recordings)
export LLAMA_STACK_TEST_RECORDING_DIR=/path/to/recordings

# Run tests
pytest tests/integration/
```

</TabItem>
<TabItem value="pytest" label="Pytest Integration">

The system integrates automatically based on environment variables, requiring **no changes** to test code.

```python
# Your test code remains unchanged
async def test_chat_completion():
    response = await client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": "Hello"}]
    )
    assert response.choices[0].message.content
```

</TabItem>
</Tabs>

### Recording New Tests

<Tabs>
<TabItem value="local-record" label="Local Recording">

```bash
# Record new interactions locally
LLAMA_STACK_TEST_INFERENCE_MODE=record pytest test_new_feature.py

# Record specific test
LLAMA_STACK_TEST_INFERENCE_MODE=record pytest test_file.py::test_function
```

</TabItem>
<TabItem value="remote-record" label="Remote Recording">

Use the automated GitHub workflow for easier recording:

```bash
# Record tests for specific subdirectories
./scripts/github/schedule-record-workflow.sh --test-subdirs "agents,inference"

# Record with specific provider
./scripts/github/schedule-record-workflow.sh --test-subdirs "agents" --test-provider vllm
```

</TabItem>
</Tabs>

## Debugging Recordings

### Inspecting Storage

<Tabs>
<TabItem value="sqlite" label="SQLite Queries">

```bash
# See what's recorded
sqlite3 recordings/index.sqlite "SELECT endpoint, model, timestamp FROM recordings LIMIT 10;"

# Find recordings by endpoint
sqlite3 recordings/index.sqlite "SELECT * FROM recordings WHERE endpoint='/v1/chat/completions';"

# Check for specific model
sqlite3 recordings/index.sqlite "SELECT * FROM recordings WHERE model='gpt-3.5-turbo';"
```

</TabItem>
<TabItem value="json" label="JSON Inspection">

```bash
# View specific response
cat recordings/responses/abc123def456.json | jq '.response.body'

# Compare request details
cat recordings/responses/abc123.json | jq '.request'

# Pretty print entire recording
cat recordings/responses/abc123.json | jq '.'
```

</TabItem>
</Tabs>

### Common Issues

<Tabs>
<TabItem value="hash-mismatch" label="Hash Mismatches">

**Problem:** Request parameters changed slightly between record and replay

**Solution:**
```bash
# Compare request details
cat recordings/responses/abc123.json | jq '.request'

# Re-record with updated parameters
rm recordings/responses/failing_hash.json
LLAMA_STACK_TEST_INFERENCE_MODE=record pytest test_failing.py
```

</TabItem>
<TabItem value="serialization" label="Serialization Errors">

**Problem:** Response types changed between versions

**Solution:**
```bash
# Re-record with updated types
rm recordings/responses/failing_hash.json
LLAMA_STACK_TEST_INFERENCE_MODE=record pytest test_failing.py
```

</TabItem>
<TabItem value="missing" label="Missing Recordings">

**Problem:** New test or changed parameters

**Solution:**
```bash
# Record the missing interaction
LLAMA_STACK_TEST_INFERENCE_MODE=record pytest test_new.py
```

</TabItem>
</Tabs>

## Design Decisions

### Why Not Mocks?

Traditional mocking breaks down with AI APIs because:

- **Complex structures** - Response structures are complex and evolve frequently
- **Streaming behavior** - Hard to mock streaming responses correctly
- **Edge cases** - Real API edge cases get missed in mocks
- **Maintenance burden** - Mocks become brittle and hard to maintain

### Why Precise Hashing?

Loose hashing (normalizing whitespace, rounding floats) seems convenient but **hides bugs**. If a test changes slightly, you want to know about it rather than accidentally getting the wrong cached response.

### Why JSON + SQLite?

The hybrid storage approach provides the best of both worlds:

- **JSON** - Human readable, diff-friendly, easy to inspect and modify
- **SQLite** - Fast indexed lookups without loading response bodies
- **Combined** - Optimal for both performance and debugging

## Advanced Usage

### Custom Recording Contexts

```python
# Record specific API calls only
with inference_recording(
    mode=InferenceMode.RECORD,
    storage_dir="./custom_recordings",
    filter_endpoints=["/v1/chat/completions"]
):
    # Only chat completions will be recorded
    response = await client.chat.completions.create(...)
```

### Conditional Recording

```python
# Record only if not exists
mode = InferenceMode.REPLAY
if not recording_exists(request_hash):
    mode = InferenceMode.RECORD

with inference_recording(mode=mode):
    response = await client.chat.completions.create(...)
```

### Recording Validation

```python
# Validate recordings during CI
def validate_recordings():
    for recording_file in glob("recordings/responses/*.json"):
        with open(recording_file) as f:
            data = json.load(f)
            assert "request" in data
            assert "response" in data
            # Additional validation...
```

## Best Practices

### 🎯 **Recording Strategy**
- Record comprehensive test scenarios once
- Use replay mode for regular development
- Re-record when API contracts change
- Keep recordings in version control

### 🔧 **Development Workflow**
- Start with LIVE mode for new features
- Switch to RECORD when tests are stable
- Use REPLAY for fast iteration
- Re-record when responses change

### 🚨 **Debugging Tips**
- Inspect JSON files for response details
- Use SQLite queries to find specific recordings
- Compare request hashes when tests fail
- Clear recordings to force re-recording

### 📊 **Maintenance**
- Regularly review and clean old recordings
- Update recordings when API versions change
- Document which recordings are critical
- Monitor recording file sizes

## Related Resources

- **[Contributing Overview](./index)** - General contribution guidelines
- **[Testing Guide](/docs/testing)** - Complete testing documentation
- **[Integration Tests](https://github.com/meta-llama/llama-stack/tree/main/tests/integration)** - Example test implementations
- **[GitHub Workflows](https://github.com/meta-llama/llama-stack/tree/main/.github/workflows)** - Automated recording workflows
