Tracing API

Overview

The Tracing API provides structured execution traces that capture the full hierarchy of an Orchestra run — from top-level execution through branches, agent steps, LLM generations, tool calls, and validation decisions.

Core Classes

TracingConfig

Configuration for execution tracing.

Import

from marsys.coordination.tracing.config import TracingConfig

Constructor

TracingConfig(
    enabled: bool = False,
    output_dir: str = "./traces",
    detail_level: str = "verbose",
    include_generation_details: bool = True,
    include_message_content: bool = True,
    include_tool_results: bool = True,
    max_content_length: int = 0
)

Parameters

TraceCollector

EventBus consumer that builds hierarchical span trees from execution events.

Import

from marsys.coordination.tracing import TraceCollector

Constructor

TraceCollector(
    event_bus: EventBus,
    config: TracingConfig,
    writers: Optional[List[TraceWriter]] = None
)

Parameters

Key Methods

finalize(session_id) -> Optional[TraceTree]

async def finalize(session_id: str) -> Optional[TraceTree]

Finalize the trace for a session. Closes any open spans (marking them as error), computes durations, and writes via all registered writers. Called automatically by Orchestra in a try/finally block.

close() -> None

async def close() -> None

Shut down all writers and release resources.

Span

A single unit of work in the execution trace.

Import

from marsys.coordination.tracing.types import Span

Fields

Key Methods

close(end_time, status) -> None

def close(end_time: Optional[float] = None, status: Optional[str] = None) -> None

Close the span, computing duration_ms from start_time to end_time.

add_event(name, attributes) -> None

def add_event(name: str, attributes: Optional[Dict[str, Any]] = None) -> None

Add an instant event (e.g., validation decision) to this span.

to_dict() -> Dict[str, Any]

def to_dict() -> Dict[str, Any]

Serialize the span tree to a nested dict for JSON output.

TraceTree

A complete execution trace rooted at an execution span.

Import

from marsys.coordination.tracing.types import TraceTree

Fields

to_dict() -> Dict[str, Any]

def to_dict() -> Dict[str, Any]

Serialize the full trace tree for JSON output.

TraceWriter

Abstract base for trace output backends.

Import

from marsys.coordination.tracing.writers.base import TraceWriter

Methods

JSONFileTraceWriter

Writes traces as structured JSON files.

Import

from marsys.coordination.tracing.writers import JSONFileTraceWriter

Constructor

JSONFileTraceWriter(config: TracingConfig)

Output: {config.output_dir}/{session_id}.json

The writer respects detail_level from the config:

• minimal: Span hierarchy and timing only
• standard: All attributes, strings truncated to max_content_length
• verbose: Full content, no truncation

Trace Events

Events emitted by execution components and consumed by TraceCollector. All extend StatusEvent.

ExecutionStartEvent

Emitted when Orchestra.execute() begins.

GenerationEvent

Emitted after each LLM generation completes.

ValidationDecisionEvent

Emitted after response validation determines the next action.

ConvergenceEvent

Emitted when parallel branches converge. The collector attaches convergence links and events to both the parent branch span and the next step span on that branch (the convergence step that receives aggregated results).

Usage Patterns

Enable Tracing

from marsys.coordination.config import ExecutionConfig
from marsys.coordination.tracing.config import TracingConfig

config = ExecutionConfig(
    tracing=TracingConfig(enabled=True, output_dir="./traces"),
)

Custom Writer

from marsys.coordination.tracing.writers.base import TraceWriter
from marsys.coordination.tracing.types import TraceTree

class MyTraceWriter(TraceWriter):
    async def write(self, trace: TraceTree) -> None:
        data = trace.to_dict()
        # Send to your observability backend
        await send_to_backend(data)

    async def close(self) -> None:
        pass

Access Trace Programmatically

# After Orchestra.execute(), if you have access to the Orchestra instance:
if orchestra.trace_collector:
    trace = await orchestra.trace_collector.finalize(session_id)
    if trace:
        trace_dict = trace.to_dict()

Best Practices

• Use detail_level="minimal" in production for low overhead
• Use detail_level="verbose" (default) during development for full visibility
• Use detail_level="standard" with max_content_length to limit trace size if needed
• Use meaningful session_id values for trace file identification
• Add ./traces/ to .gitignore

Related Documentation

• Tracing Concepts — Overview and span hierarchy
• Configuration API — ExecutionConfig reference
• Architecture Overview — How tracing fits in the system
• State Management API — Related persistence module