Move recording and Discord voice to PulseAudio

.env.example

@@ -16,12 +16,23 @@ C2_URL=http://localhost:8888
 # API key is provisioned automatically via MQTT after admin approves the node
 
 # Icecast (local container — usually no need to change)
+# Live listening only. Call recording and Discord voice use PulseAudio instead.
 ICECAST_SOURCE_PASSWORD=hackme
 ICECAST_ADMIN_PASSWORD=admin
 ICECAST_HOST=localhost
 ICECAST_PORT=8000
 ICECAST_MOUNT=/radio
 
+# PulseAudio capture (usually no need to change)
+# Monitor of the drb_sink null sink that Liquidsoap writes into.
+PULSE_SOURCE=drb_sink.monitor
+# Seconds to wait for the shared PulseAudio socket before giving up and retrying.
+PULSE_WAIT_TIMEOUT=30
+
+# Call segmentation: seconds of radio silence before the current recording is
+# closed. Grants on the same talkgroup within this window stay in ONE recording.
+CALL_IDLE_TIMEOUT=3
+
 # OP25 container (usually no need to change)
 OP25_API_URL=http://localhost:8001
 OP25_TERMINAL_URL=http://localhost:8081

Makefile

@@ -8,9 +8,9 @@ setup:
 test:
 	docker compose run --no-deps --rm edge-node pytest -v
 
-# Build all images locally and start.
+# Build all images locally and start (dev mode with local code mounted).
 up:
-	docker compose up -d
+	docker compose up -d --build
 
 # Pull pre-built images from the registry and start (no local build).
 # Requires IMAGE_REGISTRY, DOCKER_ORG, DOCKER_REPO set in .env.
@@ -21,6 +21,12 @@ up-prebuilt:
 pull:
 	docker compose pull
 
+# Helper to pull the latest git commits and rebuild/restart the dev stack.
+update-git:
+	git pull
+	docker compose build
+	docker compose up -d
+
 down:
 	docker compose down
 

drb-edge-node/app/config.py

@@ -18,12 +18,27 @@ class Settings(BaseSettings):
     # C2 server (audio upload destination); None disables upload
     c2_url: Optional[str] = None
 
-    # Local Icecast
+    # Local Icecast — live listening only (frontend / mobile).
+    # NOT used for call recording or Discord voice: it lags 1s and drifts to 100s+.
     icecast_host: str = "localhost"
     icecast_port: int = 8000
     icecast_mount: str = "/radio"
     icecast_source_password: str = "hackme"
 
+    # PulseAudio — the low-latency path used for call recording and Discord voice.
+    # Liquidsoap (op25 container) writes into the `drb_sink` null sink; we capture
+    # its monitor. Addressed explicitly rather than via "default" because the op25
+    # entrypoint starts pulseaudio with -n and never applies system.pa's
+    # `set-default-source` line.
+    pulse_source: str = "drb_sink.monitor"
+    # Bounded wait for the shared PulseAudio socket before launching FFmpeg.
+    pulse_wait_timeout: float = 30.0
+
+    # Call segmentation — seconds with no active transmission before the current
+    # recording is closed out. Consecutive grants on the SAME talkgroup inside this
+    # window are kept in one recording so back-and-forth traffic stays together.
+    call_idle_timeout: float = 3.0
+
     # OP25 container
     op25_api_url: str = "http://localhost:8001"
     op25_terminal_url: str = "http://localhost:8081"
@@ -32,6 +47,9 @@ class Settings(BaseSettings):
     config_path: str = "/configs"
     recordings_path: str = "/recordings"
 
+    # Offline call buffer — how many call_end events to keep while disconnected
+    offline_call_buffer_size: int = 35
+
     class Config:
         env_file = ".env"
 

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/mqtt_manager.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 from datetime import datetime, timezone
+from collections import deque
 from typing import Optional, Callable, Awaitable, Dict, Any
 import paho.mqtt.client as mqtt
 from app.config import settings
@@ -23,6 +24,8 @@ class MQTTManager:
         self.on_config_push: Optional[ConfigCallback] = None
         self.on_api_key: Optional[ApiKeyCallback] = None
 
+        self._offline_buffer = deque(maxlen=settings.offline_call_buffer_size)
+
         nid = settings.node_id
         self._t_checkin = f"nodes/{nid}/checkin"
         self._t_status = f"nodes/{nid}/status"
@@ -64,6 +67,7 @@ class MQTTManager:
             logger.info("MQTT connected.")
             asyncio.run_coroutine_threadsafe(self._publish_checkin(), self._loop)
             asyncio.run_coroutine_threadsafe(self._maybe_request_key(), self._loop)
+            asyncio.run_coroutine_threadsafe(self._flush_offline_buffer(), self._loop)
         else:
             logger.error(f"MQTT connect refused: {reason_code}")
 
@@ -130,7 +134,23 @@ class MQTTManager:
             "timestamp": datetime.now(timezone.utc).isoformat(),
             **data,
         }
-        self._publish(self._t_metadata, payload, qos=1)
+        if not self._connected:
+            if event_type == "call_end":
+                self._offline_buffer.append((self._t_metadata, payload))
+                logger.warning(f"MQTT offline. Buffered call_end event for {data.get('call_id')}")
+            else:
+                logger.debug(f"MQTT offline. Dropping metadata event: {event_type}")
+        else:
+            self._publish(self._t_metadata, payload, qos=1)
+
+    async def _flush_offline_buffer(self):
+        if not self._offline_buffer:
+            return
+        count = len(self._offline_buffer)
+        logger.info(f"Relaying {count} buffered call_end events from offline queue.")
+        while self._offline_buffer:
+            topic, payload = self._offline_buffer.popleft()
+            self._publish(topic, payload, qos=1)
 
     async def _maybe_request_key(self):
         """After connecting, wait for any retained api_key message to arrive.

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

drb-edge-node/app/main.py

@@ -24,12 +24,18 @@ async def on_call_start(data: dict):
     radio_bot.start_stream()
     await mqtt_manager.publish_status("recording")
     await mqtt_manager.publish_metadata("call_start", data)
-    await call_recorder.start_recording(data["call_id"])
+    # started_at_epoch is OP25's own call_log timestamp — the recorder slices the
+    # ring buffer back to it (minus pre-roll), so however late we detected the
+    # grant, the audio still starts in the right place.
+    await call_recorder.start_recording(
+        data["call_id"],
+        start_epoch=data.get("started_at_epoch"),
+    )
 
 
 async def on_call_end(data: dict):
     radio_bot.stop_stream()
-    file_path = await call_recorder.stop_recording()
+    file_path = await call_recorder.stop_recording(end_epoch=data.get("ended_at_epoch"))
     if file_path:
         node_cfg = load_node_config()
         audio_url = await call_recorder.upload_recording(
@@ -46,7 +52,8 @@ async def on_call_end(data: dict):
     else:
         logger.warning(
             f"No recording file generated for call {data['call_id']} "
-            "— call may have been too short or Icecast unreachable."
+            "— PulseAudio capture may be down (check the op25 container and "
+            f"the {settings.pulse_source} source)."
         )
     await mqtt_manager.publish_metadata("call_end", data)
     await mqtt_manager.publish_status("online")
@@ -191,7 +198,7 @@ async def lifespan(app: FastAPI):
     # Start services (radio_bot starts on-demand when a discord_join command arrives)
     await mqtt_manager.connect()
     await metadata_watcher.start()
-    await call_recorder.start()  # persistent Icecast stream buffer
+    await call_recorder.start()  # persistent PulseAudio ring buffer
 
     # Start system caching in background
     from app.internal.system_cacher import fetch_and_cache_systems

drb-edge-node/app/models.py

@@ -37,6 +37,7 @@ class NodeConfig(BaseModel):
     enforce_override_timeout: bool = True
     override_system_id: Optional[str] = None
     override_config: Optional[SystemConfig] = None
+    offline_call_buffer_size: int = 35  # max call_end events to buffer while MQTT is offline
 
 
 class CallEvent(BaseModel):

drb-edge-node/app/routers/api.py

@@ -48,6 +48,10 @@ async def get_status():
         "assigned_system_id": node_cfg.assigned_system_id,
         "system_name": system_name,
         "is_recording": call_recorder.is_recording,
+        # Health of the PulseAudio capture that feeds every recording — the single
+        # most useful signal when recordings come back empty.
+        "audio_capture": call_recorder.is_capturing,
+        "buffered_seconds": round(call_recorder.buffered_seconds, 1),
         "active_tgid": active_tgid,
         "active_tgid_name": active_tgid_name,
         "active_call_id": metadata_watcher.active_call_id,

drb-edge-node/tests/test_call_recorder.py

@@ -0,0 +1,203 @@
+"""
+Unit tests for the CallRecorder ring buffer and per-call slicing.
+
+No FFmpeg and no PulseAudio: the buffer is filled directly with timestamped
+chunks, which is exactly what _ingest() produces at runtime.
+"""
+import time
+from unittest.mock import patch
+
+import pytest
+
+from app.internal.call_recorder import (
+    CallRecorder,
+    MAX_RECORDING_SECONDS,
+    PRE_ROLL_SECONDS,
+    RING_BUFFER_SECONDS,
+)
+
+T0 = 1_700_000_000.0
+CHUNK_INTERVAL = 0.1  # seconds of audio per synthetic chunk
+
+
+@pytest.fixture
+def recorder(tmp_path):
+    r = CallRecorder()
+    r._recordings_dir = tmp_path
+    r._capturing = True
+    return r
+
+
+def fill(recorder, start: float, end: float, marker: bytes = b"A"):
+    """Append one chunk every CHUNK_INTERVAL seconds over [start, end)."""
+    ts = start
+    index = 0
+    while ts < end:
+        recorder._buffer.append((ts, marker + str(index).encode() + b";"))
+        index += 1
+        ts = round(ts + CHUNK_INTERVAL, 6)
+
+
+def timestamps(recorder):
+    return [ts for ts, _ in recorder._buffer]
+
+
+# ---------------------------------------------------------------------------
+# Ring buffer trimming
+# ---------------------------------------------------------------------------
+
+def test_idle_buffer_keeps_only_the_rolling_window(recorder):
+    with patch("app.internal.call_recorder.time.time", return_value=T0):
+        for offset in range(0, int(RING_BUFFER_SECONDS) + 20):
+            with patch("app.internal.call_recorder.time.time", return_value=T0 + offset):
+                recorder._ingest(b"x" * 16)
+
+    assert len(recorder._buffer) <= RING_BUFFER_SECONDS + 1
+    assert min(timestamps(recorder)) >= (T0 + RING_BUFFER_SECONDS + 19) - RING_BUFFER_SECONDS
+
+
+@pytest.mark.asyncio
+async def test_buffer_is_not_trimmed_below_the_active_slice(recorder):
+    fill(recorder, T0, T0 + 5.0)
+    await recorder.start_recording("call-1", start_epoch=T0 + 1.0)
+
+    # Ingest far past the normal rolling window; the slice start must survive.
+    with patch("app.internal.call_recorder.time.time", return_value=T0 + RING_BUFFER_SECONDS + 10):
+        recorder._ingest(b"z")
+
+    assert min(timestamps(recorder)) <= (T0 + 1.0) - PRE_ROLL_SECONDS + CHUNK_INTERVAL
+
+
+# ---------------------------------------------------------------------------
+# Pre-roll and slicing
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_slice_starts_pre_roll_before_the_op25_timestamp(recorder):
+    fill(recorder, T0, T0 + 10.0)
+    grant_time = T0 + 5.0
+
+    await recorder.start_recording("call-1", start_epoch=grant_time)
+    assert recorder._slice_start == pytest.approx(grant_time - PRE_ROLL_SECONDS)
+
+    path = await recorder.stop_recording(end_epoch=grant_time + 2.0)
+    assert path is not None and path.exists()
+
+    # Reconstruct which chunks landed in the file.
+    kept = path.read_bytes().decode().strip(";").split(";")
+    first_index = int(kept[0][1:])
+    first_ts = T0 + first_index * CHUNK_INTERVAL
+
+    # A chunk stamped `ts` holds the audio that arrived over [ts - interval, ts],
+    # so the audio actually covered must begin at or before the requested slice
+    # start — erring early is the safe direction, erring late loses speech.
+    assert first_ts - CHUNK_INTERVAL <= grant_time - PRE_ROLL_SECONDS + 1e-6
+    # ...and no more than one chunk of extra pre-roll is dragged in.
+    assert first_ts >= grant_time - PRE_ROLL_SECONDS - 1e-6
+
+
+@pytest.mark.asyncio
+async def test_tail_chunk_straddling_the_end_is_included(recorder):
+    fill(recorder, T0, T0 + 10.0)
+    await recorder.start_recording("call-1", start_epoch=T0 + 1.0)
+
+    # End halfway through a chunk interval.
+    path = await recorder.stop_recording(end_epoch=T0 + 3.05)
+    kept = path.read_bytes().decode().strip(";").split(";")
+    last_ts = T0 + int(kept[-1][1:]) * CHUNK_INTERVAL
+
+    assert last_ts >= T0 + 3.05, "the chunk covering the end instant must be kept"
+
+
+@pytest.mark.asyncio
+async def test_pre_roll_earlier_than_buffer_start_is_clamped(recorder, caplog):
+    """A grant older than anything buffered must still produce a file."""
+    fill(recorder, T0 + 5.0, T0 + 10.0)  # buffer only covers T0+5 onwards
+
+    await recorder.start_recording("call-1", start_epoch=T0)  # 5 s before the head
+    path = await recorder.stop_recording(end_epoch=T0 + 8.0)
+
+    assert path is not None and path.stat().st_size > 0
+    kept = path.read_bytes().decode().strip(";").split(";")
+    assert kept[0] == "A0", "slice should begin at the buffer head, not fail"
+
+
+@pytest.mark.asyncio
+async def test_no_buffered_audio_returns_none(recorder):
+    await recorder.start_recording("call-1", start_epoch=T0)
+    assert await recorder.stop_recording(end_epoch=T0 + 2.0) is None
+
+
+@pytest.mark.asyncio
+async def test_start_epoch_omitted_falls_back_to_now(recorder):
+    now = time.time()
+    fill(recorder, now - 5.0, now)
+
+    await recorder.start_recording("call-1")
+    assert recorder._slice_start == pytest.approx(now - PRE_ROLL_SECONDS, abs=1.0)
+
+
+@pytest.mark.asyncio
+async def test_max_recording_seconds_caps_the_slice(recorder):
+    fill(recorder, T0, T0 + MAX_RECORDING_SECONDS + 60, marker=b"A")
+    await recorder.start_recording("call-1", start_epoch=T0 + 1.0)
+
+    path = await recorder.stop_recording(end_epoch=T0 + MAX_RECORDING_SECONDS + 50)
+    kept = path.read_bytes().decode().strip(";").split(";")
+    last_ts = T0 + int(kept[-1][1:]) * CHUNK_INTERVAL
+
+    assert last_ts <= T0 + 1.0 + MAX_RECORDING_SECONDS + CHUNK_INTERVAL
+
+
+# ---------------------------------------------------------------------------
+# Recording lifecycle
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_second_start_is_rejected_while_recording(recorder):
+    fill(recorder, T0, T0 + 5.0)
+    assert await recorder.start_recording("call-1", start_epoch=T0 + 1.0) is True
+    assert await recorder.start_recording("call-2", start_epoch=T0 + 2.0) is False
+    assert recorder.is_recording
+
+
+@pytest.mark.asyncio
+async def test_stop_without_start_is_a_noop(recorder):
+    assert await recorder.stop_recording() is None
+    assert not recorder.is_recording
+
+
+@pytest.mark.asyncio
+async def test_split_then_immediate_restart_keeps_both_slices(recorder):
+    """
+    A tgid change closes one recording and opens the next at the same instant —
+    the second must still find its pre-roll in the buffer.
+    """
+    fill(recorder, T0, T0 + 10.0)
+    split = T0 + 5.0
+
+    await recorder.start_recording("call-1", start_epoch=T0 + 1.0)
+    first = await recorder.stop_recording(end_epoch=split)
+
+    await recorder.start_recording("call-2", start_epoch=split)
+    second = await recorder.stop_recording(end_epoch=T0 + 8.0)
+
+    assert first is not None and first.stat().st_size > 0
+    assert second is not None and second.stat().st_size > 0
+    assert first != second
+
+
+# ---------------------------------------------------------------------------
+# FFmpeg invocation
+# ---------------------------------------------------------------------------
+
+def test_ffmpeg_command_reads_pulse_and_flushes_packets(recorder):
+    cmd = recorder._ffmpeg_command()
+    joined = " ".join(cmd)
+
+    assert "-f pulse" in joined
+    assert "drb_sink.monitor" in joined, "must address the monitor explicitly, not 'default'"
+    # Without -flush_packets the mp3 muxer buffers 32 KB (~16 s at 16 kbps) before
+    # writing, which would destroy the ring buffer's timestamp resolution.
+    assert "-flush_packets" in cmd
+    assert cmd[-1] == "-" and cmd[-2] == "mp3", "must emit MP3 on stdout for /upload"

drb-edge-node/tests/test_metadata_watcher.py

@@ -1,190 +1,487 @@
 """
-Unit tests for MetadataWatcher state machine.
-All OP25 HTTP calls are mocked — no running services required.
+Unit tests for the event-driven MetadataWatcher state machine.
+
+Call START comes from OP25 `call_log` entries (stamped with OP25's own
+time.time()); call END comes from the srcaddr != 0 -> srcaddr == 0 transition in
+`channel_update`.  All OP25 HTTP calls are mocked — no running services required.
 """
 import pytest
 from unittest.mock import AsyncMock, patch
-from app.internal.metadata_watcher import MetadataWatcher, HANG_THRESHOLD
+
+from app.config import settings
+from app.internal.metadata_watcher import (
+    MetadataWatcher,
+    TAIL_PAD_SECONDS,
+    OP25_OFFLINE_GRACE,
+)
+from app.internal.op25_client import TerminalUpdate, parse_terminal_messages
+
+
+class FakeClock:
+    """Manually advanced clock so idle timeouts are testable without sleeping."""
+
+    def __init__(self, start: float = 1_700_000_000.0):
+        self.now = start
+
+    def __call__(self) -> float:
+        return self.now
+
+    def advance(self, seconds: float) -> float:
+        self.now += seconds
+        return self.now
 
 
 @pytest.fixture
-def watcher():
+def clock():
+    return FakeClock()
+
+
+@pytest.fixture
+def watcher(clock):
     w = MetadataWatcher()
+    w._clock = clock
     w.on_call_start = AsyncMock()
     w.on_call_end = AsyncMock()
     return w
 
 
+def grant(tgid: int, time_: float, tgtag: str = "", rid: int = 101, freq: int = 851_000_000):
+    """One OP25 call_log entry (see tk_p25.log_call)."""
+    return {
+        "time": time_,
+        "sysid": 1,
+        "rcvr": 0,
+        "freq": freq,
+        "slot": None,
+        "prio": 0,
+        "tgid": tgid,
+        "tgtag": tgtag,
+        "rid": rid,
+        "rtag": "",
+    }
+
+
+def channel(tgid: int = 0, srcaddr: int = 0, tag: str = "", hold_tgid: int = 0):
+    """One OP25 channel_update channel dict (see tk_p25.get_chan_status)."""
+    return {
+        "freq": 851_000_000,
+        "tgid": tgid or None,
+        "tag": tag,
+        "srcaddr": srcaddr,
+        "svcopts": bool(srcaddr),
+        "hold_tgid": hold_tgid or None,
+        "encrypted": 0,
+        "emergency": 0,
+    }
+
+
+def update(call_log=None, channels=None) -> TerminalUpdate:
+    return TerminalUpdate(channels=list(channels or []), call_log=list(call_log or []))
+
+
+def patched(result):
+    return patch(
+        "app.internal.metadata_watcher.op25_client.poll_terminal",
+        new=AsyncMock(return_value=result),
+    )
+
+
+async def tick(watcher, result):
+    with patched(result):
+        await watcher._tick()
+
+
 # ---------------------------------------------------------------------------
-# Call start
+# op25_client message parsing
+# ---------------------------------------------------------------------------
+
+def test_parser_extracts_call_log_and_channels():
+    messages = [
+        {"json_type": "trunk_update", "0": {"whatever": 1}},
+        {"json_type": "channel_update", "channels": [0], "0": channel(tgid=1234, srcaddr=555)},
+        {"json_type": "call_log", "log": [grant(1234, 100.0)]},
+    ]
+    result = parse_terminal_messages(messages)
+
+    assert len(result.channels) == 1
+    assert result.channels[0]["tgid"] == 1234
+    assert len(result.call_log) == 1
+    assert result.call_log[0]["time"] == 100.0
+
+
+def test_parser_tolerates_garbage():
+    """Unknown json_types, bare dicts and malformed entries must not raise."""
+    assert parse_terminal_messages(None).call_log == []
+    assert parse_terminal_messages("nope").channels == []
+    assert parse_terminal_messages([None, 5, {"json_type": "mystery"}]).channels == []
+
+    # A bare dict instead of a list.
+    single = parse_terminal_messages({"json_type": "call_log", "log": [grant(1, 1.0), "junk"]})
+    assert len(single.call_log) == 1
+
+    # channel_update naming a channel that isn't present.
+    missing = parse_terminal_messages([{"json_type": "channel_update", "channels": [0, 1], "0": channel(9)}])
+    assert len(missing.channels) == 1
+
+
+def test_parser_handles_multiple_channels():
+    messages = [{
+        "json_type": "channel_update",
+        "channels": [0, 1],
+        "0": channel(tgid=1111, srcaddr=1),
+        "1": channel(tgid=2222, srcaddr=2),
+    }]
+    result = parse_terminal_messages(messages)
+    assert [c["tgid"] for c in result.channels] == [1111, 2222]
+
+
+# ---------------------------------------------------------------------------
+# Call start — driven by call_log
 # ---------------------------------------------------------------------------
 
 @pytest.mark.asyncio
-async def test_call_starts_when_tgid_appears(watcher):
-    status = [{"tgid": 1234, "tag": "Police Dispatch"}]
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)):
-        await watcher._tick()
+async def test_call_log_entry_starts_call(watcher, clock):
+    grant_time = clock.now - 0.4  # OP25 logged it before our poll noticed
+    await tick(watcher, update(
+        call_log=[grant(1234, grant_time, tgtag="Police Dispatch")],
+        channels=[channel(tgid=1234, srcaddr=555)],
+    ))
 
     assert watcher.is_active
     assert watcher.current_tgid == 1234
     watcher.on_call_start.assert_called_once()
+
     payload = watcher.on_call_start.call_args[0][0]
     assert payload["tgid"] == 1234
     assert payload["tgid_name"] == "Police Dispatch"
-    assert "call_id" in payload
-    assert "started_at" in payload
+    assert payload["call_id"]
+    # The whole point: the start is OP25's timestamp, not our detection time.
+    assert payload["started_at_epoch"] == grant_time
+    assert payload["started_at"].startswith("20")
 
 
 @pytest.mark.asyncio
-async def test_tgid_zero_does_not_start_call(watcher):
-    status = [{"tgid": 0}]
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)):
-        await watcher._tick()
+async def test_channel_activity_alone_does_not_start_a_call(watcher):
+    """No call_log entry => no call, even with a live tgid on the channel."""
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
 
     assert not watcher.is_active
     watcher.on_call_start.assert_not_called()
 
 
 @pytest.mark.asyncio
-async def test_tgid_none_string_does_not_start_call(watcher):
-    status = [{"tgid": "None"}]
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)):
-        await watcher._tick()
+async def test_call_log_entry_without_tgid_is_ignored(watcher):
+    await tick(watcher, update(call_log=[grant(0, 100.0)]))
 
     assert not watcher.is_active
+    watcher.on_call_start.assert_not_called()
 
 
 @pytest.mark.asyncio
-async def test_op25_offline_does_not_start_call(watcher):
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=None)):
-        await watcher._tick()
+async def test_call_log_without_time_falls_back_to_local_clock(watcher, clock):
+    entry = grant(1234, 0.0)
+    entry.pop("time")
+    await tick(watcher, update(call_log=[entry]))
+
+    assert watcher.is_active
+    assert watcher.on_call_start.call_args[0][0]["started_at_epoch"] == clock.now
+
+
+@pytest.mark.asyncio
+async def test_op25_unreachable_does_not_start_call(watcher):
+    await tick(watcher, None)
 
     assert not watcher.is_active
     watcher.on_call_start.assert_not_called()
 
 
 # ---------------------------------------------------------------------------
-# Hang / call end
+# Call end — srcaddr edge + idle timeout
 # ---------------------------------------------------------------------------
 
 @pytest.mark.asyncio
-async def test_hang_below_threshold_keeps_call_alive(watcher):
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])):
-        await watcher._tick()
-
+async def test_srcaddr_edge_then_idle_timeout_ends_call(watcher, clock):
+    await tick(watcher, update(
+        call_log=[grant(1234, clock.now)],
+        channels=[channel(tgid=1234, srcaddr=555)],
+    ))
     assert watcher.is_active
 
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 0}])):
-        for _ in range(HANG_THRESHOLD - 1):
-            await watcher._tick()
+    # srcaddr resets to 0 while tgid is still held — OP25's end-of-call signal.
+    clock.advance(0.5)
+    edge_time = clock.now
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
+    assert watcher.is_active, "the srcaddr edge alone must not close the segment"
+    watcher.on_call_end.assert_not_called()
 
+    # Still quiet, but not long enough yet.
+    clock.advance(settings.call_idle_timeout - 0.5)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
     assert watcher.is_active
+
+    # Past the timeout.
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel()]))
+    assert not watcher.is_active
+
+    watcher.on_call_end.assert_called_once()
+    payload = watcher.on_call_end.call_args[0][0]
+    assert payload["end_reason"] == "idle_timeout"
+    # The audio ends at the last transmission plus a short pad, NOT at "now" —
+    # otherwise every recording carries call_idle_timeout seconds of silence.
+    assert payload["ended_at_epoch"] == pytest.approx(edge_time + TAIL_PAD_SECONDS)
+    assert payload["tgid"] == 1234
+
+
+@pytest.mark.asyncio
+async def test_ongoing_transmission_never_times_out(watcher, clock):
+    await tick(watcher, update(
+        call_log=[grant(1234, clock.now)],
+        channels=[channel(tgid=1234, srcaddr=555)],
+    ))
+
+    for _ in range(20):
+        clock.advance(settings.call_idle_timeout)
+        await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
+        assert watcher.is_active
+
     watcher.on_call_end.assert_not_called()
 
 
 @pytest.mark.asyncio
-async def test_hang_at_threshold_ends_call(watcher):
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])):
-        await watcher._tick()
+async def test_op25_unreachable_ends_active_call_after_grace(watcher, clock):
+    await tick(watcher, update(
+        call_log=[grant(1234, clock.now)],
+        channels=[channel(tgid=1234, srcaddr=555)],
+    ))
 
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 0}])):
-        for _ in range(HANG_THRESHOLD):
-            await watcher._tick()
+    # A single failed poll must not kill the call.
+    clock.advance(OP25_OFFLINE_GRACE / 2)
+    await tick(watcher, None)
+    assert watcher.is_active
+
+    clock.advance(OP25_OFFLINE_GRACE)
+    await tick(watcher, None)
+    assert not watcher.is_active
+    assert watcher.on_call_end.call_args[0][0]["end_reason"] == "op25_unreachable"
+
+
+@pytest.mark.asyncio
+async def test_grant_whose_call_already_ended(watcher, clock):
+    """call_log delivers a start whose transmission is already over."""
+    await tick(watcher, update(
+        call_log=[grant(1234, clock.now - 1.2)],
+        channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)],  # already idle
+    ))
+    assert watcher.is_active
+
+    clock.advance(settings.call_idle_timeout + 0.5)
+    await tick(watcher, update(channels=[channel()]))
 
     assert not watcher.is_active
     watcher.on_call_end.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_stop_closes_open_segment(watcher, clock):
+    await tick(watcher, update(
+        call_log=[grant(1234, clock.now)],
+        channels=[channel(tgid=1234, srcaddr=555)],
+    ))
+    await watcher.stop()
+
+    assert not watcher.is_active
+    assert watcher.on_call_end.call_args[0][0]["end_reason"] == "shutdown"
+
+
+# ---------------------------------------------------------------------------
+# Segmentation: same-TGID continuation, different-TGID split
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_same_tgid_grant_continues_one_recording(watcher, clock):
+    """Back-and-forth on one talkgroup must stay a single call/recording."""
+    start = clock.now
+    await tick(watcher, update(
+        call_log=[grant(1234, start)],
+        channels=[channel(tgid=1234, srcaddr=555)],
+    ))
+    call_id = watcher.active_call_id
+
+    # Transmission ends...
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
+
+    # ...and the other party keys up on the SAME tgid inside the idle window.
+    clock.advance(1.0)
+    await tick(watcher, update(
+        call_log=[grant(1234, clock.now)],
+        channels=[channel(tgid=1234, srcaddr=777)],
+    ))
+
+    assert watcher.is_active
+    assert watcher.active_call_id == call_id, "same tgid must not open a new call"
+    watcher.on_call_start.assert_called_once()
+    watcher.on_call_end.assert_not_called()
+
+    # And the whole exchange closes as one segment.
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
+    clock.advance(settings.call_idle_timeout + 0.5)
+    await tick(watcher, update(channels=[channel()]))
+
+    watcher.on_call_end.assert_called_once()
     payload = watcher.on_call_end.call_args[0][0]
-    assert "call_id" in payload
-    assert "ended_at" in payload
+    assert payload["call_id"] == call_id
+    assert payload["transmissions"] == 2
+    assert payload["started_at_epoch"] == start
 
 
 @pytest.mark.asyncio
-async def test_op25_offline_triggers_hang_and_ends_call(watcher):
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])):
-        await watcher._tick()
+async def test_different_tgid_grant_splits_recording(watcher, clock):
+    await tick(watcher, update(
+        call_log=[grant(1111, clock.now, tgtag="Fire")],
+        channels=[channel(tgid=1111, srcaddr=1)],
+    ))
+    first_id = watcher.active_call_id
 
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=None)):
-        for _ in range(HANG_THRESHOLD):
-            await watcher._tick()
-
-    assert not watcher.is_active
-    watcher.on_call_end.assert_called_once()
-
-
-@pytest.mark.asyncio
-async def test_hang_counter_resets_when_tgid_returns(watcher):
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])):
-        await watcher._tick()
-
-    # Partial hang — not enough to end
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 0}])):
-        for _ in range(HANG_THRESHOLD - 1):
-            await watcher._tick()
-
-    assert watcher.is_active
-
-    # tgid returns — counter resets
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])):
-        await watcher._tick()
-
-    assert watcher._hang_counter == 0
-    assert watcher.is_active
-
-
-# ---------------------------------------------------------------------------
-# Talkgroup changes
-# ---------------------------------------------------------------------------
-
-@pytest.mark.asyncio
-async def test_talkgroup_change_closes_old_and_opens_new(watcher):
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1111}])):
-        await watcher._tick()
-
-    first_call_id = watcher.active_call_id
-
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 2222}])):
-        await watcher._tick()
+    clock.advance(2.0)
+    split_time = clock.now
+    await tick(watcher, update(
+        call_log=[grant(2222, split_time, tgtag="EMS")],
+        channels=[channel(tgid=2222, srcaddr=2)],
+    ))
 
     assert watcher.is_active
     assert watcher.current_tgid == 2222
-    assert watcher.active_call_id != first_call_id
-    watcher.on_call_end.assert_called_once()
+    assert watcher.active_call_id != first_id
     assert watcher.on_call_start.call_count == 2
+    watcher.on_call_end.assert_called_once()
 
-
-# ---------------------------------------------------------------------------
-# Status format variations
-# ---------------------------------------------------------------------------
-
-@pytest.mark.asyncio
-async def test_single_dict_status_instead_of_list(watcher):
-    """OP25 terminal may return a bare dict instead of a list."""
-    status = {"tgid": 9999, "tag": "Fire"}
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)):
-        await watcher._tick()
-
-    assert watcher.current_tgid == 9999
+    ended = watcher.on_call_end.call_args[0][0]
+    assert ended["call_id"] == first_id
+    assert ended["tgid"] == 1111
+    assert ended["end_reason"] == "tgid_change"
+    # The outgoing segment ends exactly where the new one begins — no tail pad,
+    # or it would swallow the first moments of the new talkgroup.
+    assert ended["ended_at_epoch"] == split_time
+    assert watcher.on_call_start.call_args[0][0]["started_at_epoch"] == split_time
 
 
 @pytest.mark.asyncio
-async def test_tg_id_key_alias(watcher):
-    """Some OP25 builds use 'tg_id' instead of 'tgid'."""
-    status = [{"tg_id": 5555, "tag": "EMS"}]
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)):
-        await watcher._tick()
+async def test_multiple_call_log_entries_in_one_poll(watcher, clock):
+    """
+    The call_log deque is drained per poll and capped at 10, so one poll can
+    deliver several grants. Same-tgid ones merge, different-tgid ones split.
+    """
+    t0 = clock.now - 1.5
+    await tick(watcher, update(
+        call_log=[
+            grant(1111, t0),
+            grant(1111, t0 + 0.4),   # same tgid -> continuation
+            grant(2222, t0 + 0.9),   # different tgid -> split
+        ],
+        channels=[channel(tgid=2222, srcaddr=9)],
+    ))
 
-    assert watcher.current_tgid == 5555
+    assert watcher.current_tgid == 2222
+    assert watcher.on_call_start.call_count == 2
+    watcher.on_call_end.assert_called_once()
+
+    ended = watcher.on_call_end.call_args[0][0]
+    assert ended["tgid"] == 1111
+    assert ended["transmissions"] == 2
+    assert ended["started_at_epoch"] == t0
+    assert ended["ended_at_epoch"] == t0 + 0.9
 
 
 @pytest.mark.asyncio
-async def test_multichannel_uses_first_active(watcher):
-    """When multiple channels are returned, first active tgid wins."""
-    status = [
-        {"tgid": 0},
-        {"tgid": 7777, "tag": "Roads"},
-        {"tgid": 8888, "tag": "Other"},
-    ]
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)):
-        await watcher._tick()
+async def test_out_of_order_call_log_entries_are_sorted(watcher, clock):
+    t0 = clock.now - 2.0
+    await tick(watcher, update(
+        call_log=[grant(2222, t0 + 1.0), grant(1111, t0)],
+        channels=[channel(tgid=2222, srcaddr=9)],
+    ))
 
-    assert watcher.current_tgid == 7777
+    ended = watcher.on_call_end.call_args[0][0]
+    assert ended["tgid"] == 1111
+    assert watcher.current_tgid == 2222
+
+
+@pytest.mark.asyncio
+async def test_dropped_call_log_event_still_closes_segment(watcher, clock):
+    """
+    CALL_LOG_MAX_LEN is 10, so a slow consumer loses grants. If another talkgroup
+    is plainly transmitting we must close immediately rather than record it under
+    the wrong tgid for call_idle_timeout seconds.
+    """
+    await tick(watcher, update(
+        call_log=[grant(1111, clock.now)],
+        channels=[channel(tgid=1111, srcaddr=1)],
+    ))
+    first_id = watcher.active_call_id
+
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel(tgid=3333, srcaddr=7)]))  # no call_log
+
+    assert not watcher.is_active
+    ended = watcher.on_call_end.call_args[0][0]
+    assert ended["call_id"] == first_id
+    assert ended["end_reason"] == "tgid_change_unlogged"
+
+
+@pytest.mark.asyncio
+async def test_other_receiver_on_another_tgid_does_not_split(watcher, clock):
+    """
+    The dropped-call_log fallback must not fire on multi-receiver setups: another
+    receiver being busy says nothing about ours, and closing on it would truncate
+    every call.
+    """
+    await tick(watcher, update(
+        call_log=[grant(1111, clock.now)],
+        channels=[channel(tgid=1111, srcaddr=1)],
+    ))
+    first_id = watcher.active_call_id
+
+    clock.advance(0.5)
+    await tick(watcher, update(channels=[
+        channel(tgid=1111, srcaddr=0, hold_tgid=1111),
+        channel(tgid=3333, srcaddr=7),
+    ]))
+
+    assert watcher.is_active
+    assert watcher.active_call_id == first_id
+    watcher.on_call_end.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_second_receiver_activity_does_not_hold_segment_open(watcher, clock):
+    """Only channels on OUR talkgroup count as our transmission."""
+    await tick(watcher, update(
+        call_log=[grant(1111, clock.now)],
+        channels=[channel(tgid=1111, srcaddr=1)],
+    ))
+
+    clock.advance(0.5)
+    await tick(watcher, update(channels=[
+        channel(tgid=1111, srcaddr=0, hold_tgid=1111),
+        channel(tgid=1111, srcaddr=0),
+    ]))
+    assert watcher.is_active
+
+    clock.advance(settings.call_idle_timeout + 0.5)
+    await tick(watcher, update(channels=[channel(tgid=1111, srcaddr=0, hold_tgid=1111)]))
+    assert not watcher.is_active
+
+
+@pytest.mark.asyncio
+async def test_end_never_precedes_start(watcher, clock):
+    """A clock skew or odd grant order must never produce a negative duration."""
+    await tick(watcher, update(
+        call_log=[grant(1234, clock.now + 5.0)],  # OP25 stamp in the future
+        channels=[channel(tgid=1234, srcaddr=5)],
+    ))
+    await watcher.stop()
+
+    payload = watcher.on_call_end.call_args[0][0]
+    assert payload["ended_at_epoch"] >= payload["started_at_epoch"]