parlot.core

77@dataclass
78class BaseParlotizeResult:
79    """Resolved context from ``base_parlotize`` for framework adapters."""
80
81    context: ParlotContext
82    endpoint: str
83    api_key: str
84    tracer_provider: Any
85    agent_id: str
86    agent_version: str
87    capture_genai_content: Optional[bool]
88    capture_logs: bool | list[str] | None
89    log_level: Optional[str]

23@runtime_checkable
24class ParlotizeProtocol(Protocol):
25    """Common parameters every adapter ``parlotize()`` must accept.
26
27    Framework packages may add extra keyword-only parameters (e.g. LiveKit
28    ``record=`` / ``auto_escalate_sip=``).
29    """
30
31    def __call__(
32        self,
33        agent_id: str,
34        *,
35        endpoint: Optional[str] = None,
36        api_key: Optional[str] = None,
37        capture_genai_content: Optional[bool] = None,
38        service_name: Optional[str] = None,
39        tracer_provider: TracerProvider | None = None,
40        version: Optional[str] = None,
41        capture_logs: bool | list[str] | None = None,
42        log_level: Optional[str] = None,
43        **kwargs: Any,
44    ) -> ParlotContext:
45        """Shared parameter surface for every adapter ``parlotize()``.
46
47        Args:
48            agent_id: Required canonical deployment identity stamped on
49                ``session.agent_id``. Must be non-empty after stripping whitespace.
50            endpoint: Parlot OTLP collector base URL (e.g.
51                ``https://ingest.parlot.ai``). Spans export to
52                ``{endpoint}/v1/traces``. If omitted, reads ``PARLOT_ENDPOINT``.
53            api_key: Org-scoped API key minted in Parlot **Settings → API Keys**.
54                If omitted, reads ``PARLOT_API_KEY``. Required for remote telemetry
55                bootstrap and recording grants.
56            capture_genai_content: Process-wide override for LLM message bodies and
57                tool input/output payloads. If ``False``, payloads are omitted while
58                preserving span durations, tokens, and turn text. Precedence: job
59                metadata → this kwarg → Settings → Generative AI (default: on).
60            service_name: OpenTelemetry resource ``service.name``. Defaults to
61                ``agent_id`` or a framework-specific fallback.
62            tracer_provider: Existing OpenTelemetry ``TracerProvider`` to adopt. If
63                omitted, adapters build one with Parlot's OTLP exporter (or adopt an
64                already-registered provider when another adapter configured first).
65            version: Deployment version stamped on ``gen_ai.agent.version``.
66                Pass ``version=`` to set it; otherwise ``\"unknown\"``.
67            capture_logs: Intercept Python ``logging`` during active sessions and
68                stream to the session Logs tab. Boolean or agent-id glob patterns.
69                Precedence: job metadata → this kwarg → Settings → Logs (default: on).
70            log_level: Minimum level for session log capture (e.g. ``"INFO"``,
71                ``"WARNING"``). Defaults to ``"INFO"``.
72            **kwargs: Framework-specific options (ignored by the shared surface).
73        """
74        ...

13@dataclass
14class ParlotContext:
15    """Owns bootstrap runtime and process-local collectors for one SDK instance."""
16
17    runtime: Optional[ParlotRuntimeContext] = None
18    session_logs: SessionLogsCollector = field(default_factory=SessionLogsCollector)
19
20    def __post_init__(self) -> None:
21        self.session_logs.bind_context(self)
22
23    def shutdown(self) -> None:
24        self.session_logs.shutdown()
25        self.runtime = None

12@dataclass(frozen=True)
13class ParlotRuntimeContext:
14    endpoint: str
15    api_key: str
16    tenant_id: str
17    content_bucket: str
18    r2_endpoint: str
19    recording_globs: tuple[str, ...] = ()
20    recording_agents: tuple[tuple[str, bool], ...] = ()
21    logs_globs: tuple[str, ...] = ()
22    logs_agents: tuple[tuple[str, bool], ...] = ()
23    logs_agent_min_levels: tuple[tuple[str, str], ...] = ()
24    logs_min_level: str = DEFAULT_LOGS_MIN_LEVEL
25    # True when bootstrap JSON included a ``logs`` object (even empty).
26    logs_policy_present: bool = False
27    capture_genai_content_globs: tuple[str, ...] = ()
28    capture_genai_content_agents: tuple[tuple[str, bool], ...] = ()
29    # True when bootstrap JSON included a ``capture_genai_content`` object (even empty).
30    capture_genai_content_policy_present: bool = False
31
32    def recording_agents_map(self) -> dict[str, bool]:
33        return dict(self.recording_agents)
34
35    def logs_agents_map(self) -> dict[str, bool]:
36        return dict(self.logs_agents)
37
38    def logs_agent_min_levels_map(self) -> dict[str, str]:
39        return dict(self.logs_agent_min_levels)
40
41    def capture_genai_content_agents_map(self) -> dict[str, bool]:
42        return dict(self.capture_genai_content_agents)

241class SessionLogsCollector:
242    """Per-``ParlotContext`` session log capture pipeline."""
243
244    def __init__(self) -> None:
245        self._lock = threading.Lock()
246        self._buffer: deque[SessionLogRecord] = deque(maxlen=_MAX_BUFFER)
247        self._consecutive_failures = 0
248        self._circuit_open_until = 0.0
249        self._last_flush_at = 0.0
250        self._endpoint = ""
251        self._api_key = ""
252        self._flush_thread: threading.Thread | None = None
253        self._stop = threading.Event()
254        self._handler: Optional[SessionLogHandler] = None
255        self._installed = False
256        self._atexit_registered = False
257        self._capture_logs_config: CaptureLogsConfig = None
258        self._capture_logs_level: Optional[str] = None
259        self._session_resolver: Optional[SessionResolver] = None
260        self._agent_id_resolver: Optional[Callable[[], str]] = None
261        self._metadata_resolver: Optional[Callable[[], Optional[bool]]] = None
262        self._context_ref: Optional[weakref.ref[ParlotContext]] = None
263
264    def bind_context(self, context: ParlotContext) -> None:
265        self._context_ref = weakref.ref(context)
266
267    def _runtime(self):
268        ctx = self._context_ref() if self._context_ref is not None else None
269        return ctx.runtime if ctx is not None else None
270
271    def set_capture_logs_config(
272        self,
273        capture_logs: CaptureLogsConfig = None,
274        *,
275        log_level: Optional[str] = None,
276    ) -> None:
277        self._capture_logs_config = capture_logs
278        if log_level is not None:
279            self._capture_logs_level = normalize_log_level(log_level)
280
281    def set_resolvers(
282        self,
283        *,
284        session_resolver: Optional[SessionResolver] = None,
285        agent_id_resolver: Optional[Callable[[], str]] = None,
286        metadata_resolver: Optional[Callable[[], Optional[bool]]] = None,
287    ) -> None:
288        """Optional framework hooks (e.g. LiveKit job bootstrap fallback)."""
289        if session_resolver is not None:
290            self._session_resolver = session_resolver
291        if agent_id_resolver is not None:
292            self._agent_id_resolver = agent_id_resolver
293        if metadata_resolver is not None:
294            self._metadata_resolver = metadata_resolver
295
296    def _resolve_min_level(self) -> str:
297        if self._capture_logs_level:
298            return self._capture_logs_level
299        runtime = self._runtime()
300        if runtime is not None:
301            agent_id = ""
302            if self._agent_id_resolver is not None:
303                try:
304                    agent_id = self._agent_id_resolver() or ""
305                except Exception:
306                    agent_id = ""
307            if agent_id:
308                agent_levels = runtime.logs_agent_min_levels_map()
309                if agent_id in agent_levels:
310                    return normalize_log_level(agent_levels[agent_id])
311            if runtime.logs_min_level:
312                return normalize_log_level(runtime.logs_min_level)
313        return DEFAULT_LOGS_MIN_LEVEL
314
315    def capture_logs_enabled_for_agent(self, agent_name: str = "") -> bool:
316        runtime = self._runtime()
317        metadata = None
318        if self._metadata_resolver is not None:
319            try:
320                metadata = self._metadata_resolver()
321            except Exception:
322                metadata = None
323        name = agent_name
324        if not name and self._agent_id_resolver is not None:
325            try:
326                name = self._agent_id_resolver() or ""
327            except Exception:
328                name = ""
329        return should_capture_logs(
330            name,
331            metadata_capture_logs=metadata,
332            capture_logs_config=self._capture_logs_config,
333            bootstrap_globs=list(runtime.logs_globs) if runtime else None,
334            bootstrap_agents=runtime.logs_agents_map() if runtime else None,
335            bootstrap_present=runtime is not None and runtime.logs_policy_present,
336        )
337
338    def handle_record(
339        self, handler: logging.Handler, record: logging.LogRecord
340    ) -> None:
341        name = record.name or ""
342        if any(name == p or name.startswith(p + ".") for p in _SKIP_LOGGER_PREFIXES):
343            return
344        level_name = record.levelname or "INFO"
345        if not level_at_least(level_name, self._resolve_min_level()):
346            return
347        if not self.capture_logs_enabled_for_agent():
348            return
349        session = _resolve_session_fields(self._session_resolver)
350        if not session or not session.get("session_id"):
351            return
352        message = handler.format(record) if handler.formatter else record.getMessage()
353        if not isinstance(message, str):
354            message = str(message)
355        trace_id, span_id = _current_trace_span()
356        ts = datetime.fromtimestamp(record.created, tz=timezone.utc).strftime(
357            "%Y-%m-%d %H:%M:%S.%f"
358        )[:-3]
359        attrs: dict[str, str] = {}
360        if record.pathname:
361            attrs["pathname"] = str(record.pathname)[:512]
362        if record.lineno:
363            attrs["lineno"] = str(record.lineno)
364        if record.funcName:
365            attrs["funcName"] = str(record.funcName)[:128]
366        event = SessionLogRecord(
367            session_id=str(session["session_id"]),
368            conversation_id=str(session.get("conversation_id") or session["session_id"]),
369            ts=ts,
370            level=normalize_log_level(level_name),
371            logger_name=name,
372            message=message,
373            turn_index=int(session.get("turn_index") or 0),
374            trace_id=trace_id,
375            span_id=span_id,
376            attributes=attrs,
377        )
378        with self._lock:
379            self._buffer.append(event)
380
381    def init(self, *, endpoint: str = "", api_key: str = "") -> None:
382        """Install root handler + start background flusher."""
383        with self._lock:
384            self._endpoint = (endpoint or "").rstrip("/")
385            self._api_key = api_key or ""
386            if self._handler is None:
387                self._handler = SessionLogHandler(self)
388                self._handler.setLevel(logging.DEBUG)
389                self._handler.setFormatter(logging.Formatter("%(message)s"))
390            if not self._installed:
391                logging.root.addHandler(self._handler)
392                self._installed = True
393            if self._flush_thread is None or not self._flush_thread.is_alive():
394                self._stop.clear()
395                self._flush_thread = threading.Thread(
396                    target=self._flush_loop,
397                    name="parlot-session-logs-flush",
398                    daemon=True,
399                )
400                self._flush_thread.start()
401            if not self._atexit_registered:
402                atexit.register(self.shutdown)
403                self._atexit_registered = True
404
405    def drain(self) -> list[SessionLogRecord]:
406        with self._lock:
407            events = list(self._buffer)
408            self._buffer.clear()
409            return events
410
411    def snapshot(self) -> list[SessionLogRecord]:
412        with self._lock:
413            return list(self._buffer)
414
415    def _circuit_open(self) -> bool:
416        return time.time() < self._circuit_open_until
417
418    def _flush_loop(self) -> None:
419        while not self._stop.wait(_FLUSH_INTERVAL_S):
420            try:
421                self._try_flush()
422            except Exception:
423                continue
424
425    def _try_flush(self, *, force: bool = False) -> None:
426        if self._circuit_open():
427            with self._lock:
428                self._buffer.clear()
429            return
430        if not self._endpoint or not self._api_key:
431            return
432
433        if not force:
434            now = time.time()
435            if now - self._last_flush_at < _FLUSH_INTERVAL_S * 0.5:
436                return
437
438        events = self.drain()
439        if not events:
440            return
441
442        ok = self._post_events(events)
443        self._last_flush_at = time.time()
444        if ok:
445            self._consecutive_failures = 0
446            return
447
448        self._consecutive_failures += 1
449        if self._consecutive_failures >= _CIRCUIT_FAILURES:
450            self._circuit_open_until = time.time() + _CIRCUIT_COOLDOWN_S
451            self._consecutive_failures = 0
452
453    def _post_events(self, events: list[SessionLogRecord]) -> bool:
454        try:
455            import httpx
456        except Exception:
457            return False
458
459        url = f"{self._endpoint}/v1/logs"
460        try:
461            payload = _encode_otlp_logs(events[:_MAX_BATCH])
462        except Exception:
463            return False
464        if not payload:
465            return True
466        from parlot.core.provider import build_parlot_client_headers
467
468        headers = build_parlot_client_headers(self._api_key)
469        headers["Content-Type"] = "application/x-protobuf"
470        try:
471            with httpx.Client(timeout=3.0) as client:
472                res = client.post(
473                    url,
474                    content=payload,
475                    headers=headers,
476                )
477                return 200 <= res.status_code < 300
478        except Exception:
479            return False
480
481    def shutdown(self) -> None:
482        """Stop capture and push any remaining buffered logs. Never raises."""
483        try:
484            self._stop.set()
485            thread = self._flush_thread
486            if (
487                thread is not None
488                and thread.is_alive()
489                and thread is not threading.current_thread()
490            ):
491                thread.join(timeout=1.0)
492            self._flush_thread = None
493            with self._lock:
494                if self._installed and self._handler is not None:
495                    try:
496                        logging.root.removeHandler(self._handler)
497                    except Exception:
498                        pass
499                    self._installed = False
500            self._try_flush(force=True)
501        except Exception:
502            return

100def base_parlotize(
101    agent_id: str,
102    *,
103    endpoint: Optional[str] = None,
104    api_key: Optional[str] = None,
105    capture_genai_content: Optional[bool] = None,
106    tracer_provider: TracerProvider | None = None,
107    version: Optional[str] = None,
108    capture_logs: bool | list[str] | None = None,
109    log_level: Optional[str] = None,
110) -> BaseParlotizeResult:
111    """Execute shared telemetry configuration common across all adapters.
112
113    Creates a ``ParlotContext``, initializes the session log
114    collector, and fetches remote bootstrap into ``context.runtime``.
115
116    See ``ParlotizeProtocol`` for the shared parameter surface.
117
118    Args:
119        agent_id: Required canonical deployment identity for ``session.agent_id``.
120        endpoint: Parlot OTLP collector base URL. If omitted, reads
121            ``PARLOT_ENDPOINT``.
122        api_key: Org-scoped API key. If omitted, reads ``PARLOT_API_KEY``.
123        capture_genai_content: Process-wide GenAI content capture override.
124        tracer_provider: Existing ``TracerProvider`` to adopt, if any.
125        version: Deployment version for ``gen_ai.agent.version``. Defaults to
126            ``\"unknown\"`` when omitted or blank.
127        capture_logs: Session log capture policy (bool or agent-id globs).
128        log_level: Minimum level for session log capture.
129
130    Returns:
131        Resolved endpoint, credentials, provider, and capture settings for the
132        calling adapter.
133    """
134    from parlot.core.bootstrap import fetch_telemetry_bootstrap
135    from parlot.core.context import ParlotContext
136    from parlot.core.provider import (
137        adopt_existing_tracer_provider,
138        resolve_api_key,
139        resolve_capture_genai_content,
140        resolve_endpoint,
141    )
142
143    configure_parlot_logging()
144
145    resolved_agent_id = agent_id.strip()
146    if not resolved_agent_id:
147        raise ValueError("parlotize(agent_id) is required")
148    resolved_version = (version or "").strip() or "unknown"
149    resolved_endpoint = resolve_endpoint(endpoint)
150    resolved_api_key = resolve_api_key(api_key)
151    resolved_capture_content = resolve_capture_genai_content(capture_genai_content)
152
153    if isinstance(capture_logs, list):
154        resolved_capture_logs: bool | list[str] | None = [
155            str(item).strip() for item in capture_logs if str(item).strip()
156        ]
157    else:
158        resolved_capture_logs = capture_logs
159
160    resolved_log_level = log_level.strip().upper() if log_level else None
161
162    context = ParlotContext()
163    context.session_logs.set_capture_logs_config(
164        resolved_capture_logs,
165        log_level=resolved_log_level,
166    )
167    context.session_logs.init(endpoint=resolved_endpoint, api_key=resolved_api_key)
168
169    if not resolved_endpoint:
170        logger.error("parlot: PARLOT_ENDPOINT is not set")
171    if not resolved_api_key:
172        logger.error(
173            "parlot: PARLOT_API_KEY is not set; telemetry will not authenticate"
174        )
175    elif resolved_endpoint:
176        fetch_telemetry_bootstrap(resolved_endpoint, resolved_api_key, context)
177
178    if tracer_provider is None:
179        tracer_provider = adopt_existing_tracer_provider()
180
181    return BaseParlotizeResult(
182        context=context,
183        endpoint=resolved_endpoint,
184        api_key=resolved_api_key,
185        tracer_provider=tracer_provider,
186        agent_id=resolved_agent_id,
187        agent_version=resolved_version,
188        capture_genai_content=resolved_capture_content,
189        capture_logs=resolved_capture_logs,
190        log_level=resolved_log_level,
191    )

92def configure_parlot_logging() -> None:
93    """Configure root logging from ``PARLOT_DEBUG_LEVEL`` when no handlers exist."""
94    level_name = os.getenv("PARLOT_DEBUG_LEVEL", "INFO").upper()
95    level = getattr(logging, level_name, logging.INFO)
96    if not logging.root.handlers:
97        logging.basicConfig(level=level)

14class ExportFilterSpanExporter(SpanExporter):
15    """Pass through only Conversation Contract + GenAI + voice spans."""
16
17    def __init__(self, exporter: SpanExporter) -> None:
18        self._exporter = exporter
19
20    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
21        filtered = [span for span in spans if is_exportable_span_name(span.name)]
22        if not filtered:
23            return SpanExportResult.SUCCESS
24        return self._exporter.export(filtered)
25
26    def shutdown(self) -> None:
27        self._exporter.shutdown()
28
29    def force_flush(self, timeout_millis: int = 30000) -> bool:
30        return self._exporter.force_flush(timeout_millis)

24class ParlotBaseProcessor(SpanProcessor):
25    """
26    Subclass and override ``on_end`` (or the ``_enrich`` dispatch method) to
27    add framework-specific span enrichment. Call ``super().on_start()`` /
28    ``super().on_end()`` so shared debug logging runs.
29
30    Do NOT wrap a downstream processor — add alongside BatchSpanProcessor:
31
32        provider.add_span_processor(MyFrameworkProcessor())
33        provider.add_span_processor(BatchSpanProcessor(your_exporter))
34    """
35
36    def on_start(self, span, parent_context=None) -> None:
37        log_span_event("on_start", span)
38
39    def on_end(self, span: ReadableSpan) -> None:
40        log_span_event("on_end", span)
41
42    def shutdown(self) -> None:
43        pass
44
45    def force_flush(self, timeout_millis: int = 30_000) -> bool:
46        return True
47
48    # ------------------------------------------------------------------
49    # Shared helpers — used by all subclasses
50    # ------------------------------------------------------------------
51
52    @staticmethod
53    def _set(span: ReadableSpan, key: str, value) -> None:
54        """Write an attribute into a ReadableSpan after it has ended.
55
56        OTel SDK ≥1.44 stores attributes in an immutable ``BoundedAttributes``
57        by default; mutating it raises ``TypeError``. Copy to a plain dict
58        when needed so enrichment can still stamp session/turn fields.
59        """
60        attrs = span._attributes
61        if attrs is None:
62            span._attributes = {key: value}
63            return
64        if getattr(attrs, "_immutable", False) or not isinstance(attrs, dict):
65            span._attributes = dict(attrs)
66            span._attributes[key] = value
67            return
68        try:
69            attrs[key] = value
70        except TypeError:
71            span._attributes = dict(attrs)
72            span._attributes[key] = value
73
74    @staticmethod
75    def _add_event(span: ReadableSpan, name: str, attributes: dict) -> None:
76        """Append a span event to a ReadableSpan after it has ended."""
77        from opentelemetry.sdk.trace import Event
78        evt = Event(name=name, attributes=attributes, timestamp=time.time_ns())
79        if hasattr(span, "_events") and isinstance(span._events, list):
80            span._events.append(evt)
81
82    @staticmethod
83    def _trace_id_hex(span: ReadableSpan) -> str:
84        return format(span.context.trace_id, "032x")
85
86    @staticmethod
87    def _maybe_update(state: SessionState, attr: str, value) -> None:
88        """Set a state attribute only if it is currently falsy."""
89        if value and not getattr(state, attr, ""):
90            setattr(state, attr, str(value))