Skip to content

symfonic.platform.telemetry_sink

telemetry_sink

Run-scoped observability binding for the kernel event fan-out.

TelemetryEventSink 

TelemetryEventSink(collector: Any, sinks: tuple[Any, ...], tenant: str)

Deliver events and bind safe attribution before the first one.

Source code in src/symfonic/platform/telemetry_sink.py 
16
17
18
19
def __init__(self, collector: Any, sinks: tuple[Any, ...], tenant: str) -> None:
    self._collector = collector
    self._sinks = sinks
    self._tenant = tenant

admit_attempt 

admit_attempt(request: Any) -> None

Admit a complete tenant/session execution before any turn effects.

Source code in src/symfonic/platform/telemetry_sink.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def admit_attempt(self, request: Any) -> None:
    """Admit a complete tenant/session execution before any turn effects."""
    # A raw-ID artifact observer could mix simultaneous attempts. Reject
    # unsupported compositions before either admission or observer writes.
    for sink in self._sinks:
        for name, args, kwargs in (
            ("bind_run", (request.run_id,), dict(
                root_run_id=request.root_run_id, parent_run_id=request.parent_run_id,
                tenant_id=self._tenant, session_id=request.session_id,
                attempt_id=request.attempt_id)),
            ("capture_model_round", (request.run_id, None, ""),
             {"attempt_id": request.attempt_id}),
        ):
            method = getattr(sink, name, None)
            if callable(method):
                try:
                    inspect.signature(method).bind(*args, **kwargs)
                except (TypeError, ValueError) as exc:
                    raise TypeError(f"native {name} requires attempt_id support") from exc
        if callable(getattr(sink, "bind_run", None)) and not callable(
            getattr(sink, "observe_run", None)
        ):
            abort = getattr(sink, "abort_attempt", None)
            if not callable(abort):
                raise TypeError("native identity binder requires abort_attempt rollback")
            inspect.signature(abort).bind(request.attempt_id)
    admit = getattr(self._collector, "admit_attempt", None)
    abort_collector = getattr(self._collector, "abort_attempt", None)
    if not callable(admit) or not callable(abort_collector):
        raise TypeError("native telemetry collector must support attempt admission")
    rollback = []
    try:
        for sink in self._sinks:
            if callable(getattr(sink, "observe_run", None)):
                continue
            bind = getattr(sink, "bind_run", None)
            if callable(bind):
                # Register first: even a binder that mutates then raises
                # must release its partially admitted execution.
                rollback.append(sink.abort_attempt)
                bind(request.run_id, root_run_id=request.root_run_id,
                     parent_run_id=request.parent_run_id, tenant_id=self._tenant,
                     session_id=request.session_id, attempt_id=request.attempt_id)
        admit(request.attempt_id, run_id=request.run_id, tenant_id=self._tenant,
              session_id=request.session_id, root_run_id=request.root_run_id,
              parent_run_id=request.parent_run_id)
        rollback.append(abort_collector)
        for sink in self._sinks:
            observe = getattr(sink, "observe_run", None)
            if callable(observe):
                try:
                    observe(request.run_id, root_run_id=request.root_run_id,
                            parent_run_id=request.parent_run_id, tenant_id=self._tenant,
                            session_id=request.session_id, attempt_id=request.attempt_id)
                except Exception:  # noqa: BLE001 - best-effort observation only
                    continue
    except BaseException as failure:
        errors = []
        for abort in reversed(rollback):
            try:
                abort(request.attempt_id)
            except BaseException as cleanup_error:
                errors.append(cleanup_error)
        if errors:
            raise BaseExceptionGroup(
                "native admission rollback failed", [failure, *errors]
            ) from None
        raise

bind_run 

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

Bind only durable identifiers and attribution, never turn content.

Every binder is attempted; the first error is re-raised afterwards so one faulty binder cannot hide the run from the sinks after it.

Source code in src/symfonic/platform/telemetry_sink.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def bind_run(
    self,
    run_id: str,
    *,
    root_run_id: str,
    parent_run_id: str | None,
    session_id: str = "",
) -> None:
    """Bind only durable identifiers and attribution, never turn content.

    Every binder is attempted; the first error is re-raised afterwards so
    one faulty binder cannot hide the run from the sinks after it.
    """
    first: Exception | None = None
    steps: list[Any] = []
    lineage = getattr(self._collector, "set_lineage", None)
    if callable(lineage):
        steps.append(lambda: lineage(
            run_id, root_run_id=root_run_id, parent_run_id=parent_run_id))
    tenant = getattr(self._collector, "set_tenant", None)
    if self._tenant and callable(tenant):
        steps.append(lambda: tenant(run_id, self._tenant))
    conversation = getattr(self._collector, "set_conversation", None)
    if callable(conversation):
        steps.append(lambda: conversation(run_id, session_id or run_id))
    for sink in self._sinks:
        bind = getattr(sink, "bind_run", None)
        if callable(bind):
            steps.append(lambda bind=bind: self._bind_sink(
                bind, run_id, root_run_id, parent_run_id, session_id))
    for step in steps:
        try:
            step()
        except Exception as exc:  # noqa: BLE001 - isolate binders, re-raise first
            first = first or exc
    if first is not None:
        raise first