Pad the tgid_change closes so a talkgroup switch stops clipping the tail

The tgid_change and tgid_change_unlogged paths closed the outgoing
segment with no tail pad, on the reasoning that the new grant's
timestamp is an exact, already-known boundary. That is exact only in
control-channel time. The buffered audio lags control timestamps by
~1.5s (measured 0.84-1.62s across seven field calls), so slicing there
cut roughly the outgoing call's last 1.5s of speech - recordings ending
mid-word with ~0s trailing silence.

Both paths now pad, and the recorder's bounded tail wait blocks until
that audio has actually been captured. TAIL_WAIT_TIMEOUT_SECONDS goes
2.0 -> 4.0 so it can satisfy the 3.0s pad instead of giving up and
warning on every talkgroup switch.

The incoming call's pre-roll is served from the ring buffer, so the
delay costs it nothing. The two slices overlapping in the underlying
audio is correct: the stream genuinely contains one call's tail and
then the next call's start.

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

.env.example

@@ -16,12 +16,46 @@ 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. Raised 1.0 -> 3.0 after field measurement showed the
+# grant-to-speech offset is ~0.84-1.62s (typically ~1.5s): at 1.0s pad, short
+# calls had their recording window close before the voice even started,
+# clipping speech mid-word. Safe to be generous — trim_silence already strips
+# the extra back off long calls before upload, so only short transmissions
+# actually benefit from the larger window.
+CALL_TAIL_PAD_SECONDS=3.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/Dockerfile

@@ -5,6 +5,7 @@ RUN apt-get update && apt-get install -y \
     libopus0 \
     libopus-dev \
     libpulse0 \
+    pulseaudio-utils \
     && rm -rf /var/lib/apt/lists/*
 
 WORKDIR /app

drb-edge-node/app/config.py

@@ -18,12 +18,69 @@ 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.
+    #
+    # Raised 1.0 -> 3.0 after field measurement showed the recording WINDOW
+    # (anchored to OP25 control-channel timestamps) closing well before the
+    # actual voice audio arrives: grant->speech offset measured 0.84-1.62s
+    # across 7 calls (~1.5s typical). At the old 1.0s pad, a short
+    # transmission (e.g. a 0.97s control-channel call) had its window close
+    # at T+1.97 while voice didn't start until ~T+1.5 — leaving ~0.4s of
+    # captured speech, clipped mid-word. Confirmed by a 0.57s output file
+    # whose final 0.10s measured -12.2dB, louder than its own -18.2dB
+    # average (i.e. clipped speech, not trailing silence), and by two short
+    # calls that produced no "Trimmed" log line at all because there was no
+    # trailing silence left to trim.
+    #
+    # Safe to be generous here: trim_silence already strips trailing silence
+    # back to trim_silence_guard_seconds before upload, so a larger pad costs
+    # long calls nothing (the extra is trimmed away) while giving short
+    # transmissions enough window to actually capture the voice. Over-capture
+    # is free; under-capture loses words permanently. Do not tune this back
+    # down without new field data showing the grant->speech offset has
+    # shrunk — see DEFERRED.md for the call_idle_timeout coupling this value
+    # now sits at.
+    call_tail_pad_seconds: float = 3.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 +89,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,190 @@
+"""
+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.
+#
+# Must exceed settings.call_tail_pad_seconds (default 3.0), otherwise a
+# tgid_change close — which pads past a timestamp that is still ~now — gives up
+# before the padded audio has been captured and warns on every talkgroup switch.
+TAIL_WAIT_TIMEOUT_SECONDS = 4.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
+
+# Substrings looked for in FFmpeg's stderr to classify why capture exited.
+# "No such process" is what FFmpeg's pulse input prints when the daemon is up
+# but the named source does not exist — a real PULSE_SOURCE misconfiguration,
+# not a startup race. This is exactly the failure mode that hid unnoticed
+# behind a generic "restarting" log in the April outage: silently retrying
+# forever against a wrong source name looks identical to a normal startup
+# wait unless it is logged differently.
+_SOURCE_MISSING_MARKERS = ("no such process", "no such device")
+# "Connection refused"/"Connection failure" is what the pulse client library
+# prints when nothing is listening on the socket at all — expected while the
+# op25 container's daemon is still coming up.
+_NO_DAEMON_MARKERS = ("connection refused", "connection failure")
+
+
+@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
+
+        # Last few lines of the most recent FFmpeg stderr, used to classify
+        # why a capture process exited (see _classify_capture_exit).
+        self._last_stderr_lines: deque[str] = deque(maxlen=10)
 
     # ------------------------------------------------------------------
     # 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 +192,367 @@ 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()
+                self._log_capture_exit()
             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}")
+        self._last_stderr_lines.clear()
+        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
+                text = line.decode(errors="replace").strip()
+                self._last_stderr_lines.append(text)
+                logger.warning(f"ffmpeg(pulse): {text}")
+        except asyncio.CancelledError:
+            return
+        except Exception:
+            return
+
+    def _log_capture_exit(self) -> None:
+        """
+        Log why the just-finished capture process exited, distinguishing the
+        two failure modes that matter operationally instead of one generic
+        "restarting" line for both:
+
+          - no daemon / connection refused: infrastructure isn't up yet. This
+            is expected during startup/op25 restarts, so it stays at INFO —
+            the retry loop above already handles it.
+          - daemon up but the named source is missing: almost always a real
+            PULSE_SOURCE misconfiguration. This gets a loud, distinct ERROR
+            naming the configured source, because silently retrying forever
+            against a wrong source name is exactly how this hid in the past.
+        """
+        text = " ".join(self._last_stderr_lines).lower()
+
+        if any(marker in text for marker in _SOURCE_MISSING_MARKERS):
+            logger.error(
+                f"PulseAudio capture exited: source '{settings.pulse_source}' does not exist on the "
+                "daemon (FFmpeg reported 'No such process'). This looks like a real PULSE_SOURCE "
+                "misconfiguration or a missing drb_sink — retrying will not fix it by itself. "
+                "Restarting anyway."
+            )
+            return
+
+        if any(marker in text for marker in _NO_DAEMON_MARKERS):
+            logger.info(
+                "PulseAudio capture exited: daemon not accepting connections — "
+                "infrastructure still coming up, restarting."
+            )
+            return
+
+        logger.warning("PulseAudio capture process exited — restarting.")
+
+    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 +565,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 +583,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 +613,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,149 @@
+"""
+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 in config.py for
+    why the default moved 1.0 → 3.0 (short calls' recording window was closing
+    before the ~1.5s grant→speech offset let voice audio even start).
+
+    All three close paths use this pad: idle_timeout, tgid_change, and
+    tgid_change_unlogged. An earlier version of this docstring claimed the
+    latter two close at "an exact, already-known boundary" (the new grant's
+    timestamp, or the same poll tick) and so intentionally added no pad — THAT
+    REASONING WAS WRONG and produced real truncated recordings. The boundary is
+    exact only in CONTROL-CHANNEL time; the buffered AUDIO lags control-channel
+    timestamps by ~1.5s (measured: 0.84-1.62s of lead trimmed across 7 field
+    calls), so slicing the outgoing call at the new grant's exact timestamp cut
+    roughly the last 1.5s of its real speech — calls ending mid-word with ~0s
+    trailing silence. Do not reintroduce a zero-pad close for tgid_change or
+    tgid_change_unlogged; if the outgoing and incoming recordings end up
+    overlapping in the underlying audio because of this pad, that is correct —
+    the audio genuinely contains both. See _handle_call_log and
+    _handle_channels for how each path sources the timestamp this gets added to.
+    """
+    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 +153,243 @@ 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 CONTROL-CHANNEL end for the outgoing segment, but the buffered
+        # audio lags control by ~1.5s, so slicing exactly there cut the outgoing
+        # call's last words. Pad past it and let the recorder's bounded tail wait
+        # block until that audio has actually been captured.
+        #
+        # The incoming call's pre-roll comes from the ring buffer, so the delay
+        # costs it nothing, and the two slices overlapping in the underlying
+        # audio is correct — the stream genuinely contains one call's tail and
+        # then the next call's start.
+        await self._close_segment(started_at + _tail_pad(), 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 + _tail_pad(), 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,141 @@
+"""
+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 (bounded) for that daemon to
+actually answer before starting its own app, and the edge-node needs the same
+guarantee before launching FFmpeg: FFmpeg with `-f pulse` fails instantly if
+nothing is listening, and used to stay dead for the lifetime of the process.
+This module is the wait.
+
+HISTORY / WHY THIS CHECKS LIVENESS, NOT FILE EXISTENCE: the `pulse_socket`
+named volume survives container recreation, but the daemon process that
+created the socket does not. Observed on live hardware: a stale
+`/run/pulse/native` socket file and `/run/pulse/pid` from a killed daemon
+were still in the volume after `docker compose up -d --build` recreated the
+containers. PulseAudio refused to start ("Daemon already running") because of
+the stale pid file, so nothing was actually listening on the socket — but the
+socket *file* still existed. An earlier version of this module (and of the
+op25 entrypoint) only checked `stat.S_ISSOCK` on the path, so it reported
+"ready" against a dead daemon, FFmpeg launched anyway, and immediately failed
+with "No such process" in a tight restart loop. Readiness here means "a
+PulseAudio connection actually succeeds," never "a file exists at this path."
+
+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 shutil
+import subprocess
+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
+
+# Bounded timeout for a single `pactl info` liveness probe. Kept short: this
+# runs synchronously on the calling thread (see is_ready()), and callers of
+# is_ready() include a sync code path inside the Discord voice bot, so a slow
+# probe would stall its event loop. wait_until_ready() runs probes off-thread
+# via asyncio.to_thread and can afford this bound comfortably within its own
+# much larger PULSE_WAIT_TIMEOUT.
+PROBE_TIMEOUT_SECONDS = 1.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 _probe_env(path: str) -> dict:
+    env = dict(os.environ)
+    env["PULSE_SERVER"] = f"unix:{path}"
+    return env
+
+
+def _daemon_responds() -> bool:
+    """
+    True only when a PulseAudio daemon actually answers on the configured
+    socket. Shells out to `pactl info` (from `pulseaudio-utils`, installed
+    alongside `libpulse0` in the edge-node image) rather than re-implementing
+    the native protocol handshake in Python — this container has no other use
+    for talking to PulseAudio directly, so a subprocess call is the smallest
+    correct implementation.
+
+    Deliberately does NOT check `os.path.exists`/`stat.S_ISSOCK` first: a
+    stale socket file from a killed daemon passes that check and always did,
+    which is the exact defect this function replaces.
+    """
+    path = socket_path()
+    pactl = shutil.which("pactl")
+    if pactl is None:
+        logger.error("pactl not found in PATH — cannot verify PulseAudio liveness.")
+        return False
+    try:
+        result = subprocess.run(
+            [pactl, "info"],
+            env=_probe_env(path),
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            timeout=PROBE_TIMEOUT_SECONDS,
+        )
+        return result.returncode == 0
+    except (subprocess.TimeoutExpired, OSError):
+        return False
+
+
+def is_ready() -> bool:
+    """
+    True when a PulseAudio daemon is alive and answering right now.
+
+    Synchronous and bounded by PROBE_TIMEOUT_SECONDS — used from a sync call
+    site (discord_radio._play_stream). Prefer `wait_until_ready()` from async
+    code so the probe doesn't block the event loop.
+    """
+    return _daemon_responds()
+
+
+async def wait_until_ready(timeout: Optional[float] = None) -> bool:
+    """
+    Block until a PulseAudio daemon actually answers, or `timeout` seconds
+    elapse.
+
+    Bounded on purpose — never hang the caller forever. Returns True once a
+    live connection succeeds, False on timeout (caller decides whether to
+    retry). Each probe runs via asyncio.to_thread so the subprocess call never
+    blocks the event loop.
+    """
+    limit = settings.pulse_wait_timeout if timeout is None else timeout
+    path = socket_path()
+
+    if await asyncio.to_thread(_daemon_responds):
+        return True
+
+    logger.info(f"Waiting up to {limit:.0f}s for a live PulseAudio daemon at {path}…")
+    waited = 0.0
+    while waited < limit:
+        await asyncio.sleep(POLL_INTERVAL)
+        waited += POLL_INTERVAL
+        if await asyncio.to_thread(_daemon_responds):
+            logger.info(f"PulseAudio daemon live after {waited:.1f}s.")
+            return True
+
+    logger.error(
+        f"PulseAudio daemon at {path} not responding after {limit:.0f}s — "
+        "is the op25 container's daemon actually up? 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,484 @@
+"""
+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
+
+
+# ---------------------------------------------------------------------------
+# Capture-exit classification — the two failure modes must be told apart
+# instead of both logging the same generic "restarting" line. This is what
+# let a wrong PULSE_SOURCE hide behind normal-looking startup retries before.
+# ---------------------------------------------------------------------------
+
+def _log_levels(caplog, logger_name="drb-edge-node"):
+    return [r.levelname for r in caplog.records if r.name == logger_name]
+
+
+def test_capture_exit_logs_error_when_source_missing(recorder, caplog):
+    """FFmpeg's pulse input prints 'No such process' when the daemon is up
+    but the configured source name does not exist — a real misconfiguration,
+    not a startup race, so this must stand out as an error naming the source."""
+    recorder._last_stderr_lines.append(
+        "[pulse @ 0x...] pa_stream_connect_record failed: No such process"
+    )
+    with caplog.at_level("INFO", logger="drb-edge-node"):
+        recorder._log_capture_exit()
+
+    assert "ERROR" in _log_levels(caplog)
+    error_messages = [r.message for r in caplog.records if r.levelname == "ERROR"]
+    assert any(settings.pulse_source in m for m in error_messages)
+
+
+def test_capture_exit_logs_info_when_no_daemon(recorder, caplog):
+    """Connection refused means nothing is listening yet — expected during
+    startup, so it must NOT be logged at the same severity as a real
+    misconfiguration."""
+    recorder._last_stderr_lines.append(
+        "[pulse @ 0x...] pa_context_connect() failed: Connection refused"
+    )
+    with caplog.at_level("INFO", logger="drb-edge-node"):
+        recorder._log_capture_exit()
+
+    levels = _log_levels(caplog)
+    assert "ERROR" not in levels
+    assert "INFO" in levels
+
+
+def test_capture_exit_falls_back_to_generic_warning(recorder, caplog):
+    """An FFmpeg failure that matches neither known marker keeps the original
+    generic behavior rather than guessing."""
+    recorder._last_stderr_lines.append("[pulse @ 0x...] some other unexpected failure")
+    with caplog.at_level("INFO", logger="drb-edge-node"):
+        recorder._log_capture_exit()
+
+    assert _log_levels(caplog) == ["WARNING"]
+
+
+def test_capture_exit_with_no_stderr_captured_is_generic_warning(recorder, caplog):
+    """No stderr at all (e.g. FFmpeg killed before printing anything) must not
+    crash the classifier and must fall back to the generic message."""
+    assert list(recorder._last_stderr_lines) == []
+    with caplog.at_level("INFO", logger="drb-edge-node"):
+        recorder._log_capture_exit()
+
+    assert _log_levels(caplog) == ["WARNING"]
+    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,577 @@
 """
-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_three_seconds(watcher, clock, monkeypatch):
+    """
+    Field measurement showed the grant->speech offset runs ~0.84-1.62s, so a
+    1.0s pad let short calls' windows close before voice audio even started
+    (clipping mid-word). The default moved to 3.0 — and it has to be a
+    setting, not a magic number, so it can be tuned per node without a code
+    change (and so tests can prove it isn't hardcoded anywhere downstream).
+    """
+    assert settings.call_tail_pad_seconds == 3.0
+
+    monkeypatch.setattr(settings, "call_tail_pad_seconds", 5.0)
+
+    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)]))
+
+    # Advance well past both the idle timeout AND the monkeypatched 5.0s pad so
+    # the "now" cap in _handle_channels never masks the pad value under test.
+    clock.advance(settings.call_idle_timeout + 6.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 + 5.0)
+
+
+@pytest.mark.asyncio
+async def test_short_call_window_now_covers_delayed_voice_arrival(watcher, clock):
+    """
+    Regression test for the truncation bug: a ~0.97s control-channel call
+    (grant to srcaddr-drop) previously closed its window at
+    last_tx_end + 1.0s pad, i.e. ~1.97s after the grant — but field
+    measurement shows voice audio doesn't start until ~1.5s after the grant
+    (0.84-1.62s measured), so the old window left as little as ~0.4s of
+    captured speech and clipped it mid-word.
+
+    With the 3.0s default pad, the same short call's window must extend well
+    past the ~1.5s point where voice actually starts.
+    """
+    call_start = clock.now
+    await tick(watcher, update(
+        call_log=[grant(1234, call_start)],
+        channels=[channel(tgid=1234, srcaddr=555)],
+    ))
+
+    # The control-channel call itself is short — under 1 second.
+    clock.advance(0.97)
+    edge_time = clock.now
+    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()]))
+
+    payload = watcher.on_call_end.call_args[0][0]
+    assert payload["end_reason"] == "idle_timeout"
+
+    voice_arrival = call_start + 1.5  # measured grant->speech offset, typical case
+    assert payload["ended_at_epoch"] == pytest.approx(edge_time + settings.call_tail_pad_seconds)
+    assert payload["ended_at_epoch"] > voice_arrival, (
+        "recording window must extend past the point voice audio actually arrives, "
+        "not just past the control-channel call_log timestamps"
+    )
+
+
+@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"]

drb-edge-node/tests/test_pulse.py

@@ -0,0 +1,135 @@
+"""
+Unit tests for PulseAudio readiness helpers (app.internal.pulse).
+
+The whole point of this module is that readiness means "a PulseAudio
+connection actually succeeds," never "a socket file exists at this path" —
+that was the exact bug reproduced on live hardware: a killed daemon left its
+pid file and native socket behind in the shared `pulse_socket` volume, the
+old file-existence check reported "ready", and FFmpeg launched against a
+dead daemon.
+
+No real PulseAudio daemon or `pactl` binary is required for these tests:
+`_daemon_responds` (the one function that actually shells out) is monkeypatched
+everywhere except the dedicated subprocess-layer tests, which fake out
+`shutil.which`/`subprocess.run` directly so the "stale file, dead daemon" case
+is proven at the layer that matters.
+"""
+import subprocess
+from unittest.mock import Mock
+
+from app.internal import pulse
+
+
+# ---------------------------------------------------------------------------
+# socket_path()
+# ---------------------------------------------------------------------------
+
+def test_socket_path_defaults_when_pulse_server_unset(monkeypatch):
+    monkeypatch.delenv("PULSE_SERVER", raising=False)
+    assert pulse.socket_path() == pulse.DEFAULT_SOCKET_PATH
+
+
+def test_socket_path_parses_unix_prefixed_pulse_server(monkeypatch):
+    monkeypatch.setenv("PULSE_SERVER", "unix:/tmp/somewhere/native")
+    assert pulse.socket_path() == "/tmp/somewhere/native"
+
+
+def test_socket_path_falls_back_on_malformed_pulse_server(monkeypatch):
+    monkeypatch.setenv("PULSE_SERVER", "not-a-unix-uri")
+    assert pulse.socket_path() == pulse.DEFAULT_SOCKET_PATH
+
+
+# ---------------------------------------------------------------------------
+# is_ready() / wait_until_ready() against a monkeypatched probe
+# ---------------------------------------------------------------------------
+
+def test_is_ready_true_when_daemon_responds(monkeypatch):
+    monkeypatch.setattr(pulse, "_daemon_responds", lambda: True)
+    assert pulse.is_ready() is True
+
+
+def test_is_ready_false_when_daemon_does_not_respond(monkeypatch):
+    monkeypatch.setattr(pulse, "_daemon_responds", lambda: False)
+    assert pulse.is_ready() is False
+
+
+async def test_wait_until_ready_short_circuits_when_already_live(monkeypatch):
+    calls = Mock(return_value=True)
+    monkeypatch.setattr(pulse, "_daemon_responds", calls)
+    assert await pulse.wait_until_ready(timeout=5) is True
+    assert calls.call_count == 1
+
+
+async def test_wait_until_ready_polls_until_daemon_comes_up(monkeypatch):
+    monkeypatch.setattr(pulse, "POLL_INTERVAL", 0.01)
+    responses = iter([False, False, True])
+    monkeypatch.setattr(pulse, "_daemon_responds", lambda: next(responses))
+    assert await pulse.wait_until_ready(timeout=5) is True
+
+
+async def test_wait_until_ready_times_out_when_daemon_never_responds(monkeypatch):
+    monkeypatch.setattr(pulse, "POLL_INTERVAL", 0.01)
+    monkeypatch.setattr(pulse, "_daemon_responds", lambda: False)
+    assert await pulse.wait_until_ready(timeout=0.05) is False
+
+
+async def test_wait_until_ready_uses_settings_default_when_timeout_omitted(monkeypatch):
+    monkeypatch.setattr(pulse.settings, "pulse_wait_timeout", 0.05)
+    monkeypatch.setattr(pulse, "POLL_INTERVAL", 0.01)
+    monkeypatch.setattr(pulse, "_daemon_responds", lambda: False)
+    assert await pulse.wait_until_ready() is False
+
+
+# ---------------------------------------------------------------------------
+# _daemon_responds() at the subprocess layer — proves a stale FILE is not
+# enough, which is the actual regression this module fixes.
+# ---------------------------------------------------------------------------
+
+def test_daemon_responds_false_when_pactl_missing(monkeypatch):
+    monkeypatch.setattr(pulse.shutil, "which", lambda name: None)
+    assert pulse._daemon_responds() is False
+
+
+def test_daemon_responds_false_on_probe_timeout(monkeypatch, tmp_path):
+    monkeypatch.setattr(pulse.shutil, "which", lambda name: "/usr/bin/pactl")
+
+    def fake_run(*args, **kwargs):
+        raise subprocess.TimeoutExpired(cmd="pactl", timeout=pulse.PROBE_TIMEOUT_SECONDS)
+
+    monkeypatch.setattr(pulse.subprocess, "run", fake_run)
+    assert pulse._daemon_responds() is False
+
+
+def test_daemon_responds_false_when_stale_socket_file_exists_but_daemon_dead(monkeypatch, tmp_path):
+    """
+    The regression, reproduced at the layer that matters: a plain FILE sits
+    at the socket path (exactly what a killed daemon leaves behind), but
+    `pactl info` against it fails (nonzero exit — connection refused). This
+    must NOT be treated as ready.
+    """
+    stale_socket = tmp_path / "native"
+    stale_socket.write_bytes(b"")  # a stale file, not a live socket
+    monkeypatch.setenv("PULSE_SERVER", f"unix:{stale_socket}")
+
+    monkeypatch.setattr(pulse.shutil, "which", lambda name: "/usr/bin/pactl")
+    monkeypatch.setattr(
+        pulse.subprocess, "run",
+        lambda *a, **k: subprocess.CompletedProcess(args=a, returncode=1),
+    )
+
+    assert stale_socket.exists()  # sanity: the old file-existence check would pass
+    assert pulse._daemon_responds() is False
+
+
+def test_daemon_responds_true_when_pactl_succeeds(monkeypatch, tmp_path):
+    live_socket = tmp_path / "native"
+    live_socket.write_bytes(b"")
+    monkeypatch.setenv("PULSE_SERVER", f"unix:{live_socket}")
+
+    monkeypatch.setattr(pulse.shutil, "which", lambda name: "/usr/bin/pactl")
+    monkeypatch.setattr(
+        pulse.subprocess, "run",
+        lambda *a, **k: subprocess.CompletedProcess(args=a, returncode=0),
+    )
+
+    assert pulse._daemon_responds() is True

op25-container/docker-entrypoint.sh

@@ -1,29 +1,62 @@
 #!/bin/bash
 
+PULSE_SOCKET=/run/pulse/native
+PULSE_PIDFILE=/run/pulse/pid
+
+mkdir -p /run/pulse
+chmod 777 /run/pulse
+
+# Returns 0 (true) only when a PulseAudio daemon actually answers on
+# $PULSE_SOCKET. A socket/pid FILE existing proves nothing by itself — that
+# is exactly the bug this script works around (see stale-state check below).
+pulse_daemon_alive() {
+    PULSE_SERVER="unix:${PULSE_SOCKET}" timeout 2 pactl info >/dev/null 2>&1
+}
+
+# --- Clear stale PulseAudio state left behind by a killed daemon ---
+# The `pulse_socket` named volume survives container recreation, but the
+# PulseAudio process that owned it does not. If the previous container was
+# recreated (not gracefully stopped), its pid file and native socket are
+# still sitting in the volume; pulseaudio's pid.c sees the pid file and
+# refuses to start ("Daemon already running") even though nothing is
+# listening. Only remove these when nothing actually answers on the socket —
+# never delete a socket a live daemon is using.
+if [ -S "$PULSE_SOCKET" ] || [ -f "$PULSE_PIDFILE" ]; then
+    if pulse_daemon_alive; then
+        echo "PulseAudio daemon already alive and responding at ${PULSE_SOCKET} — leaving state as-is."
+    else
+        echo "STALE STATE: found ${PULSE_PIDFILE} / ${PULSE_SOCKET} from a previous container, but no daemon answers — clearing before start."
+        rm -f "$PULSE_SOCKET" "$PULSE_PIDFILE"
+    fi
+fi
+
 # --- Start PulseAudio Daemon ---
 # -n: skip default config (load modules inline — avoids system.pa parsing issues)
 # --system: run as system-wide daemon
 # --log-target=stderr: makes errors visible in Docker logs
 # &: background so this script continues; output still captured by Docker
 echo "Starting PulseAudio daemon..."
-mkdir -p /run/pulse
-chmod 777 /run/pulse
 pulseaudio --exit-idle-time=-1 -n --system \
-    --load="module-native-protocol-unix socket=/run/pulse/native auth-anonymous=1" \
+    --load="module-native-protocol-unix socket=${PULSE_SOCKET} auth-anonymous=1" \
     --load="module-null-sink sink_name=drb_sink sink_properties=device.description=DRB-Sink" \
     --log-target=stderr &
 
-# Wait for the socket to actually exist before continuing
+# Wait for the daemon to actually answer — NOT just for the socket file to
-echo "Waiting for PulseAudio socket..."
+# exist. A stale socket file from a killed daemon exists but nothing is
+# listening on it; a file-existence check reports "ready" against a dead
+# daemon, which is exactly how this class of bug slipped through before.
+echo "Waiting for PulseAudio to become live..."
+PULSE_LIVE=0
 for i in $(seq 1 20); do
-    if [ -S /run/pulse/native ]; then
+    if pulse_daemon_alive; then
-        echo "PulseAudio socket ready."
+        echo "PulseAudio daemon is live (pactl info succeeded)."
+        PULSE_LIVE=1
         break
     fi
     sleep 0.5
 done
-if [ ! -S /run/pulse/native ]; then
+if [ "$PULSE_LIVE" -ne 1 ]; then
-    echo "WARNING: PulseAudio socket not found after 10s — edge-node audio will fail."
+    echo "WARNING: PulseAudio daemon not responding after 10s — edge-node audio will fail until it recovers."
 fi
 ls -la /run/pulse/