Move recording and Discord voice to PulseAudio

drb-edge-node/app/internal/call_recorder.py

@@ -1,55 +1,100 @@
+"""
+Continuous PulseAudio ring buffer + per-call slicing.
+
+The ring-buffer design is deliberate and load-bearing: a persistent capture
+process runs for the lifetime of the node and every call is cut out of the
+buffer after the fact. Spawning FFmpeg per call used to lose the first 1-2 s to
+process startup, which meant short transmissions produced empty files. Nothing
+about that changes here — only the INPUT changes, from an HTTP GET on Icecast to
+a PulseAudio monitor capture.
+
+Why not Icecast: it lags ~1 s at connect and drifts progressively to 100 s+, so
+slice timestamps and audio content diverge without bound. Icecast stays in the
+stack for frontend/mobile live listening; it is not an accuracy path.
+
+CLOCK DOMAIN: chunks are stamped with time.time(), the host wall clock. OP25's
+call_log timestamps are time.time() from inside the op25 container. All client
+containers run network_mode: host and share the host kernel clock, so the two are
+the same clock and the pre-roll arithmetic below is a direct subtraction with no
+offset mapping. (A wall-clock STEP — e.g. a large NTP correction — would corrupt
+at most the calls in flight at that instant; the buffer self-heals within
+RING_BUFFER_SECONDS.)
+"""
 import asyncio
 import time
 from collections import deque
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Optional
+from typing import List, Optional, Tuple
 
 import httpx
 from app.config import settings
-from app.internal import credentials
+from app.internal import credentials, pulse
 from app.internal.logger import logger
 
-MAX_RECORDING_SECONDS = 600  # safety cap; drop call if it runs this long
-PRE_BUFFER_SECONDS = 1.0  # seconds of audio to include before call_start
-RING_BUFFER_SECONDS = 60  # how much history to keep when no call is active
-READ_CHUNK_BYTES = 4096  # bytes per httpx read
+# Safety cap on a single recording; mirrors MAX_SEGMENT_SECONDS in metadata_watcher.
+MAX_RECORDING_SECONDS = 600
+
+# Audio included ahead of OP25's call_log timestamp. The grant is logged when the
+# channel is granted, so the first syllable can land marginally before it.
+PRE_ROLL_SECONDS = 0.25
+
+# Rolling history kept when no call is active. Budget for the worst realistic
+# detection latency: 0.5 s poll interval + ~0.2 s http_server blocking floor +
+# up to 3 s httpx timeout on a stalled poll + callback work ≈ 4 s from grant to
+# start_recording(). 30 s is ~7x that margin, and at 16 kbps costs only ~60 KB of
+# RAM, so there is no reason to trim it closer. The buffer is what makes
+# detection latency harmless: however late we notice, we seek back to OP25's
+# exact timestamp.
+RING_BUFFER_SECONDS = 30
+
+# ~128 ms of audio per chunk at 16 kbps. Chunk size IS the timestamp resolution of
+# the ring buffer, so it must stay well under PRE_ROLL_SECONDS — the old 4096-byte
+# reads were ~2 s per chunk, which made sub-second slicing meaningless.
+READ_CHUNK_BYTES = 256
+
+# Encoder settings, matched on purpose to what Liquidsoap already pushes to
+# Icecast — %mp3(bitrate=16, samplerate=22050, stereo=false) — so the C2 /upload
+# endpoint keeps receiving exactly the kind of MP3 it has always received
+# (multipart "audio/mpeg", stored to GCS as .mp3, then fed to Whisper).
+# Change both of these together if you ever want higher-fidelity uploads.
+MP3_BITRATE = "16k"
+MP3_SAMPLE_RATE = "22050"
+
+# Backoff bounds for restarting a dead capture process.
+RESTART_BACKOFF_MIN = 1.0
+RESTART_BACKOFF_MAX = 15.0
 
 
 class CallRecorder:
-    """
-    Maintains a persistent HTTP connection to the Icecast stream and buffers
-    the raw MP3 bytes in a ring buffer.  When a call starts we note the
-    monotonic clock; when it ends we slice the buffer and write the file.
-
-    This approach eliminates per-call FFmpeg startup latency, which was
-    causing empty recordings for calls shorter than ~1–2 s.
-    """
+    """Continuous PulseAudio capture into a ring buffer, sliced per call."""
 
     def __init__(self):
         self._recordings_dir = Path(settings.recordings_path)
 
-        # Ring buffer: deque of (monotonic_time, bytes_chunk)
-        self._buffer: deque[tuple[float, bytes]] = deque()
+        # Ring buffer: deque of (wall_clock_epoch_at_arrival, mp3_bytes)
+        self._buffer: deque[Tuple[float, bytes]] = deque()
         self._buffer_bytes: int = 0
 
         self._stream_task: Optional[asyncio.Task] = None
+        self._proc: Optional[asyncio.subprocess.Process] = None
+        self._capturing: bool = False
 
-        # Active call state
+        # Active recording state
         self._call_id: Optional[str] = None
-        self._call_start_mono: Optional[float] = None
+        self._call_start: Optional[float] = None   # OP25 grant epoch
+        self._slice_start: Optional[float] = None  # _call_start - PRE_ROLL_SECONDS
 
     # ------------------------------------------------------------------
     # Lifecycle
     # ------------------------------------------------------------------
 
     async def start(self) -> None:
-        """Start the persistent stream buffer.  Call once from app lifespan."""
-        self._stream_task = asyncio.create_task(self._stream_loop())
-        logger.info("Stream ring-buffer started.")
+        """Start the persistent capture. Call once from app lifespan."""
+        self._stream_task = asyncio.create_task(self._capture_loop())
+        logger.info("PulseAudio ring-buffer starting.")
 
     async def stop(self) -> None:
-        """Cancel the stream reader."""
         if self._stream_task:
             self._stream_task.cancel()
             try:
@@ -57,93 +102,202 @@ class CallRecorder:
             except asyncio.CancelledError:
                 pass
             self._stream_task = None
+        await self._terminate_proc()
 
     # ------------------------------------------------------------------
-    # Stream reader
+    # Capture
     # ------------------------------------------------------------------
 
-    async def _stream_loop(self) -> None:
-        stream_url = (
-            f"http://{settings.icecast_host}:{settings.icecast_port}"
-            f"{settings.icecast_mount}"
-        )
-        timeout = httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0)
+    def _ffmpeg_command(self) -> List[str]:
+        return [
+            "ffmpeg",
+            "-hide_banner", "-nostdin", "-nostats",
+            "-loglevel", "warning",
+            "-f", "pulse", "-i", settings.pulse_source,
+            "-ac", "1",
+            "-ar", MP3_SAMPLE_RATE,
+            "-b:a", MP3_BITRATE,
+            # Without this, the MP3 muxer fills its 32 KB AVIO buffer before
+            # writing anything — 16 s of audio per burst at 16 kbps, which would
+            # destroy the arrival timestamps the slicing depends on.
+            "-flush_packets", "1",
+            "-f", "mp3", "-",
+        ]
+
+    async def _capture_loop(self) -> None:
+        backoff = RESTART_BACKOFF_MIN
         while True:
             try:
-                async with httpx.AsyncClient(timeout=timeout) as client:
-                    async with client.stream("GET", stream_url) as response:
-                        response.raise_for_status()
-                        logger.info(f"Stream buffer connected to {stream_url}")
-                        async for chunk in response.aiter_bytes(READ_CHUNK_BYTES):
-                            self._ingest(chunk)
+                # The original bug this path was abandoned for: FFmpeg was launched
+                # with -f pulse before the shared socket existed, failed instantly,
+                # and never recovered. Wait for it, bounded, every time.
+                if not await pulse.wait_until_ready():
+                    await asyncio.sleep(backoff)
+                    backoff = min(backoff * 2, RESTART_BACKOFF_MAX)
+                    continue
+
+                await self._run_capture()
+                logger.warning("PulseAudio capture process exited — restarting.")
             except asyncio.CancelledError:
+                await self._terminate_proc()
                 raise
             except Exception as e:
-                logger.warning(f"Stream buffer disconnected ({e}) — retrying in 3 s")
-                await asyncio.sleep(3)
+                logger.warning(f"PulseAudio capture error ({e}) — restarting.")
+
+            self._capturing = False
+            await asyncio.sleep(backoff)
+            backoff = min(backoff * 2, RESTART_BACKOFF_MAX)
+
+    async def _run_capture(self) -> None:
+        cmd = self._ffmpeg_command()
+        logger.info(f"Starting capture: ffmpeg -f pulse -i {settings.pulse_source}")
+        proc = await asyncio.create_subprocess_exec(
+            *cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        self._proc = proc
+        stderr_task = asyncio.create_task(self._drain_stderr(proc))
+        try:
+            assert proc.stdout is not None
+            while True:
+                chunk = await proc.stdout.read(READ_CHUNK_BYTES)
+                if not chunk:
+                    break  # EOF — FFmpeg died or the source went away
+                if not self._capturing:
+                    self._capturing = True
+                    logger.info("PulseAudio capture is producing audio.")
+                self._ingest(chunk)
+        finally:
+            self._capturing = False
+            # _drain_stderr swallows its own CancelledError, so it finishes cleanly
+            # and never needs awaiting here.
+            stderr_task.cancel()
+            await self._terminate_proc()
+
+    async def _drain_stderr(self, proc: asyncio.subprocess.Process) -> None:
+        """Surface FFmpeg's diagnostics instead of letting the pipe fill and block."""
+        if proc.stderr is None:
+            return
+        try:
+            while True:
+                line = await proc.stderr.readline()
+                if not line:
+                    return
+                logger.warning(f"ffmpeg(pulse): {line.decode(errors='replace').strip()}")
+        except asyncio.CancelledError:
+            return
+        except Exception:
+            return
+
+    async def _terminate_proc(self) -> None:
+        proc, self._proc = self._proc, None
+        if proc is None or proc.returncode is not None:
+            return
+        try:
+            # Synchronous, so the signal lands even if we are being cancelled and
+            # the reap below never gets to run.
+            proc.terminate()
+        except Exception:
+            return
+        try:
+            await asyncio.wait_for(proc.wait(), timeout=5)
+        except asyncio.TimeoutError:
+            try:
+                proc.kill()
+            except Exception:
+                pass
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            pass
 
     def _ingest(self, chunk: bytes) -> None:
-        """Append a chunk and trim stale data from the front of the buffer."""
-        now = time.monotonic()
+        """Append a chunk and trim stale audio off the front of the buffer."""
+        now = time.time()
         self._buffer.append((now, chunk))
         self._buffer_bytes += len(chunk)
 
-        # During a call, never trim data newer than (call_start - pre_buffer).
-        # Between calls, keep a rolling RING_BUFFER_SECONDS window.
-        if self._call_start_mono is not None:
-            keep_from = self._call_start_mono - PRE_BUFFER_SECONDS
+        # While recording, never trim anything the current slice still needs — but
+        # keep a hard ceiling so a recording that somehow never closes can't grow
+        # the buffer without bound.
+        if self._slice_start is not None:
+            keep_from = max(
+                self._slice_start,
+                now - (MAX_RECORDING_SECONDS + RING_BUFFER_SECONDS),
+            )
         else:
             keep_from = now - RING_BUFFER_SECONDS
 
-        while self._buffer:
-            ts, old = self._buffer[0]
-            if ts >= keep_from:
-                break
-            self._buffer.popleft()
+        while self._buffer and self._buffer[0][0] < keep_from:
+            _, old = self._buffer.popleft()
             self._buffer_bytes -= len(old)
 
     # ------------------------------------------------------------------
-    # Call recording API (same interface as before)
+    # Recording API
     # ------------------------------------------------------------------
 
-    async def start_recording(self, call_id: str) -> bool:
+    async def start_recording(self, call_id: str, start_epoch: Optional[float] = None) -> bool:
+        """
+        Open a recording. `start_epoch` is OP25's call_log timestamp (host wall
+        clock); the slice begins PRE_ROLL_SECONDS before it. Omit it only when no
+        OP25 timestamp is available — then we fall back to "now", losing precision.
+        """
         if self._call_id:
-            logger.warning("Recording already active — ignoring start.")
+            logger.warning(f"Recording already active ({self._call_id}) — ignoring start for {call_id}.")
             return False
+
         self._call_id = call_id
-        self._call_start_mono = time.monotonic()
-        logger.info(f"Recording started (ring-buffer): {call_id}")
+        self._call_start = start_epoch if start_epoch else time.time()
+        self._slice_start = self._call_start - PRE_ROLL_SECONDS
+
+        if not self._capturing:
+            logger.warning(f"Recording {call_id} opened while PulseAudio capture is down — audio may be missing.")
+
+        oldest = self._buffer[0][0] if self._buffer else None
+        if oldest is not None and self._slice_start < oldest:
+            # Pre-roll predates the buffer: node just started, capture restarted,
+            # or OP25's timestamp is far in the past. Clamp and say so.
+            logger.warning(
+                f"Pre-roll for call {call_id} predates buffered audio by "
+                f"{oldest - self._slice_start:.2f}s — recording starts at the buffer head."
+            )
+
+        logger.info(f"Recording started: {call_id} (slice from {self._slice_start:.3f})")
         return True
 
-    async def stop_recording(self) -> Optional[Path]:
+    async def stop_recording(self, end_epoch: Optional[float] = None) -> Optional[Path]:
+        """Close the recording and write the slice. `end_epoch` is host wall clock."""
         if not self._call_id:
             return None
 
         call_id = self._call_id
-        call_start = self._call_start_mono
+        call_start = self._call_start
+        slice_start = self._slice_start
         self._call_id = None
-        self._call_start_mono = None
+        self._call_start = None
+        self._slice_start = None
 
-        # Slice: everything from (call_start - pre_buffer) to now
-        cutoff = (call_start - PRE_BUFFER_SECONDS) if call_start else 0.0
-        chunks = [chunk for ts, chunk in self._buffer if ts >= cutoff]
-
-        # Safety cap: if the call ran very long, truncate to MAX_RECORDING_SECONDS
+        end = end_epoch if end_epoch else time.time()
         if call_start is not None:
-            cap_cutoff = call_start + MAX_RECORDING_SECONDS
-            now = time.monotonic()
-            if now > cap_cutoff:
-                # Approximate: trim chunks that arrived after the cap
-                cap_keep_until = call_start + MAX_RECORDING_SECONDS
-                chunks = [
-                    chunk for ts, chunk in self._buffer
-                    if cutoff <= ts <= cap_keep_until
-                ]
+            end = min(end, call_start + MAX_RECORDING_SECONDS)
+        if slice_start is None:
+            return None
+
+        chunks: List[bytes] = []
+        for ts, chunk in self._buffer:
+            if ts < slice_start:
+                continue
+            chunks.append(chunk)
+            if ts >= end:
+                # Include the chunk straddling `end` so the tail is never clipped,
+                # then stop.
+                break
 
         if not chunks:
             logger.warning(
-                f"No buffered audio for call {call_id} — "
-                "stream may not have been connected yet."
+                f"No buffered audio for call {call_id} "
+                f"(window {slice_start:.3f}–{end:.3f}) — PulseAudio capture may be down."
             )
             return None
 
@@ -151,12 +305,11 @@ class CallRecorder:
         ts_str = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
         output_path = self._recordings_dir / f"{ts_str}_{call_id}.mp3"
 
-        data = b"".join(chunks)
-        output_path.write_bytes(data)
+        output_path.write_bytes(b"".join(chunks))
 
         size = output_path.stat().st_size
         if size > 0:
-            logger.info(f"Recording saved: {output_path.name} ({size} bytes)")
+            logger.info(f"Recording saved: {output_path.name} ({size} bytes, {end - slice_start:.2f}s window)")
             return output_path
 
         output_path.unlink(missing_ok=True)
@@ -213,9 +366,24 @@ class CallRecorder:
             except Exception:
                 pass
 
+    # ------------------------------------------------------------------
+    # State
+    # ------------------------------------------------------------------
+
     @property
     def is_recording(self) -> bool:
         return self._call_id is not None
 
+    @property
+    def is_capturing(self) -> bool:
+        """True when FFmpeg is alive and audio is actually arriving."""
+        return self._capturing
+
+    @property
+    def buffered_seconds(self) -> float:
+        if len(self._buffer) < 2:
+            return 0.0
+        return self._buffer[-1][0] - self._buffer[0][0]
+
 
 call_recorder = CallRecorder()

drb-edge-node/app/internal/discord_radio.py

@@ -2,11 +2,14 @@ import asyncio
 from typing import Optional
 import discord
 from discord.ext import commands
+from app.config import settings
+from app.internal import pulse
 from app.internal.logger import logger
 
 BOT_READY_TIMEOUT = 15   # seconds to wait for Discord bot to become ready
 WATCHDOG_INTERVAL = 30   # seconds between voice-connection health checks
 REJOIN_DELAY = 5  # seconds to wait before attempting a rejoin
+STREAM_RETRY_DELAY = 5   # seconds to back off before re-arming the audio source
 
 
 class RadioBot:
@@ -47,6 +50,10 @@ class RadioBot:
             # Remember where we are so the watchdog can rejoin if we drop
             self._guild_id = guild_id
             self._channel_id = channel_id
+            # Bounded wait for the shared PulseAudio socket. Historically FFmpeg was
+            # launched before the op25 container had created it, failed instantly,
+            # and the bot sat silently connected forever.
+            await pulse.wait_until_ready()
             self._play_stream()
             if system_name:
                 await self._bot.change_presence(
@@ -108,19 +115,48 @@ class RadioBot:
         self._ready_event = None
 
     def _play_stream(self):
+        """
+        Feed Discord voice straight from the PulseAudio monitor.
+
+        Icecast is NOT used here: it lags ~1 s at connect and drifts to 100 s+,
+        which is unusable for live listening. Icecast remains the frontend/mobile
+        listening path only.
+        """
         if not self._voice_client:
             return
-        from app.config import settings
-        stream_url = f"http://{settings.icecast_host}:{settings.icecast_port}{settings.icecast_mount}"
+
+        if not pulse.is_ready():
+            logger.error(
+                f"PulseAudio socket {pulse.socket_path()} missing — "
+                f"cannot start Discord audio; retrying in {STREAM_RETRY_DELAY}s."
+            )
+            self._schedule_restart()
+            return
+
+        # before_options land ahead of -i, so this becomes:
+        #   ffmpeg -f pulse -i drb_sink.monitor …
         source = discord.FFmpegPCMAudio(
-            stream_url,
-            before_options="-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5",
+            settings.pulse_source,
+            before_options="-f pulse",
         )
         self._voice_client.play(
             discord.PCMVolumeTransformer(source, volume=1.0),
             after=self._on_stream_end,
         )
 
+    def _schedule_restart(self, delay: float = STREAM_RETRY_DELAY):
+        """Re-arm the audio source after a delay — safe to call from any thread."""
+        if not self._loop:
+            return
+
+        async def _delayed_restart():
+            await asyncio.sleep(delay)
+            vc = self._voice_client
+            if vc and vc.is_connected() and not vc.is_playing():
+                self._play_stream()
+
+        self._loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_delayed_restart()))
+
     def _on_stream_end(self, error):
         if error:
             logger.error(f"Stream ended with error: {error}")
@@ -128,12 +164,9 @@ class RadioBot:
         if not (self._loop and vc and vc.is_connected() and not vc.is_playing()):
             return
         if error:
-            # Back off before retrying — prevents tight loop when PulseAudio is unavailable
-            async def _delayed_restart():
-                await asyncio.sleep(5)
-                if self._voice_client and self._voice_client.is_connected() and not self._voice_client.is_playing():
-                    self._play_stream()
-            self._loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_delayed_restart()))
+            # Back off before retrying — prevents a tight loop when PulseAudio is
+            # unavailable (FFmpeg exits immediately in that case).
+            self._schedule_restart()
         else:
             self._loop.call_soon_threadsafe(self._play_stream)
 
@@ -232,6 +265,9 @@ class RadioBot:
                     else:
                         self._voice_client = await vc.connect()
                     self._channel_id = vc.id
+                    # A fresh connect() has no audio source attached yet.
+                    if not self._voice_client.is_playing():
+                        self._play_stream()
                     await message.reply(f"Joined {vc.name}.")
                 except Exception as e:
                     logger.error(f"joinme failed: {e}")

drb-edge-node/app/internal/metadata_watcher.py

@@ -1,38 +1,125 @@
+"""
+Event-driven call state machine.
+
+Replaces the old hang-counter inference (which derived call start from "a tgid
+appeared in channel_update" and call end from N polls of silence) with the two
+authoritative signals OP25 actually exposes:
+
+  START — a `call_log` entry. OP25 appends one at channel-grant time stamped with
+          its own time.time(). This is an exact start timestamp, not the moment
+          our poll happened to notice, so recordings can be sliced back to it.
+
+  END   — the `srcaddr` != 0 → `srcaddr` == 0 transition in `channel_update`.
+          OP25 never reports call termination externally: internally it ends a
+          call on the P25 Terminator Data Unit (duid15) or 3 voice-framing
+          timeouts, but neither becomes a log entry. What *is* observable is that
+          `srcaddr`/`svcopts` reset to 0/false the instant the call ends, while
+          `tgid`/`hold_tgid` keep showing the just-ended talkgroup for
+          TGID_HOLD_TIME (2 s). So the srcaddr edge is a real state change, not a
+          timeout heuristic.
+
+SEGMENTS: one emitted call (= one recording, one Firestore doc) spans a whole
+conversation, not a single transmission. It stays open across repeated grants on
+the same talkgroup and closes when the talkgroup changes or the radio goes quiet
+for settings.call_idle_timeout seconds.
+
+CLOCKS: `call_log["time"]` is time.time() inside the op25 container. All three
+client containers run network_mode: host and share the host kernel clock, so that
+value is directly comparable to time.time() here — no offset mapping needed. The
+call recorder's ring buffer is stamped with the same clock for the same reason.
+"""
 import asyncio
+import time
 import uuid
 from datetime import datetime, timezone
-from typing import Optional, Callable, Awaitable
+from typing import Optional, Callable, Awaitable, Any, List, Dict
+
+from app.config import settings
 from app.internal.op25_client import op25_client
 from app.internal.logger import logger
 
 CallbackFn = Callable[[dict], Awaitable[None]]
 
-HANG_THRESHOLD = 2   # polls before declaring a call ended (0.5s poll → 1s hang time)
-POLL_INTERVAL = 0.5  # seconds
+# 500 ms. Do NOT lower: start precision already comes from OP25's own timestamp,
+# and http_server.py's request handler has a ~200 ms blocking floor anyway.
+POLL_INTERVAL = 0.5
+
+# Seconds of unreachable OP25 before an open segment is force-closed.
+OP25_OFFLINE_GRACE = 3.0
+
+# Audio kept after the observed end of the last transmission, so the srcaddr edge
+# (up to one poll late) never clips the tail.
+TAIL_PAD_SECONDS = 0.5
+
+# Hard ceiling on a single segment; mirrors MAX_RECORDING_SECONDS in call_recorder
+# so a talkgroup that never goes quiet cannot produce an unbounded recording.
+MAX_SEGMENT_SECONDS = 600
+
+
+def _as_int(value: Any) -> Optional[int]:
+    """Coerce an OP25 field to a positive int, or None. Rejects 0/""/"None"."""
+    if value is None:
+        return None
+    try:
+        number = int(value)
+    except (TypeError, ValueError):
+        return None
+    return number if number > 0 else None
+
+
+def _as_float(value: Any) -> Optional[float]:
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _iso(epoch: Optional[float]) -> Optional[str]:
+    if epoch is None:
+        return None
+    return datetime.fromtimestamp(epoch, timezone.utc).isoformat()
 
 
 class MetadataWatcher:
     def __init__(self):
         self._running = False
+
+        # Open segment state
+        self._active_call_id: Optional[str] = None
         self._current_tgid: Optional[int] = None
         self._current_tgid_name: Optional[str] = None
-        self._hang_counter: int = 0
-        self._active_call_id: Optional[str] = None
-        self._call_started_at: Optional[datetime] = None
+        self._current_freq: Any = None
+        self._current_srcaddr: Optional[int] = None
+        self._started_at: Optional[float] = None   # OP25 epoch of the first grant
+        self._transmissions: int = 0
+
+        # Transmission tracking within the open segment
+        self._tx_active: bool = False              # last poll saw srcaddr != 0
+        self._last_activity: float = 0.0           # epoch of last evidence of traffic
+        self._last_tx_end: Optional[float] = None  # epoch of the srcaddr 1→0 edge
+        self._last_ok_poll: float = 0.0
+
+        # Injectable for tests; production is always the host wall clock.
+        self._clock: Callable[[], float] = time.time
 
         # Set these before calling start()
         self.on_call_start: Optional[CallbackFn] = None
         self.on_call_end: Optional[CallbackFn] = None
 
+    # ------------------------------------------------------------------
+    # Lifecycle
+    # ------------------------------------------------------------------
+
     async def start(self):
         self._running = True
+        self._last_ok_poll = self._clock()
         asyncio.create_task(self._poll_loop())
-        logger.info("Metadata watcher started.")
+        logger.info("Metadata watcher started (call_log driven).")
 
     async def stop(self):
         self._running = False
         if self._active_call_id:
-            await self._end_call()
+            await self._close_segment(self._clock(), reason="shutdown")
 
     async def _poll_loop(self):
         while self._running:
@@ -42,79 +129,218 @@ class MetadataWatcher:
                 logger.warning(f"Metadata poll error: {e}")
             await asyncio.sleep(POLL_INTERVAL)
 
-    async def _tick(self):
-        status = await op25_client.get_terminal_status()
+    # ------------------------------------------------------------------
+    # One poll
+    # ------------------------------------------------------------------
 
-        if not status:
-            # OP25 not responding — hang-out any active call
-            if self._active_call_id:
-                self._hang_counter += 1
-                if self._hang_counter >= HANG_THRESHOLD:
-                    await self._end_call()
+    async def _tick(self):
+        now = self._clock()
+        update = await op25_client.poll_terminal()
+
+        if update is None:
+            # OP25 unreachable. Don't kill an open segment on a single blip.
+            if self._active_call_id and (now - self._last_ok_poll) >= OP25_OFFLINE_GRACE:
+                await self._close_segment(now, reason="op25_unreachable")
             return
 
-        # OP25 terminal returns either a list of channels or a single dict
-        channels = status if isinstance(status, list) else [status]
-        active_tgid: Optional[int] = None
-        active_meta: dict = {}
+        self._last_ok_poll = now
 
-        for ch in channels:
-            tgid = ch.get("tgid") or ch.get("tg_id")
-            if tgid and str(tgid) not in ("0", "", "None"):
-                active_tgid = int(tgid)
-                active_meta = ch
-                break
+        # 1. call_log first — these are the authoritative starts, and processing
+        #    them before the channel scan means a same-poll grant+state pair is
+        #    already attributed to the new segment by the time we scan channels.
+        #    Sorted defensively: multi-receiver setups append per receiver.
+        for entry in sorted(update.call_log, key=lambda e: _as_float(e.get("time")) or 0.0):
+            await self._handle_call_log(entry, now)
 
-        if active_tgid:
-            self._hang_counter = 0
-            if self._current_tgid != active_tgid:
-                # Talkgroup changed — close previous call and open a new one
-                if self._active_call_id:
-                    await self._end_call()
-                self._current_tgid = active_tgid
-                await self._start_call(active_tgid, active_meta)
-        else:
-            # No active talkgroup
-            if self._active_call_id:
-                self._hang_counter += 1
-                if self._hang_counter >= HANG_THRESHOLD:
-                    await self._end_call()
+        # 2. channel_update — the only external end signal.
+        await self._handle_channels(update.channels, now)
 
-    async def _start_call(self, tgid: int, meta: dict):
+    async def _handle_call_log(self, entry: Dict[str, Any], now: float) -> None:
+        tgid = _as_int(entry.get("tgid"))
+        if tgid is None:
+            return  # a grant with no talkgroup is nothing we can record or label
+
+        # OP25's own stamp. Fall back to now only if the field is missing/garbage.
+        started_at = _as_float(entry.get("time"))
+        if started_at is None:
+            logger.warning(f"call_log entry for tgid={tgid} has no usable time — using local clock.")
+            started_at = now
+
+        if self._active_call_id is None:
+            await self._open_segment(entry, tgid, started_at, now)
+            return
+
+        if tgid == self._current_tgid:
+            # CONTINUE: same talkgroup, keep one recording so the back-and-forth
+            # of a single conversation lands in one file.
+            self._transmissions += 1
+            self._tx_active = True
+            self._last_tx_end = None
+            self._last_activity = now
+            self._refresh_meta_from_log(entry)
+            return
+
+        # SPLIT: different talkgroup. The new grant's OP25 timestamp is the most
+        # precise end available for the outgoing segment — the new call's audio
+        # starts exactly there, so no tail pad.
+        await self._close_segment(started_at, reason="tgid_change")
+        await self._open_segment(entry, tgid, started_at, now)
+
+    async def _handle_channels(self, channels: List[Dict[str, Any]], now: float) -> None:
+        if self._active_call_id is None:
+            return
+
+        tx_active = False
+        foreign_active_tgid: Optional[int] = None
+
+        for channel in channels:
+            srcaddr = _as_int(channel.get("srcaddr"))
+            chan_tgid = _as_int(channel.get("tgid"))
+            if srcaddr is None:
+                continue
+            if chan_tgid == self._current_tgid:
+                tx_active = True
+                self._current_srcaddr = srcaddr
+                self._refresh_meta_from_channel(channel)
+            elif chan_tgid is not None:
+                foreign_active_tgid = chan_tgid
+
+        if tx_active:
+            self._tx_active = True
+            self._last_tx_end = None
+            self._last_activity = now
+        elif self._tx_active:
+            # The srcaddr != 0 → 0 edge: OP25 has torn the call down.
+            self._tx_active = False
+            self._last_tx_end = now
+            self._last_activity = now
+
+        # Safety net for a dropped call_log event (deque is capped at 10): the one
+        # receiver we have is plainly on another talkgroup, so our segment is over
+        # even though we never saw its grant. Close now rather than record
+        # call_idle_timeout seconds of the wrong tgid.
+        #
+        # Restricted to single-receiver setups on purpose: with several receivers,
+        # another channel being busy says nothing about ours, and closing on it
+        # would truncate every call whenever a second receiver is active.
+        if not tx_active and foreign_active_tgid is not None and len(channels) == 1:
+            logger.warning(
+                f"tgid {foreign_active_tgid} active without a call_log entry — "
+                f"closing segment for tgid {self._current_tgid} (call_log event likely dropped)."
+            )
+            await self._close_segment(now, reason="tgid_change_unlogged")
+            return
+
+        if (now - self._last_activity) >= settings.call_idle_timeout:
+            # STOP: quiet for long enough. End the audio at the last transmission
+            # plus a short pad, not at "now" — otherwise every recording carries
+            # call_idle_timeout seconds of silence.
+            end = (self._last_tx_end + TAIL_PAD_SECONDS) if self._last_tx_end is not None else now
+            await self._close_segment(min(end, now), reason="idle_timeout")
+            return
+
+        if self._started_at is not None and (now - self._started_at) >= MAX_SEGMENT_SECONDS:
+            logger.warning(f"Segment for tgid {self._current_tgid} hit the {MAX_SEGMENT_SECONDS}s cap — closing.")
+            await self._close_segment(now, reason="max_length")
+
+    # ------------------------------------------------------------------
+    # Segment open / close
+    # ------------------------------------------------------------------
+
+    def _refresh_meta_from_log(self, entry: Dict[str, Any]) -> None:
+        if not self._current_tgid_name:
+            self._current_tgid_name = entry.get("tgtag") or ""
+        if entry.get("freq"):
+            self._current_freq = entry.get("freq")
+        rid = _as_int(entry.get("rid"))
+        if rid is not None:
+            self._current_srcaddr = rid
+
+    def _refresh_meta_from_channel(self, channel: Dict[str, Any]) -> None:
+        if not self._current_tgid_name:
+            self._current_tgid_name = channel.get("tag") or ""
+        if not self._current_freq and channel.get("freq"):
+            self._current_freq = channel.get("freq")
+
+    async def _open_segment(self, entry: Dict[str, Any], tgid: int, started_at: float, now: float) -> None:
         self._active_call_id = str(uuid.uuid4())
-        self._call_started_at = datetime.now(timezone.utc)
-        self._current_tgid_name = meta.get("tag") or meta.get("tgid_tag") or ""
+        self._current_tgid = tgid
+        self._current_tgid_name = entry.get("tgtag") or ""
+        self._current_freq = entry.get("freq")
+        self._current_srcaddr = _as_int(entry.get("rid"))
+        self._started_at = started_at
+        self._transmissions = 1
+
+        # Assume the transmission is still up: we learn otherwise from the next
+        # channel scan. A grant whose call already ended before we polled simply
+        # closes on the very next tick via the idle timeout.
+        self._tx_active = True
+        self._last_tx_end = None
+        self._last_activity = now
+
         payload = {
             "call_id": self._active_call_id,
             "tgid": tgid,
             "tgid_name": self._current_tgid_name,
-            "freq": meta.get("freq"),
-            "srcaddr": meta.get("srcaddr"),
-            "started_at": self._call_started_at.isoformat(),
+            "freq": self._current_freq,
+            "srcaddr": self._current_srcaddr,
+            "started_at": _iso(started_at),
+            # Raw epoch for the recorder's ring-buffer slice — same clock domain.
+            "started_at_epoch": started_at,
         }
-        logger.info(f"Call start: tgid={tgid} id={self._active_call_id}")
+        logger.info(
+            f"Call start: tgid={tgid} id={self._active_call_id} "
+            f"(op25 t={started_at:.3f}, detected {now - started_at:+.2f}s later)"
+        )
         if self.on_call_start:
             await self.on_call_start(payload)
 
-    async def _end_call(self):
+    async def _close_segment(self, end_epoch: float, reason: str) -> None:
         if not self._active_call_id:
             return
+
+        started_at = self._started_at
+        if started_at is not None:
+            end_epoch = max(end_epoch, started_at)
+
         payload = {
             "call_id": self._active_call_id,
             "tgid": self._current_tgid,
             "tgid_name": self._current_tgid_name or "",
-            "started_at": self._call_started_at.isoformat() if self._call_started_at else None,
-            "ended_at": datetime.now(timezone.utc).isoformat(),
+            "freq": self._current_freq,
+            "srcaddr": self._current_srcaddr,
+            "started_at": _iso(started_at),
+            "started_at_epoch": started_at,
+            "ended_at": _iso(end_epoch),
+            "ended_at_epoch": end_epoch,
+            "transmissions": self._transmissions,
+            "end_reason": reason,
         }
-        logger.info(f"Call end: id={self._active_call_id}")
+        duration = (end_epoch - started_at) if started_at is not None else 0.0
+        logger.info(
+            f"Call end: id={self._active_call_id} tgid={self._current_tgid} "
+            f"reason={reason} transmissions={self._transmissions} duration={duration:.2f}s"
+        )
+
+        # Clear state before awaiting so a re-entrant tick can't see a half-closed
+        # segment (and so an immediately-following _open_segment is clean).
         self._active_call_id = None
         self._current_tgid = None
         self._current_tgid_name = None
-        self._hang_counter = 0
-        self._call_started_at = None
+        self._current_freq = None
+        self._current_srcaddr = None
+        self._started_at = None
+        self._transmissions = 0
+        self._tx_active = False
+        self._last_tx_end = None
+
         if self.on_call_end:
             await self.on_call_end(payload)
 
+    # ------------------------------------------------------------------
+    # Public state (consumed by routers/api.py, main.py and the dashboards)
+    # ------------------------------------------------------------------
+
     @property
     def active_call_id(self) -> Optional[str]:
         return self._active_call_id

drb-edge-node/app/internal/op25_client.py

@@ -1,8 +1,34 @@
 import httpx
-from typing import Optional, Dict, Any
+from dataclasses import dataclass, field
+from typing import Optional, Dict, Any, List
 from app.config import settings
 from app.internal.logger import logger
 
+# The OP25 HTTP terminal answers a single "update" command with a LIST of
+# messages, each tagged with a `json_type`. We care about two of them:
+#
+#   channel_update — current receiver state. `channels` holds the channel ids and
+#                    each id is also a top-level key holding that channel's dict
+#                    (freq/tgid/tag/srcaddr/svcopts/hold_tgid/…).
+#
+#   call_log       — an EVENT QUEUE, not a snapshot. `log` holds entries appended
+#                    by tk_p25.log_call() at channel-grant time, each stamped with
+#                    OP25's own time.time(). get_call_log() DRAINS the deque, so
+#                    every entry is delivered exactly once and a missed poll loses
+#                    it forever. The deque is capped at CALL_LOG_MAX_LEN = 10, so
+#                    the consumer must keep up.
+#
+# Everything else (trunk_update, rx_update, terminal_config, …) is ignored.
+TERMINAL_UPDATE_COMMAND = [{"command": "update", "arg1": 0, "arg2": 0}]
+
+
+@dataclass
+class TerminalUpdate:
+    """One decoded poll of the OP25 HTTP terminal."""
+
+    channels: List[Dict[str, Any]] = field(default_factory=list)
+    call_log: List[Dict[str, Any]] = field(default_factory=list)
+
 
 class OP25Client:
     def __init__(self):
@@ -49,24 +75,67 @@ class OP25Client:
             logger.error(f"OP25 generate-config failed: {e}")
             return False
 
-    async def get_terminal_status(self) -> Optional[Any]:
-        """Poll the OP25 HTTP terminal for current call metadata."""
+    async def poll_terminal(self) -> Optional[TerminalUpdate]:
+        """
+        Poll the OP25 HTTP terminal once and decode every message we understand.
+
+        Returns None only when OP25 is unreachable / returned garbage — callers
+        use that to distinguish "no traffic" from "no OP25".
+        """
         try:
             async with httpx.AsyncClient(timeout=3) as client:
-                r = await client.post(
-                    self.terminal_url,
-                    json=[{"command": "update", "arg1": 0, "arg2": 0}],
-                )
+                r = await client.post(self.terminal_url, json=TERMINAL_UPDATE_COMMAND)
                 r.raise_for_status()
-                messages = r.json()
-                for msg in messages:
-                    if msg.get("json_type") == "channel_update":
-                        channels = msg.get("channels", [])
-                        if channels:
-                            return msg.get(str(channels[0]), {})
-                return None
+                return parse_terminal_messages(r.json())
         except Exception:
             return None
 
+    async def get_terminal_status(self) -> Optional[Dict[str, Any]]:
+        """
+        Compatibility shim: the first channel's state dict, as this used to return.
+
+        Prefer poll_terminal() — this discards the call_log, which is the only
+        source of exact call-start timestamps.
+        """
+        update = await self.poll_terminal()
+        if not update or not update.channels:
+            return None
+        return update.channels[0]
+
+
+def parse_terminal_messages(messages: Any) -> TerminalUpdate:
+    """
+    Decode an OP25 terminal response into channel state + call-log events.
+
+    Deliberately permissive: the response may be a bare dict instead of a list,
+    may contain json_type values we have never seen, and individual entries may
+    be malformed. Anything unrecognised is skipped rather than raising, because
+    dropping a whole poll would drop call_log events that are never re-sent.
+    """
+    update = TerminalUpdate()
+
+    if isinstance(messages, dict):
+        messages = [messages]
+    if not isinstance(messages, list):
+        return update
+
+    for msg in messages:
+        if not isinstance(msg, dict):
+            continue
+        json_type = msg.get("json_type")
+
+        if json_type == "channel_update":
+            for chan_id in msg.get("channels") or []:
+                channel = msg.get(str(chan_id))
+                if isinstance(channel, dict):
+                    update.channels.append(channel)
+
+        elif json_type == "call_log":
+            for entry in msg.get("log") or []:
+                if isinstance(entry, dict):
+                    update.call_log.append(entry)
+
+    return update
+
 
 op25_client = OP25Client()

drb-edge-node/app/internal/pulse.py

@@ -0,0 +1,76 @@
+"""
+PulseAudio readiness helpers.
+
+The PulseAudio daemon lives in the `op25` container and exposes its native
+socket on the shared `pulse_socket` docker volume (mounted at /run/pulse in
+both containers, with PULSE_SERVER=unix:/run/pulse/native).
+
+`op25-container/docker-entrypoint.sh` waits up to ~10 s for that socket before
+starting its own app, but the edge-node historically had *no* equivalent wait:
+FFmpeg would be launched with `-f pulse` before the socket existed, fail
+instantly, and the audio path would stay dead for the lifetime of the process.
+This module is the missing wait.
+
+NOTE on the source name: the op25 entrypoint starts pulseaudio with `-n`, which
+skips /etc/pulse/system.pa entirely and loads modules from the command line
+instead.  That means the `set-default-source drb_sink.monitor` line in system.pa
+is NOT applied at runtime, so `-i default` is unreliable.  Always address the
+monitor explicitly via settings.pulse_source (default "drb_sink.monitor").
+"""
+import asyncio
+import os
+import stat
+from typing import Optional
+
+from app.config import settings
+from app.internal.logger import logger
+
+DEFAULT_SOCKET_PATH = "/run/pulse/native"
+POLL_INTERVAL = 0.5
+
+
+def socket_path() -> str:
+    """Resolve the PulseAudio socket path from PULSE_SERVER (`unix:/path` form)."""
+    server = os.environ.get("PULSE_SERVER", "")
+    if server.startswith("unix:"):
+        candidate = server[len("unix:"):].strip()
+        if candidate:
+            return candidate
+    return DEFAULT_SOCKET_PATH
+
+
+def is_ready() -> bool:
+    """True when the PulseAudio native socket exists and really is a socket."""
+    try:
+        return stat.S_ISSOCK(os.stat(socket_path()).st_mode)
+    except OSError:
+        return False
+
+
+async def wait_until_ready(timeout: Optional[float] = None) -> bool:
+    """
+    Block until the PulseAudio socket appears, or `timeout` seconds elapse.
+
+    Bounded on purpose — never hang the caller forever.  Returns True if the
+    socket is present, False on timeout (caller decides whether to retry).
+    """
+    limit = settings.pulse_wait_timeout if timeout is None else timeout
+    path = socket_path()
+
+    if is_ready():
+        return True
+
+    logger.info(f"Waiting up to {limit:.0f}s for PulseAudio socket at {path}…")
+    waited = 0.0
+    while waited < limit:
+        await asyncio.sleep(POLL_INTERVAL)
+        waited += POLL_INTERVAL
+        if is_ready():
+            logger.info(f"PulseAudio socket ready after {waited:.1f}s.")
+            return True
+
+    logger.error(
+        f"PulseAudio socket {path} not present after {limit:.0f}s — "
+        "is the op25 container running? Audio capture will retry."
+    )
+    return False