symfonic.observability.otel.tracer

tracer

Core OTEL tracer state for symfonic-core.

CONTAINS opentelemetry imports at module top level. Loaded only when OTelExporter.build() is called (i.e. when FrameworkConfig.otel_enabled is True). Keeping these imports here -- and out of exporter.py -- preserves the zero-cost guarantee on the cold path.

The tracer owns the TracerProvider + OTLP exporter + the root-span context manager. Per-event helpers (start/end node/llm/tool spans, record consolidation/hydration/routing) live in ._span_recorder.SpanRecorder so this file stays focused on SDK plumbing. See the design note at .claude/docs/2026-05-22-otel-exporter-design.md sections 7 and 8.

OTelTracer

OTelTracer(
    *,
    endpoint: str | None,
    service_name: str,
    capture_prompts: bool,
    span_exporter: SpanExporter | None = None,
    use_batch_processor: bool | None = None,
)

Owns the TracerProvider + OTLP exporter and the run-span lifecycle.

Lifetime is tied to the owning SymfonicAgent; shutdown() is called from the agent's aclose. Per-event helpers are accessible via self.recorder.

Construct the tracer and install a processor on the provider.

Parameters:

Name	Type	Description	Default
endpoint	str | None	OTLP target URL. None falls back to the OTEL_EXPORTER_OTLP_ENDPOINT env var, then the SDK default.	required
service_name	str	service.name resource attribute.	required
capture_prompts	bool	Whether to populate privacy-gated attributes (prompt text, tool args, tool results).	required
span_exporter	SpanExporter | None	Optional pre-built exporter. Tests inject an InMemorySpanExporter here. When provided, the OTLP gRPC exporter is NOT constructed (so unit tests do not require opentelemetry-exporter-otlp-proto-grpc).	None
use_batch_processor	bool | None	When True (default in production), spans are flushed via a BatchSpanProcessor. Tests pass False for deterministic span readback via a SimpleSpanProcessor.	None

Source code in src/symfonic/observability/otel/tracer.py

def __init__(
    self,
    *,
    endpoint: str | None,
    service_name: str,
    capture_prompts: bool,
    span_exporter: SpanExporter | None = None,
    use_batch_processor: bool | None = None,
) -> None:
    """Construct the tracer and install a processor on the provider.

    Args:
        endpoint: OTLP target URL. ``None`` falls back to the
            ``OTEL_EXPORTER_OTLP_ENDPOINT`` env var, then the SDK default.
        service_name: ``service.name`` resource attribute.
        capture_prompts: Whether to populate privacy-gated attributes
            (prompt text, tool args, tool results).
        span_exporter: Optional pre-built exporter. Tests inject an
            ``InMemorySpanExporter`` here. When provided, the OTLP gRPC
            exporter is NOT constructed (so unit tests do not require
            ``opentelemetry-exporter-otlp-proto-grpc``).
        use_batch_processor: When ``True`` (default in production), spans
            are flushed via a ``BatchSpanProcessor``. Tests pass
            ``False`` for deterministic span readback via a
            ``SimpleSpanProcessor``.
    """
    self._capture_prompts = capture_prompts
    self._service_name = service_name
    self._endpoint = endpoint or os.environ.get(
        "OTEL_EXPORTER_OTLP_ENDPOINT",
    )

    resource = Resource.create({SERVICE_NAME: service_name})
    self._provider = TracerProvider(resource=resource)

    if span_exporter is None:
        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
            OTLPSpanExporter,
        )

        exporter: SpanExporter = (
            OTLPSpanExporter(endpoint=self._endpoint)
            if self._endpoint
            else OTLPSpanExporter()
        )
        processor: Any = BatchSpanProcessor(exporter)
    else:
        exporter = span_exporter
        use_batch = (
            bool(use_batch_processor) if use_batch_processor is not None else False
        )
        processor = (
            BatchSpanProcessor(exporter)
            if use_batch
            else SimpleSpanProcessor(exporter)
        )

    self._provider.add_span_processor(processor)
    self._exporter: SpanExporter = exporter
    self._tracer = self._provider.get_tracer(_TRACER_NAME)
    self._recorder = SpanRecorder(self._tracer, capture_prompts=capture_prompts)

capture_prompts property

capture_prompts: bool

True when privacy-gated attributes should be populated.

provider property

provider: TracerProvider

Direct access for tests that need to flush spans synchronously.

recorder property

recorder: SpanRecorder

Per-event span helpers used by the two bridges.

attach_for_background

attach_for_background() -> otel_context.Context | None

Snapshot the current OTEL Context for background-task re-attach.

Source code in src/symfonic/observability/otel/tracer.py

def attach_for_background(self) -> otel_context.Context | None:
    """Snapshot the current OTEL Context for background-task re-attach."""
    try:
        return otel_context.get_current()
    except Exception:
        return None

force_flush

force_flush(*, timeout_millis: int = 5000) -> bool

Force-flush all processors. Used by tests for deterministic readback.

Source code in src/symfonic/observability/otel/tracer.py

def force_flush(self, *, timeout_millis: int = 5_000) -> bool:
    """Force-flush all processors. Used by tests for deterministic readback."""
    try:
        return bool(self._provider.force_flush(timeout_millis))
    except Exception:
        logger.warning("OTel tracer force_flush raised", exc_info=True)
        return False

shutdown

shutdown(*, timeout_millis: int = 5000) -> None

Flush pending spans and tear down the TracerProvider.

Closes any interrupt spans still open at shutdown (orphaned by a process exit before the matching resolve/expire). Without this sweep the leaked spans would never reach the exporter.

Source code in src/symfonic/observability/otel/tracer.py

def shutdown(self, *, timeout_millis: int = 5_000) -> None:
    """Flush pending spans and tear down the TracerProvider.

    Closes any interrupt spans still open at shutdown (orphaned by
    a process exit before the matching resolve/expire). Without
    this sweep the leaked spans would never reach the exporter.
    """
    try:
        self._recorder.end_all_open_interrupts()
    except Exception:
        logger.warning(
            "OTel tracer end_all_open_interrupts raised", exc_info=True,
        )
    try:
        self._provider.shutdown()
    except Exception:
        logger.warning("OTel tracer shutdown raised", exc_info=True)

start_run_span

start_run_span(
    *,
    run_id: str,
    tenant_id: str | None,
    session_id: str | None,
    query: str,
    entry_point: str,
) -> Iterator[Span]

Open the symfonic.run root span for the duration of a run().

The context manager sets the OTEL Context so child spans nest underneath, and stashes a RunSpanContext on the symfonic-side contextvar so emit-site helpers can populate identity attributes without threading state.

Source code in src/symfonic/observability/otel/tracer.py

@contextlib.contextmanager
def start_run_span(
    self,
    *,
    run_id: str,
    tenant_id: str | None,
    session_id: str | None,
    query: str,
    entry_point: str,
) -> Iterator[Span]:
    """Open the ``symfonic.run`` root span for the duration of a run().

    The context manager sets the OTEL ``Context`` so child spans
    nest underneath, and stashes a ``RunSpanContext`` on the
    symfonic-side contextvar so emit-site helpers can populate
    identity attributes without threading state.
    """
    attributes: dict[str, Any] = {
        SYMFONIC_RUN_ID: run_id,
        SYMFONIC_ENTRY_POINT: entry_point,
    }
    if tenant_id:
        attributes[SYMFONIC_TENANT_ID] = tenant_id
    if session_id:
        attributes[SYMFONIC_SESSION_ID] = session_id
    if query:
        attributes[SYMFONIC_QUERY_HASH] = hash_query(query)
        if self._capture_prompts:
            attributes[SYMFONIC_QUERY_TEXT] = truncate_for_attribute(query)

    run_ctx = RunSpanContext(
        run_id=run_id,
        tenant_id=tenant_id,
        session_id=session_id,
        entry_point=entry_point,
    )
    token = set_run_context(run_ctx)
    # v7.23.3 (Jarvio CORTEX tool-span gap): install the active
    # SpanRecorder on the contextvar that ``InstrumentedToolNode``
    # reads at dispatch time.  Without this, the engine's wire-up of
    # ``make_tool_node`` would see no recorder and fall through to
    # plain ``ToolNode`` semantics -- which is the gap this release
    # closes.  Reset paired in the outer finally so the contextvar
    # tracks the run span lifecycle exactly.
    from symfonic.observability.tool_span_seam import (
        reset_active_recorder,
        set_active_recorder,
    )
    recorder_token = set_active_recorder(self._recorder)
    try:
        with self._tracer.start_as_current_span(
            "symfonic.run",
            kind=SpanKind.INTERNAL,
            attributes=attributes,
        ) as span:
            try:
                yield span
            except Exception as exc:
                # v7.23.0 (2026-06-09 external bug report): narrowed
                # from ``BaseException`` to ``Exception`` so Python's
                # system-control exceptions (``GeneratorExit``,
                # ``KeyboardInterrupt``, ``SystemExit``) pass through
                # without span-level error attribution.  ``GeneratorExit``
                # in particular fires every time an async generator is
                # garbage-collected mid-execution -- a normal cleanup
                # signal, not a runtime error.  Pre-fix code marked
                # those as ERROR status, polluting trace dashboards
                # with false-positive failures on every interrupted /
                # cancelled run.
                span.set_status(Status(StatusCode.ERROR, type(exc).__name__))
                span.set_attribute(SYMFONIC_ERROR_TYPE, type(exc).__name__)
                span.set_attribute(
                    SYMFONIC_ERROR_MESSAGE,
                    truncate_for_attribute(str(exc)),
                )
                raise
    finally:
        reset_run_context(token)
        # v7.23.3 — pair with ``set_active_recorder`` above.
        reset_active_recorder(recorder_token)