[NET-1049] feat: Add instrumentation support for LiveKit

netra/__init__.py

@@ -257,7 +257,12 @@ def shutdown(cls) -> None:
                     meter_provider = otel_metrics.get_meter_provider()
                     if hasattr(meter_provider, "force_flush"):
                         meter_provider.force_flush()
-                    if hasattr(meter_provider, "shutdown"):
+                    # _NetraOwnedMeterProvider.shutdown() is a no-op for third-party
+                    # callers (see netra/meter.py); Netra's own teardown must use the
+                    # owner entry point or metrics are never flushed at exit.
+                    if hasattr(meter_provider, "shutdown_as_owner"):
+                        meter_provider.shutdown_as_owner()
+                    elif hasattr(meter_provider, "shutdown"):
                         meter_provider.shutdown()
                 except Exception:
                     pass

netra/config.py

@@ -1,18 +1,43 @@
 import json
+import logging
 import os
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, FrozenSet, List, Optional
 
 from opentelemetry.util.re import parse_env_headers
 
 from netra.version import __version__
 
+logger = logging.getLogger(__name__)
+
 # Fallback limits used when no Config has been activated yet (e.g. code paths that
 # run before ``Netra.init()``, or tests that never call it). Once ``init()`` runs,
 # the active Config instance's values (resolved from env at init time) take over.
 _DEFAULT_ATTRIBUTE_MAX_LEN = 50000
 _DEFAULT_CONVERSATION_CONTENT_MAX_LEN = 50000
 _DEFAULT_TRIAL_BLOCK_DURATION_SECONDS = 15 * 60
 
+# --- Voice-agent audio capture (env-only; no Netra.init() parameter) ------------
+# The path segment appended to the OTLP endpoint when no explicit audio endpoint
+# is given. Deliberately not derived via UsageHttpClient._resolve_base_url, which
+# strips a "/telemetry" suffix — correct for the REST APIs, wrong here.
+_AUDIO_CHUNK_PATH = "/v1/audio/chunk"
+
+# Header names that count as an audio-ingest credential. An unauthenticated PCM
+# POST is never attempted.
+_AUDIO_AUTH_HEADERS = ("x-api-key", "Authorization")
+
+# The only recognised speaker roles.
+AUDIO_ROLES: FrozenSet[str] = frozenset({"user", "agent"})
+
+_DEFAULT_AUDIO_BATCH_BYTES = 32768
+_DEFAULT_AUDIO_BATCH_INTERVAL_MS = 1000
+_DEFAULT_AUDIO_BUFFER_BYTES = 2097152
+_DEFAULT_AUDIO_MAX_REQUEST_BYTES = 262144
+
+_MIN_AUDIO_BATCH_BYTES = 1024
+_MIN_AUDIO_BATCH_INTERVAL_MS = 100
+_MAX_AUDIO_BATCH_INTERVAL_MS = 30000
+
 
 class Config:
     """
@@ -97,8 +122,170 @@ def __init__(
             None, "TRIAL_BLOCK_DURATION_SECONDS", default=_DEFAULT_TRIAL_BLOCK_DURATION_SECONDS
         )
 
+        self._resolve_audio_settings()
+
         self._set_trace_content_env()
 
+    def _resolve_audio_settings(self) -> None:
+        """Resolve and validate the voice-agent audio-capture settings.
+
+        Env-only by design: there is no ``capture_audio`` parameter on
+        ``Netra.init()``.  Whether audio is captured at all is decided by
+        :attr:`audio_capture_enabled`, not by a flag.
+
+        Every validation failure logs a ``WARNING`` naming the setting and the
+        value actually used, then falls back to a safe value.  A bad number MUST
+        NOT raise out of ``Netra.init()``.
+        """
+        self.audio_endpoint_override = os.getenv("NETRA_AUDIO_ENDPOINT")
+        self.audio_batch_bytes = self._get_int_config(
+            None, "NETRA_AUDIO_BATCH_BYTES", default=_DEFAULT_AUDIO_BATCH_BYTES
+        )
+        self.audio_batch_interval_ms = self._get_int_config(
+            None, "NETRA_AUDIO_BATCH_INTERVAL_MS", default=_DEFAULT_AUDIO_BATCH_INTERVAL_MS
+        )
+        self.audio_buffer_bytes = self._get_int_config(
+            None, "NETRA_AUDIO_BUFFER_BYTES", default=_DEFAULT_AUDIO_BUFFER_BYTES
+        )
+        self.audio_max_request_bytes = self._get_int_config(
+            None, "NETRA_AUDIO_MAX_REQUEST_BYTES", default=_DEFAULT_AUDIO_MAX_REQUEST_BYTES
+        )
+        self.audio_roles = self._get_role_set("NETRA_AUDIO_ROLES")
+        self.audio_save_local = self._get_bool_config(None, "NETRA_AUDIO_SAVE_LOCAL", default=False)
+
+        # Order matters: audio_batch_bytes is clamped against the resolved
+        # max-request size first, then the two ceilings are raised to whatever
+        # batch size survived. Doing it the other way round lets a tiny
+        # max_request_bytes silently shrink the batch below its floor.
+        if self.audio_max_request_bytes < _MIN_AUDIO_BATCH_BYTES:
+            logger.warning(
+                "netra.audio: NETRA_AUDIO_MAX_REQUEST_BYTES=%d is below the minimum batch size; using %d",
+                self.audio_max_request_bytes,
+                _MIN_AUDIO_BATCH_BYTES,
+            )
+            self.audio_max_request_bytes = _MIN_AUDIO_BATCH_BYTES
+
+        clamped_batch = min(max(self.audio_batch_bytes, _MIN_AUDIO_BATCH_BYTES), self.audio_max_request_bytes)
+        if clamped_batch != self.audio_batch_bytes:
+            logger.warning(
+                "netra.audio: NETRA_AUDIO_BATCH_BYTES=%d out of range [%d, %d]; using %d",
+                self.audio_batch_bytes,
+                _MIN_AUDIO_BATCH_BYTES,
+                self.audio_max_request_bytes,
+                clamped_batch,
+            )
+            self.audio_batch_bytes = clamped_batch
+
+        clamped_interval = min(
+            max(self.audio_batch_interval_ms, _MIN_AUDIO_BATCH_INTERVAL_MS),
+            _MAX_AUDIO_BATCH_INTERVAL_MS,
+        )
+        if clamped_interval != self.audio_batch_interval_ms:
+            logger.warning(
+                "netra.audio: NETRA_AUDIO_BATCH_INTERVAL_MS=%d out of range [%d, %d]; using %d",
+                self.audio_batch_interval_ms,
+                _MIN_AUDIO_BATCH_INTERVAL_MS,
+                _MAX_AUDIO_BATCH_INTERVAL_MS,
+                clamped_interval,
+            )
+            self.audio_batch_interval_ms = clamped_interval
+
+        if self.audio_buffer_bytes < self.audio_batch_bytes:
+            logger.warning(
+                "netra.audio: NETRA_AUDIO_BUFFER_BYTES=%d is below the batch size; using %d",
+                self.audio_buffer_bytes,
+                self.audio_batch_bytes,
+            )
+            self.audio_buffer_bytes = self.audio_batch_bytes
+
+        if self.audio_max_request_bytes < self.audio_batch_bytes:
+            logger.warning(
+                "netra.audio: NETRA_AUDIO_MAX_REQUEST_BYTES=%d is below the batch size; using %d",
+                self.audio_max_request_bytes,
+                self.audio_batch_bytes,
+            )
+            self.audio_max_request_bytes = self.audio_batch_bytes
+
+        if not self.audio_roles:
+            logger.warning(
+                "netra.audio: NETRA_AUDIO_ROLES resolved empty; no call audio will be captured. "
+                "Traces are unaffected."
+            )
+
+        if self.audio_save_local:
+            logger.warning(
+                "netra.audio: NETRA_AUDIO_SAVE_LOCAL is enabled. Local WAV capture is a "
+                "development-only aid and retains full-session PCM in memory."
+            )
+
+    def _get_role_set(self, env_var: str) -> FrozenSet[str]:
+        """Parse a comma-separated speaker-role list, dropping unknown roles.
+
+        An explicitly empty value (``NETRA_AUDIO_ROLES=``) is the documented way
+        to disable audio capture without affecting traces, so it resolves to an
+        empty set rather than the default.
+
+        Args:
+            env_var: Name of the environment variable holding the role list.
+
+        Returns:
+            The recognised roles, or the full default set when *env_var* is unset.
+        """
+        raw = os.getenv(env_var)
+        if raw is None:
+            return AUDIO_ROLES
+
+        requested = {part.strip().lower() for part in raw.split(",") if part.strip()}
+        unknown = requested - AUDIO_ROLES
+        if unknown:
+            logger.warning(
+                "netra.audio: %s contains unknown role(s) %s; recognised roles are %s",
+                env_var,
+                sorted(unknown),
+                sorted(AUDIO_ROLES),
+            )
+        return frozenset(requested & AUDIO_ROLES)
+
+    def audio_endpoint(self) -> Optional[str]:
+        """Resolve the audio ingest URL, or None if audio must not be sent.
+
+        This is the ONLY gate on audio capture: there is no ``capture_audio``
+        flag.  A non-None return means audio WILL be captured and streamed once a
+        LiveKit session starts.  Returns None unless a concrete endpoint resolves
+        AND an auth header is present.
+
+        Callers treat None as "disable capture entirely", not "retry later" — the
+        result is resolved from init-time state and does not change during the
+        process.
+
+        Returns:
+            The absolute audio ingest URL, or ``None`` when audio must not be sent.
+        """
+        if self.audio_endpoint_override:
+            url = self.audio_endpoint_override
+        elif self.otlp_endpoint:
+            url = self.otlp_endpoint.rstrip("/") + _AUDIO_CHUNK_PATH
+        else:
+            return None
+
+        if not any(header in self.headers for header in _AUDIO_AUTH_HEADERS):
+            logger.warning(
+                "netra.audio: an audio endpoint resolved but no credential is configured; "
+                "audio capture is disabled. Set NETRA_API_KEY or pass an auth header."
+            )
+            return None
+
+        return url
+
+    @property
+    def audio_capture_enabled(self) -> bool:
+        """Whether call audio will be captured and streamed.
+
+        The single derived predicate behind audio capture, so the instrumentor,
+        the session hooks and the startup log line cannot disagree about it.
+        """
+        return self.audio_endpoint() is not None and bool(self.audio_roles)
+
     def _get_app_name(self, app_name: Optional[str]) -> str:
         """Get application name from param or environment variables."""
         return app_name or os.getenv("NETRA_APP_NAME") or os.getenv("OTEL_SERVICE_NAME") or "llm_tracing_service"

netra/instrumentation/__init__.py

@@ -196,6 +196,10 @@ def init_instrumentations(
     if CustomInstruments.DEEPGRAM in netra_custom_instruments:
         init_deepgram_instrumentation()
 
+    # Initialize LiveKit voice-agent instrumentation.
+    if CustomInstruments.LIVEKIT in netra_custom_instruments:
+        init_livekit_instrumentation()
+
     # Initialize ADK instrumentation.
     if CustomInstruments.ADK in netra_custom_instruments:
         init_adk_instrumentation()
@@ -430,6 +434,25 @@ def init_deepgram_instrumentation() -> bool:
         return False
 
 
+def init_livekit_instrumentation() -> bool:
+    """Initialize LiveKit voice-agent instrumentation.
+
+    Returns:
+        bool: True if initialization was successful, False otherwise.
+    """
+    try:
+        if is_package_installed("livekit-agents"):
+            from netra.instrumentation.livekit import NetraLiveKitInstrumentor
+
+            instrumentor = NetraLiveKitInstrumentor()
+            if not instrumentor.is_instrumented_by_opentelemetry:
+                instrumentor.instrument()
+        return True
+    except Exception:
+        logging.exception("Error initializing LiveKit instrumentor")
+        return False
+
+
 def init_adk_instrumentation() -> bool:
     """Initialize ADK instrumentation.
 

netra/instrumentation/instruments.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import Any, Optional, Type
+from typing import Any, Dict, Optional, Type
 
 from traceloop.sdk import Instruments
 
@@ -74,6 +74,7 @@ class CustomInstruments(Enum):
     ELEVENLABS = "elevenlabs"
     CLAUDE_AGENT_SDK = "claude_agent_sdk"
     HERMES_AGENT = "hermes_agent"
+    LIVEKIT = "livekit"
 
 
 class InstrumentSet(Enum):
@@ -146,6 +147,7 @@ def __new__(cls, value: Any, origin: Optional[Type[Enum]] = None) -> "Instrument
     LANCEDB = ("lancedb", Instruments)
     LANGCHAIN = ("langchain", Instruments)
     LITELLM = ("litellm", CustomInstruments)
+    LIVEKIT = ("livekit", CustomInstruments)
     LLAMA_INDEX = ("llama_index", Instruments)
     LOGGING = ("logging", CustomInstruments)
     MARQO = ("marqo", Instruments)
@@ -197,6 +199,23 @@ def __new__(cls, value: Any, origin: Optional[Type[Enum]] = None) -> "Instrument
 NetraInstruments = InstrumentSet
 
 
+# Instrumentation scopes that Netra enables but does not author, so their scope
+# name does not follow the ``netra.instrumentation.*`` /
+# ``opentelemetry.instrumentation.*`` convention that the span processors key
+# off.  Mapping the scope to its ``InstrumentSet`` value is what puts these
+# spans under the same instrument-name machinery as every other
+# instrumentation — ``root_instruments`` filtering and the
+# ``netra.instrumentation.name`` attribute.
+#
+# Matched exactly, never as a prefix: an alias claims one specific scope, and a
+# prefix match here would start pulling in unrelated third-party tracers.
+THIRD_PARTY_INSTRUMENTATION_SCOPES: Dict[str, str] = {
+    # livekit-agents emits its own span tree (agent_session -> agent_turn ->
+    # llm_node / tts_node / function_tool) under this scope.
+    "livekit-agents": InstrumentSet.LIVEKIT.value,
+}
+
+
 # Default instrument sets
 
 # These sets are intentionally independent.  Removing an
@@ -220,6 +239,7 @@ def __new__(cls, value: Any, origin: Optional[Type[Enum]] = None) -> "Instrument
         InstrumentSet.GROQ,
         InstrumentSet.LANGCHAIN,
         InstrumentSet.LITELLM,
+        InstrumentSet.LIVEKIT,
         InstrumentSet.CEREBRAS,
         InstrumentSet.MISTRALAI,
         InstrumentSet.OPENAI,
@@ -260,6 +280,11 @@ def __new__(cls, value: Any, origin: Optional[Type[Enum]] = None) -> "Instrument
 )
 
 # Subset of DEFAULT_INSTRUMENTS allowed to produce root-level spans.
+#
+# InstrumentSet.LIVEKIT must stay listed here: ``agent_session`` is the root of
+# every voice trace, so dropping LiveKit from the root allow-list peels that
+# span, then recursively peels ``agent_turn`` / ``llm_node`` / ... — the whole
+# voice tree — leaving only the provider spans underneath as orphaned roots.
 DEFAULT_INSTRUMENTS_FOR_ROOT: frozenset[InstrumentSet] = frozenset(
     {
         InstrumentSet.ANTHROPIC,
@@ -274,6 +299,7 @@ def __new__(cls, value: Any, origin: Optional[Type[Enum]] = None) -> "Instrument
         InstrumentSet.GROQ,
         InstrumentSet.LANGCHAIN,
         InstrumentSet.LITELLM,
+        InstrumentSet.LIVEKIT,
         InstrumentSet.CEREBRAS,
         InstrumentSet.MISTRALAI,
         InstrumentSet.OPENAI,