parlot.instrumentation.pipecat

 25def parlotize(
 26    agent_id: str,
 27    *,
 28    endpoint: Optional[str] = None,
 29    api_key: Optional[str] = None,
 30    capture_genai_content: Optional[bool] = None,
 31    service_name: Optional[str] = None,
 32    tracer_provider: TracerProvider | None = None,
 33    version: Optional[str] = None,
 34    capture_logs: bool | list[str] | None = None,
 35    log_level: Optional[str] = None,
 36    record: bool | list[str] | None = None,
 37) -> ParlotContext:
 38    """Parlotize Pipecat instrumentation and OTLP export.
 39
 40    Call before constructing ``Pipeline`` / ``PipelineWorker``. Builds a
 41    ``TracerProvider`` with an OTLP exporter (unless one is adopted), registers
 42    it as the global provider (so Pipecat's ``enable_tracing=True`` path uses
 43    it), and patches ``PipelineWorker`` to default metrics + turn tracing on.
 44
 45    Do **not** call Pipecat's ``setup_tracing()`` afterward — it replaces the
 46    global provider and would drop Parlot processors. ``parlotize`` is the
 47    Parlot equivalent of that setup.
 48
 49    Shared parameters match every adapter — see ``parlot.core.ParlotizeProtocol``.
 50
 51    Args:
 52        agent_id: Shared — required canonical ``session.agent_id``.
 53        endpoint: Shared — Parlot OTLP base URL (or ``PARLOT_ENDPOINT``).
 54        api_key: Shared — org API key (or ``PARLOT_API_KEY``).
 55        capture_genai_content: Shared — GenAI payload capture override.
 56        service_name: Shared — OTel ``service.name`` (defaults to ``agent_id``).
 57        tracer_provider: Shared — existing ``TracerProvider``, or build one.
 58        version: Shared — ``gen_ai.agent.version``.
 59        capture_logs: Shared — session log capture.
 60        log_level: Shared — minimum level for session log capture.
 61        record: Audio recording policy. Boolean or agent-id glob patterns.
 62            When enabled, place ``ParlotRecordingProcessor`` after
 63            ``transport.output()`` (WAV segments by default).
 64
 65    Returns:
 66        The ``ParlotContext`` for this process.
 67    """
 68    global _configured, _configured_agent_id, _configured_agent_version
 69    global _configured_record, _parlot_context
 70    if _configured:
 71        logger.debug("parlot-instrumentation.pipecat already configured — skipping")
 72        assert _parlot_context is not None
 73        return _parlot_context
 74
 75    from parlot.core import base_parlotize
 76    from parlot.core.parlotize import normalize_glob_or_bool
 77    from parlot.core.sdk_version import resolve_parlot_sdk_version
 78
 79    _configured_agent_version = (version or "").strip() or "unknown"
 80    _configured_record = normalize_glob_or_bool(record)
 81
 82    res = base_parlotize(
 83        agent_id,
 84        endpoint=endpoint,
 85        api_key=api_key,
 86        capture_genai_content=capture_genai_content,
 87        tracer_provider=tracer_provider,
 88        version=_configured_agent_version,
 89        capture_logs=capture_logs,
 90        log_level=log_level,
 91    )
 92    _parlot_context = res.context
 93    _configured_agent_id = res.agent_id
 94
 95    tracer_provider = res.tracer_provider
 96    if tracer_provider is None:
 97        if not res.endpoint:
 98            raise ValueError(
 99                "No OTLP endpoint configured. Pass endpoint= or set the "
100                "PARLOT_ENDPOINT environment variable."
101            )
102        tracer_provider = _build_provider(
103            endpoint=res.endpoint,
104            api_key=res.api_key,
105            service_name=service_name or res.agent_id,
106            service_version=_configured_agent_version or None,
107            context=res.context,
108        )
109    else:
110        # Adopt existing provider: still attach our processor when possible.
111        _attach_processor_if_needed(tracer_provider, res.context)
112
113    from parlot.core.provider import try_set_global_tracer_provider
114    from parlot.core.parlotize import log_parlotize_ready
115
116    try_set_global_tracer_provider(tracer_provider)
117
118    from ._hooks import install_pipeline_worker_hooks, install_tool_call_hooks
119    from ._session import set_identity, set_tracer
120
121    set_identity(_configured_agent_id, _configured_agent_version)
122    set_tracer(
123        tracer_provider.get_tracer(
124            "parlot.instrumentation.pipecat",
125            resolve_parlot_sdk_version() or None,
126        )
127    )
128    _wire_session_logs(res.context, _configured_agent_id)
129    install_pipeline_worker_hooks()
130    install_tool_call_hooks()
131    _neutralize_pipecat_setup_tracing()
132
133    _configured = True
134    log_parlotize_ready(
135        "pipecat",
136        agent_id=res.agent_id,
137        endpoint=res.endpoint or "",
138        api_key=res.api_key or "",
139        log=logger,
140    )
141    return res.context

365def close_session(conversation_id: str, *, reason: str = "completed") -> None:
366    """Public helper to close a Pipecat-owned session by conversation_id."""
367    from ._session import close_session as _close
368
369    # Best-effort sync bridge: schedule finish if a loop is running.
370    try:
371        loop = asyncio.get_running_loop()
372        loop.create_task(finish_active_recordings())
373    except RuntimeError:
374        try:
375            asyncio.run(finish_active_recordings())
376        except Exception:
377            logger.debug("parlot: sync recording finish skipped", exc_info=True)
378
379    _close(conversation_id, reason=reason)

 99class PipecatGenAIProcessor(ParlotBaseProcessor):
100    """Enrich and remap Pipecat OTEL spans for Parlot export."""
101
102    def __init__(self, *, context: ParlotContext | None = None) -> None:
103        self._context = context
104        # conversation_id → (start_ns by turn_number) for duration-preserving remint
105        self._turn_starts: dict[str, dict[int, int]] = {}
106        # Live parent spans (turn/conversation) so children can read attrs set
107        # after start_span (Pipecat sets conversation.id / turn.number post-start).
108        self._open_spans: dict[str, Any] = {}
109        # conversation_id → turn_number → utterance + timing/latency buffers
110        self._utterances: dict[str, dict[int, dict[str, Any]]] = {}
111        self._active_conversation_id: str = ""
112        self._active_turn_number: int | None = None
113
114    def _bucket(self, cid: str, turn_n: int) -> dict[str, Any]:
115        return self._utterances.setdefault(cid, {}).setdefault(turn_n, {})
116
117    @staticmethod
118    def _debug_log_span(name: str) -> bool:
119        """Skip FastAPI/httpx noise when PARLOT_DEBUG_LEVEL=DEBUG."""
120        n = (name or "").lower()
121        if not n:
122            return False
123        if n in (NATIVE_CONVERSATION, NATIVE_TURN) or n.startswith("parlot."):
124            return True
125        if n in ("stt", "tts", "llm") or n.startswith(
126            ("stt", "tts", "llm", "chat", "execute_tool")
127        ):
128            return True
129        return False
130
131    def on_start(self, span, parent_context=None) -> None:  # noqa: ANN001
132        # SDK Span (not API Span) — leave untyped like LiveKit processor.
133        name = getattr(span, "name", None) or ""
134        if self._debug_log_span(name):
135            super().on_start(span, parent_context)
136        sid = _span_id_hex(span)
137        if sid and name in (NATIVE_CONVERSATION, NATIVE_TURN):
138            self._open_spans[sid] = span
139        # Activate session as early as possible so INFO logs correlate.
140        if name == NATIVE_TURN:
141            parent_attrs = self._parent_attrs(span)
142            cid = str(
143                (getattr(span, "attributes", None) or {}).get(ATTR_PIPECAT_CONVERSATION_ID)
144                or parent_attrs.get(ATTR_PIPECAT_CONVERSATION_ID)
145                or self._active_conversation_id
146                or ""
147            ).strip()
148            if cid:
149                self._active_conversation_id = cid
150                ensure_session(cid)
151
152    def on_end(self, span: ReadableSpan) -> None:
153        name = span.name or ""
154        if self._debug_log_span(name):
155            super().on_end(span)
156        attrs = span.attributes or {}
157        sid = ""
158        try:
159            ctx = span.get_span_context()
160            if ctx is not None:
161                sid = format(int(ctx.span_id), "016x")
162        except Exception:
163            sid = ""
164        if sid:
165            self._open_spans.pop(sid, None)
166
167        # Passthrough contract / GenAI spans we mint (emit_turn / ensure_session).
168        if name in CONTRACT_SPAN_NAMES or is_genai_span_name(name):
169            return
170
171        if name == NATIVE_CONVERSATION:
172            cid = str(attrs.get(ATTR_PIPECAT_CONVERSATION_ID) or "").strip()
173            if cid:
174                self._active_conversation_id = cid
175                ensure_session(cid)
176                span._name = "__parlot_drop_conversation__"
177                close_session(cid, reason="completed")
178                self._utterances.pop(cid, None)
179                self._turn_starts.pop(cid, None)
180            return
181
182        if name == NATIVE_TURN:
183            self._note_turn_context(attrs, span)
184            self._mint_contract_turn(span)
185            span._name = "__parlot_drop_turn__"
186            return
187
188        exported = remap_pipecat_span_name(name, attrs)
189        if exported is None:
190            span._name = f"__parlot_drop_{name}__"
191            return
192
193        apply_pipecat_span_rename(span)
194        self._stamp_session_linkage(span)
195        self._stamp_agent_role(span)
196        self._map_pipecat_metrics(span)
197        self._buffer_span_timing(span)
198        self._map_transcript_text(span)
199        self._record_usage_metrics(span)
200        # Orphan / tool-continuation TTS: mint agent before the next user STT
201        # arrives so indices stay chronological (user N, agent N+1).
202        if self._is_tts(name):
203            self._maybe_emit_orphan_agent(span)
204
205    def _note_turn_context(
206        self, attrs: Mapping[str, AttributeValue], span: ReadableSpan
207    ) -> None:
208        cid = str(attrs.get(ATTR_PIPECAT_CONVERSATION_ID) or "").strip()
209        turn_n = self._turn_number(attrs)
210        if cid:
211            self._active_conversation_id = cid
212            ensure_session(cid)
213        if turn_n is not None:
214            self._active_turn_number = turn_n
215            if cid:
216                starts = self._turn_starts.setdefault(cid, {})
217                if turn_n not in starts and span.start_time:
218                    starts[turn_n] = int(span.start_time)
219
220    def _parent_attrs(self, span: ReadableSpan) -> Mapping[str, AttributeValue]:
221        parent_id = _parent_span_id_hex(span)
222        if not parent_id:
223            return {}
224        parent = self._open_spans.get(parent_id)
225        if parent is None:
226            return {}
227        return getattr(parent, "attributes", None) or {}
228
229    def _resolve_conversation_and_turn(
230        self, span: ReadableSpan
231    ) -> tuple[str, int | None]:
232        attrs = span.attributes or {}
233        parent_attrs = self._parent_attrs(span)
234        cid = str(
235            attrs.get(ATTR_PIPECAT_CONVERSATION_ID)
236            or parent_attrs.get(ATTR_PIPECAT_CONVERSATION_ID)
237            or self._active_conversation_id
238            or ""
239        ).strip()
240        if not cid:
241            state = _pipecat_state(get_active_session())
242            if state and state.conversation_id:
243                cid = state.conversation_id
244        turn_n = self._turn_number(attrs) or self._turn_number(parent_attrs)
245        if turn_n is None:
246            turn_n = self._active_turn_number
247        if turn_n is None:
248            state = lookup_session(cid) if cid else _pipecat_state(get_active_session())
249            if state and state.open_turn_index is not None:
250                turn_n = state.open_turn_index
251        return cid, turn_n
252
253    def _conversation_id_from_context(
254        self, attrs: Mapping[str, AttributeValue]
255    ) -> str:
256        cid = str(attrs.get(ATTR_PIPECAT_CONVERSATION_ID) or "").strip()
257        if cid:
258            return cid
259        if self._active_conversation_id:
260            return self._active_conversation_id
261        state = _pipecat_state(get_active_session())
262        if state and state.conversation_id:
263            return state.conversation_id
264        return ""
265
266    @staticmethod
267    def _turn_number(attrs: Mapping[str, AttributeValue]) -> int | None:
268        raw = attrs.get(ATTR_PIPECAT_TURN_NUMBER)
269        if raw is None or isinstance(raw, (list, tuple)):
270            return None
271        try:
272            return int(raw)  # type: ignore[arg-type]
273        except (TypeError, ValueError):
274            return None
275
276    def _find_agent_bucket(
277        self, cid: str, prefer_turn: int | None = None
278    ) -> tuple[int | None, dict[str, Any]]:
279        """Locate a buffer that still holds unemitted agent text."""
280        buckets = self._utterances.get(cid) or {}
281        if prefer_turn is not None and prefer_turn in buckets:
282            b = buckets[prefer_turn]
283            if str(b.get("agent") or "").strip() and not b.get("agent_emitted"):
284                return prefer_turn, b
285        for key, b in list(buckets.items()):
286            if str(b.get("agent") or "").strip() and not b.get("agent_emitted"):
287                return key if isinstance(key, int) else None, b
288        return prefer_turn, {}
289
290    def _flush_pending_agent_only(
291        self,
292        state: _PipecatSessionState,
293        *,
294        cid: str,
295        prefer_turn: int | None = None,
296        interrupted: bool = False,
297        e2e: float | None = None,
298    ) -> bool:
299        """Emit a reserved agent-only turn (tool continuation) before a new user.
300
301        Without this, STT allocates ``user = open_agent + 1`` while keeping the
302        old agent index — timeline then sorts user (earlier STT) before agent
303        (later LLM) despite a lower agent turn index.
304        """
305        if state.open_agent_turn_index is None:
306            return False
307        if state.open_user_turn_index is not None:
308            return False
309        agent_idx = state.open_agent_turn_index
310        _key, buffered = self._find_agent_bucket(cid, prefer_turn)
311        agent_text = str(buffered.get("agent") or "").strip()
312        if not agent_text:
313            # Indices reserved by LLM/TTS with no text yet — drop reservation.
314            clear_open_turn_indices(state)
315            return False
316        agent_start = buffered.get("agent_start_ns")
317        agent_end = buffered.get("agent_end_ns")
318        llm_ttft = buffered.get("llm_ttft_s")
319        tts_ttfb = buffered.get("tts_ttfb_s")
320        emit_turn(
321            state,
322            turn_index=agent_idx,
323            role="agent",
324            interrupted=interrupted,
325            e2e_latency_s=e2e if e2e is not None else _as_float(buffered.get("e2e")),
326            llm_ttft_s=float(llm_ttft) if llm_ttft is not None else None,
327            tts_ttfb_s=float(tts_ttfb) if tts_ttfb is not None else None,
328            agent_text=agent_text,
329            start_time_ns=int(agent_start) if agent_start else None,
330            end_time_ns=int(agent_end) if agent_end else None,
331        )
332        buffered["agent_emitted"] = True
333        buffered["agent"] = ""
334        for key in (
335            "agent_start_ns",
336            "agent_end_ns",
337            "llm_ttft_s",
338            "tts_ttfb_s",
339            "e2e",
340        ):
341            buffered.pop(key, None)
342        clear_open_turn_indices(state)
343        return True
344
345    def _emit_user_from_stt(
346        self,
347        state: _PipecatSessionState,
348        *,
349        cid: str,
350        turn_n: int | None,
351        text: str,
352        start_ns: int | None,
353        end_ns: int | None,
354    ) -> None:
355        """Mint the user turn as soon as STT finals (LiveKit-style ordering)."""
356        self._flush_pending_agent_only(state, cid=cid, prefer_turn=turn_n)
357        if state.user_turn_emitted and state.open_user_turn_index is not None:
358            return
359        user_idx = allocate_user_turn_index(state)
360        allocate_agent_turn_index(state)  # reserve agent slot after user
361        emit_turn(
362            state,
363            turn_index=user_idx,
364            role="user",
365            user_text=text,
366            start_time_ns=start_ns,
367            end_time_ns=end_ns,
368        )
369        state.user_turn_emitted = True
370        if turn_n is not None:
371            self._bucket(cid, turn_n)["user_emitted"] = True
372
373    def _native_turn_open(self, cid: str) -> bool:
374        for sp in self._open_spans.values():
375            if (getattr(sp, "name", None) or "") != NATIVE_TURN:
376                continue
377            attrs = getattr(sp, "attributes", None) or {}
378            sp_cid = str(attrs.get(ATTR_PIPECAT_CONVERSATION_ID) or "").strip()
379            if sp_cid == cid or (not sp_cid and self._active_conversation_id == cid):
380                return True
381        return False
382
383    def _maybe_emit_orphan_agent(self, span: ReadableSpan) -> None:
384        """Mint agent speech that lands after its native turn already closed.
385
386        Late tool-follow-up TTS otherwise buffers onto the *next* turn and gets
387        paired with the next user STT (inverted indices / timeline order).
388        """
389        cid, turn_n = self._resolve_conversation_and_turn(span)
390        state = lookup_session(cid) if cid else _pipecat_state(get_active_session())
391        if state is None or not cid:
392            return
393        if state.open_user_turn_index is not None:
394            return
395        if self._native_turn_open(cid):
396            return
397        bucket = self._bucket(cid, turn_n) if turn_n is not None else {}
398        agent_text = str(bucket.get("agent") or "").strip()
399        if not agent_text or bucket.get("agent_emitted"):
400            return
401        if state.open_agent_turn_index is None:
402            allocate_agent_turn_index(state)
403        self._flush_pending_agent_only(state, cid=cid, prefer_turn=turn_n)
404
405    def _mint_contract_turn(self, span: ReadableSpan) -> None:
406        attrs = span.attributes or {}
407        cid = str(attrs.get(ATTR_PIPECAT_CONVERSATION_ID) or "").strip()
408        state = lookup_session(cid) if cid else _pipecat_state(get_active_session())
409        if state is None or not state.conversation_id:
410            if not cid:
411                return
412            state = ensure_session(cid)
413        if state is None:
414            return
415        pipecat_turn = self._turn_number(attrs)
416        if pipecat_turn is not None:
417            self._active_turn_number = pipecat_turn
418        self._active_conversation_id = state.conversation_id
419        interrupted = bool(attrs.get(ATTR_PIPECAT_TURN_INTERRUPTED))
420        e2e = _as_float(attrs.get(ATTR_PIPECAT_TURN_USER_BOT_LATENCY))
421        start_ns = None
422        starts = self._turn_starts.get(state.conversation_id) or {}
423        if pipecat_turn is not None and pipecat_turn in starts:
424            start_ns = starts.pop(pipecat_turn, None)
425        end_ns = span.end_time
426        buf_key = pipecat_turn if pipecat_turn is not None else -1
427        # Flush any prior agent-only reservation *before* popping this turn's
428        # bucket (flush reads from ``_utterances``).
429        if state.open_user_turn_index is None and state.open_agent_turn_index is not None:
430            # Peek whether this turn has user text; if not, we'll emit agent below.
431            peek = (self._utterances.get(state.conversation_id) or {}).get(buf_key) or {}
432            peek_user = str(peek.get("user") or "").strip()
433            if peek_user:
434                self._flush_pending_agent_only(
435                    state, cid=state.conversation_id, prefer_turn=None
436                )
437
438        buffered = (self._utterances.get(state.conversation_id) or {}).pop(buf_key, {})
439        user_text = str(buffered.get("user") or "").strip()
440        agent_text = str(buffered.get("agent") or "").strip()
441        user_already = bool(buffered.get("user_emitted") or state.user_turn_emitted)
442        agent_already = bool(buffered.get("agent_emitted"))
443        stt_start = buffered.get("stt_start_ns")
444        stt_end = buffered.get("stt_end_ns")
445        agent_start = buffered.get("agent_start_ns")
446        agent_end = buffered.get("agent_end_ns")
447        llm_ttft = buffered.get("llm_ttft_s")
448        tts_ttfb = buffered.get("tts_ttfb_s")
449
450        user_idx = state.open_user_turn_index
451        agent_idx = state.open_agent_turn_index
452        if user_text and not user_already and user_idx is None:
453            user_idx = allocate_user_turn_index(state)
454            agent_idx = allocate_agent_turn_index(state)
455        if agent_text and not agent_already and agent_idx is None:
456            agent_idx = allocate_agent_turn_index(state)
457
458        # Split wall clocks: shared native-turn bounds made user+agent share
459        # the same offset (inflated "gaps"). Prefer STT / LLM+TTS windows.
460        user_start = int(stt_start) if stt_start else start_ns
461        user_end = int(stt_end) if stt_end else (
462            int(agent_start) if agent_start else end_ns
463        )
464        agent_start_ns = int(agent_start) if agent_start else (
465            int(stt_end) if stt_end else start_ns
466        )
467        agent_end_ns = int(agent_end) if agent_end else end_ns
468
469        if user_text and not user_already and user_idx is not None:
470            emit_turn(
471                state,
472                turn_index=user_idx,
473                role="user",
474                user_text=user_text,
475                start_time_ns=user_start,
476                end_time_ns=user_end,
477            )
478            state.user_turn_emitted = True
479        if agent_text and not agent_already and agent_idx is not None:
480            emit_turn(
481                state,
482                turn_index=agent_idx,
483                role="agent",
484                interrupted=interrupted,
485                e2e_latency_s=e2e,
486                llm_ttft_s=float(llm_ttft) if llm_ttft is not None else None,
487                tts_ttfb_s=float(tts_ttfb) if tts_ttfb is not None else None,
488                agent_text=agent_text,
489                start_time_ns=agent_start_ns,
490                end_time_ns=agent_end_ns,
491            )
492        clear_open_turn_indices(state)
493
494    @staticmethod
495    def _is_stt(name: str) -> bool:
496        return name == "stt" or name.startswith("stt")
497
498    @staticmethod
499    def _is_tts(name: str) -> bool:
500        return name == "tts" or name.startswith("tts")
501
502    @staticmethod
503    def _is_llm(name: str) -> bool:
504        # Native ``llm`` before rename, or GenAI ``chat`` / ``chat {model}`` after.
505        if name.startswith("llm_tool"):
506            return False
507        return (
508            name == "llm"
509            or name.startswith("llm")
510            or name == "chat"
511            or name.startswith("chat")
512        )
513
514    @staticmethod
515    def _is_tool(name: str) -> bool:
516        return name.startswith("execute_tool") or name.startswith("llm_tool")
517
518    def _parlot_turn_index_for_span(
519        self, span: ReadableSpan, state: _PipecatSessionState
520    ) -> int | None:
521        """Map pipeline spans onto Parlot turn indices for timeline nesting.
522
523        The app nests ``stt`` / ``llm`` / ``tts`` under **agent** turns only.
524        STT at a user-only turn_index shows as an orphan row (odd indentation).
525        So we allocate the user index for the utterance row, then stamp STT
526        (and LLM/TTS) on the paired agent index.
527        """
528        name = span.name or ""
529        if self._is_stt(name):
530            # Tool-continuation may have reserved an agent index with no user.
531            # Flush it first so this STT becomes user N / agent N+1 (not N+1 / N).
532            self._flush_pending_agent_only(state, cid=state.conversation_id)
533            allocate_user_turn_index(state)
534            return allocate_agent_turn_index(state)
535        if self._is_tts(name) or self._is_llm(name) or self._is_tool(name):
536            return allocate_agent_turn_index(state)
537        return state.open_agent_turn_index or state.open_user_turn_index
538
539    def _stamp_session_linkage(self, span: ReadableSpan) -> None:
540        cid, pipecat_turn = self._resolve_conversation_and_turn(span)
541        state = lookup_session(cid) if cid else _pipecat_state(get_active_session())
542        if state is None and cid:
543            state = ensure_session(cid)
544        if state is None:
545            return
546        if cid:
547            self._active_conversation_id = cid
548        if pipecat_turn is not None:
549            self._active_turn_number = pipecat_turn
550        parlot_turn = self._parlot_turn_index_for_span(span, state)
551        self._set(span, ATTR_SESSION_ID, state.session_id)
552        self._set(span, ATTR_SESSION_CONVERSATION_ID, state.conversation_id)
553        if parlot_turn is not None:
554            self._set(span, ATTR_TURN_INDEX, parlot_turn)
555        agent_id = configured_agent_id()
556        if agent_id:
557            self._set(span, ATTR_GEN_AI_AGENT_ID, agent_id)
558            self._set(span, ATTR_GEN_AI_AGENT_NAME, agent_id)
559            self._set(span, ATTR_SESSION_AGENT_ID, agent_id)
560        stamp_pipecat_platform_refs(span, conversation_id=state.conversation_id)
561
562    def _stamp_agent_role(self, span: ReadableSpan) -> None:
563        name = span.name or ""
564        attrs = span.attributes or {}
565        if attrs.get(ATTR_AGENT_ROLE):
566            return
567        if self._is_stt(name):
568            self._set(span, ATTR_AGENT_ROLE, "stt")
569        elif self._is_tts(name):
570            self._set(span, ATTR_AGENT_ROLE, "tts")
571        elif self._is_llm(name):
572            self._set(span, ATTR_AGENT_ROLE, "llm")
573        elif self._is_tool(name):
574            self._set(span, ATTR_AGENT_ROLE, "tool")
575
576    def _map_pipecat_metrics(self, span: ReadableSpan) -> None:
577        attrs = span.attributes or {}
578        name = span.name or ""
579        ttfb = _as_float(attrs.get(ATTR_PIPECAT_METRICS_TTFB))
580        # Only TTS owns gen_ai.tts.ttfb_s — do not stamp LLM TTFB there.
581        if (
582            ttfb is not None
583            and self._is_tts(name)
584            and not attrs.get(ATTR_GEN_AI_TTS_TTFB_S)
585        ):
586            self._set(span, ATTR_GEN_AI_TTS_TTFB_S, ttfb)
587        if not attrs.get(ATTR_GEN_AI_MODEL):
588            model = attrs.get("gen_ai.request.model") or attrs.get("model")
589            if model is not None and not isinstance(model, (list, tuple)):
590                self._set(span, ATTR_GEN_AI_MODEL, str(model))
591
592    def _buffer_span_timing(self, span: ReadableSpan) -> None:
593        """Record STT/LLM/TTS windows + TTFB for reminted turn bounds / latency."""
594        cid, turn_n = self._resolve_conversation_and_turn(span)
595        if not cid or turn_n is None:
596            return
597        name = span.name or ""
598        bucket = self._bucket(cid, turn_n)
599        start = span.start_time
600        end = span.end_time
601        ttfb = _as_float((span.attributes or {}).get(ATTR_PIPECAT_METRICS_TTFB))
602        if self._is_stt(name):
603            if start:
604                bucket["stt_start_ns"] = int(start)
605            if end:
606                bucket["stt_end_ns"] = int(end)
607            return
608        if self._is_llm(name) or self._is_tts(name):
609            # After an agent-only flush, the next LLM/TTS is a new utterance —
610            # do not keep the previous window via min()/max().
611            if bucket.get("agent_emitted"):
612                bucket["agent_start_ns"] = int(start) if start else None
613                bucket["agent_end_ns"] = int(end) if end else None
614                bucket.pop("llm_ttft_s", None)
615                bucket.pop("tts_ttfb_s", None)
616            else:
617                if start:
618                    prev = bucket.get("agent_start_ns")
619                    bucket["agent_start_ns"] = (
620                        min(int(prev), int(start)) if prev else int(start)
621                    )
622                if end:
623                    prev_end = bucket.get("agent_end_ns")
624                    bucket["agent_end_ns"] = (
625                        max(int(prev_end), int(end)) if prev_end else int(end)
626                    )
627            if ttfb is not None:
628                if self._is_llm(name):
629                    bucket["llm_ttft_s"] = ttfb
630                elif bucket.get("tts_ttfb_s") is None:
631                    # Prefer first successful TTFB (retry after "no audio" is noisier).
632                    bucket["tts_ttfb_s"] = ttfb
633
634    def _record_usage_metrics(self, span: ReadableSpan) -> None:
635        recorder = get_metrics_recorder()
636        if recorder is None:
637            return
638        attrs = span.attributes or {}
639        session_id = str(attrs.get(ATTR_SESSION_ID) or "").strip()
640        if not session_id:
641            return
642        name = span.name or ""
643        model = str(attrs.get(ATTR_GEN_AI_MODEL) or attrs.get("gen_ai.request.model") or "")
644        provider = str(attrs.get("gen_ai.provider.name") or "")
645        if self._is_stt(name):
646            audio = _as_float(attrs.get(ATTR_PIPECAT_METRICS_AUDIO_SECONDS))
647            if audio:
648                recorder.record_stt_audio(
649                    session_id=session_id,
650                    audio_seconds=audio,
651                    model=model,
652                    provider=provider,
653                )
654        elif self._is_tts(name):
655            raw_chars = attrs.get(ATTR_PIPECAT_METRICS_CHAR_COUNT)
656            try:
657                chars = int(raw_chars) if raw_chars is not None else 0  # type: ignore[arg-type]
658            except (TypeError, ValueError):
659                chars = 0
660            if chars:
661                recorder.record_tts_characters(
662                    session_id=session_id,
663                    characters=chars,
664                    model=model,
665                    provider=provider,
666                )
667        elif self._is_llm(name):
668            try:
669                tin = int(attrs.get("gen_ai.usage.input_tokens") or 0)  # type: ignore[arg-type]
670            except (TypeError, ValueError):
671                tin = 0
672            try:
673                tout = int(attrs.get("gen_ai.usage.output_tokens") or 0)  # type: ignore[arg-type]
674            except (TypeError, ValueError):
675                tout = 0
676            if tin or tout:
677                recorder.record_llm_tokens(
678                    session_id=session_id,
679                    input_tokens=tin,
680                    output_tokens=tout,
681                    model=model,
682                    provider=provider,
683                )
684                # Keep session totals for session.* close histograms.
685                cid = str(
686                    attrs.get(ATTR_SESSION_CONVERSATION_ID)
687                    or attrs.get(ATTR_PIPECAT_CONVERSATION_ID)
688                    or ""
689                ).strip()
690                sess = lookup_session(cid) if cid else _pipecat_state(get_active_session())
691                if sess is not None and sess.session_id == session_id:
692                    if tin:
693                        sess.total_input_tokens += tin
694                    if tout:
695                        sess.total_output_tokens += tout
696
697    def _map_transcript_text(self, span: ReadableSpan) -> None:
698        attrs = span.attributes or {}
699        name = span.name or ""
700        # Prefer final transcripts; interim STT still updates the buffer but we
701        # only remint the user turn on finals (or when is_final is absent).
702        is_final = attrs.get("is_final")
703        transcript = (
704            attrs.get(ATTR_PIPECAT_TRANSCRIPT)
705            or attrs.get(ATTR_PIPECAT_TEXT)
706            or attrs.get("output")
707        )
708        if transcript is None or isinstance(transcript, (list, tuple)):
709            return
710        text = str(transcript).strip()
711        if not text:
712            return
713        cid, turn_n = self._resolve_conversation_and_turn(span)
714        if self._is_stt(name):
715            if not attrs.get(ATTR_TURN_USER_TEXT):
716                self._set(span, ATTR_TURN_USER_TEXT, text)
717            if not cid:
718                return
719            if turn_n is not None:
720                self._bucket(cid, turn_n)["user"] = text
721            # Interim results: keep buffer only.
722            if is_final is False:
723                return
724            state = lookup_session(cid) or _pipecat_state(get_active_session())
725            if state is None:
726                state = ensure_session(cid)
727            if state is None:
728                return
729            self._emit_user_from_stt(
730                state,
731                cid=cid,
732                turn_n=turn_n,
733                text=text,
734                start_ns=int(span.start_time) if span.start_time else None,
735                end_ns=int(span.end_time) if span.end_time else None,
736            )
737        elif self._is_tts(name) or self._is_llm(name):
738            if not attrs.get(ATTR_TURN_AGENT_TEXT):
739                self._set(span, ATTR_TURN_AGENT_TEXT, text)
740            if cid and turn_n is None:
741                return
742            bucket = self._bucket(cid, turn_n) if cid and turn_n is not None else None
743            if bucket is None:
744                return
745            if self._is_tts(name):
746                prev = str(bucket.get("agent") or "")
747                # Avoid "hello hello" when Pipecat retries / double-fires TTS.
748                if prev == text or prev.endswith(text):
749                    return
750                if text.startswith(prev) and prev:
751                    bucket["agent"] = text
752                else:
753                    bucket["agent"] = f"{prev} {text}".strip() if prev else text
754                # New speech after a prior flush (tool then follow-up answer).
755                bucket["agent_emitted"] = False
756            elif not bucket.get("agent"):
757                # LLM output as fallback when TTS never lands text.
758                bucket["agent"] = text
759                bucket["agent_emitted"] = False