Trace Store Module

April 3, 2026 · View on GitHub

Import: from selectools.trace import trace_to_html Stability: stable

from selectools import Agent, tool
from selectools.providers.stubs import LocalProvider
from selectools.trace import trace_to_html

@tool(description="Add two numbers")
def add(a: int, b: int) -> str:
    return str(a + b)

agent = Agent(tools=[add], provider=LocalProvider())
result = agent.run("What is 2 + 3?")

# Render the execution trace as an interactive HTML waterfall
html = trace_to_html(result.trace)
with open("trace.html", "w") as f:
    f.write(html)
print(f"Trace saved with {len(result.trace.steps)} steps")

!!! tip "See Also" - Agent Module -- the Agent class that produces traces, and observer events - Serve Module -- deploy agents with automatic trace persistence

Added in: v0.19.0 Package: src/selectools/observe/ Protocol: TraceStore Classes: InMemoryTraceStore, SQLiteTraceStore, JSONLTraceStore, TraceSummary, TraceFilter

Overview
Quick Start
TraceStore Protocol
Backends
TraceFilter
TraceSummary
Integration with Serve
Patterns
API Reference
Examples

Overview

The trace store module persists and queries AgentTrace objects -- the detailed execution logs produced by every agent run. Instead of traces disappearing when the process exits, you can save them to SQLite, append them to a JSONL file, or hold them in memory for the duration of a session.

Why Trace Store?

Without Trace Store	With Trace Store
Traces lost on process exit	Persisted to disk or database
No way to compare runs	Query by date, step count, metadata
Manual logging of costs and durations	Automatic structured storage
No audit trail	Full execution history for compliance

Design Philosophy

Protocol-based. TraceStore is a Protocol class. Implement 5 methods and any backend works.
Three built-in backends. InMemory for dev, SQLite for production, JSONL for export/archive.
Thread-safe. All backends use locks or WAL mode for concurrent access.
Zero required dependencies. All three backends use Python stdlib only.

Quick Start

from selectools import Agent, AgentConfig
from selectools.observe import SQLiteTraceStore

# Create a trace store
store = SQLiteTraceStore("traces.db")

# Run an agent
agent = Agent(provider=provider, config=AgentConfig(model="gpt-4o"))
result = agent.run("What is the capital of France?")

# Save the trace
run_id = store.save(result.trace)
print(f"Saved trace: {run_id}")

# Load it back
trace = store.load(run_id)
print(f"Steps: {len(trace.steps)}, Duration: {trace.total_duration_ms:.0f}ms")

# List recent traces
for summary in store.list(limit=10):
    print(f"  {summary.run_id}: {summary.steps} steps, {summary.total_ms:.0f}ms")

TraceStore Protocol

File: src/selectools/observe/trace_store.py

Any class that implements these 5 methods satisfies the TraceStore protocol:

from typing import Protocol, runtime_checkable

@runtime_checkable
class TraceStore(Protocol):
    def save(self, trace: AgentTrace) -> str:
        """Persist a trace. Returns the run_id."""
        ...

    def load(self, run_id: str) -> AgentTrace:
        """Load a trace by run_id. Raises ValueError if not found."""
        ...

    def list(self, limit: int = 50, offset: int = 0) -> List[TraceSummary]:
        """List trace summaries, newest first."""
        ...

    def query(self, filters: TraceFilter) -> List[TraceSummary]:
        """Query traces matching filter criteria."""
        ...

    def delete(self, run_id: str) -> bool:
        """Delete a trace. Returns True if deleted."""
        ...

The protocol is @runtime_checkable, so you can use isinstance(obj, TraceStore) to verify compliance.

Backends

InMemoryTraceStore

In-memory storage for development and testing. Traces are lost when the process exits.

from selectools.observe import InMemoryTraceStore

store = InMemoryTraceStore(max_size=1000)

Parameter	Type	Default	Description
`max_size`	`int`	`1000`	Maximum traces to keep. Oldest evicted when full.

Characteristics:

Thread-safe (uses threading.Lock).
LRU eviction when max_size is exceeded.
Fastest backend -- no I/O overhead.

SQLiteTraceStore

SQLite-backed storage for production use. Uses WAL mode for concurrent read/write access.

from selectools.observe import SQLiteTraceStore

store = SQLiteTraceStore("traces.db")

Parameter	Type	Default	Description
`db_path`	`str`	(required)	Path to the SQLite database file. Created if it does not exist.

Characteristics:

Thread-safe (per-thread connections via threading.local).
WAL journal mode for concurrent readers.
Indexed on created_at DESC for fast listing.
Traces serialized as JSON in the trace_json column.
Supports SQL-level filtering for min_steps, max_steps, since, until.
Metadata filtering done in Python (not SQL) for flexibility.

Schema:

CREATE TABLE IF NOT EXISTS traces (
    run_id TEXT PRIMARY KEY,
    steps INTEGER NOT NULL,
    total_ms REAL NOT NULL,
    created_at TEXT NOT NULL,
    metadata TEXT NOT NULL DEFAULT '{}',
    trace_json TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_traces_created ON traces(created_at DESC);

JSONLTraceStore

Append-only JSONL (JSON Lines) storage for export and archival. Each trace is stored as a single JSON line.

from selectools.observe import JSONLTraceStore

store = JSONLTraceStore("traces.jsonl")

Parameter	Type	Default	Description
`path`	`str`	(required)	Path to the JSONL file. Parent directories created automatically.

Characteristics:

Thread-safe (uses threading.Lock).
Append-only writes -- fast saves with no seek overhead.
Delete rewrites the entire file (expensive, by design -- JSONL is for archival).
Human-readable -- each line is valid JSON.
Easy to process with standard tools (jq, grep, Python line iteration).

Line format:

{"run_id": "abc-123", "steps": 5, "total_ms": 1234.5, "created_at": "2026-03-27T10:00:00+00:00", "metadata": {"user_id": "alice"}, "trace": {...}}

Choosing a Backend

Backend	Best For	Persistence	Query Speed	Write Speed
`InMemoryTraceStore`	Dev, testing, short-lived processes	None	Fast	Fast
`SQLiteTraceStore`	Production, dashboards, analytics	Disk	Fast (indexed)	Fast (WAL)
`JSONLTraceStore`	Export, archival, log shipping	Disk	Slow (full scan)	Fast (append)

TraceFilter

Filter criteria for querying traces. All fields are optional -- unset fields are not applied.

from selectools.observe import TraceFilter
from datetime import datetime, timezone, timedelta

# Traces from the last 24 hours with at least 3 steps
filters = TraceFilter(
    since=datetime.now(timezone.utc) - timedelta(hours=24),
    min_steps=3,
)
results = store.query(filters)

# Traces for a specific user
filters = TraceFilter(
    metadata_match={"user_id": "alice"},
)
results = store.query(filters)

# Combine all criteria
filters = TraceFilter(
    metadata_match={"environment": "production"},
    min_steps=2,
    max_steps=20,
    since=datetime(2026, 3, 1, tzinfo=timezone.utc),
    until=datetime(2026, 3, 31, tzinfo=timezone.utc),
)
results = store.query(filters)

Fields

Field	Type	Default	Description
`metadata_match`	`Optional[Dict[str, Any]]`	`None`	All key-value pairs must match the trace's metadata.
`min_steps`	`Optional[int]`	`None`	Minimum number of trace steps (inclusive).
`max_steps`	`Optional[int]`	`None`	Maximum number of trace steps (inclusive).
`since`	`Optional[datetime]`	`None`	Only traces created at or after this time.
`until`	`Optional[datetime]`	`None`	Only traces created at or before this time.

TraceSummary

A lightweight summary of a stored trace, returned by list() and query(). Avoids loading the full trace JSON for listing operations.

for summary in store.list(limit=10):
    print(f"Run: {summary.run_id}")
    print(f"  Steps: {summary.steps}")
    print(f"  Duration: {summary.total_ms:.0f}ms")
    print(f"  Created: {summary.created_at}")
    print(f"  Metadata: {summary.metadata}")

Fields

Field	Type	Description
`run_id`	`str`	Unique run identifier.
`steps`	`int`	Number of trace steps.
`total_ms`	`float`	Total execution duration in milliseconds.
`created_at`	`datetime`	When the trace was saved (UTC).
`metadata`	`Dict[str, Any]`	Trace metadata (user_id, environment, etc.).

Integration with Serve

When using the Serve Module, traces can be persisted automatically and queried via HTTP. Wire a TraceStore into your served agent for full observability:

from selectools import Agent, AgentConfig
from selectools.observe import SQLiteTraceStore
from selectools.serve import AgentRouter, create_app

# Agent with trace store
store = SQLiteTraceStore("traces.db")
agent = Agent(provider=provider, config=AgentConfig(model="gpt-4o"))

# Custom endpoint that saves traces
router = AgentRouter(agent)

# After each invoke, save the trace
original_invoke = router.handle_invoke

def invoke_with_trace(body):
    result = original_invoke(body)
    # The agent's last trace is available after run()
    if hasattr(agent, '_last_trace') and agent._last_trace:
        store.save(agent._last_trace)
    return result

router.handle_invoke = invoke_with_trace

app = create_app(agent)
app.serve()

Querying Traces Programmatically

# List recent traces
traces = store.list(limit=20)
for t in traces:
    print(f"{t.run_id}: {t.steps} steps, {t.total_ms:.0f}ms")

# Find slow traces
slow = store.query(TraceFilter(min_steps=10))
for t in slow:
    trace = store.load(t.run_id)
    for step in trace.steps:
        if step.duration_ms and step.duration_ms > 5000:
            print(f"  Slow step: {step.type.value} at {step.node_name} ({step.duration_ms:.0f}ms)")

Patterns

Automatic Trace Saving with Observer

Use an AgentObserver to save traces automatically after every run:

from selectools import AgentObserver

class TraceSavingObserver(AgentObserver):
    def __init__(self, store: TraceStore):
        self.store = store

    def on_run_end(self, run_id, result, duration_ms):
        if result.trace:
            self.store.save(result.trace)

store = SQLiteTraceStore("traces.db")
agent = Agent(
    provider=provider,
    config=AgentConfig(model="gpt-4o"),
    observers=[TraceSavingObserver(store)],
)

Cost Analytics

Query traces to compute cost analytics:

from selectools.observe import TraceFilter, SQLiteTraceStore
from datetime import datetime, timezone, timedelta

store = SQLiteTraceStore("traces.db")

# Get all traces from the last 7 days
week_ago = datetime.now(timezone.utc) - timedelta(days=7)
traces = store.query(TraceFilter(since=week_ago))

total_ms = sum(t.total_ms for t in traces)
avg_steps = sum(t.steps for t in traces) / max(len(traces), 1)

print(f"Runs: {len(traces)}")
print(f"Total duration: {total_ms / 1000:.1f}s")
print(f"Avg steps per run: {avg_steps:.1f}")

Export to JSONL for Analysis

Use JSONLTraceStore to export traces for offline analysis:

from selectools.observe import SQLiteTraceStore, JSONLTraceStore

sqlite_store = SQLiteTraceStore("traces.db")
export_store = JSONLTraceStore("export/traces_2026_03.jsonl")

# Export all traces from March
from datetime import datetime, timezone
filters = TraceFilter(
    since=datetime(2026, 3, 1, tzinfo=timezone.utc),
    until=datetime(2026, 3, 31, tzinfo=timezone.utc),
)

for summary in sqlite_store.query(filters):
    trace = sqlite_store.load(summary.run_id)
    export_store.save(trace)

print(f"Exported {len(sqlite_store.query(filters))} traces")

Custom Backend

Implement the TraceStore protocol for any storage system:

class RedisTraceStore:
    """Redis-backed trace store (example)."""

    def __init__(self, redis_client, prefix="trace:"):
        self.redis = redis_client
        self.prefix = prefix

    def save(self, trace: AgentTrace) -> str:
        run_id = trace.run_id
        data = json.dumps(trace.to_dict(), default=str)
        self.redis.set(f"{self.prefix}{run_id}", data)
        self.redis.zadd(f"{self.prefix}index", {run_id: time.time()})
        return run_id

    def load(self, run_id: str) -> AgentTrace:
        data = self.redis.get(f"{self.prefix}{run_id}")
        if data is None:
            raise ValueError(f"Trace {run_id!r} not found")
        return AgentTrace.from_dict(json.loads(data))

    def list(self, limit=50, offset=0):
        ids = self.redis.zrevrange(f"{self.prefix}index", offset, offset + limit - 1)
        return [self._to_summary(rid.decode()) for rid in ids]

    def query(self, filters: TraceFilter):
        # Implement filtering logic
        ...

    def delete(self, run_id: str) -> bool:
        result = self.redis.delete(f"{self.prefix}{run_id}")
        self.redis.zrem(f"{self.prefix}index", run_id)
        return result > 0

API Reference

InMemoryTraceStore.init()

Parameter	Type	Default	Description
`max_size`	`int`	`1000`	Maximum traces to store before LRU eviction.

SQLiteTraceStore.init()

Parameter	Type	Default	Description
`db_path`	`str`	(required)	Path to SQLite database file.

JSONLTraceStore.init()

Parameter	Type	Default	Description
`path`	`str`	(required)	Path to JSONL file.

TraceStore Methods

Method	Description
`save(trace)`	Persist a trace. Returns `run_id` (str).
`load(run_id)`	Load full trace by run_id. Raises `ValueError` if not found.
`list(limit=50, offset=0)`	List `TraceSummary` objects, newest first.
`query(filters)`	Query traces matching a `TraceFilter`. Returns `List[TraceSummary]`.
`delete(run_id)`	Delete a trace. Returns `True` if deleted.

TraceFilter Fields

Field	Type	Default	Description
`metadata_match`	`Optional[Dict[str, Any]]`	`None`	Metadata key-value pairs to match.
`min_steps`	`Optional[int]`	`None`	Minimum step count.
`max_steps`	`Optional[int]`	`None`	Maximum step count.
`since`	`Optional[datetime]`	`None`	Created at or after.
`until`	`Optional[datetime]`	`None`	Created at or before.

TraceSummary Fields

Field	Type	Description
`run_id`	`str`	Unique run identifier.
`steps`	`int`	Number of trace steps.
`total_ms`	`float`	Total duration in milliseconds.
`created_at`	`datetime`	When the trace was saved.
`metadata`	`Dict[str, Any]`	Trace metadata.

Examples

Example	File	Description
69	`69_trace_store.py`	Save, query, and export traces with all 3 backends

#	File	Description
74	`74_trace_to_html.py`	Render traces as interactive HTML waterfalls
69	`69_trace_store.py`	Save, query, and export traces with all 3 backends
24	`24_traces_and_reasoning.py`	Inspect agent traces and reasoning

Trace Store Module

Table of Contents

Overview

Why Trace Store?

Design Philosophy

Quick Start

TraceStore Protocol

Backends

InMemoryTraceStore

SQLiteTraceStore

JSONLTraceStore

Choosing a Backend

TraceFilter

Fields

TraceSummary

Fields

Integration with Serve

Querying Traces Programmatically

Patterns

Automatic Trace Saving with Observer

Cost Analytics

Export to JSONL for Analysis

Custom Backend

API Reference

InMemoryTraceStore.init()

SQLiteTraceStore.init()

JSONLTraceStore.init()

TraceStore Methods

TraceFilter Fields

TraceSummary Fields

Examples

Further Reading