from __future__ import annotations import asyncio import hashlib import inspect import json import os import re from collections.abc import Awaitable, Callable, Mapping, Sequence from dataclasses import dataclass from types import MethodType from typing import TYPE_CHECKING, Any, TypedDict, cast from livekit.agents.job import get_job_context from livekit.agents.types import NOT_GIVEN from livekit.agents.voice.agent_session import AgentSession from livekit.agents.voice.events import CloseEvent from ._payload import ( CALL_ID_STRATEGY_CUSTOM, CALL_ID_STRATEGY_PARTICIPANT_IDENTITY, CALL_ID_STRATEGY_PARTICIPANT_METADATA, CALL_ID_STRATEGY_ROOM_NAME, CallIdResolver, PayloadBuildConfig, SessionReportLike, build_livekit_monitoring_envelope, ) from ._transport import ConnectionPolicy, HammingTransport from .log import logger from .version import __version__ if TYPE_CHECKING: from livekit.agents.job import JobContext RECORDING_MODE_NONE = "none" RECORDING_MODE_SESSION_AUDIO = "session_audio" RECORDING_MODE_ROOM_COMPOSITE = "room_composite" RECORDING_MODE_PARTICIPANT_EGRESS = "participant_egress" SUPPORTED_RECORDING_MODES = { RECORDING_MODE_NONE, RECORDING_MODE_SESSION_AUDIO, RECORDING_MODE_ROOM_COMPOSITE, RECORDING_MODE_PARTICIPANT_EGRESS, } RECORDING_SOURCE_EXTERNAL_URLS = "external_urls" RECORDING_SOURCE_AUTO_EGRESS = "auto_egress" LEGACY_RECORDING_SOURCE_CUSTOMER_MANAGED = "customer_managed" LEGACY_RECORDING_SOURCE_PLUGIN_MANAGED = "plugin_managed" SUPPORTED_RECORDING_SOURCES = { RECORDING_SOURCE_EXTERNAL_URLS, RECORDING_SOURCE_AUTO_EGRESS, } RECORDING_DELIVERY_AUTO = "auto" RECORDING_DELIVERY_URL = "url" RECORDING_DELIVERY_UPLOAD = "upload" SUPPORTED_RECORDING_DELIVERIES = { RECORDING_DELIVERY_AUTO, RECORDING_DELIVERY_URL, RECORDING_DELIVERY_UPLOAD, } SUPPORTED_CALL_ID_STRATEGIES = { CALL_ID_STRATEGY_ROOM_NAME, CALL_ID_STRATEGY_PARTICIPANT_IDENTITY, CALL_ID_STRATEGY_PARTICIPANT_METADATA, CALL_ID_STRATEGY_CUSTOM, } SUPPORTED_CAPTURE_MODES = {"none", "full"} SAMPLING_VERSION = "v1" STREAMING_MODE_NONE = "none" STREAMING_MODE_TRACK_EGRESS = "track_egress" SUPPORTED_STREAMING_MODES = { STREAMING_MODE_NONE, STREAMING_MODE_TRACK_EGRESS, } SESSION_MODE_MONITORING = "monitoring" SESSION_MODE_TESTING = "testing" SUPPORTED_SESSION_MODES = { SESSION_MODE_MONITORING, SESSION_MODE_TESTING, } TEST_CASE_RUN_ID_METADATA_KEYS = ( "test_case_run_id", "testCaseRunId", "conversation_id", "conversationId", ) PLUGIN_MANAGED_RECORDINGS_DIR = "recordings" PLUGIN_MANAGED_ROOM_COMPOSITE_FILENAME_PREFIX = "hamming_plugin_room_composite" PLUGIN_MANAGED_ROOM_COMPOSITE_EXTENSION = "ogg" class RecordingContext(TypedDict, total=False): call_id: str room_name: str customer_conversation_id: str provider_recording_id: str provider_recording_filepath: str metadata: dict[str, str] class RecordingArtifacts(TypedDict, total=False): recording_url: str agent_recording_url: str user_recording_url: str local_paths: list[str] class RecordingConfigInput(TypedDict, total=False): mode: str source: str delivery: str resolver: RecordingResolver class StreamingConfigInput(TypedDict, total=False): mode: str source: str class CaptureConfigInput(TypedDict, total=False): agent_context: bool prompts: str tools: str interim_transcripts: bool class RedactConfigInput(TypedDict, total=False): pii: bool tool_args_paths: list[str] tool_output_paths: list[str] class LiveKitConfigInput(TypedDict, total=False): url: str api_key: str api_secret: str class ExternalLinkInput(TypedDict, total=False): label: str url: str source: str description: str @dataclass(frozen=True) class RecordingResolutionContext: session: AgentSession[Any] report: SessionReportLike job_ctx: JobContext | None close_event: CloseEvent | None call_id: str room_name: str participant_identity: str | None participant_metadata_raw: str | None external_agent_id: str recording_context: RecordingContext | None RecordingResolverResult = RecordingArtifacts | Awaitable[RecordingArtifacts] RecordingResolver = Callable[[RecordingResolutionContext], RecordingResolverResult] @dataclass(frozen=True) class ResolvedRecordingConfig: mode: str = RECORDING_MODE_NONE source: str = RECORDING_SOURCE_AUTO_EGRESS delivery: str = RECORDING_DELIVERY_AUTO resolver: RecordingResolver | None = None @dataclass(frozen=True) class ResolvedStreamingConfig: mode: str = STREAMING_MODE_NONE source: str = RECORDING_SOURCE_EXTERNAL_URLS @dataclass(frozen=True) class ResolvedCaptureConfig: agent_context: bool = True prompts: str = "full" tools: str = "full" interim_transcripts: bool = False @dataclass(frozen=True) class ResolvedRedactConfig: pii: bool = True tool_args_paths: tuple[str, ...] = () tool_output_paths: tuple[str, ...] = () @dataclass(frozen=True) class ResolvedLiveKitConfig: url: str | None = None api_key: str | None = None api_secret: str | None = None @dataclass(frozen=True) class ResolvedSamplingConfig: rate: float = 1.0 key: str = "participant_identity" @dataclass(frozen=True) class PluginManagedRoomCompositeState: room_name: str filepath: str start_task: asyncio.Task[str] @dataclass(frozen=True) class HammingConfig: api_key: str base_url: str = "https://app.hamming.ai" call_review_enabled: bool = True flush_on_shutdown: bool = True auto_record_audio: bool = False external_agent_id: str | None = None plugin_api_version: str = "1.0.0" request_timeout_seconds: float = 10.0 max_retries: int = 3 retry_backoff_seconds: float = 0.5 payload_schema_version: str = "2026-03-02" call_id_strategy: str = CALL_ID_STRATEGY_ROOM_NAME call_id_metadata_key: str = "call_id" resolve_call_id: CallIdResolver | None = None sampling: ResolvedSamplingConfig = ResolvedSamplingConfig() capture: ResolvedCaptureConfig = ResolvedCaptureConfig() redact: ResolvedRedactConfig = ResolvedRedactConfig() livekit: ResolvedLiveKitConfig = ResolvedLiveKitConfig() recording: ResolvedRecordingConfig = ResolvedRecordingConfig() streaming: ResolvedStreamingConfig = ResolvedStreamingConfig() class _SessionMonitor: def __init__( self, *, runtime: HammingRuntime, session: AgentSession[Any], participant_identity: str | None, participant_metadata: str | None, external_agent_id: str, customer_metadata: Mapping[str, Any] | None = None, external_links: Sequence[ExternalLinkInput] | None = None, session_mode: str = SESSION_MODE_MONITORING, job_ctx: JobContext | None, session_key: int, recording_context: RecordingContext | None, ) -> None: self._runtime = runtime self._session = session self._participant_identity = participant_identity self._participant_metadata = participant_metadata self._external_agent_id = external_agent_id self._customer_metadata = ( _normalize_json_object(customer_metadata, field_name="customer_metadata") if customer_metadata else None ) self._external_links = _normalize_external_links(external_links) self._session_mode = _normalize_session_mode(session_mode) self._job_ctx = job_ctx self._session_key = session_key self._recording_context = ( cast(RecordingContext, dict(recording_context)) if recording_context else None ) self._plugin_managed_room_composite: PluginManagedRoomCompositeState | None = None self._close_event: CloseEvent | None = None self._close_event_ready = asyncio.Event() self._send_lock = asyncio.Lock() self._send_task: asyncio.Task[None] | None = None def on_close(self, event: CloseEvent) -> None: self._close_event = event self._close_event_ready.set() self._ensure_send_task(from_shutdown=False) async def on_shutdown(self, reason: str | None = None) -> None: if not self._runtime.config.flush_on_shutdown: return task = self._ensure_send_task(from_shutdown=True) try: await asyncio.wait_for(task, timeout=5.0) except asyncio.TimeoutError: logger.warning("timed out waiting for hamming final payload flush") except asyncio.CancelledError: logger.warning( "hamming shutdown callback cancelled before flush completed", extra={"reason": reason}, ) except Exception: logger.exception( "hamming shutdown callback failed while waiting for flush", extra={"reason": reason}, ) def start_plugin_managed_room_composite_if_needed(self) -> None: recording = self._runtime.config.recording if ( recording.mode != RECORDING_MODE_ROOM_COMPOSITE or recording.source != RECORDING_SOURCE_AUTO_EGRESS ): return if self._job_ctx is None: logger.warning( "plugin-managed room composite requires JobContext; falling back to resolver path", ) return filepath = _build_plugin_managed_room_composite_filepath( room_name=self._job_ctx.room.name, participant_metadata_raw=self._participant_metadata, recording_context=self._recording_context, session_key=self._session_key, ) logger.info( "scheduling plugin-managed room composite egress", extra={ "room_name": self._job_ctx.room.name, "filepath": filepath, **_recording_debug_context( participant_metadata_raw=self._participant_metadata, recording_context=self._recording_context, ), }, ) start_task = asyncio.create_task( self._start_plugin_managed_room_composite_egress(filepath=filepath) ) self._runtime.track_task(start_task) self._plugin_managed_room_composite = PluginManagedRoomCompositeState( room_name=self._job_ctx.room.name, filepath=filepath, start_task=start_task, ) def _ensure_send_task(self, *, from_shutdown: bool) -> asyncio.Task[None]: if self._send_task is None: self._send_task = asyncio.create_task( self._send_final_payload(wait_for_close_event=from_shutdown) ) self._runtime.track_task(self._send_task) return self._send_task async def _send_final_payload(self, *, wait_for_close_event: bool) -> None: try: async with self._send_lock: close_event = await self._wait_for_close_event_if_needed( wait_for_close_event=wait_for_close_event ) report = self._build_report_for_export() if report is None: return envelope = self._build_monitoring_envelope(report=report, close_event=close_event) payload_record = envelope.get("payload") if not isinstance(payload_record, dict): await self._send_envelope(envelope=envelope, report=report, payload_record=None) return if self._should_skip_for_sampling( envelope=envelope, payload_record=payload_record, report=report, ): return await self._populate_recording_payload( report=report, payload_record=payload_record, close_event=close_event, ) self._log_recording_resolution_outcome( report=report, payload_record=payload_record, ) await self._send_envelope( envelope=envelope, report=report, payload_record=payload_record, ) except Exception as exc: logger.error("unhandled exception in hamming final payload task: %r", exc) logger.exception("unhandled exception in hamming final payload task") finally: self._runtime.remove_monitor(self._session_key) async def _wait_for_close_event_if_needed( self, *, wait_for_close_event: bool, ) -> CloseEvent | None: if wait_for_close_event and self._close_event is None: try: await asyncio.wait_for(self._close_event_ready.wait(), timeout=0.5) except asyncio.TimeoutError: pass return self._close_event def _build_report_for_export(self) -> SessionReportLike | None: if not self._runtime.config.call_review_enabled: logger.debug("hamming call review export disabled; skipping final payload") return None try: return self._runtime.build_report(self._session, self._job_ctx) except Exception: logger.exception("failed to build session report for hamming payload") return None def _build_monitoring_envelope( self, *, report: SessionReportLike, close_event: CloseEvent | None, ) -> dict[str, Any]: envelope = build_livekit_monitoring_envelope( config=PayloadBuildConfig( external_agent_id=self._external_agent_id, plugin_api_version=self._runtime.config.plugin_api_version, plugin_version=__version__, payload_schema_version=self._runtime.config.payload_schema_version, call_id_strategy=self._runtime.config.call_id_strategy, call_id_metadata_key=self._runtime.config.call_id_metadata_key, resolve_call_id=self._runtime.config.resolve_call_id, capture_manifest=_build_capture_manifest(self._runtime.config), customer_metadata=self._customer_metadata, external_links=self._external_links, session_mode=self._session_mode, ), report=report, participant_identity=self._participant_identity, participant_metadata_raw=self._participant_metadata, recording_context=self._recording_context, close_event=close_event, ) return envelope def _should_skip_for_sampling( self, *, envelope: dict[str, Any], payload_record: dict[str, Any], report: SessionReportLike, ) -> bool: sampling_metadata = _build_sampling_metadata( config=self._runtime.config, call_id=_string_or_fallback(payload_record.get("call_id"), report.room), room_name=report.room, participant_identity=self._participant_identity, participant_metadata_raw=self._participant_metadata, external_agent_id=self._external_agent_id, ) metadata_record = envelope.setdefault("metadata", {}) if isinstance(metadata_record, dict): metadata_record["sampling"] = sampling_metadata if sampling_metadata["decision"] != "excluded": return False logger.info( "skipping hamming payload due to deterministic sampling", extra={ "call_id": payload_record.get("call_id"), "room_name": report.room, "sampling_key": sampling_metadata["key"], "sampling_rate": sampling_metadata["rate"], }, ) return True async def _send_envelope( self, *, envelope: dict[str, Any], report: SessionReportLike, payload_record: dict[str, Any] | None, ) -> None: try: await self._runtime.transport.send_capture(envelope) except Exception: logger.exception("failed to send hamming payload") return logger.info( "sent hamming payload", extra={ "call_id": payload_record.get("call_id") if payload_record is not None else None, "room_name": report.room, }, ) async def _populate_recording_payload( self, *, report: SessionReportLike, payload_record: dict[str, Any], close_event: CloseEvent | None, ) -> None: recording = self._runtime.config.recording if recording.mode == RECORDING_MODE_NONE: return if recording.mode == RECORDING_MODE_SESSION_AUDIO: await self._attach_inline_recording_capture( report=report, payload_record=payload_record ) return if recording.mode not in { RECORDING_MODE_ROOM_COMPOSITE, RECORDING_MODE_PARTICIPANT_EGRESS, }: logger.warning("unsupported hamming recording mode", extra={"mode": recording.mode}) return resolution_context = self._build_recording_resolution_context( report=report, payload_record=payload_record, close_event=close_event, ) if await self._populate_builtin_recording_payload( payload_record=payload_record, resolution_context=resolution_context, ): return resolved_artifacts = await self._resolve_recording_artifacts( payload_record=payload_record, resolution_context=resolution_context, ) if resolved_artifacts is None: return self._apply_resolved_recording_artifacts( payload_record=payload_record, resolved_artifacts=resolved_artifacts, ) def _build_recording_resolution_context( self, *, report: SessionReportLike, payload_record: dict[str, Any], close_event: CloseEvent | None, ) -> RecordingResolutionContext: return RecordingResolutionContext( session=self._session, report=report, job_ctx=self._job_ctx, close_event=close_event, call_id=_string_or_fallback(payload_record.get("call_id"), report.room), room_name=report.room, participant_identity=self._participant_identity, participant_metadata_raw=self._participant_metadata, external_agent_id=self._external_agent_id, recording_context=self._recording_context, ) async def _populate_builtin_recording_payload( self, *, payload_record: dict[str, Any], resolution_context: RecordingResolutionContext, ) -> bool: recording = self._runtime.config.recording if recording.mode != RECORDING_MODE_ROOM_COMPOSITE: return False if recording.source == RECORDING_SOURCE_AUTO_EGRESS: recording_url = await self._resolve_plugin_managed_room_composite_recording_url( resolution_context=resolution_context, ) if recording_url: payload_record["recording_url"] = recording_url return True recording_url = await self._resolve_hamming_room_composite_recording_url( resolution_context=resolution_context, ) if recording_url: payload_record["recording_url"] = recording_url return True return False async def _resolve_recording_artifacts( self, *, payload_record: dict[str, Any], resolution_context: RecordingResolutionContext, ) -> RecordingArtifacts | None: recording = self._runtime.config.recording resolver = recording.resolver if resolver is None: logger.warning( "recording mode requires resolver but none was configured", extra={"mode": recording.mode, "source": recording.source}, ) return None try: resolved_artifacts = resolver(resolution_context) if inspect.isawaitable(resolved_artifacts): resolved_artifacts = await resolved_artifacts except Exception: logger.exception( "failed to resolve recording artifacts for hamming payload", extra={"call_id": payload_record.get("call_id"), "mode": recording.mode}, ) return None if not isinstance(resolved_artifacts, dict): logger.warning( "recording resolver returned unsupported artifact payload", extra={"mode": recording.mode, "type": type(resolved_artifacts).__name__}, ) return None return resolved_artifacts def _apply_resolved_recording_artifacts( self, *, payload_record: dict[str, Any], resolved_artifacts: RecordingArtifacts, ) -> None: recording = self._runtime.config.recording if recording.mode == RECORDING_MODE_ROOM_COMPOSITE: self._apply_room_composite_recording_artifacts( payload_record=payload_record, resolved_artifacts=resolved_artifacts, ) return self._apply_participant_egress_recording_artifacts( payload_record=payload_record, resolved_artifacts=resolved_artifacts, ) def _apply_room_composite_recording_artifacts( self, *, payload_record: dict[str, Any], resolved_artifacts: RecordingArtifacts, ) -> None: recording_url = resolved_artifacts.get("recording_url") if isinstance(recording_url, str) and recording_url.strip(): payload_record["recording_url"] = recording_url.strip() return logger.warning( "room_composite resolver did not return recording_url", extra={"call_id": payload_record.get("call_id")}, ) def _apply_participant_egress_recording_artifacts( self, *, payload_record: dict[str, Any], resolved_artifacts: RecordingArtifacts, ) -> None: recording_payload = payload_record.get("recording") if not isinstance(recording_payload, dict): recording_payload = {} payload_record["recording"] = recording_payload agent_recording_url = resolved_artifacts.get("agent_recording_url") user_recording_url = resolved_artifacts.get("user_recording_url") if isinstance(agent_recording_url, str) and agent_recording_url.strip(): recording_payload["agent_recording_url"] = agent_recording_url.strip() if isinstance(user_recording_url, str) and user_recording_url.strip(): recording_payload["user_recording_url"] = user_recording_url.strip() if ( "agent_recording_url" not in recording_payload or "user_recording_url" not in recording_payload ): logger.warning( "participant_egress resolver did not return both recording URLs", extra={"call_id": payload_record.get("call_id")}, ) async def _attach_inline_recording_capture( self, *, report: SessionReportLike, payload_record: dict[str, Any], ) -> None: if not report.audio_recording_path: return try: recording_capture = await self._runtime.transport.build_inline_recording_capture( recording_path=report.audio_recording_path, ) except Exception: logger.exception( "failed to capture inline recording for hamming payload", extra={"call_id": payload_record.get("call_id")}, ) return if recording_capture: payload_record["recording_capture"] = recording_capture async def _resolve_hamming_room_composite_recording_url( self, *, resolution_context: RecordingResolutionContext, ) -> str | None: test_case_run_id = _extract_test_case_run_id( participant_metadata_raw=resolution_context.participant_metadata_raw, recording_context=resolution_context.recording_context, ) if not test_case_run_id: logger.info( "skipping hamming test-run room composite lookup because no test-case correlation id was found", extra={ "call_id": resolution_context.call_id, "room_name": resolution_context.room_name, **_recording_debug_context( participant_metadata_raw=resolution_context.participant_metadata_raw, recording_context=resolution_context.recording_context, ), }, ) return None max_attempts = _env_int("HAMMING_ROOM_COMPOSITE_LOOKUP_ATTEMPTS", default=90) poll_interval_seconds = _env_float( "HAMMING_ROOM_COMPOSITE_LOOKUP_INTERVAL_SECONDS", default=1.0, ) for attempt in range(1, max_attempts + 1): try: recording_url = await self._runtime.transport.fetch_test_case_run_recording_url( test_case_run_id=test_case_run_id, ) except Exception: logger.exception( "failed to fetch room composite recording from hamming test run", extra={ "test_case_run_id": test_case_run_id, "call_id": resolution_context.call_id, "attempt": attempt, }, ) if attempt >= max_attempts: return None await asyncio.sleep(poll_interval_seconds) continue if recording_url: logger.info( "resolved room composite recording from hamming test run", extra={ "test_case_run_id": test_case_run_id, "call_id": resolution_context.call_id, "attempt": attempt, }, ) return recording_url if attempt < max_attempts: await asyncio.sleep(poll_interval_seconds) return None async def _start_plugin_managed_room_composite_egress( self, *, filepath: str, ) -> str: if self._job_ctx is None: raise RuntimeError("JobContext is required for plugin-managed room composite egress") egress_id = await self._runtime.transport.start_plugin_managed_room_composite_egress( room_name=self._job_ctx.room.name, filepath=filepath, ) if self._recording_context is None: self._recording_context = cast(RecordingContext, {}) self._recording_context["provider_recording_id"] = egress_id self._recording_context["provider_recording_filepath"] = filepath self._recording_context["room_name"] = self._job_ctx.room.name logger.info( "plugin-managed room composite egress started for session", extra={ "room_name": self._job_ctx.room.name, "filepath": filepath, "egress_id": egress_id, **_recording_debug_context( participant_metadata_raw=self._participant_metadata, recording_context=self._recording_context, ), }, ) return egress_id async def _resolve_plugin_managed_room_composite_recording_url( self, *, resolution_context: RecordingResolutionContext, ) -> str | None: state = self._plugin_managed_room_composite if state is None: logger.info( "plugin-managed room composite state was not initialized for session", extra={ "call_id": resolution_context.call_id, "room_name": resolution_context.room_name, **_recording_debug_context( participant_metadata_raw=resolution_context.participant_metadata_raw, recording_context=resolution_context.recording_context, ), }, ) return None try: egress_id = await state.start_task except Exception: logger.exception( "failed to start plugin-managed room composite egress", extra={"room_name": state.room_name, "filepath": state.filepath}, ) return None try: logger.info( "waiting for plugin-managed room composite egress recording URL", extra={ "egress_id": egress_id, "room_name": state.room_name, "filepath": state.filepath, "call_id": resolution_context.call_id, "max_attempts": _env_int( "HAMMING_PLUGIN_MANAGED_ROOM_COMPOSITE_LOOKUP_ATTEMPTS", default=90, ), "poll_interval_seconds": _env_float( "HAMMING_PLUGIN_MANAGED_ROOM_COMPOSITE_LOOKUP_INTERVAL_SECONDS", default=1.0, ), }, ) recording_url = await self._runtime.transport.stop_plugin_managed_room_composite_egress_and_wait_for_url( egress_id=egress_id, filepath=state.filepath, max_attempts=_env_int( "HAMMING_PLUGIN_MANAGED_ROOM_COMPOSITE_LOOKUP_ATTEMPTS", default=90, ), poll_interval_seconds=_env_float( "HAMMING_PLUGIN_MANAGED_ROOM_COMPOSITE_LOOKUP_INTERVAL_SECONDS", default=1.0, ), ) except Exception: logger.exception( "failed to finalize plugin-managed room composite egress", extra={ "egress_id": egress_id, "room_name": state.room_name, "call_id": resolution_context.call_id, }, ) return None if recording_url: logger.info( "resolved room composite recording from plugin-managed egress", extra={ "egress_id": egress_id, "room_name": state.room_name, "call_id": resolution_context.call_id, "recording_url": recording_url, }, ) else: logger.warning( "plugin-managed room composite egress did not produce a recording URL", extra={ "egress_id": egress_id, "room_name": state.room_name, "call_id": resolution_context.call_id, "filepath": state.filepath, }, ) return recording_url def _log_recording_resolution_outcome( self, *, report: SessionReportLike, payload_record: dict[str, Any], ) -> None: recording = self._runtime.config.recording recording_field_names = ( "recording_url", "recording_capture", ) nested_recording_field_names = ( "agent_recording_url", "user_recording_url", ) resolved_fields = [ field_name for field_name in recording_field_names if payload_record.get(field_name) ] nested_recording = payload_record.get("recording") if isinstance(nested_recording, dict): resolved_fields.extend( field_name for field_name in nested_recording_field_names if nested_recording.get(field_name) ) log_method = logger.info if resolved_fields else logger.warning log_method( "hamming recording payload resolution summary", extra={ "call_id": payload_record.get("call_id"), "room_name": report.room, "recording_mode": recording.mode, "recording_source": recording.source, "resolved_fields": resolved_fields, **_recording_debug_context( participant_metadata_raw=self._participant_metadata, recording_context=self._recording_context, ), }, ) class HammingRuntime: def __init__(self, config: HammingConfig) -> None: self.config = config self.transport = HammingTransport( base_url=config.base_url, api_key=config.api_key, policy=ConnectionPolicy( timeout_seconds=config.request_timeout_seconds, max_retries=config.max_retries, retry_backoff_seconds=config.retry_backoff_seconds, ), ) self._monitors: dict[int, _SessionMonitor] = {} self._tasks: set[asyncio.Task[Any]] = set() def track_task(self, task: asyncio.Task[Any]) -> None: self._tasks.add(task) task.add_done_callback(self._tasks.discard) def remove_monitor(self, session_key: int) -> None: self._monitors.pop(session_key, None) if self._monitors: return close_task = asyncio.create_task(self.aclose()) self.track_task(close_task) async def aclose(self) -> None: await self.transport.aclose() def attach_session( self, session: AgentSession[Any], *, job_ctx: JobContext | None = None, participant_identity: str | None = None, participant_metadata: str | None = None, external_agent_id: str | None = None, customer_metadata: Mapping[str, Any] | None = None, external_links: Sequence[ExternalLinkInput] | None = None, session_mode: str = SESSION_MODE_MONITORING, recording_context: RecordingContext | None = None, ) -> None: session_key = id(session) if session_key in self._monitors: return resolved_job_ctx = job_ctx or _safe_get_job_context() resolved_participant_identity = participant_identity or _participant_identity_from_job_ctx( resolved_job_ctx ) resolved_participant_metadata = participant_metadata or _participant_metadata_from_job_ctx( resolved_job_ctx ) resolved_external_agent_id = _resolve_external_agent_id( config=self.config, job_ctx=resolved_job_ctx, explicit_external_agent_id=external_agent_id, ) if self.config.auto_record_audio: _wrap_session_start_for_recording(session) elif self.config.recording.mode == RECORDING_MODE_SESSION_AUDIO: _wrap_session_start_for_recording(session) monitor = _SessionMonitor( runtime=self, session=session, participant_identity=resolved_participant_identity, participant_metadata=resolved_participant_metadata, external_agent_id=resolved_external_agent_id, customer_metadata=customer_metadata, external_links=external_links, session_mode=session_mode, job_ctx=resolved_job_ctx, session_key=session_key, recording_context=recording_context, ) session.on("close", monitor.on_close) if self.config.flush_on_shutdown and resolved_job_ctx is not None: resolved_job_ctx.add_shutdown_callback(monitor.on_shutdown) elif self.config.flush_on_shutdown: logger.warning( "flush_on_shutdown is enabled but JobContext is unavailable; final payload will " "flush only on session close" ) monitor.start_plugin_managed_room_composite_if_needed() self._monitors[session_key] = monitor def build_report( self, session: AgentSession[Any], job_ctx: JobContext | None, ) -> SessionReportLike: active_job_ctx = job_ctx or _safe_get_job_context() if active_job_ctx is None: raise RuntimeError("JobContext not found; cannot build session report") make_session_report = getattr(active_job_ctx, "make_session_report", None) if not callable(make_session_report): raise RuntimeError( "JobContext.make_session_report() is unavailable; " "livekit-agents>=1.4.4 is required for Hamming monitoring exports" ) return cast(SessionReportLike, make_session_report(session)) _RUNTIME: HammingRuntime | None = None def configure_runtime(config: HammingConfig) -> HammingRuntime: global _RUNTIME if _RUNTIME is not None: logger.warning("hamming already configured, keeping existing runtime") return _RUNTIME _validate_config(config) _apply_runtime_environment_defaults(config) runtime = HammingRuntime(config) _RUNTIME = runtime return runtime def attach_session( session: AgentSession[Any], *, job_ctx: JobContext | None = None, participant_identity: str | None = None, participant_metadata: str | None = None, external_agent_id: str | None = None, customer_metadata: Mapping[str, Any] | None = None, external_links: Sequence[ExternalLinkInput] | None = None, session_mode: str = SESSION_MODE_MONITORING, recording_context: RecordingContext | None = None, ) -> None: runtime = get_runtime() if runtime is None: raise RuntimeError("hamming is not configured. Call configure_hamming(...) first.") runtime.attach_session( session, job_ctx=job_ctx, participant_identity=participant_identity, participant_metadata=participant_metadata, external_agent_id=external_agent_id, customer_metadata=customer_metadata, external_links=external_links, session_mode=session_mode, recording_context=recording_context, ) def get_runtime() -> HammingRuntime | None: return _RUNTIME def build_config( *, api_key: str | None = None, base_url: str | None = None, call_review_enabled: bool = True, flush_on_shutdown: bool = True, auto_record_audio: bool = False, external_agent_id: str | None = None, plugin_api_version: str = "1.0.0", payload_schema_version: str = "2026-03-02", call_id_strategy: str = CALL_ID_STRATEGY_ROOM_NAME, call_id_metadata_key: str = "call_id", resolve_call_id: CallIdResolver | None = None, recording_mode: str | None = None, sampling_rate: float = 1.0, sampling_key: str = "participant_identity", capture: CaptureConfigInput | None = None, redact: RedactConfigInput | None = None, livekit: LiveKitConfigInput | None = None, livekit_url: str | None = None, livekit_api_key: str | None = None, livekit_api_secret: str | None = None, include_interim_transcripts: bool | None = None, request_timeout_seconds: float = 10.0, max_retries: int = 3, retry_backoff_seconds: float = 0.5, recording: RecordingConfigInput | None = None, streaming: StreamingConfigInput | None = None, ) -> HammingConfig: resolved_api_key = api_key or os.getenv("HAMMING_API_KEY") if not resolved_api_key: raise ValueError( "Hamming API key required. Pass api_key= or set HAMMING_API_KEY environment variable." ) resolved_base_url = ( base_url or os.getenv("HAMMING_BASE_URL") or "https://app.hamming.ai" ).rstrip("/") config = HammingConfig( api_key=resolved_api_key, base_url=resolved_base_url, call_review_enabled=call_review_enabled, flush_on_shutdown=flush_on_shutdown, auto_record_audio=auto_record_audio, external_agent_id=external_agent_id, plugin_api_version=plugin_api_version, payload_schema_version=payload_schema_version, call_id_strategy=_build_call_id_strategy( call_id_strategy=call_id_strategy, resolve_call_id=resolve_call_id, ), call_id_metadata_key=_normalize_optional_string(call_id_metadata_key) or "call_id", resolve_call_id=resolve_call_id, sampling=_build_sampling_config(rate=sampling_rate, key=sampling_key), capture=_build_capture_config( capture=capture, include_interim_transcripts=include_interim_transcripts, ), redact=_build_redact_config(redact=redact), livekit=_build_livekit_config( livekit=livekit, livekit_url=livekit_url, livekit_api_key=livekit_api_key, livekit_api_secret=livekit_api_secret, ), request_timeout_seconds=request_timeout_seconds, max_retries=max_retries, retry_backoff_seconds=retry_backoff_seconds, recording=_build_recording_config( recording=recording, auto_record_audio=auto_record_audio, recording_mode=recording_mode, ), streaming=_build_streaming_config(streaming=streaming), ) _validate_config(config) return config def _validate_config(config: HammingConfig) -> None: if not config.call_review_enabled: logger.warning("call_review_enabled is False; hamming payload export is disabled") _validate_payload_config(config) _validate_recording_config(config) _validate_streaming_config(config) def _validate_payload_config(config: HammingConfig) -> None: if not config.plugin_api_version: raise ValueError("plugin_api_version is required") if not config.payload_schema_version: raise ValueError("payload_schema_version is required") if config.call_id_strategy not in SUPPORTED_CALL_ID_STRATEGIES: raise ValueError( "Unsupported call_id_strategy. Expected one of: " f"{', '.join(sorted(SUPPORTED_CALL_ID_STRATEGIES))}." ) if config.call_id_strategy == CALL_ID_STRATEGY_CUSTOM and config.resolve_call_id is None: raise ValueError("call_id_strategy='custom' requires resolve_call_id") def _validate_recording_config(config: HammingConfig) -> None: recording = config.recording if recording.mode == RECORDING_MODE_PARTICIPANT_EGRESS: _validate_participant_egress_config(recording) return if recording.mode != RECORDING_MODE_ROOM_COMPOSITE: return if recording.delivery == RECORDING_DELIVERY_UPLOAD: raise ValueError( f"recording mode '{RECORDING_MODE_ROOM_COMPOSITE}' does not support delivery='upload' yet" ) if recording.source == RECORDING_SOURCE_EXTERNAL_URLS and recording.resolver is None: raise ValueError( f"recording mode '{RECORDING_MODE_ROOM_COMPOSITE}' requires a callable " "recording resolver when source='external_urls'" ) if recording.source == RECORDING_SOURCE_AUTO_EGRESS: missing_env_names = _missing_auto_egress_runtime_env_names(config) if missing_env_names: raise ValueError( "recording_mode='auto_egress' requires runtime configuration for: " f"{', '.join(missing_env_names)}" ) def _validate_participant_egress_config(recording: ResolvedRecordingConfig) -> None: if recording.source != RECORDING_SOURCE_EXTERNAL_URLS: raise ValueError( f"recording mode '{RECORDING_MODE_PARTICIPANT_EGRESS}' currently supports only " f"source='{RECORDING_SOURCE_EXTERNAL_URLS}'" ) if recording.delivery == RECORDING_DELIVERY_UPLOAD: raise ValueError( f"recording mode '{RECORDING_MODE_PARTICIPANT_EGRESS}' does not support delivery='upload' yet" ) if recording.resolver is None: raise ValueError( f"recording mode '{RECORDING_MODE_PARTICIPANT_EGRESS}' requires a callable recording resolver" ) def _validate_streaming_config(config: HammingConfig) -> None: if config.streaming.mode == STREAMING_MODE_TRACK_EGRESS: logger.info( "track_egress streaming is customer-managed; hamming plugin will not start streaming", ) def _safe_get_job_context() -> JobContext | None: try: return get_job_context() except RuntimeError: return None def _participant_identity_from_job_ctx(job_ctx: JobContext | None) -> str | None: if job_ctx is None: return None for participant in job_ctx.room.remote_participants.values(): identity = getattr(participant, "identity", None) if isinstance(identity, str) and identity: return identity identity = getattr(job_ctx, "local_participant_identity", None) if isinstance(identity, str) and identity: return identity agent = getattr(job_ctx, "agent", None) agent_identity = getattr(agent, "identity", None) if isinstance(agent_identity, str) and agent_identity: return agent_identity return None def _participant_metadata_from_job_ctx(job_ctx: JobContext | None) -> str | None: if job_ctx is None: return None for participant in job_ctx.room.remote_participants.values(): metadata = _serialize_participant_metadata(participant) if metadata: return metadata return None def _serialize_participant_metadata(participant: object) -> str | None: metadata = getattr(participant, "metadata", None) attributes = getattr(participant, "attributes", None) parsed_metadata = _parse_participant_metadata_json(metadata) merged_metadata = _merge_participant_attributes(parsed_metadata, attributes) if merged_metadata: return json.dumps(merged_metadata) if parsed_metadata: return json.dumps(parsed_metadata) if isinstance(metadata, str) and metadata: return metadata return None def _parse_participant_metadata_json(metadata: object) -> dict[str, Any]: if not isinstance(metadata, str) or not metadata: return {} try: parsed = json.loads(metadata) except json.JSONDecodeError: return {} if isinstance(parsed, dict): return parsed return {} def _merge_participant_attributes( metadata: dict[str, Any], attributes: object, ) -> dict[str, Any]: if not isinstance(attributes, dict) or not attributes: return {} merged = {**metadata} for key, value in attributes.items(): if isinstance(key, str): merged[key] = value return merged def _env_flag(name: str, *, default: bool) -> bool: raw = os.getenv(name) if raw is None: return default normalized = raw.strip().lower() if normalized in {"1", "true", "yes", "on"}: return True if normalized in {"0", "false", "no", "off"}: return False return default def _env_int(name: str, *, default: int) -> int: raw = os.getenv(name) if raw is None: return default try: return max(1, int(raw.strip())) except ValueError: return default def _env_float(name: str, *, default: float) -> float: raw = os.getenv(name) if raw is None: return default try: return max(0.1, float(raw.strip())) except ValueError: return default def _default_auto_recording_options() -> dict[str, bool]: return { "audio": True, "traces": False, "logs": False, "transcript": False, } def _build_recording_config( *, recording: RecordingConfigInput | None, auto_record_audio: bool, recording_mode: str | None, ) -> ResolvedRecordingConfig: if recording is None and recording_mode is not None: normalized_recording_mode = _normalize_recording_source( recording_mode, fallback=RECORDING_SOURCE_AUTO_EGRESS, ) recording = cast( RecordingConfigInput, { "mode": RECORDING_MODE_ROOM_COMPOSITE, "source": normalized_recording_mode, "delivery": RECORDING_DELIVERY_URL, }, ) if recording is None: if auto_record_audio: return ResolvedRecordingConfig(mode=RECORDING_MODE_SESSION_AUDIO) return ResolvedRecordingConfig() raw_mode = _normalize_lower_string(recording.get("mode"), fallback=RECORDING_MODE_NONE) if raw_mode not in SUPPORTED_RECORDING_MODES: raise ValueError( "Unsupported recording mode. Expected one of: " f"{', '.join(sorted(SUPPORTED_RECORDING_MODES))}." ) default_source = ( RECORDING_SOURCE_AUTO_EGRESS if raw_mode in {RECORDING_MODE_NONE, RECORDING_MODE_SESSION_AUDIO} else RECORDING_SOURCE_EXTERNAL_URLS ) source = _normalize_recording_source(recording.get("source"), fallback=default_source) if source not in SUPPORTED_RECORDING_SOURCES: raise ValueError( "Unsupported recording source. Expected one of: " f"{', '.join(sorted(SUPPORTED_RECORDING_SOURCES))}." ) delivery = _normalize_lower_string(recording.get("delivery"), fallback=RECORDING_DELIVERY_AUTO) if delivery not in SUPPORTED_RECORDING_DELIVERIES: raise ValueError( "Unsupported recording delivery. Expected one of: " f"{', '.join(sorted(SUPPORTED_RECORDING_DELIVERIES))}." ) resolver = recording.get("resolver") if resolver is not None and not callable(resolver): raise ValueError("recording.resolver must be callable") return ResolvedRecordingConfig( mode=raw_mode, source=source, delivery=delivery, resolver=resolver, ) def _build_streaming_config( *, streaming: StreamingConfigInput | None, ) -> ResolvedStreamingConfig: if streaming is None: return ResolvedStreamingConfig(mode=STREAMING_MODE_NONE) mode = _normalize_lower_string(streaming.get("mode"), fallback=STREAMING_MODE_NONE) if mode not in SUPPORTED_STREAMING_MODES: raise ValueError( "Unsupported streaming mode. Expected one of: " f"{', '.join(sorted(SUPPORTED_STREAMING_MODES))}." ) source = _normalize_lower_string( streaming.get("source"), fallback=RECORDING_SOURCE_EXTERNAL_URLS, ) if source == LEGACY_RECORDING_SOURCE_CUSTOMER_MANAGED: source = RECORDING_SOURCE_EXTERNAL_URLS elif source == LEGACY_RECORDING_SOURCE_PLUGIN_MANAGED: source = RECORDING_SOURCE_AUTO_EGRESS if source not in SUPPORTED_RECORDING_SOURCES: raise ValueError( "Unsupported streaming source. Expected one of: " f"{', '.join(sorted(SUPPORTED_RECORDING_SOURCES))}." ) return ResolvedStreamingConfig(mode=mode, source=source) def _build_call_id_strategy( *, call_id_strategy: str, resolve_call_id: CallIdResolver | None, ) -> str: normalized_strategy = _normalize_lower_string( call_id_strategy, fallback=CALL_ID_STRATEGY_ROOM_NAME, ) if normalized_strategy not in SUPPORTED_CALL_ID_STRATEGIES: raise ValueError( "Unsupported call_id_strategy. Expected one of: " f"{', '.join(sorted(SUPPORTED_CALL_ID_STRATEGIES))}." ) if normalized_strategy == CALL_ID_STRATEGY_CUSTOM and resolve_call_id is None: raise ValueError("call_id_strategy='custom' requires resolve_call_id") return normalized_strategy def _build_capture_config( *, capture: CaptureConfigInput | None, include_interim_transcripts: bool | None, ) -> ResolvedCaptureConfig: capture = capture or {} prompts = _normalize_lower_string(capture.get("prompts"), fallback="full") tools = _normalize_lower_string(capture.get("tools"), fallback="full") if prompts not in SUPPORTED_CAPTURE_MODES: raise ValueError("capture.prompts must be one of: full, none") if tools not in SUPPORTED_CAPTURE_MODES: raise ValueError("capture.tools must be one of: full, none") interim_transcripts = capture.get("interim_transcripts") if include_interim_transcripts is not None: interim_transcripts = include_interim_transcripts return ResolvedCaptureConfig( agent_context=bool(capture.get("agent_context", True)), prompts=prompts, tools=tools, interim_transcripts=bool(interim_transcripts), ) def _build_redact_config(*, redact: RedactConfigInput | None) -> ResolvedRedactConfig: redact = redact or {} return ResolvedRedactConfig( pii=bool(redact.get("pii", True)), tool_args_paths=_normalize_string_list(redact.get("tool_args_paths")), tool_output_paths=_normalize_string_list(redact.get("tool_output_paths")), ) def _build_livekit_config( *, livekit: LiveKitConfigInput | None, livekit_url: str | None, livekit_api_key: str | None, livekit_api_secret: str | None, ) -> ResolvedLiveKitConfig: livekit = livekit or {} return ResolvedLiveKitConfig( url=_normalize_optional_string(livekit.get("url") or livekit_url), api_key=_normalize_optional_string(livekit.get("api_key") or livekit_api_key), api_secret=_normalize_optional_string(livekit.get("api_secret") or livekit_api_secret), ) def _build_sampling_config(*, rate: float, key: str) -> ResolvedSamplingConfig: if rate < 0 or rate > 1: raise ValueError("sampling_rate must be between 0.0 and 1.0") normalized_key = _normalize_lower_string(key, fallback="participant_identity") return ResolvedSamplingConfig(rate=rate, key=normalized_key) def _wrap_session_start_for_recording(session: AgentSession[Any]) -> None: if getattr(session, "_hamming_auto_record_wrapped", False): return original_start = session.start async def _wrapped_start(self: AgentSession[Any], *args: Any, **kwargs: Any) -> Any: record = kwargs.get("record", NOT_GIVEN) if record is NOT_GIVEN or record is None: kwargs["record"] = _default_auto_recording_options() if inspect.ismethod(original_start): return await cast(Callable[..., Awaitable[Any]], original_start)(*args, **kwargs) return await cast(Callable[..., Awaitable[Any]], original_start)(self, *args, **kwargs) session_runtime = cast(Any, session) session_runtime.start = MethodType(_wrapped_start, session) session_runtime._hamming_auto_record_wrapped = True def _resolve_external_agent_id( *, config: HammingConfig, job_ctx: JobContext | None, explicit_external_agent_id: str | None, ) -> str: if explicit_external_agent_id: return explicit_external_agent_id if config.external_agent_id: return config.external_agent_id env_value = os.getenv("HAMMING_EXTERNAL_AGENT_ID") if env_value: return env_value if job_ctx is not None: agent_name = job_ctx.job.agent_name if agent_name: return agent_name raise ValueError( "external_agent_id is required. Pass external_agent_id= to configure_hamming(...) or " "attach_session(...), set HAMMING_EXTERNAL_AGENT_ID, or ensure job.agent_name is set." ) def _normalize_lower_string(value: object, *, fallback: str) -> str: if isinstance(value, str): normalized = value.strip().lower() if normalized: return normalized return fallback def _normalize_optional_string(value: object) -> str | None: if isinstance(value, str): normalized = value.strip() if normalized: return normalized return None def _normalize_string_list(values: object) -> tuple[str, ...]: if not isinstance(values, list): return () normalized_values: list[str] = [] for value in values: if isinstance(value, str): candidate = value.strip() if candidate: normalized_values.append(candidate) return tuple(normalized_values) def _normalize_json_object(value: object, *, field_name: str) -> dict[str, Any]: if value is None: return {} if not isinstance(value, dict): raise ValueError(f"{field_name} must be a JSON object") try: normalized = json.loads(json.dumps(value)) except (TypeError, ValueError) as exc: raise ValueError(f"{field_name} must be JSON serializable") from exc if not isinstance(normalized, dict): raise ValueError(f"{field_name} must be a JSON object") return normalized def _normalize_external_links( external_links: Sequence[ExternalLinkInput] | None, ) -> list[dict[str, str]] | None: if external_links is None: return None normalized_links: list[dict[str, str]] = [] for index, link in enumerate(external_links): if not isinstance(link, dict): raise ValueError(f"external_links[{index}] must be an object") label = _normalize_optional_string(link.get("label")) url = _normalize_optional_string(link.get("url")) if not label or not url: continue normalized_link = { "label": label, "url": url, } source = _normalize_optional_string(link.get("source")) if source: normalized_link["source"] = source description = _normalize_optional_string(link.get("description")) if description: normalized_link["description"] = description normalized_links.append(normalized_link) return normalized_links def _normalize_session_mode(session_mode: object) -> str: normalized_mode = _normalize_lower_string(session_mode, fallback=SESSION_MODE_MONITORING) if normalized_mode not in SUPPORTED_SESSION_MODES: raise ValueError( "Unsupported session_mode. Expected one of: " f"{', '.join(sorted(SUPPORTED_SESSION_MODES))}." ) return normalized_mode def _normalize_recording_source(value: object, *, fallback: str) -> str: normalized = _normalize_lower_string(value, fallback=fallback) if normalized == LEGACY_RECORDING_SOURCE_CUSTOMER_MANAGED: return RECORDING_SOURCE_EXTERNAL_URLS if normalized == LEGACY_RECORDING_SOURCE_PLUGIN_MANAGED: return RECORDING_SOURCE_AUTO_EGRESS return normalized def _build_capture_manifest(config: HammingConfig) -> dict[str, Any]: return { "agent_context": config.capture.agent_context, "prompts": config.capture.prompts, "tools": config.capture.tools, "interim_transcripts": config.capture.interim_transcripts, "recording_mode": config.recording.source, "recording_shape": config.recording.mode, "fallback_behavior": ( "resolver_required" if config.recording.source == RECORDING_SOURCE_EXTERNAL_URLS else "plugin_managed_egress" ), } def _build_sampling_metadata( *, config: HammingConfig, call_id: str, room_name: str, participant_identity: str | None, participant_metadata_raw: str | None, external_agent_id: str, ) -> dict[str, Any]: sample_key_value = _resolve_sampling_key_value( key=config.sampling.key, call_id=call_id, room_name=room_name, participant_identity=participant_identity, participant_metadata_raw=participant_metadata_raw, external_agent_id=external_agent_id, ) hash_input = f"{SAMPLING_VERSION}:{config.sampling.key}:{sample_key_value}" hash_hex = hashlib.sha256(hash_input.encode("utf-8")).hexdigest() hash_value = int(hash_hex, 16) threshold = int(config.sampling.rate * ((1 << 256) - 1)) decision = "included" if hash_value <= threshold else "excluded" return { "rate": config.sampling.rate, "key": config.sampling.key, "decision": decision, "hash": hash_hex[:8], "version": SAMPLING_VERSION, } def _resolve_sampling_key_value( *, key: str, call_id: str, room_name: str, participant_identity: str | None, participant_metadata_raw: str | None, external_agent_id: str, ) -> str: participant_metadata = _parse_json_object(participant_metadata_raw) if key in {"participant_identity", "identity"} and participant_identity: return participant_identity if key in {"call_id", "call"}: return call_id if key in {"room_name", "livekit_room_name"}: return room_name if key == "external_agent_id": return external_agent_id if key.startswith("participant_metadata."): metadata_key = key.split(".", 1)[1] value = participant_metadata.get(metadata_key) if value is not None: return str(value) return call_id def _apply_runtime_environment_defaults(config: HammingConfig) -> None: env_defaults = { "LIVEKIT_URL": config.livekit.url, "LIVEKIT_API_KEY": config.livekit.api_key, "LIVEKIT_API_SECRET": config.livekit.api_secret, } for env_name, env_value in env_defaults.items(): if env_value and not os.getenv(env_name): os.environ[env_name] = env_value def _missing_auto_egress_runtime_env_names(config: HammingConfig) -> list[str]: available_values = { "LIVEKIT_URL": config.livekit.url or os.getenv("LIVEKIT_URL", "").strip() or None, "LIVEKIT_API_KEY": config.livekit.api_key or os.getenv("LIVEKIT_API_KEY", "").strip() or None, "LIVEKIT_API_SECRET": config.livekit.api_secret or os.getenv("LIVEKIT_API_SECRET", "").strip() or None, "AWS_RECORDINGS_BUCKET": os.getenv("AWS_RECORDINGS_BUCKET", "").strip() or None, "AWS_REGION": os.getenv("AWS_REGION", "").strip() or None, "AWS_ACCESS_KEY_ID": os.getenv("AWS_ACCESS_KEY_ID", "").strip() or None, "AWS_SECRET_ACCESS_KEY": os.getenv("AWS_SECRET_ACCESS_KEY", "").strip() or None, } return [name for name, value in available_values.items() if not value] def _build_plugin_managed_room_composite_filepath( *, room_name: str, participant_metadata_raw: str | None, recording_context: RecordingContext | None, session_key: int, ) -> str: participant_metadata = _parse_json_object(participant_metadata_raw) call_id = participant_metadata.get("call_id") candidate = _extract_test_case_run_id( participant_metadata_raw=participant_metadata_raw, recording_context=recording_context, ) if candidate is None and isinstance(call_id, str) and call_id.strip(): candidate = call_id.strip() if candidate is None: candidate = room_name safe_candidate = _sanitize_recording_path_token(candidate) return ( f"{PLUGIN_MANAGED_RECORDINGS_DIR}/" f"{PLUGIN_MANAGED_ROOM_COMPOSITE_FILENAME_PREFIX}_{safe_candidate}_{session_key}." f"{PLUGIN_MANAGED_ROOM_COMPOSITE_EXTENSION}" ) def _sanitize_recording_path_token(value: str) -> str: sanitized = re.sub(r"[^A-Za-z0-9._-]+", "-", value.strip()) sanitized = re.sub(r"-{2,}", "-", sanitized) sanitized = sanitized.strip("-") return sanitized or "session" def _extract_test_case_run_id( *, participant_metadata_raw: str | None, recording_context: RecordingContext | None, ) -> str | None: participant_metadata = _parse_json_object(participant_metadata_raw) for key in TEST_CASE_RUN_ID_METADATA_KEYS: value = participant_metadata.get(key) if isinstance(value, str) and value.strip(): return value.strip() if recording_context: customer_conversation_id = recording_context.get("customer_conversation_id") if isinstance(customer_conversation_id, str) and customer_conversation_id.strip(): return customer_conversation_id.strip() return None def _recording_debug_context( *, participant_metadata_raw: str | None, recording_context: RecordingContext | None, ) -> dict[str, Any]: participant_metadata = _parse_json_object(participant_metadata_raw) metadata_keys = sorted(key for key in participant_metadata.keys() if isinstance(key, str)) return { "test_case_run_id": _extract_test_case_run_id( participant_metadata_raw=participant_metadata_raw, recording_context=recording_context, ), "participant_metadata_keys": metadata_keys, "customer_conversation_id": ( recording_context.get("customer_conversation_id") if recording_context else None ), "provider_recording_id": ( recording_context.get("provider_recording_id") if recording_context else None ), "provider_recording_filepath": ( recording_context.get("provider_recording_filepath") if recording_context else None ), "recording_context_room_name": ( recording_context.get("room_name") if recording_context else None ), } def _parse_json_object(value: str | None) -> dict[str, Any]: if not value: return {} try: parsed = json.loads(value) except json.JSONDecodeError: return {} if isinstance(parsed, dict): return parsed return {} def _string_or_fallback(value: object, fallback: str) -> str: if isinstance(value, str) and value.strip(): return value.strip() return fallback def _reset_runtime_for_tests() -> None: global _RUNTIME runtime = _RUNTIME _RUNTIME = None if runtime is None: return try: loop = asyncio.get_running_loop() except RuntimeError: asyncio.run(runtime.aclose()) return loop.create_task(runtime.aclose())