Trace

Trace recording, replay, and time-travel debugging.

Recorder

agentprobe.trace.recorder

Trace recorder: captures agent execution events into structured Traces.

Provides an async context manager for recording LLM calls, tool invocations, and message exchanges during agent execution.

TraceRecordingContext

Mutable context for recording events during an agent run.

Accumulates LLM calls, tool calls, and turns that will be assembled into a frozen Trace by TraceRecorder.finalize().

Source code in src/agentprobe/trace/recorder.py

class TraceRecordingContext:
    """Mutable context for recording events during an agent run.

    Accumulates LLM calls, tool calls, and turns that will be
    assembled into a frozen Trace by ``TraceRecorder.finalize()``.
    """

    def __init__(self, agent_name: str, model: str | None = None) -> None:
        self.agent_name = agent_name
        self.model = model
        self.llm_calls: list[LLMCall] = []
        self.tool_calls: list[ToolCall] = []
        self.turns: list[Turn] = []
        self.tags: list[str] = []
        self.metadata: dict[str, Any] = {}
        self._start_time = time.monotonic()

    def record_llm_call(
        self,
        *,
        model: str,
        input_tokens: int = 0,
        output_tokens: int = 0,
        input_text: str = "",
        output_text: str = "",
        latency_ms: int = 0,
        metadata: dict[str, Any] | None = None,
    ) -> LLMCall:
        """Record an LLM call event.

        Args:
            model: Model identifier.
            input_tokens: Number of input tokens.
            output_tokens: Number of output tokens.
            input_text: Prompt text.
            output_text: Response text.
            latency_ms: Call latency in milliseconds.
            metadata: Additional metadata.

        Returns:
            The recorded LLMCall object.
        """
        call = LLMCall(
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            input_text=input_text,
            output_text=output_text,
            latency_ms=latency_ms,
            metadata=metadata or {},
        )
        self.llm_calls.append(call)
        self.turns.append(Turn(turn_type=TurnType.LLM_CALL, content=output_text, llm_call=call))
        if self.model is None:
            self.model = model
        return call

    def record_tool_call(
        self,
        *,
        tool_name: str,
        tool_input: dict[str, Any] | None = None,
        tool_output: Any = None,
        success: bool = True,
        error: str | None = None,
        latency_ms: int = 0,
    ) -> ToolCall:
        """Record a tool call event.

        Args:
            tool_name: Name of the tool.
            tool_input: Arguments passed to the tool.
            tool_output: Output from the tool.
            success: Whether the call succeeded.
            error: Error message if failed.
            latency_ms: Call latency in milliseconds.

        Returns:
            The recorded ToolCall object.
        """
        call = ToolCall(
            tool_name=tool_name,
            tool_input=tool_input or {},
            tool_output=tool_output,
            success=success,
            error=error,
            latency_ms=latency_ms,
        )
        self.tool_calls.append(call)
        self.turns.append(
            Turn(
                turn_type=TurnType.TOOL_CALL,
                content=str(tool_output) if tool_output is not None else "",
                tool_call=call,
            )
        )
        return call

    @property
    def elapsed_ms(self) -> int:
        """Return elapsed time since recording started."""
        return int((time.monotonic() - self._start_time) * 1000)

elapsed_ms property

Return elapsed time since recording started.

record_llm_call(*, model, input_tokens=0, output_tokens=0, input_text='', output_text='', latency_ms=0, metadata=None)

Record an LLM call event.

Parameters:

Name	Type	Description	Default
model	str	Model identifier.	required
input_tokens	int	Number of input tokens.	0
output_tokens	int	Number of output tokens.	0
input_text	str	Prompt text.	''
output_text	str	Response text.	''
latency_ms	int	Call latency in milliseconds.	0
metadata	dict[str, Any] | None	Additional metadata.	None

Returns:

Type	Description
LLMCall	The recorded LLMCall object.

Source code in src/agentprobe/trace/recorder.py

def record_llm_call(
    self,
    *,
    model: str,
    input_tokens: int = 0,
    output_tokens: int = 0,
    input_text: str = "",
    output_text: str = "",
    latency_ms: int = 0,
    metadata: dict[str, Any] | None = None,
) -> LLMCall:
    """Record an LLM call event.

    Args:
        model: Model identifier.
        input_tokens: Number of input tokens.
        output_tokens: Number of output tokens.
        input_text: Prompt text.
        output_text: Response text.
        latency_ms: Call latency in milliseconds.
        metadata: Additional metadata.

    Returns:
        The recorded LLMCall object.
    """
    call = LLMCall(
        model=model,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        input_text=input_text,
        output_text=output_text,
        latency_ms=latency_ms,
        metadata=metadata or {},
    )
    self.llm_calls.append(call)
    self.turns.append(Turn(turn_type=TurnType.LLM_CALL, content=output_text, llm_call=call))
    if self.model is None:
        self.model = model
    return call

record_tool_call(*, tool_name, tool_input=None, tool_output=None, success=True, error=None, latency_ms=0)

Record a tool call event.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool.	required
tool_input	dict[str, Any] | None	Arguments passed to the tool.	None
tool_output	Any	Output from the tool.	None
success	bool	Whether the call succeeded.	True
error	str | None	Error message if failed.	None
latency_ms	int	Call latency in milliseconds.	0

Returns:

Type	Description
ToolCall	The recorded ToolCall object.

Source code in src/agentprobe/trace/recorder.py

def record_tool_call(
    self,
    *,
    tool_name: str,
    tool_input: dict[str, Any] | None = None,
    tool_output: Any = None,
    success: bool = True,
    error: str | None = None,
    latency_ms: int = 0,
) -> ToolCall:
    """Record a tool call event.

    Args:
        tool_name: Name of the tool.
        tool_input: Arguments passed to the tool.
        tool_output: Output from the tool.
        success: Whether the call succeeded.
        error: Error message if failed.
        latency_ms: Call latency in milliseconds.

    Returns:
        The recorded ToolCall object.
    """
    call = ToolCall(
        tool_name=tool_name,
        tool_input=tool_input or {},
        tool_output=tool_output,
        success=success,
        error=error,
        latency_ms=latency_ms,
    )
    self.tool_calls.append(call)
    self.turns.append(
        Turn(
            turn_type=TurnType.TOOL_CALL,
            content=str(tool_output) if tool_output is not None else "",
            tool_call=call,
        )
    )
    return call

TraceRecorder

Records agent execution events into a structured Trace.

Use as an async context manager via recording() to capture events, then call finalize() to produce the frozen Trace.

Attributes:

Name	Type	Description
agent_name		Name of the agent being recorded.
model		Primary model being used.
tags		Optional tags for filtering.

Example

recorder = TraceRecorder(agent_name="support")

async with recorder.recording() as ctx:
    ctx.record_llm_call(model="claude-sonnet-4-5-20250929", ...)
    ctx.record_tool_call(tool_name="search", ...)

trace = recorder.finalize(output="Done")

Source code in src/agentprobe/trace/recorder.py

class TraceRecorder:
    """Records agent execution events into a structured Trace.

    Use as an async context manager via ``recording()`` to capture
    events, then call ``finalize()`` to produce the frozen Trace.

    Attributes:
        agent_name: Name of the agent being recorded.
        model: Primary model being used.
        tags: Optional tags for filtering.

    Example:
        ```python
        recorder = TraceRecorder(agent_name="support")

        async with recorder.recording() as ctx:
            ctx.record_llm_call(model="claude-sonnet-4-5-20250929", ...)
            ctx.record_tool_call(tool_name="search", ...)

        trace = recorder.finalize(output="Done")
        ```
    """

    def __init__(
        self,
        agent_name: str,
        model: str | None = None,
        tags: list[str] | None = None,
    ) -> None:
        """Initialize a new trace recorder.

        Args:
            agent_name: Identifier for the agent being recorded.
            model: Primary model name.
            tags: Optional tags for categorization.

        Raises:
            ValueError: If agent_name is empty.
        """
        if not agent_name:
            msg = "agent_name must not be empty"
            raise ValueError(msg)
        self._agent_name = agent_name
        self._model = model
        self._tags = tags or []
        self._context: TraceRecordingContext | None = None

    @asynccontextmanager
    async def recording(self) -> AsyncGenerator[TraceRecordingContext, None]:
        """Start a recording session.

        Yields a TraceRecordingContext for recording events.

        Yields:
            A mutable recording context.
        """
        self._context = TraceRecordingContext(
            agent_name=self._agent_name,
            model=self._model,
        )
        self._context.tags = list(self._tags)
        logger.debug("Started recording for agent '%s'", self._agent_name)
        try:
            yield self._context
        except Exception:
            logger.exception("Error during recording for agent '%s'", self._agent_name)
            raise
        finally:
            logger.debug(
                "Recording ended for agent '%s': %d LLM calls, %d tool calls",
                self._agent_name,
                len(self._context.llm_calls),
                len(self._context.tool_calls),
            )

    def finalize(
        self,
        *,
        input_text: str = "",
        output: str = "",
    ) -> Trace:
        """Produce a frozen Trace from the recorded events.

        Args:
            input_text: The input given to the agent.
            output: The final output from the agent.

        Returns:
            An immutable Trace object.

        Raises:
            TraceError: If no recording session was started.
        """
        if self._context is None:
            raise TraceError("No recording session — call recording() first")

        ctx = self._context
        total_input = sum(c.input_tokens for c in ctx.llm_calls)
        total_output = sum(c.output_tokens for c in ctx.llm_calls)

        trace = Trace(
            agent_name=ctx.agent_name,
            model=ctx.model,
            input_text=input_text,
            output_text=output,
            turns=tuple(ctx.turns),
            llm_calls=tuple(ctx.llm_calls),
            tool_calls=tuple(ctx.tool_calls),
            total_input_tokens=total_input,
            total_output_tokens=total_output,
            total_latency_ms=ctx.elapsed_ms,
            tags=tuple(ctx.tags),
            metadata=ctx.metadata,
            created_at=datetime.now(UTC),
        )

        self._context = None
        return trace

__init__(agent_name, model=None, tags=None)

Initialize a new trace recorder.

Parameters:

Name	Type	Description	Default
agent_name	str	Identifier for the agent being recorded.	required
model	str | None	Primary model name.	None
tags	list[str] | None	Optional tags for categorization.	None

Raises:

Type	Description
ValueError	If agent_name is empty.

Source code in src/agentprobe/trace/recorder.py

def __init__(
    self,
    agent_name: str,
    model: str | None = None,
    tags: list[str] | None = None,
) -> None:
    """Initialize a new trace recorder.

    Args:
        agent_name: Identifier for the agent being recorded.
        model: Primary model name.
        tags: Optional tags for categorization.

    Raises:
        ValueError: If agent_name is empty.
    """
    if not agent_name:
        msg = "agent_name must not be empty"
        raise ValueError(msg)
    self._agent_name = agent_name
    self._model = model
    self._tags = tags or []
    self._context: TraceRecordingContext | None = None

recording() async

Start a recording session.

Yields a TraceRecordingContext for recording events.

Yields:

Type	Description
AsyncGenerator[TraceRecordingContext, None]	A mutable recording context.

Source code in src/agentprobe/trace/recorder.py

@asynccontextmanager
async def recording(self) -> AsyncGenerator[TraceRecordingContext, None]:
    """Start a recording session.

    Yields a TraceRecordingContext for recording events.

    Yields:
        A mutable recording context.
    """
    self._context = TraceRecordingContext(
        agent_name=self._agent_name,
        model=self._model,
    )
    self._context.tags = list(self._tags)
    logger.debug("Started recording for agent '%s'", self._agent_name)
    try:
        yield self._context
    except Exception:
        logger.exception("Error during recording for agent '%s'", self._agent_name)
        raise
    finally:
        logger.debug(
            "Recording ended for agent '%s': %d LLM calls, %d tool calls",
            self._agent_name,
            len(self._context.llm_calls),
            len(self._context.tool_calls),
        )

finalize(*, input_text='', output='')

Produce a frozen Trace from the recorded events.

Parameters:

Name	Type	Description	Default
input_text	str	The input given to the agent.	''
output	str	The final output from the agent.	''

Returns:

Type	Description
Trace	An immutable Trace object.

Raises:

Type	Description
TraceError	If no recording session was started.

Source code in src/agentprobe/trace/recorder.py

def finalize(
    self,
    *,
    input_text: str = "",
    output: str = "",
) -> Trace:
    """Produce a frozen Trace from the recorded events.

    Args:
        input_text: The input given to the agent.
        output: The final output from the agent.

    Returns:
        An immutable Trace object.

    Raises:
        TraceError: If no recording session was started.
    """
    if self._context is None:
        raise TraceError("No recording session — call recording() first")

    ctx = self._context
    total_input = sum(c.input_tokens for c in ctx.llm_calls)
    total_output = sum(c.output_tokens for c in ctx.llm_calls)

    trace = Trace(
        agent_name=ctx.agent_name,
        model=ctx.model,
        input_text=input_text,
        output_text=output,
        turns=tuple(ctx.turns),
        llm_calls=tuple(ctx.llm_calls),
        tool_calls=tuple(ctx.tool_calls),
        total_input_tokens=total_input,
        total_output_tokens=total_output,
        total_latency_ms=ctx.elapsed_ms,
        tags=tuple(ctx.tags),
        metadata=ctx.metadata,
        created_at=datetime.now(UTC),
    )

    self._context = None
    return trace

Replay Engine

agentprobe.trace.replay

Trace replay engine for re-executing recorded traces.

Supports pure replay from recorded data, with optional mock overrides for tool calls and outputs. Computes a ReplayDiff showing differences between original and replayed results.

ReplayEngine

Replays recorded traces for comparison and testing.

In pure replay mode, tool calls return their recorded outputs. Optional mock functions can override specific tools.

Attributes:

Name	Type	Description
mock_tools		Mapping of tool names to mock functions.
mock_output		If set, override the replay output text.

Source code in src/agentprobe/trace/replay.py

class ReplayEngine:
    """Replays recorded traces for comparison and testing.

    In pure replay mode, tool calls return their recorded outputs.
    Optional mock functions can override specific tools.

    Attributes:
        mock_tools: Mapping of tool names to mock functions.
        mock_output: If set, override the replay output text.
    """

    def __init__(
        self,
        *,
        mock_tools: dict[str, Callable[..., Any]] | None = None,
        mock_output: str | None = None,
    ) -> None:
        """Initialize the replay engine.

        Args:
            mock_tools: Optional tool name to mock function mapping.
            mock_output: If set, override the output text.
        """
        self._mock_tools = mock_tools or {}
        self._mock_output = mock_output

    def replay(self, trace: Trace) -> Trace:
        """Replay a trace, applying any mock overrides.

        Args:
            trace: The original trace to replay.

        Returns:
            A new trace with mock overrides applied.
        """
        if not self._mock_tools and self._mock_output is None:
            return trace

        modified_calls: list[ToolCall] = []
        for tc in trace.tool_calls:
            mock_fn = self._mock_tools.get(tc.tool_name)
            if mock_fn is not None:
                try:
                    mock_result = mock_fn(tc.tool_input)
                    modified_calls.append(tc.model_copy(update={"tool_output": mock_result}))
                except Exception as exc:
                    modified_calls.append(
                        tc.model_copy(
                            update={
                                "success": False,
                                "error": f"Mock error: {exc}",
                                "tool_output": None,
                            }
                        )
                    )
            else:
                modified_calls.append(tc)

        output = self._mock_output if self._mock_output is not None else trace.output_text

        return trace.model_copy(
            update={
                "tool_calls": tuple(modified_calls),
                "output_text": output,
            },
        )

    def diff(self, original: Trace, replay: Trace) -> ReplayDiff:
        """Compute the diff between an original trace and a replay.

        Args:
            original: The original trace.
            replay: The replayed trace.

        Returns:
            A ReplayDiff showing the differences.
        """
        tool_diffs: list[DiffItem] = []

        max_len = max(len(original.tool_calls), len(replay.tool_calls))
        for i in range(max_len):
            if i < len(original.tool_calls) and i < len(replay.tool_calls):
                orig_tc = original.tool_calls[i]
                replay_tc = replay.tool_calls[i]
                sim = _tool_call_similarity(orig_tc, replay_tc)
                tool_diffs.append(
                    DiffItem(
                        dimension=f"tool_call_{i}",
                        expected=orig_tc.tool_name,
                        actual=replay_tc.tool_name,
                        similarity=round(sim, 4),
                    )
                )
            elif i < len(original.tool_calls):
                tool_diffs.append(
                    DiffItem(
                        dimension=f"tool_call_{i}",
                        expected=original.tool_calls[i].tool_name,
                        actual=None,
                        similarity=0.0,
                    )
                )
            else:
                tool_diffs.append(
                    DiffItem(
                        dimension=f"tool_call_{i}",
                        expected=None,
                        actual=replay.tool_calls[i].tool_name,
                        similarity=0.0,
                    )
                )

        output_matches = original.output_text == replay.output_text

        return ReplayDiff(
            original_trace_id=original.trace_id,
            replay_trace_id=replay.trace_id,
            tool_call_diffs=tuple(tool_diffs),
            output_matches=output_matches,
            original_output=original.output_text,
            replay_output=replay.output_text,
        )

__init__(*, mock_tools=None, mock_output=None)

Initialize the replay engine.

Parameters:

Name	Type	Description	Default
mock_tools	dict[str, Callable[..., Any]] | None	Optional tool name to mock function mapping.	None
mock_output	str | None	If set, override the output text.	None

Source code in src/agentprobe/trace/replay.py

def __init__(
    self,
    *,
    mock_tools: dict[str, Callable[..., Any]] | None = None,
    mock_output: str | None = None,
) -> None:
    """Initialize the replay engine.

    Args:
        mock_tools: Optional tool name to mock function mapping.
        mock_output: If set, override the output text.
    """
    self._mock_tools = mock_tools or {}
    self._mock_output = mock_output

replay(trace)

Replay a trace, applying any mock overrides.

Parameters:

Name	Type	Description	Default
trace	Trace	The original trace to replay.	required

Returns:

Type	Description
Trace	A new trace with mock overrides applied.

Source code in src/agentprobe/trace/replay.py

def replay(self, trace: Trace) -> Trace:
    """Replay a trace, applying any mock overrides.

    Args:
        trace: The original trace to replay.

    Returns:
        A new trace with mock overrides applied.
    """
    if not self._mock_tools and self._mock_output is None:
        return trace

    modified_calls: list[ToolCall] = []
    for tc in trace.tool_calls:
        mock_fn = self._mock_tools.get(tc.tool_name)
        if mock_fn is not None:
            try:
                mock_result = mock_fn(tc.tool_input)
                modified_calls.append(tc.model_copy(update={"tool_output": mock_result}))
            except Exception as exc:
                modified_calls.append(
                    tc.model_copy(
                        update={
                            "success": False,
                            "error": f"Mock error: {exc}",
                            "tool_output": None,
                        }
                    )
                )
        else:
            modified_calls.append(tc)

    output = self._mock_output if self._mock_output is not None else trace.output_text

    return trace.model_copy(
        update={
            "tool_calls": tuple(modified_calls),
            "output_text": output,
        },
    )

diff(original, replay)

Compute the diff between an original trace and a replay.

Parameters:

Name	Type	Description	Default
original	Trace	The original trace.	required
replay	Trace	The replayed trace.	required

Returns:

Type	Description
ReplayDiff	A ReplayDiff showing the differences.

Source code in src/agentprobe/trace/replay.py

def diff(self, original: Trace, replay: Trace) -> ReplayDiff:
    """Compute the diff between an original trace and a replay.

    Args:
        original: The original trace.
        replay: The replayed trace.

    Returns:
        A ReplayDiff showing the differences.
    """
    tool_diffs: list[DiffItem] = []

    max_len = max(len(original.tool_calls), len(replay.tool_calls))
    for i in range(max_len):
        if i < len(original.tool_calls) and i < len(replay.tool_calls):
            orig_tc = original.tool_calls[i]
            replay_tc = replay.tool_calls[i]
            sim = _tool_call_similarity(orig_tc, replay_tc)
            tool_diffs.append(
                DiffItem(
                    dimension=f"tool_call_{i}",
                    expected=orig_tc.tool_name,
                    actual=replay_tc.tool_name,
                    similarity=round(sim, 4),
                )
            )
        elif i < len(original.tool_calls):
            tool_diffs.append(
                DiffItem(
                    dimension=f"tool_call_{i}",
                    expected=original.tool_calls[i].tool_name,
                    actual=None,
                    similarity=0.0,
                )
            )
        else:
            tool_diffs.append(
                DiffItem(
                    dimension=f"tool_call_{i}",
                    expected=None,
                    actual=replay.tool_calls[i].tool_name,
                    similarity=0.0,
                )
            )

    output_matches = original.output_text == replay.output_text

    return ReplayDiff(
        original_trace_id=original.trace_id,
        replay_trace_id=replay.trace_id,
        tool_call_diffs=tuple(tool_diffs),
        output_matches=output_matches,
        original_output=original.output_text,
        replay_output=replay.output_text,
    )

Time Travel

agentprobe.trace.time_travel

Time-travel debugger for step-by-step trace inspection.

Provides indexed access to individual turns in a trace with cumulative metrics at each step, enabling debugging and analysis of agent execution flow.

TimeTravel

Step-by-step trace inspector with cumulative metrics.

Pre-computes a list of TraceStep objects on construction, providing indexed access and iteration over the trace timeline with cumulative token, cost, and latency metrics at each step.

Attributes:

Name	Type	Description
trace	Trace	The trace being inspected.

Source code in src/agentprobe/trace/time_travel.py

class TimeTravel:
    """Step-by-step trace inspector with cumulative metrics.

    Pre-computes a list of TraceStep objects on construction,
    providing indexed access and iteration over the trace timeline
    with cumulative token, cost, and latency metrics at each step.

    Attributes:
        trace: The trace being inspected.
    """

    def __init__(
        self, trace: Trace, *, cost_per_1k_input: float = 0.0, cost_per_1k_output: float = 0.0
    ) -> None:
        """Initialize the time-travel debugger.

        Args:
            trace: The trace to inspect.
            cost_per_1k_input: Cost per 1K input tokens for cumulative cost.
            cost_per_1k_output: Cost per 1K output tokens for cumulative cost.
        """
        self._trace = trace
        self._steps = self._build_steps(trace, cost_per_1k_input, cost_per_1k_output)

    @property
    def trace(self) -> Trace:
        """Return the underlying trace."""
        return self._trace

    @property
    def total_steps(self) -> int:
        """Return the total number of steps."""
        return len(self._steps)

    @staticmethod
    def _build_steps(
        trace: Trace,
        cost_per_1k_input: float,
        cost_per_1k_output: float,
    ) -> list[TraceStep]:
        """Build the list of trace steps with cumulative metrics."""
        steps: list[TraceStep] = []
        cum_input = 0
        cum_output = 0
        cum_cost = 0.0
        cum_latency = 0

        for i, turn in enumerate(trace.turns):
            if turn.turn_type == TurnType.LLM_CALL and turn.llm_call is not None:
                cum_input += turn.llm_call.input_tokens
                cum_output += turn.llm_call.output_tokens
                cum_cost += (
                    turn.llm_call.input_tokens / 1000.0 * cost_per_1k_input
                    + turn.llm_call.output_tokens / 1000.0 * cost_per_1k_output
                )
                cum_latency += turn.llm_call.latency_ms
            elif turn.turn_type == TurnType.TOOL_CALL and turn.tool_call is not None:
                cum_latency += turn.tool_call.latency_ms

            steps.append(
                TraceStep(
                    step_index=i,
                    turn=turn,
                    cumulative_input_tokens=cum_input,
                    cumulative_output_tokens=cum_output,
                    cumulative_cost_usd=round(cum_cost, 6),
                    cumulative_latency_ms=cum_latency,
                )
            )

        return steps

    def __len__(self) -> int:
        return len(self._steps)

    def __getitem__(self, index: int) -> TraceStep:
        """Get a step by index.

        Args:
            index: Zero-based step index. Supports negative indexing.

        Returns:
            The TraceStep at the given index.

        Raises:
            IndexError: If the index is out of range.
        """
        return self._steps[index]

    def __iter__(self) -> Iterator[TraceStep]:
        return iter(self._steps)

    def steps(self) -> list[TraceStep]:
        """Return all steps as a list."""
        return list(self._steps)

    def rerun_from(self, step_index: int) -> list[TraceStep]:
        """Return all steps from a given index onward.

        Args:
            step_index: Zero-based starting index.

        Returns:
            Steps from the given index to the end.

        Raises:
            IndexError: If step_index is out of range.
        """
        if step_index < 0 or step_index >= len(self._steps):
            raise IndexError(f"Step index {step_index} out of range [0, {len(self._steps)})")
        return list(self._steps[step_index:])

trace property

Return the underlying trace.

total_steps property

Return the total number of steps.

__init__(trace, *, cost_per_1k_input=0.0, cost_per_1k_output=0.0)

Initialize the time-travel debugger.

Parameters:

Name	Type	Description	Default
trace	Trace	The trace to inspect.	required
cost_per_1k_input	float	Cost per 1K input tokens for cumulative cost.	0.0
cost_per_1k_output	float	Cost per 1K output tokens for cumulative cost.	0.0

Source code in src/agentprobe/trace/time_travel.py

def __init__(
    self, trace: Trace, *, cost_per_1k_input: float = 0.0, cost_per_1k_output: float = 0.0
) -> None:
    """Initialize the time-travel debugger.

    Args:
        trace: The trace to inspect.
        cost_per_1k_input: Cost per 1K input tokens for cumulative cost.
        cost_per_1k_output: Cost per 1K output tokens for cumulative cost.
    """
    self._trace = trace
    self._steps = self._build_steps(trace, cost_per_1k_input, cost_per_1k_output)

__getitem__(index)

Get a step by index.

Parameters:

Name	Type	Description	Default
index	int	Zero-based step index. Supports negative indexing.	required

Returns:

Type	Description
TraceStep	The TraceStep at the given index.

Raises:

Type	Description
IndexError	If the index is out of range.

Source code in src/agentprobe/trace/time_travel.py

def __getitem__(self, index: int) -> TraceStep:
    """Get a step by index.

    Args:
        index: Zero-based step index. Supports negative indexing.

    Returns:
        The TraceStep at the given index.

    Raises:
        IndexError: If the index is out of range.
    """
    return self._steps[index]

steps()

Return all steps as a list.

Source code in src/agentprobe/trace/time_travel.py

def steps(self) -> list[TraceStep]:
    """Return all steps as a list."""
    return list(self._steps)

rerun_from(step_index)

Return all steps from a given index onward.

Parameters:

Name	Type	Description	Default
step_index	int	Zero-based starting index.	required

Returns:

Type	Description
list[TraceStep]	Steps from the given index to the end.

Raises:

Type	Description
IndexError	If step_index is out of range.

Source code in src/agentprobe/trace/time_travel.py

def rerun_from(self, step_index: int) -> list[TraceStep]:
    """Return all steps from a given index onward.

    Args:
        step_index: Zero-based starting index.

    Returns:
        Steps from the given index to the end.

    Raises:
        IndexError: If step_index is out of range.
    """
    if step_index < 0 or step_index >= len(self._steps):
        raise IndexError(f"Step index {step_index} out of range [0, {len(self._steps)})")
    return list(self._steps[step_index:])