Skip to content

symfonic.observability.otel.kernel_sink

kernel_sink

OpenTelemetry projection of the kernel's single ordered event stream.

KernelOTelSink 

KernelOTelSink(*, endpoint: str | None, service_name: str, policy: TracePolicy, span_exporter: SpanExporter | None = None)

Create a root turn span and an ordered child span for every event.

Source code in src/symfonic/observability/otel/kernel_sink.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(
    self,
    *,
    endpoint: str | None,
    service_name: str,
    policy: TracePolicy,
    span_exporter: SpanExporter | None = None,
) -> None:
    self._policy = policy
    provider = TracerProvider(resource=Resource.create({SERVICE_NAME: service_name}))
    if span_exporter is None:
        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
            OTLPSpanExporter,
        )

        span_exporter = OTLPSpanExporter(endpoint=endpoint) if endpoint else OTLPSpanExporter()
    provider.add_span_processor(BatchSpanProcessor(span_exporter))
    self._provider = provider
    self._tracer = provider.get_tracer("symfonic.kernel")
    self._roots: dict[str, Span] = {}
    self._contexts: dict[str, Any] = {}
    self._lineages: dict[str, tuple[str, str | None, str, str]] = {}
    self._raw_runs: dict[str, str] = {}

bind_run 

bind_run(run_id: str, *, root_run_id: str = '', parent_run_id: str | None = None, tenant_id: str = '', session_id: str = '', attempt_id: str = '') -> None

Add attribution once the telemetry stage knows the conversation.

Source code in src/symfonic/observability/otel/kernel_sink.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def bind_run(
    self,
    run_id: str,
    *,
    root_run_id: str = "",
    parent_run_id: str | None = None,
    tenant_id: str = "",
    session_id: str = "",
    attempt_id: str = "",
) -> None:
    """Add attribution once the telemetry stage knows the conversation."""
    raw_run_id = run_id
    run_id = attempt_id or run_id
    self._raw_runs[run_id] = raw_run_id
    self._lineages[run_id] = (root_run_id or raw_run_id, parent_run_id, tenant_id, session_id)
    root = self._roots.get(run_id)
    if root is None:
        return
    root.set_attribute("symfonic.root_run.id", root_run_id or raw_run_id)
    if parent_run_id:
        root.set_attribute("symfonic.parent_run.id", parent_run_id)
    if tenant_id:
        root.set_attribute("symfonic.tenant.id", tenant_id)
    if session_id:
        root.set_attribute("symfonic.conversation.id", session_id)

capture_model_round 

capture_model_round(run_id: str, transcript: Any, output: str, *, attempt_id: str = '') -> None

Attach policy-gated content to this execution's root.

Source code in src/symfonic/observability/otel/kernel_sink.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def capture_model_round(self, run_id: str, transcript: Any, output: str,
                        *, attempt_id: str = "") -> None:
    """Attach policy-gated content to this execution's root."""
    if not self._policy.captures_content:
        return
    root, parent = self._root_for(attempt_id or run_id)
    messages = self._messages_of(transcript)
    prompt_text = "\n".join(str(item.get("content", "")) for item in messages)
    payload = strip_trace_reasoning({
        "model_input": messages,
        "model_output": output,
        "memory_recall": _RECALL_BLOCK.findall(prompt_text),
    })
    with self._tracer.start_as_current_span(
        "symfonic.model.artifacts",
        context=parent,
        kind=SpanKind.INTERNAL,
        attributes={
            "symfonic.run.id": run_id,
            "symfonic.event.kind": "model_artifacts",
            "symfonic.payload": self._encoded_payload(payload),
        },
    ):
        pass

force_flush 

force_flush() -> bool

Flush spans for deterministic tests and controlled shutdowns.

Source code in src/symfonic/observability/otel/kernel_sink.py 
298
299
300
def force_flush(self) -> bool:
    """Flush spans for deterministic tests and controlled shutdowns."""
    return bool(self._provider.force_flush())