Fix tail truncation, decouple call length from buffer, trim silence

Measured six recordings off a live P25 node and found two independent
defects causing lost audio at the end of calls:

- stop_recording() sliced the ring buffer immediately, so if the MP3
  muxer had not yet delivered the tail the file was silently short.
  Now waits (bounded, 2s) until buffered audio covers the end epoch.
- TGID-change closes used the new grant's epoch as the end with no pad
  at all, guaranteeing truncation on every split. Tail pad is now a
  setting, default raised 0.5s -> 1.0s.

The ring buffer also capped maximum call length: a call longer than the
buffer had its front silently clamped. The ring now serves the pre-roll
only, with a per-call accumulator for the rest, bounded at 4.8MB.
Clamping is loudly warned rather than silent.

Uploads averaged 63% silence, which inflates STT cost and is a known
Whisper hallucination trigger. Leading/trailing silence is now trimmed
conservatively (-40dB, 0.25s guard, internal pauses untouched).
started_at/ended_at still describe the call; new audio_* fields carry
the trimmed audio bounds so playback can map back to wall clock.
All-silence recordings are skipped and logged instead of uploaded.

Also: log measured control-channel idle on idle-timeout closes so
CALL_IDLE_TIMEOUT can be tuned from data, and quiet the httpx logger
which emitted ~170k lines/day of poll noise.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.env.example

@@ -16,12 +16,42 @@ 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.
+# Tune ONLY from the "measured control-channel idle" line the edge node logs on
+# every idle-timeout close — silence measured in the audio is a different clock
+# (it also contains the ~1.9s P25 grant-to-speech delay).
+CALL_IDLE_TIMEOUT=3
+
+# Seconds of audio kept after the last transmission ends. This is the only
+# headroom protecting the final word of a transmission — usually the disposition
+# or the address. Measured at 0.5s it left ~0.3s of real margin and one recording
+# ended mid-word, hence 1.0.
+CALL_TAIL_PAD_SECONDS=1.0
+
+# Strip leading/trailing dead air before upload. ~63% of an untrimmed recording
+# is silence, which costs Whisper spend and makes it hallucinate text that was
+# never spoken. Only the head and tail are touched, with a guard margin so no
+# syllable is clipped. Set to false to upload raw audio.
+TRIM_SILENCE=true
+# dBFS below which audio counts as silence for detection.
+TRIM_SILENCE_THRESHOLD_DB=-40
+# Seconds of audio kept either side of detected speech.
+TRIM_SILENCE_GUARD_SECONDS=0.25
+
 # 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,50 @@ 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.
+    #
+    # Do NOT tune this against measured *audio* silence: audio gaps also contain
+    # the ~1.9 s P25 grant→speech delay, so they are always longer than the
+    # control-channel idle this timer measures. metadata_watcher logs the real
+    # measured idle on every idle-timeout close — tune from that.
+    call_idle_timeout: float = 3.0
+
+    # Audio kept after the observed end of the last transmission. The srcaddr
+    # 1→0 edge can be up to one poll (0.5 s) late and the encoder adds its own
+    # latency, so this is the only headroom protecting the last word of a
+    # transmission — which is usually the disposition or the address. Field
+    # measurement at 0.5 s left only 0.29–0.37 s of real trailing margin and one
+    # recording ended mid-word, hence 1.0 s.
+    call_tail_pad_seconds: float = 1.0
+
+    # Strip leading/trailing dead air before upload. ~63% of a typical recording
+    # is silence (the grant→speech delay plus the tail pad), which inflates
+    # Whisper cost and is a well-documented trigger for hallucinated transcript
+    # text. Trimming is conservative — see internal/audio_trim.py.
+    trim_silence: bool = True
+    # Anything quieter than this counts as silence for detection purposes.
+    trim_silence_threshold_db: float = -40.0
+    # Guard margin kept around detected speech so no syllable is clipped.
+    trim_silence_guard_seconds: float = 0.25
+
     # OP25 container
     op25_api_url: str = "http://localhost:8001"
     op25_terminal_url: str = "http://localhost:8081"
@@ -32,6 +70,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/audio_trim.py

@@ -0,0 +1,270 @@
+"""
+Conservative leading/trailing silence removal for finished call recordings.
+
+WHY: P25 grants the channel, radios tune, and only then does a human start
+talking. Measured on six real recordings from a live node, that costs 1.71–2.45 s
+of dead air at the head of every single file, and the tail pad adds its own
+~0.3–1.0 s. Roughly 63% of every uploaded MP3 was silence. That is not just
+wasted Whisper spend: silence is a well-documented trigger for Whisper
+hallucinating text that was never spoken, and a hallucinated sentence poisons
+entity extraction and then incident correlation downstream.
+
+WHY IT IS SAFE: only the head and tail are touched, never the middle, and a
+guard margin is kept around the detected speech so no syllable can be clipped.
+If detection says the whole file is silent we do NOT emit a zero-length file —
+the caller is told and decides (see call_recorder: it skips the upload and logs).
+
+TIMING: trimming changes the audio's duration relative to the call's wall-clock
+start/end, so every trim reports exactly how much was removed from each end.
+Callers must carry those offsets forward — `started_at`/`ended_at` keep meaning
+the CALL's bounds, and the trimmed audio's own bounds are reported separately.
+
+Implementation is two FFmpeg passes (detect, then cut). FFmpeg is already a hard
+dependency of this container and is already running the capture, so this adds no
+new moving parts. `silenceremove` in one pass was rejected on purpose: it gives
+no way to learn how much it removed, which would make the timing metadata above
+impossible to produce.
+"""
+import asyncio
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from app.config import settings
+from app.internal.logger import logger
+
+# Minimum run of quiet before FFmpeg calls it a silence region. Below ~0.3 s this
+# starts firing on the natural pauses between words, which is not what we want —
+# we only care about the big block of dead air at each end.
+MIN_SILENCE_SECONDS = 0.3
+
+# A silence region only counts as "leading" if it begins essentially at the file
+# head. One chunk of slop.
+HEAD_EPSILON_SECONDS = 0.15
+
+# ...and only as "trailing" if it reaches the end of the audio. Do NOT assume a
+# missing `silence_end` marks that case: FFmpeg 6.x flushes a closing
+# `silence_end` at EOF, so an all-silence file looks exactly like a file with one
+# closed silence region. Verified against ffmpeg 6.1.1 — the reported end lands
+# ~0.03 s short of the (offset-corrected) duration, hence this epsilon. The
+# residual risk is clipping <=0.1 s of audio that follows a >=0.3 s gap right at
+# EOF, which the guard margin below more than covers.
+TAIL_EPSILON_SECONDS = 0.10
+
+# Don't bother re-encoding to reclaim less than this — a re-encode costs a CPU
+# spike on a Pi and a generation of MP3 quality, which is a bad trade for
+# a fraction of a second.
+MIN_TRIM_SECONDS = 0.20
+
+# Bounded so a wedged FFmpeg can never stall the upload path.
+FFMPEG_TIMEOUT_SECONDS = 30.0
+
+_SILENCE_START_RE = re.compile(r"silence_start:\s*(-?[0-9.]+)")
+_SILENCE_END_RE = re.compile(r"silence_end:\s*(-?[0-9.]+)")
+_DURATION_RE = re.compile(r"Duration:\s*(\d+):(\d\d):(\d\d\.\d+)")
+_START_RE = re.compile(r"Duration:[^\n]*?start:\s*(-?[0-9.]+)")
+
+
+@dataclass(frozen=True)
+class TrimResult:
+    """Outcome of a trim attempt. `lead`/`tail` are seconds actually removed."""
+
+    path: Optional[Path]
+    lead: float = 0.0
+    tail: float = 0.0
+    duration_before: float = 0.0
+    duration_after: float = 0.0
+    all_silence: bool = False
+    applied: bool = False
+
+    @property
+    def trimmed_seconds(self) -> float:
+        return self.lead + self.tail
+
+
+async def _run(cmd: List[str]) -> Tuple[int, str]:
+    """Run FFmpeg and return (returncode, stderr). FFmpeg reports on stderr."""
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        stdout=asyncio.subprocess.DEVNULL,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    try:
+        _, stderr = await asyncio.wait_for(proc.communicate(), timeout=FFMPEG_TIMEOUT_SECONDS)
+    except asyncio.TimeoutError:
+        try:
+            proc.kill()
+        except Exception:
+            pass
+        raise
+    return proc.returncode or 0, stderr.decode(errors="replace")
+
+
+def _parse_duration(stderr: str) -> Optional[float]:
+    """
+    Length of the decodable audio, in silencedetect's coordinates.
+
+    MP3 carries encoder delay/padding, which FFmpeg reports as a container
+    `start:` offset — the container duration is that much longer than the audio
+    silencedetect actually timestamps. Subtracting it is what lets
+    TAIL_EPSILON_SECONDS stay tight enough to be safe.
+    """
+    match = _DURATION_RE.search(stderr)
+    if not match:
+        return None
+    hours, minutes, seconds = match.groups()
+    duration = int(hours) * 3600 + int(minutes) * 60 + float(seconds)
+
+    start = _START_RE.search(stderr)
+    if start:
+        offset = float(start.group(1))
+        if 0.0 <= offset < duration:
+            duration -= offset
+    return duration
+
+
+def _parse_silences(stderr: str) -> List[Tuple[float, Optional[float]]]:
+    """
+    Extract silence regions as (start, end) with end=None when it runs to EOF.
+
+    FFmpeg emits `silence_start:` and a later `silence_end:` per region, and
+    simply never emits the closing line for a region that reaches EOF.
+    """
+    regions: List[Tuple[float, Optional[float]]] = []
+    pending: Optional[float] = None
+    for line in stderr.splitlines():
+        if "silencedetect" not in line:
+            continue
+        start = _SILENCE_START_RE.search(line)
+        if start:
+            pending = float(start.group(1))
+            continue
+        end = _SILENCE_END_RE.search(line)
+        if end and pending is not None:
+            regions.append((pending, float(end.group(1))))
+            pending = None
+    if pending is not None:
+        regions.append((pending, None))
+    return regions
+
+
+def speech_bounds(
+    regions: List[Tuple[float, Optional[float]]],
+    duration: float,
+    guard: float,
+) -> Tuple[float, float, bool]:
+    """
+    Turn detected silence regions into the [start, end] window to keep.
+
+    Pure and side-effect free so the decision logic is unit-testable without
+    FFmpeg. Returns (start, end, all_silence).
+    """
+    speech_start = 0.0
+    speech_end = duration
+
+    if regions:
+        head_start, head_end = regions[0]
+        if head_start <= HEAD_EPSILON_SECONDS and head_end is not None:
+            speech_start = head_end
+
+        tail_start, tail_end = regions[-1]
+        # `None` = pre-6.x FFmpeg, which simply stopped reporting at EOF.
+        reaches_eof = tail_end is None or (duration - tail_end) <= TAIL_EPSILON_SECONDS
+        if reaches_eof:
+            speech_end = min(speech_end, tail_start)
+
+    if speech_end <= speech_start:
+        # Detection says there is no speech anywhere in the file.
+        return 0.0, duration, True
+
+    # Guard margin: never trim right up against the first/last syllable.
+    keep_start = max(0.0, speech_start - guard)
+    keep_end = min(duration, speech_end + guard)
+    return keep_start, keep_end, False
+
+
+async def trim_silence(
+    path: Path,
+    sample_rate: str,
+    bitrate: str,
+    threshold_db: Optional[float] = None,
+    guard: Optional[float] = None,
+) -> TrimResult:
+    """
+    Trim leading/trailing silence in place, preserving the original on failure.
+
+    Never raises: any problem degrades to "leave the file exactly as it was",
+    because shipping an untrimmed recording is far better than shipping none.
+    """
+    threshold = settings.trim_silence_threshold_db if threshold_db is None else threshold_db
+    margin = settings.trim_silence_guard_seconds if guard is None else guard
+
+    try:
+        _, stderr = await _run([
+            "ffmpeg", "-hide_banner", "-nostdin", "-nostats",
+            "-i", str(path),
+            "-af", f"silencedetect=noise={threshold}dB:d={MIN_SILENCE_SECONDS}",
+            "-f", "null", "-",
+        ])
+    except Exception as e:
+        logger.warning(f"Silence detection failed for {path.name} ({e}) — uploading untrimmed.")
+        return TrimResult(path=path)
+
+    duration = _parse_duration(stderr)
+    if duration is None or duration <= 0:
+        logger.warning(f"Could not determine duration of {path.name} — uploading untrimmed.")
+        return TrimResult(path=path)
+
+    regions = _parse_silences(stderr)
+    keep_start, keep_end, all_silence = speech_bounds(regions, duration, margin)
+
+    if all_silence:
+        # Deliberately NOT trimmed to nothing. The caller decides what to do with
+        # a recording that contains no speech at all — that is itself a signal
+        # (squelch misconfigured, wrong sink, dead audio path).
+        logger.warning(
+            f"{path.name} is entirely silence ({duration:.2f}s, threshold {threshold}dB) — "
+            "no speech detected."
+        )
+        return TrimResult(path=path, duration_before=duration, duration_after=duration, all_silence=True)
+
+    lead = keep_start
+    tail = duration - keep_end
+    if (lead + tail) < MIN_TRIM_SECONDS:
+        return TrimResult(path=path, duration_before=duration, duration_after=duration)
+
+    trimmed = path.with_name(f"{path.stem}_trimmed{path.suffix}")
+    try:
+        code, err = await _run([
+            "ffmpeg", "-hide_banner", "-nostdin", "-nostats",
+            "-loglevel", "warning", "-y",
+            "-ss", f"{keep_start:.3f}",
+            "-i", str(path),
+            "-t", f"{keep_end - keep_start:.3f}",
+            "-ac", "1", "-ar", sample_rate, "-b:a", bitrate,
+            "-f", "mp3", str(trimmed),
+        ])
+    except Exception as e:
+        logger.warning(f"Silence trim failed for {path.name} ({e}) — uploading untrimmed.")
+        trimmed.unlink(missing_ok=True)
+        return TrimResult(path=path, duration_before=duration, duration_after=duration)
+
+    if code != 0 or not trimmed.exists() or trimmed.stat().st_size == 0:
+        logger.warning(f"Silence trim produced no output for {path.name} ({err.strip()}) — uploading untrimmed.")
+        trimmed.unlink(missing_ok=True)
+        return TrimResult(path=path, duration_before=duration, duration_after=duration)
+
+    trimmed.replace(path)
+    logger.info(
+        f"Trimmed {path.name}: -{lead:.2f}s lead, -{tail:.2f}s tail "
+        f"({duration:.2f}s → {keep_end - keep_start:.2f}s)"
+    )
+    return TrimResult(
+        path=path,
+        lead=lead,
+        tail=tail,
+        duration_before=duration,
+        duration_after=keep_end - keep_start,
+        applied=True,
+    )

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

@@ -1,55 +1,169 @@
+"""
+Continuous PulseAudio capture: a ring buffer for PRE-ROLL, a per-call
+accumulator for the call itself.
+
+A persistent capture process runs for the lifetime of the node. Spawning FFmpeg
+per call used to lose the first 1-2 s to process startup, which meant short
+transmissions produced empty files, so capture never stops.
+
+TWO BUFFERS, TWO JOBS — this split is load-bearing:
+
+  RING BUFFER   holds the last RING_BUFFER_SECONDS of audio at all times. Its
+                only job is PRE-ROLL: however late we notice a grant, we can
+                still seek back to OP25's exact timestamp. It is sized for
+                detection latency, nothing else.
+
+  ACCUMULATOR   opened by start_recording(), fed by every subsequent chunk, and
+                closed by stop_recording(). Call length is therefore bounded by
+                MAX_RECORDING_SECONDS alone — NOT by the ring buffer size. The
+                old design sliced the finished call back out of the ring buffer,
+                which silently clamped the front of any call longer than
+                RING_BUFFER_SECONDS.
+
+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 dataclasses import dataclass, field
 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 audio_trim, credentials, pulse
 from app.internal.logger import logger
 
-MAX_RECORDING_SECONDS = 600  # safety cap; drop call if it runs this long
+# Safety cap on a single recording; mirrors MAX_SEGMENT_SECONDS in metadata_watcher.
-PRE_BUFFER_SECONDS = 1.0  # seconds of audio to include before call_start
+MAX_RECORDING_SECONDS = 600
-RING_BUFFER_SECONDS = 60  # how much history to keep when no call is active
+
-READ_CHUNK_BYTES = 4096  # bytes per httpx read
+# 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.
+#
+# Kept small on purpose: measurement on a live node shows 1.71–2.45 s of real
+# grant→speech delay on every call, so there is no clipping risk at the head and
+# a larger pre-roll would only add dead air.
+PRE_ROLL_SECONDS = 0.25
+
+# Rolling history kept for PRE-ROLL ONLY. 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. This value does NOT bound call length — the accumulator does.
+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"
+
+# Hard memory ceiling for one call's accumulator. 16 kbps is 2 KB/s, so 600 s of
+# call is ~1.2 MB; 4x that is the ceiling, which both leaves room for encoder
+# overshoot and guarantees a runaway call can never eat a Pi's RAM.
+_MP3_BYTES_PER_SECOND = 16_000 // 8
+MAX_RECORDING_BYTES = MAX_RECORDING_SECONDS * _MP3_BYTES_PER_SECOND * 4
+
+# How long stop_recording() will wait for captured audio to actually reach the
+# call's end timestamp. PulseAudio → FFmpeg → MP3 encoder → muxer → our pipe read
+# is a pipeline with latency, so at the instant a call ends the newest buffered
+# chunk is typically a few hundred ms OLDER than the end epoch. Slicing
+# immediately therefore cuts the tail short — which costs the last word of the
+# transmission, usually the disposition or the address. Bounded so a dead capture
+# can never hang the upload path.
+TAIL_WAIT_TIMEOUT_SECONDS = 2.0
+TAIL_WAIT_POLL_SECONDS = 0.05
+
+# Backoff bounds for restarting a dead capture process.
+RESTART_BACKOFF_MIN = 1.0
+RESTART_BACKOFF_MAX = 15.0
+
+
+@dataclass
+class _ActiveRecording:
+    """Audio accumulating for the call currently being recorded."""
+
+    call_id: str
+    call_start: float           # OP25 grant epoch
+    slice_start: float          # call_start - PRE_ROLL_SECONDS
+    chunks: List[Tuple[float, bytes]] = field(default_factory=list)
+    total_bytes: int = 0
+    # Seconds of requested pre-roll that were not in the buffer at open time.
+    clamped_seconds: float = 0.0
+    # True once the byte ceiling was hit and audio started being dropped.
+    truncated_by_cap: bool = False
+
+
+@dataclass
+class Recording:
+    """
+    A finished recording plus the timing metadata needed to map audio position
+    back to wall clock.
+
+    `started_at`/`ended_at` upstream keep meaning the CALL's bounds. These are
+    the AUDIO's bounds, which differ once silence is trimmed:
+
+        wall_clock_of(audio_offset_t) == audio_start_epoch + t
+
+    Bounds are accurate to ±one capture chunk (~128 ms).
+    """
+
+    call_id: str
+    path: Optional[Path]
+    audio_start_epoch: float
+    audio_end_epoch: float
+    lead_trimmed: float = 0.0
+    tail_trimmed: float = 0.0
+    clamped_seconds: float = 0.0
+    all_silence: bool = False
 
 
 class CallRecorder:
-    """
+    """Continuous PulseAudio capture: ring buffer for pre-roll, accumulator per call."""
-    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.
-    """
 
     def __init__(self):
         self._recordings_dir = Path(settings.recordings_path)
 
-        # Ring buffer: deque of (monotonic_time, bytes_chunk)
+        # Ring buffer: deque of (wall_clock_epoch_at_arrival, mp3_bytes).
-        self._buffer: deque[tuple[float, bytes]] = deque()
+        # Pre-roll only — see the module docstring.
+        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 (None when idle)
-        self._call_id: Optional[str] = None
+        self._active: Optional[_ActiveRecording] = None
-        self._call_start_mono: Optional[float] = None
 
     # ------------------------------------------------------------------
     # Lifecycle
     # ------------------------------------------------------------------
 
     async def start(self) -> None:
-        """Start the persistent stream buffer.  Call once from app lifespan."""
+        """Start the persistent capture. Call once from app lifespan."""
-        self._stream_task = asyncio.create_task(self._stream_loop())
+        self._stream_task = asyncio.create_task(self._capture_loop())
-        logger.info("Stream ring-buffer started.")
+        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,111 +171,330 @@ class CallRecorder:
             except asyncio.CancelledError:
                 pass
             self._stream_task = None
+        await self._terminate_proc()
 
     # ------------------------------------------------------------------
-    # Stream reader
+    # Capture
     # ------------------------------------------------------------------
 
-    async def _stream_loop(self) -> None:
+    def _ffmpeg_command(self) -> List[str]:
-        stream_url = (
+        return [
-            f"http://{settings.icecast_host}:{settings.icecast_port}"
+            "ffmpeg",
-            f"{settings.icecast_mount}"
+            "-hide_banner", "-nostdin", "-nostats",
-        )
+            "-loglevel", "warning",
-        timeout = httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0)
+            "-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:
+                # The original bug this path was abandoned for: FFmpeg was launched
-                    async with client.stream("GET", stream_url) as response:
+                # with -f pulse before the shared socket existed, failed instantly,
-                        response.raise_for_status()
+                # and never recovered. Wait for it, bounded, every time.
-                        logger.info(f"Stream buffer connected to {stream_url}")
+                if not await pulse.wait_until_ready():
-                        async for chunk in response.aiter_bytes(READ_CHUNK_BYTES):
+                    await asyncio.sleep(backoff)
-                            self._ingest(chunk)
+                    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")
+                logger.warning(f"PulseAudio capture error ({e}) — restarting.")
-                await asyncio.sleep(3)
+
+            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."""
+        """Append a chunk to the ring buffer and, if recording, the accumulator."""
-        now = time.monotonic()
+        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).
+        # The ring buffer serves pre-roll only, so it is trimmed to a fixed
-        # Between calls, keep a rolling RING_BUFFER_SECONDS window.
+        # window unconditionally — an open recording no longer pins it, because
-        if self._call_start_mono is not None:
+        # the accumulator owns that audio.
-            keep_from = self._call_start_mono - PRE_BUFFER_SECONDS
+        keep_from = now - RING_BUFFER_SECONDS
-        else:
+        while self._buffer and self._buffer[0][0] < keep_from:
-            keep_from = now - RING_BUFFER_SECONDS
+            _, old = self._buffer.popleft()
-
-        while self._buffer:
-            ts, old = self._buffer[0]
-            if ts >= keep_from:
-                break
-            self._buffer.popleft()
             self._buffer_bytes -= len(old)
 
+        active = self._active
+        if active is None:
+            return
+
+        if active.total_bytes + len(chunk) > MAX_RECORDING_BYTES:
+            if not active.truncated_by_cap:
+                active.truncated_by_cap = True
+                logger.warning(
+                    f"Recording {active.call_id} hit the {MAX_RECORDING_BYTES} byte memory ceiling "
+                    f"after {now - active.slice_start:.1f}s — further audio is being dropped."
+                )
+            return
+
+        active.chunks.append((now, chunk))
+        active.total_bytes += len(chunk)
+
     # ------------------------------------------------------------------
-    # 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:
-        if self._call_id:
+        """
-            logger.warning("Recording already active — ignoring start.")
+        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._active is not None:
+            logger.warning(f"Recording already active ({self._active.call_id}) — ignoring start for {call_id}.")
             return False
-        self._call_id = call_id
+
-        self._call_start_mono = time.monotonic()
+        call_start = start_epoch if start_epoch else time.time()
-        logger.info(f"Recording started (ring-buffer): {call_id}")
+        slice_start = 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.")
+
+        # Seed the accumulator with the pre-roll already sitting in the ring
+        # buffer. No await between reading the buffer and publishing _active, so
+        # the capture task cannot slip a chunk in between and double-count it.
+        clamped = 0.0
+        oldest = self._buffer[0][0] if self._buffer else None
+        if oldest is not None and 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 LOUDLY —
+            # this is silent audio loss otherwise.
+            clamped = oldest - slice_start
+            logger.warning(
+                f"BUFFER CLAMP: pre-roll for call {call_id} predates buffered audio by "
+                f"{clamped:.2f}s — recording starts at the buffer head and that audio is lost."
+            )
+
+        seeded = [(ts, chunk) for ts, chunk in self._buffer if ts >= slice_start]
+        self._active = _ActiveRecording(
+            call_id=call_id,
+            call_start=call_start,
+            slice_start=slice_start,
+            chunks=seeded,
+            total_bytes=sum(len(c) for _, c in seeded),
+            clamped_seconds=clamped,
+        )
+
+        logger.info(f"Recording started: {call_id} (slice from {slice_start:.3f})")
         return True
 
-    async def stop_recording(self) -> Optional[Path]:
+    async def stop_recording(self, end_epoch: Optional[float] = None) -> Optional[Recording]:
-        if not self._call_id:
+        """
+        Close the recording and write the file. `end_epoch` is host wall clock.
+
+        Waits (bounded) for captured audio to actually cover `end_epoch` before
+        slicing — see TAIL_WAIT_TIMEOUT_SECONDS. Returns None only when there was
+        no recording open or no audio at all.
+        """
+        active = self._active
+        if active is None:
             return None
 
-        call_id = self._call_id
+        call_id = active.call_id
-        call_start = self._call_start_mono
+        slice_start = active.slice_start
-        self._call_id = None
-        self._call_start_mono = None
 
-        # Slice: everything from (call_start - pre_buffer) to now
+        end = end_epoch if end_epoch else time.time()
-        cutoff = (call_start - PRE_BUFFER_SECONDS) if call_start else 0.0
+        end = min(end, active.call_start + MAX_RECORDING_SECONDS)
-        chunks = [chunk for ts, chunk in self._buffer if ts >= cutoff]
 
-        # Safety cap: if the call ran very long, truncate to MAX_RECORDING_SECONDS
+        # The accumulator keeps filling during this wait — that is the point.
-        if call_start is not None:
+        await self._await_tail(end, call_id)
-            cap_cutoff = call_start + MAX_RECORDING_SECONDS
+        self._active = None
-            now = time.monotonic()
+
-            if now > cap_cutoff:
+        chunks: List[bytes] = []
-                # Approximate: trim chunks that arrived after the cap
+        last_ts = slice_start
-                cap_keep_until = call_start + MAX_RECORDING_SECONDS
+        for ts, chunk in active.chunks:
-                chunks = [
+            if ts < slice_start:
-                    chunk for ts, chunk in self._buffer
+                continue
-                    if cutoff <= ts <= cap_keep_until
+            chunks.append(chunk)
-                ]
+            last_ts = ts
+            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} — "
+                f"No buffered audio for call {call_id} "
-                "stream may not have been connected yet."
+                f"(window {slice_start:.3f}–{end:.3f}) — PulseAudio capture may be down."
             )
             return None
 
+        if last_ts < end - TAIL_WAIT_POLL_SECONDS:
+            logger.warning(
+                f"BUFFER CLAMP: call {call_id} ends {end - last_ts:.2f}s after the newest captured "
+                "audio — the tail is short. Capture may be stalled or restarting."
+            )
+
         self._recordings_dir.mkdir(parents=True, exist_ok=True)
         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(b"".join(chunks))
-        output_path.write_bytes(data)
 
         size = output_path.stat().st_size
-        if size > 0:
+        if size <= 0:
-            logger.info(f"Recording saved: {output_path.name} ({size} bytes)")
+            output_path.unlink(missing_ok=True)
-            return output_path
+            logger.warning(f"Recording for call {call_id} produced an empty file.")
+            return None
 
-        output_path.unlink(missing_ok=True)
+        audio_end = min(end, last_ts)
-        logger.warning(f"Recording for call {call_id} produced an empty file.")
+        logger.info(f"Recording saved: {output_path.name} ({size} bytes, {audio_end - slice_start:.2f}s window)")
-        return None
+
+        recording = Recording(
+            call_id=call_id,
+            path=output_path,
+            audio_start_epoch=slice_start,
+            audio_end_epoch=audio_end,
+            clamped_seconds=active.clamped_seconds,
+        )
+        return await self._apply_trim(recording)
+
+    async def _await_tail(self, end: float, call_id: str) -> float:
+        """
+        Block until captured audio reaches `end`, or the bounded timeout expires.
+
+        Returns seconds waited. Logs whenever a wait was actually needed so the
+        real pipeline latency is observable in the field.
+        """
+        if not self._buffer:
+            return 0.0
+        if self._buffer[-1][0] >= end:
+            return 0.0
+
+        started = time.monotonic()
+        deadline = started + TAIL_WAIT_TIMEOUT_SECONDS
+        while time.monotonic() < deadline:
+            await asyncio.sleep(TAIL_WAIT_POLL_SECONDS)
+            if self._buffer and self._buffer[-1][0] >= end:
+                waited = time.monotonic() - started
+                logger.info(f"Waited {waited:.2f}s for the tail of call {call_id} to reach the buffer.")
+                return waited
+            if not self._capturing:
+                break  # capture died mid-wait; nothing more is coming
+
+        waited = time.monotonic() - started
+        newest = self._buffer[-1][0] if self._buffer else end
+        logger.warning(
+            f"Tail wait for call {call_id} gave up after {waited:.2f}s — captured audio is still "
+            f"{max(0.0, end - newest):.2f}s short of the call end. Tail may be clipped."
+        )
+        return waited
+
+    async def _apply_trim(self, recording: Recording) -> Recording:
+        """
+        Strip leading/trailing dead air and keep the timing metadata honest.
+
+        An all-silence recording is NOT uploaded: it carries no information and
+        silence is exactly what makes Whisper hallucinate. It is logged instead,
+        because it also means something is wrong with the audio path.
+        """
+        if not settings.trim_silence or recording.path is None:
+            return recording
+
+        result = await audio_trim.trim_silence(
+            recording.path,
+            sample_rate=MP3_SAMPLE_RATE,
+            bitrate=MP3_BITRATE,
+        )
+
+        if result.all_silence:
+            logger.warning(
+                f"Call {recording.call_id} contains no speech at all — skipping upload. "
+                "Check squelch, the Liquidsoap output and the drb_sink monitor."
+            )
+            recording.path.unlink(missing_ok=True)
+            recording.path = None
+            recording.all_silence = True
+            return recording
+
+        if result.applied:
+            recording.lead_trimmed = result.lead
+            recording.tail_trimmed = result.tail
+            # Wall clock of the trimmed audio's first and last sample.
+            recording.audio_start_epoch += result.lead
+            recording.audio_end_epoch -= result.tail
+        return recording
 
     # ------------------------------------------------------------------
     # Upload (unchanged interface)
@@ -174,6 +507,8 @@ class CallRecorder:
         talkgroup_id: Optional[int] = None,
         talkgroup_name: Optional[str] = None,
         system_id: Optional[str] = None,
+        audio_start_epoch: Optional[float] = None,
+        audio_end_epoch: Optional[float] = None,
     ) -> Optional[str]:
         if not settings.c2_url:
             logger.info("No C2_URL configured — skipping upload.")
@@ -190,6 +525,13 @@ class CallRecorder:
             form["talkgroup_name"] = talkgroup_name
         if system_id:
             form["system_id"] = system_id
+        # Where this audio really sits on the wall clock once silence is trimmed.
+        # C2 does not declare these Form fields yet, so FastAPI ignores them —
+        # they cost nothing and are here for when playback/correlation want them.
+        if audio_start_epoch is not None:
+            form["audio_start_epoch"] = f"{audio_start_epoch:.3f}"
+        if audio_end_epoch is not None:
+            form["audio_end_epoch"] = f"{audio_end_epoch:.3f}"
 
         try:
             async with httpx.AsyncClient(timeout=120) as client:
@@ -213,9 +555,24 @@ class CallRecorder:
             except Exception:
                 pass
 
+    # ------------------------------------------------------------------
+    # State
+    # ------------------------------------------------------------------
+
     @property
     def is_recording(self) -> bool:
-        return self._call_id is not None
+        return self._active 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,
+            settings.pulse_source,
-            before_options="-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5",
+            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
+            # Back off before retrying — prevents a tight loop when PulseAudio is
-            async def _delayed_restart():
+            # unavailable (FFmpeg exits immediately in that case).
-                await asyncio.sleep(5)
+            self._schedule_restart()
-                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()))
         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/logger.py

@@ -7,4 +7,12 @@ logging.basicConfig(
     handlers=[logging.StreamHandler(sys.stdout)],
 )
 
+# The metadata watcher polls the OP25 terminal twice a second and httpx logs
+# every one of those requests at INFO ("HTTP Request: POST http://... 200 OK").
+# That is ~170k lines/day of pure noise which buries real events and makes field
+# log-reading useless. WARNING keeps genuine transport failures visible.
+# httpcore is the transport layer underneath httpx and is just as chatty.
+for _noisy in ("httpx", "httpcore"):
+    logging.getLogger(_noisy).setLevel(logging.WARNING)
+
 logger = logging.getLogger("drb-edge-node")

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

@@ -1,38 +1,133 @@
+"""
+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)
+# 500 ms. Do NOT lower: start precision already comes from OP25's own timestamp,
-POLL_INTERVAL = 0.5  # seconds
+# 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
+
+# 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 _tail_pad() -> float:
+    """
+    Audio kept after the observed end of the last transmission, so the srcaddr
+    edge (up to one poll late) plus encoder latency never clips the tail.
+
+    Read live from settings (env CALL_TAIL_PAD_SECONDS) rather than frozen into a
+    module constant, so it is tunable per node. See the setting for why the
+    default moved 0.5 → 1.0.
+    """
+    return settings.call_tail_pad_seconds
+
+
+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._current_freq: Any = None
-        self._active_call_id: Optional[str] = None
+        self._current_srcaddr: Optional[int] = None
-        self._call_started_at: Optional[datetime] = 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 +137,236 @@ 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:
+    async def _tick(self):
-            # OP25 not responding — hang-out any active call
+        now = self._clock()
-            if self._active_call_id:
+        update = await op25_client.poll_terminal()
-                self._hang_counter += 1
+
-                if self._hang_counter >= HANG_THRESHOLD:
+        if update is None:
-                    await self._end_call()
+            # 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
+        self._last_ok_poll = now
-        channels = status if isinstance(status, list) else [status]
-        active_tgid: Optional[int] = None
-        active_meta: dict = {}
 
-        for ch in channels:
+        # 1. call_log first — these are the authoritative starts, and processing
-            tgid = ch.get("tgid") or ch.get("tg_id")
+        #    them before the channel scan means a same-poll grant+state pair is
-            if tgid and str(tgid) not in ("0", "", "None"):
+        #    already attributed to the new segment by the time we scan channels.
-                active_tgid = int(tgid)
+        #    Sorted defensively: multi-receiver setups append per receiver.
-                active_meta = ch
+        for entry in sorted(update.call_log, key=lambda e: _as_float(e.get("time")) or 0.0):
-                break
+            await self._handle_call_log(entry, now)
 
-        if active_tgid:
+        # 2. channel_update — the only external end signal.
-            self._hang_counter = 0
+        await self._handle_channels(update.channels, now)
-            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()
 
-    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.
+            #
+            # The measured idle below is the CONTROL-CHANNEL idle (srcaddr 1→0
+            # edge → now). It is NOT comparable to silence measured in the audio,
+            # which additionally contains the ~1.9 s P25 grant→speech delay.
+            # Tune settings.call_idle_timeout from THIS number and nothing else.
+            if self._last_tx_end is not None:
+                measured_idle = now - self._last_tx_end
+                end = self._last_tx_end + _tail_pad()
+                logger.info(
+                    f"Idle timeout for tgid {self._current_tgid}: measured control-channel idle "
+                    f"{measured_idle:.2f}s (threshold {settings.call_idle_timeout:.2f}s, "
+                    f"tail pad {_tail_pad():.2f}s)."
+                )
+            else:
+                end = now
+                logger.info(
+                    f"Idle timeout for tgid {self._current_tgid}: no srcaddr end edge observed, "
+                    f"idle {now - self._last_activity:.2f}s measured from last activity."
+                )
+            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 = tgid
-        self._current_tgid_name = meta.get("tag") or meta.get("tgid_tag") or ""
+        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"),
+            "freq": self._current_freq,
-            "srcaddr": meta.get("srcaddr"),
+            "srcaddr": self._current_srcaddr,
-            "started_at": self._call_started_at.isoformat(),
+            "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,
+            "freq": self._current_freq,
-            "ended_at": datetime.now(timezone.utc).isoformat(),
+            "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._current_freq = None
-        self._call_started_at = 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]:
+    async def poll_terminal(self) -> Optional[TerminalUpdate]:
-        """Poll the OP25 HTTP terminal for current call metadata."""
+        """
+        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(
+                r = await client.post(self.terminal_url, json=TERMINAL_UPDATE_COMMAND)
-                    self.terminal_url,
-                    json=[{"command": "update", "arg1": 0, "arg2": 0}],
-                )
                 r.raise_for_status()
-                messages = r.json()
+                return parse_terminal_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
         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

@@ -1,5 +1,8 @@
 import asyncio
 from contextlib import asynccontextmanager
+from datetime import datetime, timezone
+from typing import Optional
+
 from fastapi import FastAPI
 from app.config import settings
 from app.models import SystemConfig
@@ -20,33 +23,71 @@ from app.routers import api, ui
 # Event handlers wired up at startup
 # ---------------------------------------------------------------------------
 
+def _iso(epoch: Optional[float]) -> Optional[str]:
+    """Epoch → UTC ISO-8601, matching metadata_watcher's timestamp format."""
+    if epoch is None:
+        return None
+    return datetime.fromtimestamp(epoch, timezone.utc).isoformat()
+
+
 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()
+    recording = await call_recorder.stop_recording(end_epoch=data.get("ended_at_epoch"))
-    if file_path:
+
+    if recording is not None and recording.path is not None:
+        # Silence trimming shortens the audio, so the audio's own bounds no
+        # longer equal the call's. `started_at`/`ended_at` keep meaning the CALL
+        # (what OP25 observed on the control channel) — these extra fields carry
+        # the AUDIO's wall-clock bounds so playback, correlation and incident
+        # timelines can still map an audio offset back to real time:
+        #     wall_clock_of(audio_offset_t) == audio_start_epoch + t
+        data["audio_start_at"] = _iso(recording.audio_start_epoch)
+        data["audio_end_at"] = _iso(recording.audio_end_epoch)
+        data["audio_start_epoch"] = recording.audio_start_epoch
+        data["audio_end_epoch"] = recording.audio_end_epoch
+        data["audio_lead_trimmed"] = round(recording.lead_trimmed, 3)
+        data["audio_tail_trimmed"] = round(recording.tail_trimmed, 3)
+        if recording.clamped_seconds:
+            data["audio_clamped_seconds"] = round(recording.clamped_seconds, 3)
+
+    if recording is not None and recording.path is not None:
         node_cfg = load_node_config()
         audio_url = await call_recorder.upload_recording(
-            file_path,
+            recording.path,
             data["call_id"],
             talkgroup_id=data.get("tgid"),
             talkgroup_name=data.get("tgid_name"),
             system_id=node_cfg.assigned_system_id,
+            audio_start_epoch=recording.audio_start_epoch,
+            audio_end_epoch=recording.audio_end_epoch,
         )
         if audio_url:
             data["audio_url"] = audio_url
         else:
             logger.error(f"Audio upload failed for call {data['call_id']}. Verify C2_URL and Node API Key.")
+    elif recording is not None and recording.all_silence:
+        # Explicit policy: an all-silence recording is not uploaded. It has no
+        # transcript value and silence is what makes Whisper invent text.
+        data["audio_skipped"] = "all_silence"
+        logger.warning(f"Call {data['call_id']} was pure silence — no upload. Investigate the audio path.")
     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 +232,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_audio_trim.py

@@ -0,0 +1,160 @@
+"""
+Unit tests for silence-trim decision logic.
+
+`speech_bounds` is pure on purpose so the "what do we keep" decision — the part
+that can destroy a transmission if it is wrong — is testable without FFmpeg.
+The numbers below come from ffmpeg silencedetect run against six real recordings
+off a live P25 node: 1.71–2.45 s of leading silence and 0.00–1.11 s trailing.
+"""
+import pytest
+
+from app.internal.audio_trim import (
+    TrimResult,
+    _parse_duration,
+    _parse_silences,
+    speech_bounds,
+)
+
+GUARD = 0.25
+
+
+def test_leading_silence_is_trimmed_with_a_guard_margin():
+    # Real shape of file f4bfaa1f: 1.85s lead, 0.34s trail, 4.54s total.
+    regions = [(0.0, 1.85), (4.20, None)]
+    start, end, all_silence = speech_bounds(regions, duration=4.54, guard=GUARD)
+
+    assert not all_silence
+    assert start == pytest.approx(1.85 - GUARD)
+    assert end == pytest.approx(4.20 + GUARD)
+    # The guard must never eat into detected speech.
+    assert start < 1.85 and end > 4.20
+
+
+def test_guard_margin_never_runs_past_the_file_bounds():
+    regions = [(0.0, 0.10), (3.95, None)]
+    start, end, _ = speech_bounds(regions, duration=4.0, guard=1.0)
+
+    assert start == 0.0
+    assert end == 4.0
+
+
+def test_trailing_silence_is_trimmed_when_ffmpeg_closes_the_region_at_eof():
+    """
+    FFmpeg 6.x flushes a `silence_end` at EOF, so a trailing region looks closed.
+    Treating "no silence_end" as the only trailing signal silently disabled tail
+    trimming entirely — verified against ffmpeg 6.1.1.
+    """
+    # Real ffmpeg 6.1.1 output for a 2s-silence + 1.5s-tone + 1s-silence file.
+    regions = [(0.0, 2.05361), (3.56367, 4.63102)]
+    start, end, all_silence = speech_bounds(regions, duration=4.65, guard=GUARD)
+
+    assert not all_silence
+    assert start == pytest.approx(2.05361 - GUARD)
+    assert end == pytest.approx(3.56367 + GUARD), "the trailing second must be trimmed"
+
+
+def test_all_silence_survives_ffmpeg_closing_the_region_at_eof():
+    # Real ffmpeg 6.1.1 output for a 4s file of pure silence.
+    _, _, all_silence = speech_bounds([(0.0, 4.0)], duration=4.03, guard=GUARD)
+    assert all_silence
+
+
+def test_trailing_silence_that_does_not_reach_eof_is_left_alone():
+    """
+    A silence region with a closing silence_end is an internal pause between
+    transmissions, not dead air at the tail. Trimming it would cut the middle
+    out of a conversation.
+    """
+    regions = [(0.0, 1.9), (5.0, 7.5)]
+    start, end, _ = speech_bounds(regions, duration=12.0, guard=GUARD)
+
+    assert start == pytest.approx(1.9 - GUARD)
+    assert end == 12.0, "an internal pause must not shorten the file"
+
+
+def test_no_silence_detected_keeps_the_whole_file():
+    start, end, all_silence = speech_bounds([], duration=6.0, guard=GUARD)
+
+    assert (start, end) == (0.0, 6.0)
+    assert not all_silence
+
+
+def test_silence_starting_late_is_not_treated_as_leading():
+    """Only a region at the very head counts as leading silence."""
+    regions = [(1.20, 2.00)]
+    start, end, _ = speech_bounds(regions, duration=5.0, guard=GUARD)
+
+    assert start == 0.0, "speech before 1.20s must not be trimmed away"
+    assert end == 5.0
+
+
+def test_all_silence_is_reported_not_trimmed_to_nothing():
+    # One region covering the whole file and running to EOF.
+    regions = [(0.0, None)]
+    start, end, all_silence = speech_bounds(regions, duration=4.0, guard=GUARD)
+
+    assert all_silence
+    assert (start, end) == (0.0, 4.0), "an all-silence file must not become zero-length"
+
+
+def test_all_silence_when_head_and_tail_regions_overlap():
+    regions = [(0.0, 3.2), (3.0, None)]
+    _, _, all_silence = speech_bounds(regions, duration=4.0, guard=GUARD)
+
+    assert all_silence
+
+
+# ---------------------------------------------------------------------------
+# FFmpeg output parsing
+# ---------------------------------------------------------------------------
+
+# Verbatim shape of ffmpeg 6.1.1 output.
+FFMPEG_STDERR = """
+Input #0, mp3, from '/recordings/x.mp3':
+  Duration: 00:00:04.70, start: 0.050113, bitrate: 16 kb/s
+[silencedetect @ 0000029160e63f40] silence_start: 0
+[silencedetect @ 0000029160e63f40] silence_end: 2.05361 | silence_duration: 2.05361
+[silencedetect @ 0000029160e63f40] silence_start: 3.56367
+[silencedetect @ 0000029160e63f40] silence_end: 4.63102 | silence_duration: 1.06735
+[out#0/null @ 0x2] video:0kB audio:97kB
+"""
+
+FFMPEG_STDERR_OPEN_TAIL = """
+  Duration: 00:00:04.54, start: 0.000000, bitrate: 16 kb/s
+[silencedetect @ 0x1] silence_start: 0
+[silencedetect @ 0x1] silence_end: 1.85042 | silence_duration: 1.85042
+[silencedetect @ 0x1] silence_start: 4.20134
+"""
+
+
+def test_duration_is_corrected_for_the_mp3_container_start_offset():
+    """
+    MP3 encoder delay makes the container duration longer than the audio
+    silencedetect timestamps. Without this correction the trailing-region test
+    needs a slack epsilon big enough to clip real speech.
+    """
+    assert _parse_duration(FFMPEG_STDERR) == pytest.approx(4.70 - 0.050113)
+
+
+def test_duration_is_none_when_absent():
+    assert _parse_duration("no duration here") is None
+
+
+def test_silence_regions_are_parsed():
+    regions = _parse_silences(FFMPEG_STDERR)
+
+    assert len(regions) == 2
+    assert regions[0] == (pytest.approx(0.0), pytest.approx(2.05361))
+    assert regions[1] == (pytest.approx(3.56367), pytest.approx(4.63102))
+
+
+def test_a_region_with_no_silence_end_is_still_parsed():
+    """Older FFmpeg simply stopped reporting at EOF — keep handling that."""
+    regions = _parse_silences(FFMPEG_STDERR_OPEN_TAIL)
+
+    assert regions[-1][1] is None
+
+
+def test_trim_result_reports_total_trimmed():
+    result = TrimResult(path=None, lead=1.9, tail=0.35)
+    assert result.trimmed_seconds == pytest.approx(2.25)

drb-edge-node/tests/test_call_recorder.py

@@ -0,0 +1,424 @@
+"""
+Unit tests for the CallRecorder pre-roll ring buffer and per-call accumulator.
+
+No FFmpeg and no PulseAudio: chunks are pushed through _ingest() with a patched
+clock, which is exactly what the capture loop does at runtime. Silence trimming
+is disabled by default here and exercised separately with a stubbed trimmer.
+"""
+import asyncio
+import itertools
+import time
+from typing import List
+from unittest.mock import patch
+
+import pytest
+
+from app.config import settings
+from app.internal import call_recorder as recorder_mod
+from app.internal.audio_trim import TrimResult
+from app.internal.call_recorder import (
+    CallRecorder,
+    MAX_RECORDING_BYTES,
+    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, monkeypatch):
+    monkeypatch.setattr(settings, "trim_silence", False)
+    r = CallRecorder()
+    r._recordings_dir = tmp_path
+    r._capturing = True
+    return r
+
+
+def ingest(recorder, start: float, end: float, marker: bytes = b"A", index: int = 0) -> int:
+    """Feed one chunk every CHUNK_INTERVAL seconds over [start, end) through _ingest."""
+    stamps: List[float] = []
+    chunks: List[bytes] = []
+    ts = start
+    while ts < end:
+        stamps.append(ts)
+        chunks.append(marker + str(index).encode() + b";")
+        index += 1
+        ts = round(ts + CHUNK_INTERVAL, 6)
+
+    with patch("app.internal.call_recorder.time.time", side_effect=stamps):
+        for chunk in chunks:
+            recorder._ingest(chunk)
+    return index
+
+
+def fill(recorder, start: float, end: float, marker: bytes = b"A", index: int = 0) -> int:
+    """Alias kept for readability where the accumulator is not the point."""
+    return ingest(recorder, start, end, marker=marker, index=index)
+
+
+def timestamps(recorder):
+    return [ts for ts, _ in recorder._buffer]
+
+
+def markers(path) -> List[str]:
+    return path.read_bytes().decode().strip(";").split(";")
+
+
+def indices(path) -> List[int]:
+    return [int(m[1:]) for m in markers(path) if m[1:].isdigit()]
+
+
+# ---------------------------------------------------------------------------
+# Ring buffer trimming (pre-roll duty only)
+# ---------------------------------------------------------------------------
+
+def test_idle_buffer_keeps_only_the_rolling_window(recorder):
+    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_ring_buffer_is_trimmed_even_while_recording(recorder):
+    """
+    The ring buffer serves PRE-ROLL only. An open recording must no longer pin
+    it — that was the mechanism that made call length depend on buffer size.
+    """
+    index = ingest(recorder, T0, T0 + 2.0)
+    await recorder.start_recording("call-1", start_epoch=T0 + 1.0)
+    ingest(recorder, T0 + 2.0, T0 + 2.0 + RING_BUFFER_SECONDS + 10, index=index)
+
+    assert recorder.buffered_seconds <= RING_BUFFER_SECONDS + 1
+    # ...and the audio the ring buffer dropped is safe in the accumulator.
+    assert recorder._active is not None
+    assert recorder._active.chunks[0][0] == pytest.approx(T0 + 1.0 - PRE_ROLL_SECONDS, abs=CHUNK_INTERVAL)
+
+
+# ---------------------------------------------------------------------------
+# Call length must not be bounded by the ring buffer
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_call_longer_than_the_ring_buffer_is_captured_whole(recorder):
+    call_length = RING_BUFFER_SECONDS * 2 + 5  # 65s against a 30s ring buffer
+    grant = T0 + 1.0
+    end = grant + call_length
+
+    index = ingest(recorder, T0, grant)
+    await recorder.start_recording("call-long", start_epoch=grant)
+    ingest(recorder, grant, end + 1.0, index=index)
+
+    rec = await recorder.stop_recording(end_epoch=end)
+    assert rec is not None and rec.path is not None
+
+    kept = indices(rec.path)
+    # Contiguous: no hole anywhere in the middle of a 65s call.
+    assert kept == list(range(kept[0], kept[-1] + 1))
+    span = (kept[-1] - kept[0]) * CHUNK_INTERVAL
+    assert span > RING_BUFFER_SECONDS, "call length must not be clamped by the ring buffer"
+    assert span == pytest.approx(call_length + PRE_ROLL_SECONDS, abs=2 * CHUNK_INTERVAL)
+
+
+@pytest.mark.asyncio
+async def test_accumulator_stops_growing_at_the_memory_ceiling(recorder, caplog):
+    """A runaway call must not be able to exhaust RAM on a Pi."""
+    await recorder.start_recording("call-runaway", start_epoch=T0)
+
+    big = b"z" * 64_000
+    needed = (MAX_RECORDING_BYTES // len(big)) + 5
+    ticks = itertools.count()
+    with caplog.at_level("WARNING", logger="drb-edge-node"):
+        with patch("app.internal.call_recorder.time.time",
+                   side_effect=lambda: T0 + next(ticks) * 0.1):
+            for _ in range(needed):
+                recorder._ingest(big)
+
+    assert recorder._active is not None
+    assert recorder._active.total_bytes <= MAX_RECORDING_BYTES
+    assert recorder._active.truncated_by_cap
+    assert any("memory ceiling" in r.message for r in caplog.records)
+
+
+# ---------------------------------------------------------------------------
+# 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._active.slice_start == pytest.approx(grant_time - PRE_ROLL_SECONDS)
+
+    rec = await recorder.stop_recording(end_epoch=grant_time + 2.0)
+    assert rec is not None and rec.path is not None and rec.path.exists()
+
+    first_ts = T0 + indices(rec.path)[0] * 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.
+    rec = await recorder.stop_recording(end_epoch=T0 + 3.05)
+    last_ts = T0 + indices(rec.path)[-1] * CHUNK_INTERVAL
+
+    assert last_ts >= T0 + 3.05, "the chunk covering the end instant must be kept"
+
+
+@pytest.mark.asyncio
+async def test_max_recording_seconds_caps_the_slice(recorder):
+    index = fill(recorder, T0, T0 + 1.0)
+    await recorder.start_recording("call-1", start_epoch=T0 + 1.0)
+    ingest(recorder, T0 + 1.0, T0 + MAX_RECORDING_SECONDS + 60, index=index)
+
+    rec = await recorder.stop_recording(end_epoch=T0 + MAX_RECORDING_SECONDS + 50)
+    last_ts = T0 + indices(rec.path)[-1] * CHUNK_INTERVAL
+
+    assert last_ts <= T0 + 1.0 + MAX_RECORDING_SECONDS + CHUNK_INTERVAL
+
+
+# ---------------------------------------------------------------------------
+# Tail wait — the fix for recordings that ended mid-word
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_stop_waits_for_captured_audio_to_reach_the_call_end(recorder, caplog):
+    """
+    PulseAudio → FFmpeg → encoder → muxer → our pipe read has latency, so at the
+    instant a call ends the newest captured chunk is OLDER than the end epoch.
+    Slicing immediately cuts the last word off. stop_recording must wait for it.
+    """
+    index = ingest(recorder, T0, T0 + 4.0)
+    await recorder.start_recording("call-1", start_epoch=T0 + 1.0)
+
+    async def late_tail():
+        await asyncio.sleep(0.15)
+        with patch("app.internal.call_recorder.time.time", return_value=T0 + 4.6):
+            recorder._ingest(b"TAIL;")
+
+    task = asyncio.create_task(late_tail())
+    with caplog.at_level("INFO", logger="drb-edge-node"):
+        rec = await recorder.stop_recording(end_epoch=T0 + 4.5)
+    await task
+
+    assert rec is not None and rec.path is not None
+    assert b"TAIL" in rec.path.read_bytes(), "the late-arriving tail must be in the file"
+    assert any("Waited" in r.message and "tail" in r.message for r in caplog.records), \
+        "a tail wait must be observable in the field logs"
+    assert index  # sanity: the pre-roll fill actually ran
+
+
+@pytest.mark.asyncio
+async def test_tail_wait_is_bounded_and_warns_when_audio_never_arrives(recorder, caplog, monkeypatch):
+    monkeypatch.setattr(recorder_mod, "TAIL_WAIT_TIMEOUT_SECONDS", 0.2)
+    ingest(recorder, T0, T0 + 4.0)
+    await recorder.start_recording("call-1", start_epoch=T0 + 1.0)
+
+    started = time.monotonic()
+    with caplog.at_level("WARNING", logger="drb-edge-node"):
+        rec = await recorder.stop_recording(end_epoch=T0 + 10.0)
+    elapsed = time.monotonic() - started
+
+    assert elapsed < 2.0, "the wait must be bounded, never open-ended"
+    assert rec is not None and rec.path is not None, "a short tail still beats no recording"
+    messages = [r.message for r in caplog.records]
+    assert any("Tail wait" in m and "gave up" in m for m in messages)
+    assert any("BUFFER CLAMP" in m for m in messages), "silent truncation must be loud"
+
+
+@pytest.mark.asyncio
+async def test_no_wait_when_the_buffer_already_covers_the_end(recorder, caplog):
+    ingest(recorder, T0, T0 + 10.0)
+    await recorder.start_recording("call-1", start_epoch=T0 + 1.0)
+
+    started = time.monotonic()
+    with caplog.at_level("INFO", logger="drb-edge-node"):
+        await recorder.stop_recording(end_epoch=T0 + 3.0)
+    assert (time.monotonic() - started) < 0.1
+    assert not any("Waited" in r.message for r in caplog.records)
+
+
+# ---------------------------------------------------------------------------
+# Clamping must be loud
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_pre_roll_earlier_than_buffer_start_is_clamped_and_warned(recorder, caplog):
+    """A grant older than anything buffered must still produce a file, loudly."""
+    ingest(recorder, T0 + 5.0, T0 + 10.0)  # buffer only covers T0+5 onwards
+
+    with caplog.at_level("WARNING", logger="drb-edge-node"):
+        await recorder.start_recording("call-1", start_epoch=T0)  # 5 s before the head
+    rec = await recorder.stop_recording(end_epoch=T0 + 8.0)
+
+    assert rec is not None and rec.path is not None and rec.path.stat().st_size > 0
+    assert markers(rec.path)[0] == "A0", "slice should begin at the buffer head, not fail"
+    # Buffer head is T0+5.0, requested slice start is T0-PRE_ROLL: everything in
+    # between is audio we can never recover, and the number must be reported.
+    assert rec.clamped_seconds == pytest.approx(5.0 + PRE_ROLL_SECONDS, abs=CHUNK_INTERVAL)
+    assert any("BUFFER CLAMP" in r.message for r in caplog.records)
+
+
+@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._active.slice_start == pytest.approx(now - PRE_ROLL_SECONDS, abs=1.0)
+
+
+# ---------------------------------------------------------------------------
+# Silence trimming and timing metadata
+# ---------------------------------------------------------------------------
+
+async def _recorded(recorder, end_offset: float = 3.0):
+    ingest(recorder, T0, T0 + 10.0)
+    await recorder.start_recording("call-1", start_epoch=T0 + 1.0)
+    return await recorder.stop_recording(end_epoch=T0 + end_offset)
+
+
+@pytest.mark.asyncio
+async def test_trimming_is_off_when_the_setting_is_off(recorder, monkeypatch):
+    monkeypatch.setattr(settings, "trim_silence", False)
+    called = False
+
+    async def _never(*args, **kwargs):
+        nonlocal called
+        called = True
+        return TrimResult(path=None)
+
+    monkeypatch.setattr(recorder_mod.audio_trim, "trim_silence", _never)
+    rec = await _recorded(recorder)
+    assert rec is not None and not called
+
+
+@pytest.mark.asyncio
+async def test_trim_shifts_the_audio_bounds_but_not_the_call_bounds(recorder, monkeypatch):
+    """
+    Trimming changes audio duration, so the AUDIO's wall-clock bounds move.
+    The call's own started_at/ended_at (owned by metadata_watcher) must not be
+    redefined — the recorder only reports where the audio now sits.
+    """
+    monkeypatch.setattr(settings, "trim_silence", True)
+
+    async def _trim(path, **kwargs):
+        return TrimResult(path=path, lead=1.9, tail=0.4, duration_before=3.3,
+                          duration_after=1.0, applied=True)
+
+    monkeypatch.setattr(recorder_mod.audio_trim, "trim_silence", _trim)
+
+    rec = await _recorded(recorder)
+    assert rec is not None and rec.path is not None
+    assert rec.lead_trimmed == pytest.approx(1.9)
+    assert rec.tail_trimmed == pytest.approx(0.4)
+    # Untrimmed slice was [T0+0.75, T0+3.0]; the audio now starts 1.9s later and
+    # ends 0.4s earlier, which is exactly what downstream needs to map an audio
+    # offset back to wall clock.
+    assert rec.audio_start_epoch == pytest.approx(T0 + 1.0 - PRE_ROLL_SECONDS + 1.9, abs=CHUNK_INTERVAL)
+    assert rec.audio_end_epoch == pytest.approx(T0 + 3.0 - 0.4, abs=CHUNK_INTERVAL)
+
+
+@pytest.mark.asyncio
+async def test_all_silence_recording_is_dropped_and_logged(recorder, monkeypatch, caplog):
+    monkeypatch.setattr(settings, "trim_silence", True)
+    seen = {}
+
+    async def _trim(path, **kwargs):
+        seen["path"] = path
+        return TrimResult(path=path, duration_before=4.0, duration_after=4.0, all_silence=True)
+
+    monkeypatch.setattr(recorder_mod.audio_trim, "trim_silence", _trim)
+
+    with caplog.at_level("WARNING", logger="drb-edge-node"):
+        rec = await _recorded(recorder)
+
+    assert rec is not None
+    assert rec.all_silence is True
+    assert rec.path is None, "an all-silence recording must not be uploaded"
+    assert not seen["path"].exists(), "the file must be cleaned up, not left on disk"
+    assert any("no speech" in r.message for r in caplog.records)
+
+
+# ---------------------------------------------------------------------------
+# 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.
+    """
+    ingest(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.path.stat().st_size > 0
+    assert second is not None and second.path.stat().st_size > 0
+    assert first.path != second.path
+
+
+# ---------------------------------------------------------------------------
+# 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"
+
+
+def test_memory_ceiling_covers_the_longest_allowed_call():
+    """The cap must bound RAM without ever being able to truncate a legal call."""
+    bytes_per_second = 16_000 // 8
+    assert MAX_RECORDING_BYTES >= MAX_RECORDING_SECONDS * bytes_per_second
+    assert MAX_RECORDING_BYTES <= 8 * 1024 * 1024, "must stay small enough for a Pi"

drb-edge-node/tests/test_metadata_watcher.py

@@ -1,190 +1,535 @@
 """
-Unit tests for MetadataWatcher state machine.
+Unit tests for the event-driven MetadataWatcher state machine.
-All OP25 HTTP calls are mocked — no running services required.
+
+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,
+    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):
+async def test_call_log_entry_starts_call(watcher, clock):
-    status = [{"tgid": 1234, "tag": "Police Dispatch"}]
+    grant_time = clock.now - 0.4  # OP25 logged it before our poll noticed
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)):
+    await tick(watcher, update(
-        await watcher._tick()
+        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 payload["call_id"]
-    assert "started_at" in payload
+    # 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):
+async def test_channel_activity_alone_does_not_start_a_call(watcher):
-    status = [{"tgid": 0}]
+    """No call_log entry => no call, even with a live tgid on the channel."""
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)):
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
-        await watcher._tick()
 
     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):
+async def test_call_log_entry_without_tgid_is_ignored(watcher):
-    status = [{"tgid": "None"}]
+    await tick(watcher, update(call_log=[grant(0, 100.0)]))
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)):
-        await watcher._tick()
 
     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):
+async def test_call_log_without_time_falls_back_to_local_clock(watcher, clock):
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=None)):
+    entry = grant(1234, 0.0)
-        await watcher._tick()
+    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):
+async def test_srcaddr_edge_then_idle_timeout_ends_call(watcher, clock):
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])):
+    await tick(watcher, update(
-        await watcher._tick()
+        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}])):
+    # srcaddr resets to 0 while tgid is still held — OP25's end-of-call signal.
-        for _ in range(HANG_THRESHOLD - 1):
+    clock.advance(0.5)
-            await watcher._tick()
+    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 + settings.call_tail_pad_seconds)
+    assert payload["tgid"] == 1234
+
+
+@pytest.mark.asyncio
+async def test_tail_pad_is_configurable_and_defaults_to_one_second(watcher, clock, monkeypatch):
+    """
+    0.5s left only ~0.3s of real trailing margin in field measurement and one
+    recording ended mid-word, so the default moved to 1.0 — and it has to be a
+    setting, not a magic number, so it can be tuned per node.
+    """
+    assert settings.call_tail_pad_seconds == 1.0
+
+    monkeypatch.setattr(settings, "call_tail_pad_seconds", 2.5)
+
+    await tick(watcher, update(
+        call_log=[grant(1234, clock.now)],
+        channels=[channel(tgid=1234, srcaddr=555)],
+    ))
+    clock.advance(0.5)
+    edge_time = clock.now
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
+
+    clock.advance(settings.call_idle_timeout + 1.0)
+    await tick(watcher, update(channels=[channel()]))
+
+    payload = watcher.on_call_end.call_args[0][0]
+    assert payload["ended_at_epoch"] == pytest.approx(edge_time + 2.5)
+
+
+@pytest.mark.asyncio
+async def test_idle_close_logs_the_measured_control_channel_idle(watcher, clock, caplog):
+    """
+    The correct idle timeout can only be tuned from the CONTROL-CHANNEL idle, not
+    from silence measured in the audio (which also contains the ~1.9s P25
+    grant→speech delay). So the real measured value has to reach the log.
+    """
+    await tick(watcher, update(
+        call_log=[grant(1234, clock.now)],
+        channels=[channel(tgid=1234, srcaddr=555)],
+    ))
+    clock.advance(0.5)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
+
+    with caplog.at_level("INFO", logger="drb-edge-node"):
+        clock.advance(settings.call_idle_timeout + 0.25)
+        await tick(watcher, update(channels=[channel()]))
+
+    idle_lines = [r.message for r in caplog.records if "measured control-channel idle" in r.message]
+    assert idle_lines, "idle-timeout closes must log the measured idle for later tuning"
+    assert "3.25s" in idle_lines[0]
+
+
+@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):
+async def test_op25_unreachable_ends_active_call_after_grace(watcher, clock):
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])):
+    await tick(watcher, update(
-        await watcher._tick()
+        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}])):
+    # A single failed poll must not kill the call.
-        for _ in range(HANG_THRESHOLD):
+    clock.advance(OP25_OFFLINE_GRACE / 2)
-            await watcher._tick()
+    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 payload["call_id"] == call_id
-    assert "ended_at" in payload
+    assert payload["transmissions"] == 2
+    assert payload["started_at_epoch"] == start
 
 
 @pytest.mark.asyncio
-async def test_op25_offline_triggers_hang_and_ends_call(watcher):
+async def test_different_tgid_grant_splits_recording(watcher, clock):
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])):
+    await tick(watcher, update(
-        await watcher._tick()
+        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)):
+    clock.advance(2.0)
-        for _ in range(HANG_THRESHOLD):
+    split_time = clock.now
-            await watcher._tick()
+    await tick(watcher, update(
-
+        call_log=[grant(2222, split_time, tgtag="EMS")],
-    assert not watcher.is_active
+        channels=[channel(tgid=2222, srcaddr=2)],
-    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()
 
     assert watcher.is_active
     assert watcher.current_tgid == 2222
-    assert watcher.active_call_id != first_call_id
+    assert watcher.active_call_id != first_id
-    watcher.on_call_end.assert_called_once()
     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["call_id"] == first_id
-# Status format variations
+    assert ended["tgid"] == 1111
-# ---------------------------------------------------------------------------
+    assert ended["end_reason"] == "tgid_change"
-
+    # The outgoing segment ends exactly where the new one begins — no tail pad,
-@pytest.mark.asyncio
+    # or it would swallow the first moments of the new talkgroup.
-async def test_single_dict_status_instead_of_list(watcher):
+    assert ended["ended_at_epoch"] == split_time
-    """OP25 terminal may return a bare dict instead of a list."""
+    assert watcher.on_call_start.call_args[0][0]["started_at_epoch"] == split_time
-    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
 
 
 @pytest.mark.asyncio
-async def test_tg_id_key_alias(watcher):
+async def test_multiple_call_log_entries_in_one_poll(watcher, clock):
-    """Some OP25 builds use 'tg_id' instead of 'tgid'."""
+    """
-    status = [{"tg_id": 5555, "tag": "EMS"}]
+    The call_log deque is drained per poll and capped at 10, so one poll can
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)):
+    deliver several grants. Same-tgid ones merge, different-tgid ones split.
-        await watcher._tick()
+    """
+    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):
+async def test_out_of_order_call_log_entries_are_sorted(watcher, clock):
-    """When multiple channels are returned, first active tgid wins."""
+    t0 = clock.now - 2.0
-    status = [
+    await tick(watcher, update(
-        {"tgid": 0},
+        call_log=[grant(2222, t0 + 1.0), grant(1111, t0)],
-        {"tgid": 7777, "tag": "Roads"},
+        channels=[channel(tgid=2222, srcaddr=9)],
-        {"tgid": 8888, "tag": "Other"},
+    ))
-    ]
-    with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)):
-        await watcher._tick()
 
-    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"]