naqsha.tracing

Hierarchical QAOA Trace span helpers

naqsha.tracing

Trace store interfaces and implementations.

Span dataclass

Mutable span tracking execution metrics.

A Span represents a unit of work in the execution tree. It accumulates metrics like token counts and latency as the work progresses.

Source code in src/naqsha/tracing/span.py

@dataclass
class Span:
    """Mutable span tracking execution metrics.

    A Span represents a unit of work in the execution tree. It accumulates
    metrics like token counts and latency as the work progresses.
    """

    context: SpanContext
    token_count: int = 0
    model_latency_ms: float | None = None
    tool_exec_ms: float | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

    def add_tokens(self, count: int) -> None:
        """Add tokens to the span's total count."""
        self.token_count += count

    def set_model_latency(self, latency_ms: float) -> None:
        """Record model invocation latency."""
        self.model_latency_ms = latency_ms

    def set_tool_exec_time(self, exec_ms: float) -> None:
        """Record tool execution time."""
        self.tool_exec_ms = exec_ms

    def to_dict(self) -> dict[str, Any]:
        """Serialize span to dictionary for trace events."""
        return {
            "trace_id": self.context.trace_id,
            "span_id": self.context.span_id,
            "parent_span_id": self.context.parent_span_id,
            "agent_id": self.context.agent_id,
            "token_count": self.token_count,
            "model_latency_ms": self.model_latency_ms,
            "tool_exec_ms": self.tool_exec_ms,
            "metadata": self.metadata,
        }

context instance-attribute

context: SpanContext

token_count class-attribute instance-attribute

token_count: int = 0

model_latency_ms class-attribute instance-attribute

model_latency_ms: float | None = None

tool_exec_ms class-attribute instance-attribute

tool_exec_ms: float | None = None

metadata class-attribute instance-attribute

metadata: dict[str, Any] = field(default_factory=dict)

add_tokens

add_tokens(count: int) -> None

Add tokens to the span's total count.

Source code in src/naqsha/tracing/span.py

def add_tokens(self, count: int) -> None:
    """Add tokens to the span's total count."""
    self.token_count += count

set_model_latency

set_model_latency(latency_ms: float) -> None

Record model invocation latency.

Source code in src/naqsha/tracing/span.py

def set_model_latency(self, latency_ms: float) -> None:
    """Record model invocation latency."""
    self.model_latency_ms = latency_ms

set_tool_exec_time

set_tool_exec_time(exec_ms: float) -> None

Record tool execution time.

Source code in src/naqsha/tracing/span.py

def set_tool_exec_time(self, exec_ms: float) -> None:
    """Record tool execution time."""
    self.tool_exec_ms = exec_ms

to_dict

to_dict() -> dict[str, Any]

Serialize span to dictionary for trace events.

Source code in src/naqsha/tracing/span.py

def to_dict(self) -> dict[str, Any]:
    """Serialize span to dictionary for trace events."""
    return {
        "trace_id": self.context.trace_id,
        "span_id": self.context.span_id,
        "parent_span_id": self.context.parent_span_id,
        "agent_id": self.context.agent_id,
        "token_count": self.token_count,
        "model_latency_ms": self.model_latency_ms,
        "tool_exec_ms": self.tool_exec_ms,
        "metadata": self.metadata,
    }

SpanContext dataclass

Immutable context carrying trace and span identifiers.

SpanContext is propagated through the execution tree and attached to every trace event, enabling hierarchical trace reconstruction.

Source code in src/naqsha/tracing/span.py

@dataclass(frozen=True)
class SpanContext:
    """Immutable context carrying trace and span identifiers.

    SpanContext is propagated through the execution tree and attached to
    every trace event, enabling hierarchical trace reconstruction.
    """

    trace_id: str
    span_id: str
    parent_span_id: str | None
    agent_id: str

    def child_span(self, agent_id: str | None = None) -> SpanContext:
        """Create a child span context with this span as parent.

        Args:
            agent_id: Agent ID for the child span. If None, inherits from parent.

        Returns:
            New SpanContext with this span as parent.
        """
        return SpanContext(
            trace_id=self.trace_id,
            span_id=str(uuid4()),
            parent_span_id=self.span_id,
            agent_id=agent_id if agent_id is not None else self.agent_id,
        )

trace_id instance-attribute

trace_id: str

span_id instance-attribute

span_id: str

parent_span_id instance-attribute

parent_span_id: str | None

agent_id instance-attribute

agent_id: str

child_span

child_span(agent_id: str | None = None) -> SpanContext

Create a child span context with this span as parent.

Parameters:

Name	Type	Description	Default
agent_id	str | None	Agent ID for the child span. If None, inherits from parent.	None

Returns:

Type	Description
SpanContext	New SpanContext with this span as parent.

Source code in src/naqsha/tracing/span.py

def child_span(self, agent_id: str | None = None) -> SpanContext:
    """Create a child span context with this span as parent.

    Args:
        agent_id: Agent ID for the child span. If None, inherits from parent.

    Returns:
        New SpanContext with this span as parent.
    """
    return SpanContext(
        trace_id=self.trace_id,
        span_id=str(uuid4()),
        parent_span_id=self.span_id,
        agent_id=agent_id if agent_id is not None else self.agent_id,
    )

create_root_span

create_root_span(trace_id: str, agent_id: str) -> Span

Create a root span for a new trace.

Parameters:

Name	Type	Description	Default
trace_id	str	Unique identifier for the trace.	required
agent_id	str	Identifier for the agent starting the trace.	required

Returns:

Type	Description
Span	New Span with no parent.

Source code in src/naqsha/tracing/span.py

def create_root_span(trace_id: str, agent_id: str) -> Span:
    """Create a root span for a new trace.

    Args:
        trace_id: Unique identifier for the trace.
        agent_id: Identifier for the agent starting the trace.

    Returns:
        New Span with no parent.
    """
    context = SpanContext(
        trace_id=trace_id,
        span_id=str(uuid4()),
        parent_span_id=None,
        agent_id=agent_id,
    )
    return Span(context=context)