feat(conversation-recorder): add webhook support and enhance storage handling

ai_agents/agents/ten_packages/extension/conversation_recorder/README.md

@@ -31,6 +31,7 @@ boto3                  # Only needed for S3 storage
 | `filename` | string | auto-generated | Custom filename (works with all storage modes) |
 | `start_trigger` | string | `"on_user_joined"` | When to start recording: `"on_user_joined"` or `"on_start"` |
 | `sample_rate` | int | `24000` | Audio sample rate in Hz |
+| `source_prebuffer_ms` | int | `120` | Per-source recording jitter buffer before audio is drained to the WAV |
 
 ### GCS Storage Properties
 

ai_agents/agents/ten_packages/extension/conversation_recorder/audio_mixer.py

@@ -1,20 +1,37 @@
 import numpy as np
 import threading
+from dataclasses import dataclass, field
 from typing import Dict, Deque
 from collections import deque
 
 
+@dataclass
+class _SourceBuffer:
+    samples: Deque[float] = field(default_factory=deque)
+    ready: bool = False
+
+
 class AudioMixer:
-    def __init__(self, sample_rate=24000, channels=1, chunk_duration_ms=40):
+    def __init__(
+        self,
+        sample_rate=24000,
+        channels=1,
+        chunk_duration_ms=40,
+        source_prebuffer_ms=120,
+    ):
         self.sample_rate = sample_rate
         self.channels = channels
         # Calculate chunk size in samples.
         # e.g., 24000Hz * 0.04s = 960 samples
         self.chunk_size = int(self.sample_rate * (chunk_duration_ms / 1000.0))
+        self.prebuffer_size = max(
+            self.chunk_size,
+            int(self.sample_rate * (source_prebuffer_ms / 1000.0)),
+        )
 
-        # Buffers: map string(stream_id) -> Deque[float] (samples)
+        # Buffers: map string(stream_id) -> source playout state.
         # We use deque for efficient pop from left.
-        self.buffers: Dict[str, Deque[float]] = {}
+        self.buffers: Dict[str, _SourceBuffer] = {}
         self.lock = threading.Lock()
 
     def _resample(
@@ -65,40 +82,63 @@ def push_audio(
 
         with self.lock:
             if source_id not in self.buffers:
-                self.buffers[source_id] = deque()
+                self.buffers[source_id] = _SourceBuffer()
 
             # Extend deque with samples
-            self.buffers[source_id].extend(audio_array)
+            self.buffers[source_id].samples.extend(audio_array)
 
-    def mix_next_chunk(self) -> bytes:
+    def mix_next_chunk(self, drain: bool = False) -> bytes:
         """
         Extracts `chunk_size` samples from all buffers, mixes them, and returns bytes.
-        If a buffer has insufficient data, it contributes 0 (silence) for the missing part.
+        A source must have a small prebuffer before playout starts. This avoids
+        writing bursty model audio as alternating audio/silence fragments.
+        If a ready source underruns, it waits to rebuffer instead of consuming
+        a partial chunk. `drain=True` consumes partial tails during shutdown.
         If ALL buffers are empty, returns empty bytes (indicating no data to write).
         """
         with self.lock:
             # Check if any buffer has data
-            has_data = any(len(buf) > 0 for buf in self.buffers.values())
+            has_data = any(
+                len(source.samples) > 0 for source in self.buffers.values()
+            )
             if not has_data:
                 return b""
 
             # Initialize mixer buffer
             mixed_chunk = np.zeros(self.chunk_size, dtype=np.float32)
+            consumed_any = False
 
             # Mix each source
-            for buf in self.buffers.values():
-                # Extract up to chunk_size samples
-                count = min(len(buf), self.chunk_size)
-                if count > 0:
-                    # Create temporary array from deque slice
-                    # Iterating deque is fast enough for 960 items?
-                    # Ideally we'd chunk this better, but for python MVP this is readable.
-                    # Optimization: slice deque to list then array.
-                    samples = [buf.popleft() for _ in range(count)]
-                    samples_arr = np.array(samples, dtype=np.float32)
-
-                    # Add to mix
-                    mixed_chunk[:count] += samples_arr
+            for source in self.buffers.values():
+                available = len(source.samples)
+                if available == 0:
+                    continue
+
+                if drain:
+                    count = min(available, self.chunk_size)
+                else:
+                    if not source.ready:
+                        if available < self.prebuffer_size:
+                            continue
+                        source.ready = True
+
+                    if available < self.chunk_size:
+                        source.ready = False
+                        continue
+
+                    count = self.chunk_size
+
+                # Create temporary array from deque slice. This is small
+                # (usually 960 samples) and keeps the logic simple.
+                samples = [source.samples.popleft() for _ in range(count)]
+                samples_arr = np.array(samples, dtype=np.float32)
+
+                # Add to mix
+                mixed_chunk[:count] += samples_arr
+                consumed_any = True
+
+            if not consumed_any:
+                return b""
 
             # Clip and convert to int16
             mixed_chunk = np.clip(mixed_chunk, -32768, 32767)

ai_agents/agents/ten_packages/extension/conversation_recorder/extension.py

@@ -10,10 +10,11 @@
 import json
 import signal
 import atexit
+import urllib.request
+import os
 from .audio_mixer import AudioMixer
 from .storage import StorageFactory
 
-
 # Global registry of active recorders for signal handling
 _active_recorders = []
 
@@ -31,6 +32,22 @@ def _signal_handler(signum, _frame):
     raise SystemExit(128 + signum)
 
 
+def _send_webhook_request(ten_env, url: str, payload: dict):
+    try:
+        data = json.dumps(payload).encode('utf-8')
+        req = urllib.request.Request(
+            url,
+            data=data,
+            headers={'Content-Type': 'application/json'},
+            method='POST'
+        )
+        with urllib.request.urlopen(req, timeout=5) as response:
+            res_body = response.read().decode('utf-8')
+            ten_env.log_info(f"Webhook response status: {response.status}, body: {res_body}")
+    except Exception as e:
+        ten_env.log_warn(f"Failed to send webhook: {e}")
+
+
 class ConversationRecorderExtension(AsyncExtension):
     def __init__(self, name: str):
         super().__init__(name)
@@ -44,6 +61,11 @@ def __init__(self, name: str):
         self._flush_counter = 0
         self._signals_registered = False
 
+    def _get_loop(self) -> asyncio.AbstractEventLoop:
+        if self.loop is None:
+            self.loop = asyncio.get_running_loop()
+        return self.loop
+
     async def on_init(self, ten_env: AsyncTenEnv) -> None:
         ten_env.log_info("ConversationRecorderExtension on_init")
         config_json, _ = await ten_env.get_property_to_json()
@@ -52,8 +74,12 @@ async def on_init(self, ten_env: AsyncTenEnv) -> None:
 
         # Get sample rate from config, default to 24000Hz (Gemini output rate)
         sample_rate = self.config.get("sample_rate", 24000)
+        source_prebuffer_ms = self.config.get("source_prebuffer_ms", 120)
 
-        self.mixer = AudioMixer(sample_rate=sample_rate)
+        self.mixer = AudioMixer(
+            sample_rate=sample_rate,
+            source_prebuffer_ms=source_prebuffer_ms,
+        )
         self.storage = StorageFactory.create_storage(
             self.config.get("storage_type", "local"), self.config
         )
@@ -154,10 +180,11 @@ async def start_recording(self, ten_env: AsyncTenEnv):
 
         ten_env.log_info("Starting recording session...")
         self.is_recording = True
+        loop = self._get_loop()
 
         # Open storage in executor to avoid blocking
         if self.storage:
-            await self.loop.run_in_executor(None, self.storage.open)
+            await loop.run_in_executor(None, self.storage.open)
             if hasattr(self.storage, "actual_file_path"):
                 ten_env.log_info(
                     f"Recording to file: {self.storage.actual_file_path}"
@@ -171,16 +198,73 @@ async def stop_recording(self, ten_env: AsyncTenEnv):
 
         ten_env.log_info("Stopping recording session...")
         self.is_recording = False
+        loop = self._get_loop()
         if self.recording_task:
             await self.recording_task
             self.recording_task = None
 
         if self.storage:
+            await self._drain_mixer(ten_env)
             file_path = getattr(self.storage, "actual_file_path", None)
-            await self.loop.run_in_executor(None, self.storage.close)
+            try:
+                await loop.run_in_executor(None, self.storage.close)
+            except Exception as err:
+                ten_env.log_error(
+                    f"Failed to save recording at {file_path}: {err}"
+                )
+                raise
+            file_path = getattr(self.storage, "actual_file_path", file_path)
             ten_env.log_info(f"Recording saved to: {file_path}")
 
+            webhook_url = self.config.get("webhook_url") or os.getenv(
+                "CONVERSATION_RECORD_WEBHOOK_URL"
+            )
+            if webhook_url:
+                import time
+
+                channel_name = self.config.get("channel", "")
+                payload = {
+                    "channel_name": channel_name,
+                    "status": "uploaded",
+                    "file_path": file_path,
+                    "timestamp": int(time.time() * 1000),
+                }
+                ten_env.log_info(
+                    f"Sending recording upload webhook to {webhook_url} "
+                    f"with payload {payload}"
+                )
+                try:
+                    await loop.run_in_executor(
+                        None,
+                        _send_webhook_request,
+                        ten_env,
+                        webhook_url,
+                        payload,
+                    )
+                except Exception as webhook_err:
+                    ten_env.log_warn(
+                        f"Failed to await webhook request: {webhook_err}"
+                    )
+
+    async def _drain_mixer(self, ten_env: AsyncTenEnv):
+        if not getattr(self, "mixer", None) or not self.storage:
+            return
+
+        loop = self._get_loop()
+        while True:
+            mixed_bytes = self.mixer.mix_next_chunk(drain=True)
+            if not mixed_bytes:
+                break
+            try:
+                await loop.run_in_executor(
+                    None, self.storage.write, mixed_bytes
+                )
+            except Exception as e:
+                ten_env.log_error(f"Error draining recording mixer: {e}")
+                raise
+
     async def _recording_loop(self, ten_env: AsyncTenEnv):
+        loop = self._get_loop()
         while self.is_recording:
             try:
                 # Sleep approx one chunk duration (40ms)
@@ -191,7 +275,7 @@ async def _recording_loop(self, ten_env: AsyncTenEnv):
 
                 if mixed_bytes and self.storage:
                     # Write in thread pool
-                    await self.loop.run_in_executor(
+                    await loop.run_in_executor(
                         None, self.storage.write, mixed_bytes
                     )
 
@@ -200,9 +284,7 @@ async def _recording_loop(self, ten_env: AsyncTenEnv):
                 if self._flush_counter >= 25:
                     self._flush_counter = 0
                     if self.storage:
-                        await self.loop.run_in_executor(
-                            None, self.storage.flush
-                        )
+                        await loop.run_in_executor(None, self.storage.flush)
 
             except Exception as e:
                 ten_env.log_error(f"Error in recording loop: {e}")

ai_agents/agents/ten_packages/extension/conversation_recorder/manifest.json

@@ -26,6 +26,12 @@
     "api": {
         "property": {
             "properties": {
+                "webhook_url": {
+                    "type": "string"
+                },
+                "channel": {
+                    "type": "string"
+                },
                 "storage_type": {
                     "type": "string"
                 },
@@ -41,6 +47,9 @@
                 "sample_rate": {
                     "type": "int32"
                 },
+                "source_prebuffer_ms": {
+                    "type": "int32"
+                },
                 "gcp_bucket_name": {
                     "type": "string"
                 },
@@ -88,4 +97,4 @@
             "requirements.txt"
         ]
     }
-}
+}