symfonic.infra.trace

trace

TraceStoreProtocol -- abstract interface for execution trace persistence.

The library depends only on this protocol. Concrete adapters (PostgreSQL, Redis, etc.) live in the application layer. InMemoryTraceStore is the default implementation using a bounded deque.

ExecutionTrace dataclass

ExecutionTrace(
    run_id: str,
    source: str = "chat",
    start_time: float = time.monotonic(),
    events: list[TraceEvent] = list(),
)

Complete trace for a single request or background task.

add_event

add_event(
    event_type: str, label: str, **details: Any
) -> None

Append a timestamped event to this trace.

Source code in src/symfonic/infra/trace.py

def add_event(self, event_type: str, label: str, **details: Any) -> None:
    """Append a timestamped event to this trace."""
    elapsed = (time.monotonic() - self.start_time) * 1000
    self.events.append(
        TraceEvent(
            elapsed_ms=round(elapsed, 1),
            event_type=event_type,
            label=label,
            details=details,
        )
    )

to_dict

to_dict() -> dict[str, Any]

Serialise to a plain dict suitable for JSON responses.

Source code in src/symfonic/infra/trace.py

def to_dict(self) -> dict[str, Any]:
    """Serialise to a plain dict suitable for JSON responses."""
    return {
        "run_id": self.run_id,
        "source": self.source,
        "events": [
            {
                "elapsed_ms": e.elapsed_ms,
                "event_type": e.event_type,
                "label": e.label,
                "details": e.details,
            }
            for e in self.events
        ],
    }

InMemoryTraceStore

InMemoryTraceStore(max_traces: int = MAX_TRACES)

Default trace store using an in-memory bounded deque.

Suitable for development and single-process deployments. Data is lost on restart. For production, inject a PostgresTraceStore or similar adapter that satisfies TraceStoreProtocol.

Source code in src/symfonic/infra/trace.py

def __init__(self, max_traces: int = MAX_TRACES) -> None:
    self._traces: deque[ExecutionTrace] = deque(maxlen=max_traces)
    self._current: dict[str, ExecutionTrace] = {}

finish_trace

finish_trace(run_id: str) -> None

Append a done event and move the trace to completed storage.

Source code in src/symfonic/infra/trace.py

def finish_trace(self, run_id: str) -> None:
    """Append a done event and move the trace to completed storage."""
    trace = self._current.pop(run_id, None)
    if trace:
        trace.add_event("done", "Complete")
        self._traces.append(trace)
        logger.debug("trace finished: run_id=%s", run_id)
    else:
        logger.debug("finish_trace called for unknown run_id=%s (no-op)", run_id)

get_trace

get_trace(run_id: str) -> ExecutionTrace | None

Return the active trace for run_id, or None.

Source code in src/symfonic/infra/trace.py

def get_trace(self, run_id: str) -> ExecutionTrace | None:
    """Return the active trace for run_id, or None."""
    return self._current.get(run_id)

latest

latest(limit: int = 10) -> list[dict[str, Any]]

Return up to limit most-recently completed traces.

Source code in src/symfonic/infra/trace.py

def latest(self, limit: int = 10) -> list[dict[str, Any]]:
    """Return up to `limit` most-recently completed traces."""
    return [t.to_dict() for t in list(self._traces)[-limit:]]

start_trace

start_trace(
    run_id: str, source: str = "chat"
) -> ExecutionTrace

Create a new active trace and return it.

Source code in src/symfonic/infra/trace.py

def start_trace(self, run_id: str, source: str = "chat") -> ExecutionTrace:
    """Create a new active trace and return it."""
    trace = ExecutionTrace(run_id=run_id, source=source)
    trace.add_event("start", f"Start {source}")
    self._current[run_id] = trace
    logger.debug("trace started: run_id=%s source=%s", run_id, source)
    return trace

TraceEvent dataclass

TraceEvent(
    elapsed_ms: float,
    event_type: str,
    label: str,
    details: dict[str, Any] = dict(),
)

Single event in an execution trace.

TraceStoreProtocol

Bases: Protocol

Abstract interface for execution trace storage.

All concrete adapters (in-memory, PostgreSQL, Redis, …) must satisfy this structural protocol. The library never imports a concrete class.

finish_trace

finish_trace(run_id: str) -> None

Mark a trace as complete and move it from active to storage.

A "done" event is appended before archiving. Calling with an unknown run_id is a no-op.

Parameters:

Name	Type	Description	Default
run_id	str	Identifier of the trace to finish.	required

Source code in src/symfonic/infra/trace.py

def finish_trace(self, run_id: str) -> None:
    """Mark a trace as complete and move it from active to storage.

    A "done" event is appended before archiving.  Calling with an
    unknown run_id is a no-op.

    Args:
        run_id: Identifier of the trace to finish.
    """
    ...

get_trace

get_trace(run_id: str) -> ExecutionTrace | None

Return an active (in-progress) trace, or None if not found.

Parameters:

Name	Type	Description	Default
run_id	str	Identifier to look up.	required

Source code in src/symfonic/infra/trace.py

def get_trace(self, run_id: str) -> ExecutionTrace | None:
    """Return an active (in-progress) trace, or None if not found.

    Args:
        run_id: Identifier to look up.
    """
    ...

latest

latest(limit: int = 10) -> list[dict[str, Any]]

Return the most recent completed traces as plain dicts.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of traces to return.	10

Source code in src/symfonic/infra/trace.py

def latest(self, limit: int = 10) -> list[dict[str, Any]]:
    """Return the most recent completed traces as plain dicts.

    Args:
        limit: Maximum number of traces to return.
    """
    ...

start_trace

start_trace(
    run_id: str, source: str = "chat"
) -> ExecutionTrace

Begin tracking a new execution trace.

Parameters:

Name	Type	Description	Default
run_id	str	Unique identifier for this execution run.	required
source	str	Label for the caller type (e.g. "chat", "stream", "worker").	'chat'

Returns:

Type	Description
ExecutionTrace	The newly created ExecutionTrace with an initial "start" event.

Source code in src/symfonic/infra/trace.py

def start_trace(self, run_id: str, source: str = "chat") -> ExecutionTrace:
    """Begin tracking a new execution trace.

    Args:
        run_id: Unique identifier for this execution run.
        source: Label for the caller type (e.g. "chat", "stream", "worker").

    Returns:
        The newly created ExecutionTrace with an initial "start" event.
    """
    ...