traceflowlens 0.4.0

Record, replay, and diff AI agent executions. Local-first.

TraceFlowLens

Record, replay, and diff AI agent executions, local-first.

Why

AI agents fail non-deterministically and the failure is hard to reproduce. TraceFlowLens records every model call, tool call, and error to a local SQLite file as your agent runs. Replay re-runs your own code, answering every recorded call with the saved output, so you can reproduce a failure exactly and prove a fix.

Install

pip install traceflowlens

Requires Python 3.11 or later. Zero runtime dependencies.

Quickstart

The script below runs without any AI SDK installed. FakeClient stands in for a real openai.OpenAI() or anthropic.Anthropic() client.

import traceflowlens as tfl


# Stand-in for a real OpenAI or Anthropic client (no SDK required).
class _FakeResponse:
    def __init__(self, content):
        self._content = content

    def model_dump(self):
        return {"choices": [{"message": {"content": self._content, "role": "assistant"}}]}


class _FakeCompletions:
    def create(self, **kwargs):
        return _FakeResponse("Hello from the fake model.")


class _FakeChat:
    def __init__(self):
        self.completions = _FakeCompletions()


class FakeClient:
    """Minimal stand-in for openai.OpenAI() with chat.completions.create support."""

    def __init__(self):
        self.chat = _FakeChat()


def run_agent(ctx, client):
    wrapped = ctx.wrap(client)
    response = wrapped.chat.completions.create(
        model="fake-model-v1",
        messages=[{"role": "user", "content": "Hello"}],
    )

    with ctx.step("tool_call", "web_search", inputs={"query": "TraceFlowLens"}) as step:
        step.set_output({"results": ["https://example.com"]})

    ctx.log_step(
        "custom",
        "agent_decision",
        inputs={"candidates": 3},
        outputs={"chosen": 0},
    )
    return response


client = FakeClient()

# Record
with tfl.record("my-run") as trace:
    trace_id = trace.trace_id
    run_agent(trace, client)

print(f"Recorded trace: {trace_id}")

# Replay
with tfl.replay(trace_id) as session:
    run_agent(session, client)

print(f"Replay complete: {session.result}")

Tags

import traceflowlens as tfl

with tfl.record("nightly-run", tags=["regression", "nightly"]) as trace:
    trace.log_step("custom", "greeting", inputs={"text": "hi"}, outputs={"ok": True})

Tags are strings, stored verbatim under the reserved metadata key tfl.tags: stripped of surrounding whitespace, at most 64 characters each, at most 20 per trace, deduplicated exactly. They render in tfl list and tfl show.

Reserved metadata keys

Metadata keys beginning with tfl. are reserved for the SDK. Supplying a reserved key in your own metadata raises ValueError at record time. Reserved keys currently stamped: tfl.sdk_version (always), tfl.tags (when tags are passed), tfl.replay_of and tfl.replay_outcome (on traces produced by record-while-replay). A trace without tfl.replay_outcome was not produced by record-while-replay; the key's absence never means "replayed clean".

Wrapping real clients

Call trace.wrap(client) to get a recording proxy. The proxy intercepts the supported methods and records each call as a step.

import openai
import anthropic
import traceflowlens as tfl

openai_client = openai.OpenAI()
anthropic_client = anthropic.Anthropic()

with tfl.record("my-run") as trace:
    oai = trace.wrap(openai_client)
    ant = trace.wrap(anthropic_client)

    # Intercepted and recorded:
    oai.chat.completions.create(model="gpt-4o", messages=[...])
    oai.responses.create(model="gpt-4o", input="Hello")
    ant.messages.create(model="claude-sonnet-4-6", max_tokens=1024, messages=[...])

Supported methods:

• OpenAI: chat.completions.create, responses.create
• Anthropic: messages.create

Sync only. Passing stream=True raises StreamingNotSupportedError. Calling any other method passes through to the real client with a one-time warning.

During replay, use session.wrap(client) in exactly the same way. The proxy intercepts the same methods and returns the recorded outputs without calling the real API.

Replay semantics

Replay matches recorded steps in strict sequence order by kind and name. Any call-order change raises ReplayDivergence at the first divergence point, carrying the exact position, what was expected, and what was received. That pinpoint is the debugging value: the divergence tells you exactly where your fix changed behavior.

Input drift (different arguments to the same call) is recorded on ReplayResult.input_drift but is not fatal by default. Pass strict_inputs=True to tfl.replay(...) to raise ReplayDivergence on any input change.

For model calls from the OpenAI or Anthropic SDKs, replay reconstructs the typed response object using model_validate. If the SDK is not installed or the saved data no longer validates, ReplayReconstructionError is raised. Pass allow_degraded=True to get a raw dict instead of an error.

Replay never modifies the original recording. By default it writes nothing at all; record_as records the replay run as a new trace (next section).

Record while replaying

Pass record_as to record the replay run as a new trace while it replays. The example below runs as written:

import traceflowlens as tfl

with tfl.record("original-run") as trace:
    trace.log_step("custom", "plan", inputs={"goal": "demo"}, outputs={"steps": 2})
    trace.log_step("custom", "act", inputs={"step": 1}, outputs={"ok": True})
original_id = trace.trace_id

with tfl.replay(original_id, record_as="fix-attempt-1", tags=["verify"]) as session:
    session.log_step("custom", "plan", inputs={"goal": "demo"})
    session.log_step("custom", "act", inputs={"step": 1})

replay_trace = session.recorded_trace
print(f"Original: {original_id}")
print(f"Recorded replay trace: {replay_trace.trace_id}")

The recorded trace is a real trace: it appears in tfl list, can be diffed against the original (tfl diff ORIGINAL_ID REPLAY_ID), and can be pushed. It is stamped with three reserved keys: tfl.sdk_version, tfl.replay_of (the original trace id), and tfl.replay_outcome.

tfl.replay_outcome is written when the session exits. A clean, fully consumed replay stamps:

{"diverged": false, "unconsumed": 0}

diverged is true when the replay raised ReplayDivergence (even if your code caught it), and the outcome then carries a divergence object with the seq and the expected and actual kind and name, never payload contents. unconsumed is the number of recorded steps the replay never reached: a replay that exits early is not proof of a fix, and the count says so.

On divergence the partial trace is kept and finalized, marked by the outcome key. The divergence point is the debugging value, so the recording that led up to it is preserved, not discarded.

Recording is opt-in per replay. Without record_as, replay writes nothing, exactly as before.

Manual steps and the recorded trace

During record-while-replay, the steps recorded automatically are the ones that flow through the replay session: wrapped client calls, session.log_step, and session.step. A log_step against any other trace handle does not appear in the recorded replay trace. To add extra context steps to the recorded trace itself, log to session.recorded_trace:

import traceflowlens as tfl

with tfl.record("original-run-2") as trace:
    trace.log_step("custom", "plan", inputs={"goal": "demo"}, outputs={"steps": 1})
original_id = trace.trace_id

with tfl.replay(original_id, record_as="fix-attempt-2") as session:
    session.log_step("custom", "plan", inputs={"goal": "demo"})
    session.recorded_trace.log_step("custom", "note", inputs={"msg": "verified"})

print(f"Recorded replay trace: {session.recorded_trace.trace_id}")

Tags passed to tfl.replay(...) apply to the recorded trace only and are never inherited from the original; the original's tags stay on the original. Passing tags without record_as raises ValueError.

Push (opt-in)

TraceFlowLens is local-first. Push is an explicit opt-in that sends a recorded trace to the hosted API. It is off by default: nothing imports the push module unless you call it.

Library

from traceflowlens import push

result = push(
    trace_id="<TRACE_ID>",
    url="<API_BASE_URL>",    # e.g. https://api.example.com
    api_key="<YOUR_API_KEY>",
    db_path="./traceflowlens.db",  # optional, default used if omitted
)
print(result.trace_id, result.status)  # status: "created" or "already_exists"

push raises PushConfigError when url or api_key is missing, the SDK's TraceNotFoundError when the trace does not exist locally (before any network I/O), and PushError on any HTTP or transport failure. PushError carries http_status (None for transport failures), error_code, and detail from the API error body when available.

CLI

tfl push TRACE_ID [--url URL] [--key KEY] [--db PATH]

--url and --key are optional when the corresponding environment variables are set. On success the trace_id and status are printed to stdout. On error the message is printed to stderr and the exit code is 1.

Environment variables

Variable	Purpose
TFL_PUSH_URL	API base URL (fallback when --url is absent)
TFL_API_KEY	Bearer token (fallback when --key is absent)

Note on shell history: passing --key on the command line stores the value in shell history. Use the environment variable instead:

export TFL_API_KEY="<YOUR_API_KEY>"
tfl push <TRACE_ID>

CLI

The tfl command operates on the local database. Default path: ./traceflowlens.db. Override with --db PATH.

tfl list                      List all traces (id, name, status, started_at, step count).
tfl show TRACE_ID             Show trace header and step table with DEGRADED flags.
tfl replay TRACE_ID           Print a replay readiness report (not re-execution).
tfl diff TRACE_A TRACE_B      Print a step-by-step structural diff of two traces.
tfl delete TRACE_ID [--yes]   Delete a trace; prompts for confirmation without --yes.
tfl push TRACE_ID             Push a trace to the hosted API.

Note: tfl replay prints a readiness report. Actual replay is done via the library: with traceflowlens.replay(trace_id) as session:.

Data

Traces are stored in a local SQLite file (./traceflowlens.db by default). Override the path with the db_path parameter on tfl.record(...) and tfl.replay(...), or with --db on the CLI.

Full inputs and outputs are recorded by design. Nothing leaves your machine.

Scope

TraceFlowLens is sync only. Async is not supported. Streaming is not supported. There are no framework adapters (LangChain, LangGraph, CrewAI, etc.). Push to the hosted API is opt-in and off by default. These are current facts about this release, not items on a roadmap.

Versioning

TraceFlowLens follows SemVer. Before 1.0, minor version increments may include breaking changes.

Changes in 0.4.0

• Record-while-replay: new keyword-only record_as argument on replay() and start_replay(). The replay run is recorded as a new trace stamped with tfl.replay_of and tfl.replay_outcome. Replay still writes nothing without it.
• New keyword-only tags argument on replay() and start_replay(), applied to the recorded trace; requires record_as.
• New ReplaySession.recorded_trace property.
• New reserved metadata key tfl.replay_outcome: diverged, unconsumed, and a divergence summary (seq, expected and actual kind and name) on diverged runs.

Changes in 0.3.0

• push is importable only from the package root: from traceflowlens import push. The former traceflowlens.push module path is gone (the module is private now).
• The module-class interceptor workaround from 0.2.0 is removed.
• Metadata keys beginning with tfl. are now rejected at record time; this namespace is reserved for SDK-stamped keys.
• New keyword-only tags argument on record() and start_trace().
• Every trace now records the SDK version that produced it under tfl.sdk_version.

License

MIT