trunk-recorder: mark drb-upload.sh and run.sh executable (node-26#45)

TR execvp()s the uploadScript; without +x every call failed with status 127
on node-002 and nothing was uploaded.

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

.env.example

@@ -99,6 +99,32 @@ TRIM_SILENCE_THRESHOLD_DB=-40
 # Seconds of audio kept either side of detected speech.
 TRIM_SILENCE_GUARD_SECONDS=0.25
 
+# Upload spool: a recording whose upload to C2 fails is kept on disk and
+# retried, never deleted (default dir: <recordings>/spool). Bounded so a long
+# C2 outage cannot fill the host disk; the oldest recordings go first.
+# UPLOAD_SPOOL_DIR=/recordings/spool
+UPLOAD_SPOOL_MAX_BYTES=536870912
+UPLOAD_SPOOL_MAX_AGE_HOURS=72
+UPLOAD_RETRY_MIN_SECONDS=15
+UPLOAD_RETRY_MAX_SECONDS=600
+
+# --- Recorder backend (node-26#45) -------------------------------------------
+# op25 (default): OP25 + PulseAudio recorder, one call at a time.
+# trunk-recorder: Trunk Recorder records every talkgroup at once, with call
+# boundaries from the decoder. Set BOTH lines below to switch; OP25 is then
+# never started (it would hold the SDR), and the live browser stream (Icecast)
+# goes quiet — Discord live audio keeps working.
+RECORDER_BACKEND=op25
+# COMPOSE_PROFILES=trunk-recorder
+# Trunk Recorder SDR settings (only read with RECORDER_BACKEND=trunk-recorder).
+# TR_SAMPLE_RATE=2400000
+# TR_GAIN=34
+# TR_CENTER_HZ=            # override the computed centre frequency
+# TR_DIGITAL_RECORDERS=4   # simultaneous voice channels on a trunked system
+# TR_MIN_DURATION=1.0      # seconds; shorter calls are dropped by TR
+# TR_SQUELCH_DB=-50        # conventional channels; absolute dB, tune per site
+# TR_LIVE_PORT=9123        # local UDP port for simplestream -> Discord
+
 # OP25 container (usually no need to change)
 OP25_API_URL=http://localhost:8001
 OP25_TERMINAL_URL=http://localhost:8081

docker-compose.yml

@@ -51,6 +51,35 @@ services:
     volumes:
       - /dev:/dev
 
+  # Trunk Recorder (node-26#45) — the alternative recorder. Opt-in per node:
+  # it only starts when the node's .env has COMPOSE_PROFILES=trunk-recorder,
+  # and the edge-node only uses it with RECORDER_BACKEND=trunk-recorder. Set
+  # both or neither. Upstream image, pinned by tag AND digest (the digest is
+  # the multi-arch index for 5.2.1 on Docker Hub, amd64/arm64/arm/v7); bump
+  # both together. GPL-3.0: we ship it unmodified as a separate program —
+  # see trunk-recorder/README.md for the source offer.
+  #
+  # The edge-node writes ./configs/trunk-recorder/config.json; run.sh (our
+  # supervisor, replacing the image's CMD) restarts TR when it changes. It
+  # shares OP25's dongle slot, so op25 is never started on such a node.
+  trunk-recorder:
+    image: robotastic/trunk-recorder:5.2.1@sha256:c3e425858bfca16c609d8612808a83f06b0f376d950d11733d7c25b2a28f8866
+    profiles: ["trunk-recorder"]
+    restart: unless-stopped
+    network_mode: host
+    devices:
+      - /dev/bus/usb:/dev/bus/usb
+    # TR stages each transmission in tempDir (/dev/shm); Docker's 64 MB
+    # default is tight with several recorders running at once.
+    shm_size: 256m
+    entrypoint: ["/bin/sh", "/drb/run.sh"]
+    environment:
+      TZ: ${TZ:-UTC}
+    volumes:
+      - ./configs/trunk-recorder:/app
+      - ./recordings:/recordings
+      - ./trunk-recorder:/drb:ro
+
   edge-node:
     image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/edge-node:${DRB_TAG_EDGE_NODE:-latest}
     build: ./drb-edge-node

drb-edge-node/app/config.py

@@ -1,5 +1,5 @@
 from pydantic_settings import BaseSettings
-from typing import Optional
+from typing import Literal, Optional
 
 
 class Settings(BaseSettings):
@@ -153,6 +153,63 @@ class Settings(BaseSettings):
     # Offline call buffer — how many call_end events to keep while disconnected
     offline_call_buffer_size: int = 35
 
+    # Upload spool (node-26#44) — recordings whose upload to C2 failed are kept
+    # here and retried instead of deleted. Defaults to <recordings_path>/spool,
+    # i.e. on the host disk via the /recordings bind mount, so it survives a
+    # restart. Bounded both ways because a full host disk stops recording
+    # outright (node-26#41): FLAC is ~1.3 MB per minute of audio, so 512 MiB is
+    # roughly 6 h of continuous traffic. Oldest entries are evicted first.
+    upload_spool_dir: Optional[str] = None
+    upload_spool_max_bytes: int = 512 * 1024 * 1024
+    upload_spool_max_age_hours: float = 72.0
+    # Exponential backoff between retries of one spooled recording.
+    upload_retry_min_seconds: float = 15.0
+    upload_retry_max_seconds: float = 600.0
+    # Timeout for the FIRST, inline upload attempt. metadata_watcher awaits the
+    # call close, so a black-holed C2 used to stall segmentation for the full
+    # 120 s; a failure here just lands in the spool.
+    upload_inline_timeout_seconds: float = 30.0
+
+    # ------------------------------------------------------------------
+    # Recorder backend (node-26#45)
+    #
+    #   op25            OP25 decodes, Liquidsoap mixes into PulseAudio, and
+    #                   call_recorder/metadata_watcher cut calls out of that
+    #                   one stream. The default; every existing node.
+    #   trunk-recorder  Trunk Recorder owns the SDR and records every grant at
+    #                   once, with call boundaries from the decoder. Its
+    #                   uploadScript drops each finished call in tr_inbox_dir
+    #                   and app/internal/trunk_recorder.py uploads it through
+    #                   the same spool. OP25 is never started (it would fight
+    #                   TR for the dongle) and the PulseAudio recorder stays
+    #                   off, so a call is never uploaded twice. Needs the
+    #                   compose profile too: COMPOSE_PROFILES=trunk-recorder.
+    # ------------------------------------------------------------------
+    recorder_backend: Literal["op25", "trunk-recorder"] = "op25"
+    # Shared with the trunk-recorder container (mounted there as /app):
+    # config.json, the CSVs, its logs, and the supervisor's status file.
+    tr_config_dir: Optional[str] = None  # default <config_path>/trunk-recorder
+    # Where TR's uploadScript hands over finished calls (default
+    # <recordings_path>/tr-inbox). Same bind mount in both containers.
+    tr_inbox_dir: Optional[str] = None
+    # Source (SDR) settings for the generated config. 2.4 Msps is the most an
+    # RTL-SDR does without dropping samples; ~2 MHz of it is usable.
+    tr_sample_rate: int = 2_400_000
+    tr_gain: float = 34.0  # dB; matches op25-container's rtl-sdr-v3 preset (lna:34)
+    tr_center_hz: Optional[int] = None  # override the computed centre frequency
+    # Simultaneous voice channels on a trunked system. Each costs CPU (on a
+    # Pi 4 roughly 4-6 is the ceiling); conventional channels get one each.
+    tr_digital_recorders: int = 4
+    # TR deletes calls shorter than this. Matches the key-up noise the old
+    # recorder's trim + all-silence check threw away.
+    tr_min_duration: float = 1.0
+    # Conventional squelch, absolute dB (TR has no noise squelch). Used when
+    # the system config gives no squelch_threshold. Needs hand tuning per site.
+    tr_squelch_db: float = -50.0
+    # simplestream UDP port for live audio (TR -> edge-node -> PulseAudio ->
+    # Discord). Local only.
+    tr_live_port: int = 9123
+
     # ------------------------------------------------------------------
     # Local dashboard / API authentication
     #

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

@@ -64,12 +64,12 @@ from collections import deque
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple
 
-import httpx
 from app.config import settings
-from app.internal import audio_trim, credentials, pcm, pulse
+from app.internal import audio_trim, pcm, pulse, upload_spool
 from app.internal.logger import logger
+from app.internal.upload_spool import UploadMeta, UploadResult, UploadSpool
 
 # Safety cap on a single recording; mirrors MAX_SEGMENT_SECONDS in metadata_watcher.
 MAX_RECORDING_SECONDS = 600
@@ -320,16 +320,35 @@ class CallRecorder:
         # why a capture process exited (see _classify_capture_exit).
         self._last_stderr_lines: deque[str] = deque(maxlen=10)
 
+        # Failed uploads wait here for a retry instead of being deleted
+        # (node-26#44). See internal/upload_spool.py.
+        self.spool = UploadSpool(
+            Path(settings.upload_spool_dir or (Path(settings.recordings_path) / "spool")),
+            uploader=self._post,
+            max_bytes=settings.upload_spool_max_bytes,
+            max_age_seconds=settings.upload_spool_max_age_hours * 3600,
+            backoff_min=settings.upload_retry_min_seconds,
+            backoff_max=settings.upload_retry_max_seconds,
+        )
+
     # ------------------------------------------------------------------
     # Lifecycle
     # ------------------------------------------------------------------
 
-    async def start(self) -> None:
+    async def start(self, capture: bool = True) -> None:
-        """Start the persistent capture. Call once from app lifespan."""
+        """Start the persistent capture. Call once from app lifespan.
+
+        capture=False starts only the upload spool: with the trunk-recorder
+        backend (node-26#45) TR does the recording, and a second recorder on
+        the same audio would upload every call twice."""
+        if capture:
             self._stream_task = asyncio.create_task(self._capture_loop())
             logger.info("PulseAudio ring-buffer starting.")
+        if settings.c2_url:
+            self.spool.start()
 
     async def stop(self) -> None:
+        await self.spool.stop()
         if self._stream_task:
             self._stream_task.cancel()
             try:
@@ -751,6 +770,18 @@ class CallRecorder:
     # Upload (unchanged interface)
     # ------------------------------------------------------------------
 
+    async def _post(self, file_path: Path, meta: UploadMeta) -> UploadResult:
+        return await upload_spool.post_recording(file_path, meta, upload_spool.mime_for(file_path, AUDIO_MIME))
+
+    async def _post_inline(self, file_path: Path, meta: UploadMeta) -> UploadResult:
+        # metadata_watcher awaits on_call_end, so this attempt blocks the poll
+        # loop. With the spool behind it, it can afford a short timeout; the
+        # background retries get the full UPLOAD_TIMEOUT_SECONDS.
+        return await upload_spool.post_recording(
+            file_path, meta, upload_spool.mime_for(file_path, AUDIO_MIME),
+            timeout=settings.upload_inline_timeout_seconds,
+        )
+
     async def upload_recording(
         self,
         file_path: Path,
@@ -760,51 +791,40 @@ class CallRecorder:
         system_id: Optional[str] = None,
         audio_start_epoch: Optional[float] = None,
         audio_end_epoch: Optional[float] = None,
+        extra: Optional[Dict[str, str]] = None,
     ) -> Optional[str]:
+        """
+        Upload once, inline. Returns the storage URL on success.
+
+        The file is deleted ONLY after a 2xx with a URL (node-26#44). A
+        retryable failure moves it into the spool, which keeps retrying in the
+        background; C2's /upload writes `audio_gcs_uri` onto the call doc by
+        call_id whenever the upload lands, so a late upload still attaches.
+        """
         if not settings.c2_url:
             logger.info("No C2_URL configured — skipping upload.")
+            file_path.unlink(missing_ok=True)
             return None
 
-        upload_url = f"{settings.c2_url}/upload"
+        meta = UploadMeta(
-        api_key = credentials.get_api_key()
+            call_id=call_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name,
-        headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
+            system_id=system_id, audio_start_epoch=audio_start_epoch, audio_end_epoch=audio_end_epoch,
-
+            extra=extra,
-        form: dict = {"call_id": call_id, "node_id": settings.node_id}
-        if talkgroup_id is not None:
-            form["talkgroup_id"] = str(talkgroup_id)
-        if talkgroup_name:
-            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:
-                with open(file_path, "rb") as f:
-                    r = await client.post(
-                        upload_url,
-                        files={"file": (file_path.name, f, AUDIO_MIME)},
-                        data=form,
-                        headers=headers,
         )
-                    r.raise_for_status()
+        result = await self._post_inline(file_path, meta)
-                    audio_url = r.json().get("url")
+        self.spool.last_result = {"call_id": call_id, "kind": result.kind, "status": result.status,
-                    logger.info(f"Upload complete: {audio_url}")
+                                  "detail": result.detail, "epoch": time.time()}
-                    return audio_url
+        if result.kind == upload_spool.OK:
-        except Exception as e:
+            logger.info(f"Upload complete: {result.url}")
-            logger.error(f"Upload failed: {e}")
+            file_path.unlink(missing_ok=True)
+            return result.url
+        if result.kind == upload_spool.REJECT:
+            logger.error(f"C2 rejected the upload of call {call_id} ({result.detail}) — not retryable, deleting it.")
+            file_path.unlink(missing_ok=True)
+            return None
+        if self.spool.add(file_path, meta, result) is None:
+            self.spool.last_result = {**self.spool.last_result, "kind": "lost"}
         return None
-        finally:
-            try:
-                file_path.unlink()
-            except Exception:
-                pass
 
     # ------------------------------------------------------------------
     # State

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

@@ -183,6 +183,10 @@ class RadioManager:
         self._mqtt_down_since: Optional[float] = None
         self._failsafe = False
         self._paused = False
+        # The applied seq at the moment of a local pause (node-26#34). The
+        # broker redelivers the retained desired message on every reconnect
+        # with the SAME seq; only a strictly newer seq from C2 lifts the pause.
+        self._paused_at_seq: Optional[int] = None
         self._ticks = 0
 
     # ------------------------------------------------------------------
@@ -210,15 +214,25 @@ class RadioManager:
                 logger.error(f"Malformed Discord desired stream entry ignored: {s!r}")
         self._ensure_watchdog()
         self._desired = wanted
+        if self._paused and self._redelivery_of_paused_state(seq):
+            logger.info(f"Discord desired seq {seq} is a redelivery — staying paused locally.")
+        else:
             self._paused = False
+            self._paused_at_seq = None
         await self.reconcile()
         if seq is not None:
             self._applied_seq = seq
         await self.publish_actual()
 
+    def _redelivery_of_paused_state(self, seq: Optional[int]) -> bool:
+        """True when `seq` is not newer than what was applied at pause time."""
+        return seq is not None and self._paused_at_seq is not None and seq <= self._paused_at_seq
+
     async def pause_locally(self) -> None:
-        """Local-dashboard stop: log every bot out until C2 sends a new list."""
+        """Local-dashboard stop: log every bot out until C2 sends a NEW list
+        (a seq strictly greater than the one applied now — node-26#34)."""
         self._paused = True
+        self._paused_at_seq = self._applied_seq
         await self.reconcile()
         await self.publish_actual()
 

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

@@ -8,6 +8,10 @@ answers one of those for one link in the chain
     password -> C2 enrollment -> MQTT -> SDR -> what to listen to -> OP25
              -> decoding -> audio (PulseAudio) -> local stream (Icecast)
 
+With RECORDER_BACKEND=trunk-recorder (node-26#45) the last four are replaced
+by Trunk Recorder's: supervisor state, control-channel decode rate (from its
+log), recordings handed over, and the live feed to Discord.
+
 `gather_facts()` does the I/O; `build_checks()` is a pure function of those
 facts so every state (unapproved, no SDR, OP25 crashed, …) is unit-testable.
 
@@ -154,6 +158,22 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]:
             f"Not connected to {broker} — expected until the node is approved (the broker only admits approved nodes).",
             "Fixes itself once the node is approved and has its API key (see above).",
         ))
+    elif (f.get("mqtt_last_error") or {}).get("kind") == "refused":
+        err = f["mqtt_last_error"]
+        checks.append(_check(
+            "mqtt", "Live link (MQTT)", ERROR,
+            f"{broker} refused this node's login ({err.get('detail')}).",
+            "The API key on file is not the one the broker expects — every re-approval issues a new one. "
+            "Re-run install.sh to fetch the current key, or ask a DRB admin to re-approve the node.",
+        ))
+    elif (f.get("mqtt_last_error") or {}).get("kind") == "unreachable":
+        err = f["mqtt_last_error"]
+        checks.append(_check(
+            "mqtt", "Live link (MQTT)", ERROR,
+            f"Can't reach {broker} ({err.get('detail')}).",
+            f"Check this site's internet connection, DNS, and that outbound port {f['mqtt_port']} isn't "
+            "blocked by a firewall. The key is not the problem.",
+        ))
     else:
         checks.append(_check(
             "mqtt", "Live link (MQTT)", ERROR,
@@ -192,6 +212,10 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]:
         kind = "P25 conventional" if src.get("mode") == "conventional" else src["type"]
         checks.append(_check("source", "What to listen to", OK, f"{src['name']} ({kind}) — {origin}."))
 
+    if f.get("backend") == "trunk-recorder":
+        checks.extend(_trunk_recorder_checks(f, src, now))
+        return checks
+
     # 6. OP25
     op25 = f.get("op25_status")
     running = op25 == "running"
@@ -286,6 +310,104 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]:
     return checks
 
 
+def _trunk_recorder_checks(f: Dict[str, Any], src: Optional[Dict[str, Any]], now: float) -> List[Dict[str, str]]:
+    """Checks 6-9 when Trunk Recorder is the recorder (node-26#45)."""
+    tr = f.get("tr") or {}
+    checks: List[Dict[str, str]] = []
+    status = tr.get("status")
+    state = (status or {}).get("state")
+    running = False
+    label = "Radio recorder (Trunk Recorder)"
+    if tr.get("config_error"):
+        checks.append(_check(
+            "recorder", label, ERROR, f"Couldn't build its config — {tr['config_error']}",
+            "Fix the system's channel list (or TR_* settings in .env); the node retries on the next config push.",
+        ))
+    elif status is None:
+        checks.append(_check(
+            "recorder", label, ERROR, "The trunk-recorder service isn't running.",
+            "Set COMPOSE_PROFILES=trunk-recorder in the node's .env, then run `docker compose up -d`.",
+        ))
+    elif status.get("stale"):
+        checks.append(_check(
+            "recorder", label, ERROR, f"Its supervisor stopped reporting {_ago(status.get('age_seconds') or 0)}.",
+            "Check the trunk-recorder container: `docker compose logs trunk-recorder`.",
+        ))
+    elif state == "running":
+        running = True
+        up = now - float(status.get("started_at") or now)
+        restarts = int(status.get("restarts") or 0)
+        checks.append(_check(
+            "recorder", label, OK if restarts < 3 else WARN,
+            f"Running for {_ago(up).replace(' ago', '')}"
+            + (f" — restarted {restarts} times since the container started." if restarts else "."),
+            "" if restarts < 3 else "Frequent restarts: `docker compose logs trunk-recorder` says why.",
+        ))
+    elif state == "crashed":
+        checks.append(_check(
+            "recorder", label, ERROR, f"Exited (code {status.get('last_exit')}) — restarting.",
+            "Usually the SDR is unplugged or held by another program. `docker compose logs trunk-recorder`.",
+        ))
+    elif state == "stopped":
+        checks.append(_check("recorder", label, WARN, "Stopped from the dashboard.", "Press Start."))
+    elif not src:
+        checks.append(_check("recorder", label, OFF, "Waiting — choose what to listen to first."))
+    else:
+        checks.append(_check("recorder", label, WARN, "Waiting for its config…"))
+
+    mode = (src or {}).get("mode")
+    if not running:
+        checks.append(_check("decode", "Decoding", OFF, "Starts when Trunk Recorder runs."))
+    elif mode == "trunked":
+        rate = tr.get("decode_rate")
+        age = tr.get("log_age_seconds")
+        if rate is None or age is None or age > 30:
+            checks.append(_check(
+                "decode", "Decoding", WARN, "No control-channel report from Trunk Recorder yet.",
+                "Give it 30 seconds after a start; if it stays, `docker compose logs trunk-recorder`.",
+            ))
+        elif rate > 0:
+            checks.append(_check("decode", "Decoding", OK, f"Decoding the control channel ({rate:g} messages/sec)."))
+        else:
+            checks.append(_check(
+                "decode", "Decoding", ERROR, "Not decoding the control channel (0 messages/sec).",
+                "Usually a wrong control-channel frequency, a weak signal/antenna, or SDR frequency error (PPM).",
+            ))
+    else:
+        checks.append(_check(
+            "decode", "Decoding", OK, "Listening (conventional) — channels record when their squelch opens.",
+            "If calls never arrive through a busy period, the squelch (TR_SQUELCH_DB or the system's "
+            "squelch_threshold) may be set too high.",
+        ))
+
+    last = tr.get("last_call_epoch")
+    backlog = int(tr.get("inbox_backlog") or 0)
+    if not running and not last:
+        checks.append(_check("audio", "Recordings", OFF, "Nothing recorded yet."))
+    elif backlog > 20:
+        checks.append(_check(
+            "audio", "Recordings", WARN, f"{backlog} finished calls waiting to be sent.",
+            "Normal for a minute after the server link comes back; if it keeps growing, check the C2 link.",
+        ))
+    elif last:
+        extra = f", {tr['encrypted_skipped']} encrypted skipped" if tr.get("encrypted_skipped") else ""
+        checks.append(_check(
+            "audio", "Recordings", OK,
+            f"Last call {_ago(now - last)} ({tr.get('calls_ingested', 0)} since start{extra}).",
+        ))
+    else:
+        checks.append(_check("audio", "Recordings", OK, "Recorder up — no call yet."))
+
+    live = tr.get("live") or {}
+    heard = live.get("last_packet_epoch")
+    checks.append(_check(
+        "stream", "Live audio", OK if heard else OFF,
+        ("Feeding Discord — last audio " + _ago(now - heard) + "." if heard else "Feeds Discord once a call is heard.")
+        + " The local browser stream isn't available with Trunk Recorder.",
+    ))
+    return checks
+
+
 async def icecast_live() -> Optional[bool]:
     """True if the configured mount has a source, False if not, None if Icecast is down."""
     url = f"http://{settings.icecast_host}:{settings.icecast_port}/status-json.xsl"
@@ -326,13 +448,17 @@ async def gather_facts() -> Dict[str, Any]:
             sdr_count = op25_devs["count"]
 
     activity = call_recorder.audio_activity()
+    from app.internal import trunk_recorder as tr
     return {
         "now": time.time(),
+        "backend": settings.recorder_backend,
+        "tr": tr.trunk_recorder.health_facts() if tr.enabled() else None,
         "node_id": settings.node_id,
         "default_password": auth.is_using_default_password(),
         "c2_url": settings.c2_url,
         "has_api_key": bool(credentials.get_api_key()),
         "mqtt_connected": mqtt_manager.is_connected,
+        "mqtt_last_error": mqtt_manager.last_error,
         "mqtt_broker": settings.mqtt_broker,
         "mqtt_port": settings.mqtt_port,
         "sdr_count": sdr_count,

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

@@ -1,5 +1,7 @@
 import asyncio
 import json
+import shutil
+import time
 from datetime import datetime, timezone
 from collections import deque
 from typing import Optional, Callable, Awaitable, Dict, Any
@@ -8,6 +10,22 @@ from app.config import settings
 from app.internal.logger import logger
 from app.internal import credentials, key_pickup
 
+
+def disk_report(path: str) -> Dict[str, Optional[int]]:
+    """Free/total bytes of the filesystem holding `path` (node-26#41).
+
+    /recordings is a bind mount of the host checkout, so this is the host disk
+    the recorder writes to and Docker's images/build cache live on. A full disk
+    stopped node-002 recording for 17 days while it still showed Online. Nulls
+    on error, so C2 overwrites a stale reading instead of keeping it.
+    """
+    try:
+        usage = shutil.disk_usage(path)
+    except OSError:
+        return {"disk_free_bytes": None, "disk_total_bytes": None}
+    return {"disk_free_bytes": usage.free, "disk_total_bytes": usage.total}
+
+
 CommandCallback = Callable[[Dict[str, Any]], Awaitable[None]]
 ConfigCallback = Callable[[Dict[str, Any]], Awaitable[None]]
 ApiKeyCallback = Callable[[Dict[str, Any]], Awaitable[None]]
@@ -43,6 +61,12 @@ class MQTTManager:
         # install.sh writing the file). Recovery compares against this.
         self._client_key: Optional[str] = None
         self._recovery_task: Optional[asyncio.Task] = None
+        # Why the link is down, for the dashboard (node-26#21): "refused" (the
+        # broker answered and said no), "unreachable" (TCP/TLS never got that
+        # far) or "disconnected" (an established session dropped). Cleared on
+        # every successful connect. Set from paho's thread as well as the event
+        # loop — a single attribute rebind, so no lock.
+        self._last_error: Optional[Dict[str, Any]] = None
 
         self.on_command: Optional[CommandCallback] = None
         self.on_config_push: Optional[ConfigCallback] = None
@@ -129,12 +153,21 @@ class MQTTManager:
         client.reconnect_delay_set(min_delay=2, max_delay=60)
         client.on_connect = self._on_connect
         client.on_disconnect = self._on_disconnect
+        client.on_connect_fail = self._on_connect_fail
         client.on_message = self._on_message
         return client
 
+    def _set_error(self, kind: str, detail: str) -> None:
+        self._last_error = {"kind": kind, "detail": detail, "epoch": time.time()}
+
+    def _on_connect_fail(self, client, userdata):
+        # paho's own reconnect (after loop_start) could not open the socket.
+        self._set_error("unreachable", f"cannot reach {settings.mqtt_broker}:{settings.mqtt_port}")
+
     def _on_connect(self, client, userdata, flags, reason_code, properties):
         if reason_code == 0:
             self._connected = True
+            self._last_error = None
             client.subscribe(self._t_commands, qos=1)
             client.subscribe(self._t_config, qos=1)
             client.subscribe(self._t_discord_desired, qos=1)
@@ -147,6 +180,7 @@ class MQTTManager:
             asyncio.run_coroutine_threadsafe(self._flush_offline_buffer(), self._loop)
         else:
             logger.error(f"MQTT connect refused: {reason_code}")
+            self._set_error("refused", str(reason_code))
             if _is_auth_failure(reason_code) and self._loop is not None:
                 # paho thread -> event loop. A refused login never fixes itself
                 # by retrying the same password, so go get the right one.
@@ -155,6 +189,9 @@ class MQTTManager:
     def _on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties):
         self._connected = False
         logger.warning(f"MQTT disconnected: {reason_code}")
+        # A refusal is followed by a disconnect; keep the more specific reason.
+        if self._last_error is None:
+            self._set_error("disconnected", str(reason_code))
 
     def _on_message(self, client, userdata, msg):
         try:
@@ -256,6 +293,7 @@ class MQTTManager:
                 return
             except Exception as e:
                 logger.warning(f"MQTT connect failed ({e}) — retrying in {delay}s")
+                self._set_error("unreachable", f"cannot reach {settings.mqtt_broker}:{settings.mqtt_port}: {e}")
                 await asyncio.sleep(delay)
                 delay = min(delay * 2, 60)
 
@@ -372,6 +410,23 @@ class MQTTManager:
             count = devices.get("count") or None
         if count is not None:
             payload["sdr_count"] = count
+        payload.update(disk_report(settings.recordings_path))
+        # Upload spool depth + last upload outcome (node-26#44, toward #40).
+        from app.internal.call_recorder import call_recorder
+        spool = call_recorder.spool.stats()
+        payload["upload_spool_files"] = spool["files"]
+        payload["upload_spool_bytes"] = spool["bytes"]
+        last = spool["last_result"] or {}
+        payload["last_upload_status"] = last.get("status")
+        payload["last_upload_kind"] = last.get("kind")
+        # node-26#45: which recorder this node runs; C2 ignores unknown keys.
+        payload["recorder_backend"] = settings.recorder_backend
+        from app.internal import trunk_recorder as tr
+        if tr.enabled():
+            facts = tr.trunk_recorder.health_facts()
+            payload["tr_state"] = (facts["status"] or {}).get("state")
+            payload["tr_decode_rate"] = facts["decode_rate"]
+            payload["tr_inbox_backlog"] = facts["inbox_backlog"]
         self._publish(self._t_checkin, payload, qos=1)
 
     def _publish(self, topic: str, payload: dict, qos: int = 0, retain: bool = False):
@@ -390,5 +445,10 @@ class MQTTManager:
     def is_connected(self) -> bool:
         return self._connected
 
+    @property
+    def last_error(self) -> Optional[Dict[str, Any]]:
+        """{kind, detail, epoch} for the most recent failure, None while connected."""
+        return None if self._connected else self._last_error
+
 
 mqtt_manager = MQTTManager()

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

@@ -0,0 +1,55 @@
+"""
+Start/stop "the radio" without every caller knowing which recorder backend is
+active (node-26#45). Every path that used to call op25_client.start/stop — C2
+config pushes, the op25_restart command, the dashboard, SDR re-pinning — goes
+through here.
+
+With RECORDER_BACKEND=trunk-recorder, OP25 is never started: both open the
+same RTL-SDR, and OP25 holding it would starve Trunk Recorder. start() becomes
+a restart request to trunk-recorder/run.sh instead.
+"""
+import time
+from typing import Optional
+
+from app.internal import trunk_recorder as tr
+from app.internal.logger import logger
+from app.internal.op25_client import op25_client
+
+
+def name() -> str:
+    return "Trunk Recorder" if tr.enabled() else "OP25"
+
+
+async def start(wait: bool = True) -> bool:
+    return await start_with_reason(wait) is None
+
+
+async def start_with_reason(wait: bool = True) -> Optional[str]:
+    """None on success, else a human-readable reason.
+
+    For Trunk Recorder, `wait` blocks until run.sh reports TR restarted (up
+    to trunk_recorder.RESTART_WAIT_SECONDS), like op25_client.start returns
+    only once OP25 has started — callers hand the spare dongles out next.
+    """
+    if not tr.enabled():
+        return await op25_client.start_with_reason()
+    if tr.trunk_recorder.config_error:
+        return tr.trunk_recorder.config_error
+    requested = time.time()
+    if not tr.trunk_recorder.request_restart():
+        return f"Couldn't write to {tr.config_dir()} to restart Trunk Recorder."
+    if not wait:
+        return None
+    err = await tr.trunk_recorder.wait_started_since(requested)
+    if err:
+        logger.error(err)
+    return err
+
+
+async def stop() -> bool:
+    if not tr.enabled():
+        return await op25_client.stop()
+    ok = tr.trunk_recorder.request_stop()
+    if ok:
+        logger.info("Trunk Recorder stop requested.")
+    return ok

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

@@ -9,7 +9,11 @@ C2 system documents (the `config` blob is passed through untouched):
     P25 conventional  type "P25", config.mode "conventional",
                       config.channels = [{freq: MHz, name, nac?}, ...]   (node-26#24)
     Analog            type "NBFM", config.channels = [{freq, name}] — or, from
-                      before #24, config.control_channels[0]
+                      before #24, config.control_channels[0]. Optional squelch
+                      (node-26#46): config.squelch_mode "noise" (default) |
+                      "voice" | "power", config.squelch_db (noise/voice: dB
+                      of quieting to open) and config.squelch_threshold
+                      (power: absolute dB, OP25 default -60)
 
 A missing mode is trunked, so every system saved before #24 behaves exactly as
 it did. Pure functions; main._generate_op25_config does the I/O.
@@ -80,6 +84,41 @@ def conventional_channels(config: Dict[str, Any]) -> List[Dict[str, Any]]:
     return out
 
 
+SQUELCH_MODES = ("noise", "voice", "power")
+DEFAULT_SQUELCH_MODE = "noise"
+
+
+def _optional_float(config: Dict[str, Any], key: str) -> Optional[float]:
+    value = (config or {}).get(key)
+    if value is None or (isinstance(value, str) and not value.strip()):
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        raise ValueError(f"{key} {value!r} isn't a number.")
+
+
+def analog_squelch(config: Dict[str, Any]) -> Dict[str, Any]:
+    """config -> the op25-container squelch fields for an analog channel.
+
+    Noise squelch by default (node-26#46): OP25's power squelch compares an
+    absolute level that depends on dongle, gain and antenna, and left analog
+    open on static; noise squelch measures FM quieting and needs no
+    calibration. A config that says nothing gets noise mode.
+    """
+    mode = str((config or {}).get("squelch_mode") or DEFAULT_SQUELCH_MODE).strip().lower()
+    if mode not in SQUELCH_MODES:
+        raise ValueError(f"Squelch mode {mode!r} isn't one of {', '.join(SQUELCH_MODES)}.")
+    out: Dict[str, Any] = {"nbfm_squelch_mode": mode}
+    db = _optional_float(config, "squelch_db")
+    if db is not None:
+        out["nbfm_noise_squelch_db"] = db
+    threshold = _optional_float(config, "squelch_threshold")
+    if threshold is not None:
+        out["nbfm_squelch"] = threshold
+    return out
+
+
 def analog_frequency(config: Dict[str, Any]) -> Optional[int]:
     chans = (config or {}).get("channels") or []
     if chans and isinstance(chans[0], dict) and chans[0].get("freq") not in (None, ""):
@@ -117,6 +156,7 @@ def build_generate_payload(
     if mode == ANALOG:
         freq = analog_frequency(config)
         payload["channels"] = [freq] if freq else []
+        payload.update(analog_squelch(config))
         return payload
 
     talkgroups = (config or {}).get("talkgroups", []) or []

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

@@ -83,7 +83,18 @@ async def set_sdr_settings(priority: Optional[Iterable[str]] = None,
         cfg.sdr_pins = normalize_sdr_pins(pins)
     save_node_config(cfg)
 
-    if cfg.sdr_pins.get("op25") != old_op25:
+    if cfg.sdr_pins.get("op25") != old_op25 and settings.recorder_backend == "trunk-recorder":
+        # Trunk Recorder takes the "op25" dongle (node-26#45): rewrite its
+        # config with the new serial; run.sh restarts it on the change.
+        from app.internal.health import active_config
+        from app.main import _generate_op25_config
+        await secondary_sdr_client.apply([], {}, [])
+        current = active_config(cfg)
+        if current is not None and await _generate_op25_config(current):
+            from app.internal import radio_backend
+            await radio_backend.start()  # waits until TR holds the new dongle
+        logger.info(f"Trunk Recorder moved to SDR {cfg.sdr_pins.get('op25') or 'rtl=0 (first dongle)'}")
+    elif cfg.sdr_pins.get("op25") != old_op25:
         from app.internal.op25_client import op25_client
         # Free the new dongle first if a decoder holds it, then move OP25.
         await secondary_sdr_client.apply([], {}, [])

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

@@ -0,0 +1,210 @@
+"""
+Trunk Recorder call JSON -> what DRB sends for one call (node-26#45).
+
+TR writes one JSON per call next to its WAV and passes both to uploadScript.
+The shape is `create_call_json` in trunk-recorder/call_concluder/
+call_concluder.cc at v5.2.1 (keys, in order: freq, freq_error, signal, noise,
+source_num, recorder_num, tdma_slot, phase2_tdma, start_time, stop_time,
+start_time_ms, stop_time_ms, emergency, priority, mode, duplex, encrypted,
+call_length, call_length_ms, talkgroup, talkgroup_tag, talkgroup_description,
+talkgroup_group_tag, talkgroup_group, color_code, audio_type, short_name,
+patched_talkgroups?, freqList[], srcList[]). Older TR releases omit the *_ms
+keys, so both are read.
+
+Pure: trunk_recorder.py does the I/O.
+"""
+import json
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+DRIVER = "trunk-recorder"
+# Fixed namespace so a call's id is a pure function of its JSON: re-reading the
+# same file after a crash yields the same call_id, so C2 overwrites rather than
+# duplicates (trunk_recorder.py relies on this to be restart-safe).
+CALL_ID_NAMESPACE = uuid.UUID("5d0c7f4e-2b8f-4e0e-9a59-45c0de7c0045")
+
+
+def _iso(epoch: float) -> str:
+    return datetime.fromtimestamp(epoch, timezone.utc).isoformat()
+
+
+def _int(value: Any) -> Optional[int]:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _epoch(data: Dict[str, Any], key: str) -> Optional[float]:
+    ms = data.get(f"{key}_ms")
+    if isinstance(ms, (int, float)) and ms > 0:
+        return ms / 1000.0
+    s = data.get(key)
+    if isinstance(s, (int, float)) and s > 0:
+        return float(s)
+    return None
+
+
+@dataclass
+class TrCall:
+    call_id: str
+    tgid: int
+    tgid_name: str
+    freq: Optional[int]
+    start_epoch: float
+    stop_epoch: float
+    units: List[int] = field(default_factory=list)
+    emergency: bool = False
+    encrypted: bool = False
+    src_list: List[Dict[str, Any]] = field(default_factory=list)
+    freq_list: List[Dict[str, Any]] = field(default_factory=list)
+    call_length: Optional[float] = None
+    signal: Optional[int] = None
+    noise: Optional[int] = None
+    freq_error: Optional[int] = None
+    decode_errors: Optional[int] = None
+    spikes: Optional[int] = None
+    patched_talkgroups: List[int] = field(default_factory=list)
+    short_name: str = ""
+
+    @property
+    def srcaddr(self) -> Optional[int]:
+        return self.units[0] if self.units else None
+
+    def start_payload(self) -> Dict[str, Any]:
+        """MQTT call_start — the keys main.on_call_start publishes, which C2's
+        _on_call_start reads (tgid, tgid_name, freq, srcaddr, started_at)."""
+        return {
+            "call_id": self.call_id,
+            "tgid": self.tgid,
+            "tgid_name": self.tgid_name,
+            "freq": self.freq,
+            "srcaddr": self.srcaddr,
+            "started_at": _iso(self.start_epoch),
+            "started_at_epoch": self.start_epoch,
+            "attributed": True,
+            "driver": DRIVER,
+        }
+
+    def audio_end_epoch(self, wav_seconds: Optional[float] = None) -> float:
+        """Wall clock at the END of the audio file.
+
+        NOT stop_time: TR joins a call's transmissions back to back and drops
+        the gaps between them, so the WAV is shorter than stop - start by the
+        sum of the gaps. start + WAV length is what the file's last sample
+        maps to only if there were no gaps; the exact per-transmission map is
+        srcList/freqList (time = wall clock, pos = offset in the file), sent
+        as src_list / freq_list.
+        """
+        if wav_seconds is None:
+            return self.stop_epoch
+        return self.start_epoch + wav_seconds
+
+    def end_payload(self, wav_seconds: Optional[float] = None) -> Dict[str, Any]:
+        """MQTT call_end. TR cut the call on decoder boundaries with no
+        trimming, so the audio starts at the call start; see audio_end_epoch
+        for why it does not end at the call end."""
+        audio_end = self.audio_end_epoch(wav_seconds)
+        out = self.start_payload()
+        out.update({
+            "ended_at": _iso(self.stop_epoch),
+            "ended_at_epoch": self.stop_epoch,
+            "audio_start_at": _iso(self.start_epoch),
+            "audio_end_at": _iso(audio_end),
+            "audio_start_epoch": self.start_epoch,
+            "audio_end_epoch": audio_end,
+            "units": self.units,
+            "emergency": self.emergency,
+            "encrypted": self.encrypted,
+        })
+        return out
+
+    def upload_extra(self) -> Dict[str, str]:
+        """Extra /upload form fields. C2 declares none of these yet, so FastAPI
+        ignores them (same as audio_start_epoch today); they are here for when
+        it does."""
+        extra = {
+            "recorder_backend": DRIVER,
+            "units": ",".join(str(u) for u in self.units),
+            "src_list": json.dumps(self.src_list, separators=(",", ":")),
+            "freq_list": json.dumps(self.freq_list, separators=(",", ":")),
+            "emergency": "1" if self.emergency else "0",
+            "encrypted": "1" if self.encrypted else "0",
+            "started_at_epoch": f"{self.start_epoch:.3f}",
+            "ended_at_epoch": f"{self.stop_epoch:.3f}",
+        }
+        optional = {
+            "freq": self.freq, "call_length": self.call_length, "signal": self.signal, "noise": self.noise,
+            "freq_error": self.freq_error, "decode_errors": self.decode_errors, "spikes": self.spikes,
+        }
+        extra.update({k: str(v) for k, v in optional.items() if v is not None})
+        if self.patched_talkgroups:
+            extra["patched_talkgroups"] = ",".join(str(t) for t in self.patched_talkgroups)
+        return extra
+
+
+def parse_call(data: Dict[str, Any], node_id: str, file_stem: str = "") -> TrCall:
+    """One TR call JSON -> TrCall. ValueError if it has no talkgroup or start."""
+    tgid = _int(data.get("talkgroup"))
+    if not tgid or tgid <= 0:
+        raise ValueError(f"call JSON has no talkgroup ({data.get('talkgroup')!r})")
+    start = _epoch(data, "start_time")
+    if start is None:
+        raise ValueError("call JSON has no start_time")
+    stop = _epoch(data, "stop_time")
+    length = data.get("call_length_ms")
+    length_s = (length / 1000.0) if isinstance(length, (int, float)) and length > 0 else None
+    if length_s is None and isinstance(data.get("call_length"), (int, float)):
+        length_s = float(data["call_length"])
+    if stop is None or stop < start:
+        stop = start + (length_s or 0.0)
+
+    src_list: List[Dict[str, Any]] = []
+    units: List[int] = []
+    unit_emergency = False
+    for s in data.get("srcList") or []:
+        if not isinstance(s, dict):
+            continue
+        src = _int(s.get("src"))
+        entry = {"src": src, "time": s.get("time"), "pos": s.get("pos"),
+                 "emergency": bool(_int(s.get("emergency")))}
+        if s.get("tag"):
+            entry["tag"] = s.get("tag")
+        src_list.append(entry)
+        unit_emergency = unit_emergency or entry["emergency"]
+        # TR reports 0 / -1 when the unit id wasn't decoded.
+        if src and src > 0 and src not in units:
+            units.append(src)
+
+    freq_list = [f for f in data.get("freqList") or [] if isinstance(f, dict)]
+    errors = sum(_int(f.get("error_count")) or 0 for f in freq_list) if freq_list else None
+    spikes = sum(_int(f.get("spike_count")) or 0 for f in freq_list) if freq_list else None
+
+    freq = _int(data.get("freq"))
+    short_name = str(data.get("short_name") or "")
+    key = f"{node_id}|{short_name}|{tgid}|{start:.3f}|{freq}|{file_stem}"
+    return TrCall(
+        call_id=str(uuid.uuid5(CALL_ID_NAMESPACE, key)),
+        tgid=tgid,
+        tgid_name=str(data.get("talkgroup_tag") or data.get("talkgroup_description") or ""),
+        freq=freq,
+        start_epoch=start,
+        stop_epoch=stop,
+        units=units,
+        emergency=bool(_int(data.get("emergency"))) or unit_emergency,
+        encrypted=bool(_int(data.get("encrypted"))),
+        src_list=src_list,
+        freq_list=[{"freq": _int(f.get("freq")), "time": f.get("time"), "pos": f.get("pos"), "len": f.get("len"),
+                    "error_count": _int(f.get("error_count")), "spike_count": _int(f.get("spike_count"))}
+                   for f in freq_list],
+        call_length=round(length_s, 3) if length_s is not None else None,
+        signal=_int(data.get("signal")),
+        noise=_int(data.get("noise")),
+        freq_error=_int(data.get("freq_error")),
+        decode_errors=errors,
+        spikes=spikes,
+        patched_talkgroups=[t for t in (_int(x) for x in data.get("patched_talkgroups") or []) if t],
+        short_name=short_name,
+    )

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

@@ -0,0 +1,291 @@
+"""
+Trunk Recorder config.json (+ CSVs) from the C2 system config (node-26#45).
+
+Same input as radio_config.build_generate_payload — the SystemConfig C2 pushes —
+so a node can switch backends without anyone re-entering its system:
+
+    P25 trunked       -> system type "p25", control_channels, talkgroupsFile
+    P25 conventional  -> "conventionalP25", channelFile
+    Analog            -> "conventional",    channelFile, squelch
+
+Schema per Trunk Recorder v5.2.1 docs/CONFIGURE.md. Pure functions: the I/O is
+in trunk_recorder.py. Unverified against a radio — see the notes inline.
+
+Conventional channels are numbered with conventional.channel_tgid (frequency in
+100 Hz units), the same synthetic talkgroup the OP25 path reports, so C2 names
+a TR call and an OP25 call on the same channel identically.
+"""
+import csv
+import io
+from typing import Any, Dict, List, Optional, Tuple
+
+from app.internal.logger import logger
+from app.internal.conventional import channel_tgid
+from app.internal.radio_config import (
+    ANALOG, CONVENTIONAL, DMR, TRUNKED, analog_frequency, conventional_channels, system_mode, to_hz,
+)
+
+# Paths INSIDE the trunk-recorder container (see docker-compose.yml).
+TR_APP_DIR = "/app"                         # tr_config_dir is mounted here
+TR_UPLOAD_SCRIPT = "/drb/drb-upload.sh"     # ./trunk-recorder, read-only
+TR_CAPTURE_DIR = "/recordings/tr-capture"   # ./recordings
+TR_TEMP_DIR = "/dev/shm"                    # TR's recommended tmpfs (shm_size in compose)
+TALKGROUPS_FILE = "talkgroups.csv"
+CHANNELS_FILE = "channels.csv"
+SHORT_NAME = "drb"
+
+# TR splits a call longer than this; mirrors call_recorder.MAX_RECORDING_SECONDS.
+MAX_DURATION_SECONDS = 600
+# Seconds of control-channel silence before TR closes a call (TR's default).
+CALL_TIMEOUT_SECONDS = 3
+
+# Of an RTL-SDR's sample rate, the share that is usable passband (anti-alias
+# roll-off eats the edges). Channels must sit inside +-(rate * share / 2).
+USABLE_BANDWIDTH_SHARE = 0.8
+# The RTL-SDR's DC spike sits on the centre frequency; keep channels off it.
+DC_SPIKE_GUARD_HZ = 25_000
+CENTER_NUDGES_HZ = (0, 150_000, -150_000, 300_000, -300_000)
+
+
+def _fits(center: int, freqs: List[int], half_band: float) -> bool:
+    return all(abs(f - center) <= half_band for f in freqs)
+
+
+def widest_cover(freqs: List[int], rate: int) -> List[int]:
+    """The largest run of frequencies one SDR at `rate` can hold at once
+    (ties: the lowest). A trunked system lists every control channel it might
+    use, but TR needs only one live at a time, so a list wider than the SDR is
+    cut down to what fits instead of refusing the whole system."""
+    half = rate * USABLE_BANDWIDTH_SHARE / 2
+    fs = sorted(set(freqs))
+    best: List[int] = []
+    for i, lo in enumerate(fs):
+        run = [f for f in fs[i:] if f - lo <= 2 * half]
+        if len(run) > len(best):
+            best = run
+    return best
+
+
+def choose_center(freqs: List[int], rate: int, override: Optional[int] = None) -> int:
+    """A centre frequency that puts every channel in the passband and none on
+    the DC spike. ValueError if they cannot all fit on one SDR."""
+    if not freqs:
+        raise ValueError("No frequencies to tune to.")
+    half = rate * USABLE_BANDWIDTH_SHARE / 2
+    if override:
+        if not _fits(override, freqs, half):
+            raise ValueError(f"TR_CENTER_HZ {override} leaves a channel outside the SDR's {2 * half / 1e6:.1f} MHz.")
+        return int(override)
+    lo, hi = min(freqs), max(freqs)
+    if hi - lo > 2 * half:
+        raise ValueError(
+            f"Channels span {(hi - lo) / 1e6:.3f} MHz, more than one SDR at {rate / 1e6:.1f} Msps covers "
+            f"({2 * half / 1e6:.2f} MHz). Trim the channel list or raise TR_SAMPLE_RATE."
+        )
+    mid = int(round((lo + hi) / 2))
+    for nudge in CENTER_NUDGES_HZ:
+        c = mid + nudge
+        if _fits(c, freqs, half) and all(abs(f - c) >= DC_SPIKE_GUARD_HZ for f in freqs):
+            return c
+    return mid  # channels packed around the middle: accept the spike over not fitting
+
+
+def _csv(header: List[str], rows: List[List[Any]]) -> str:
+    buf = io.StringIO()
+    w = csv.writer(buf, lineterminator="\n")
+    w.writerow(header)
+    w.writerows(rows)
+    return buf.getvalue()
+
+
+def talkgroups_csv(talkgroups: List[Dict[str, Any]]) -> str:
+    """TR talkgroupsFile: header row required, "Decimal" first (CONFIGURE.md)."""
+    rows = []
+    for tg in talkgroups:
+        if tg.get("id") is None:
+            continue
+        name = str(tg.get("name") or "").strip()
+        rows.append([int(tg["id"]), "D", name, name, "", ""])
+    return _csv(["Decimal", "Mode", "Alpha Tag", "Description", "Tag", "Category"], rows)
+
+
+def channels_csv(channels: List[Dict[str, Any]], squelch: Optional[float] = None) -> str:
+    """TR channelFile: header required, "TG Number" first; Frequency in Hz
+    (no decimal point = Hz). Per-channel Squelch overrides the system's."""
+    header = ["TG Number", "Frequency", "Alpha Tag", "Description"]
+    if squelch is not None:
+        header.append("Squelch")
+    rows = []
+    for ch in channels:
+        name = ch.get("name") or f"{ch['frequency'] / 1e6:.4f} MHz"
+        row = [channel_tgid(ch["frequency"]), ch["frequency"], name, name]
+        if squelch is not None:
+            row.append(squelch)
+        rows.append(row)
+    return _csv(header, rows)
+
+
+def _analog_channels(config: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """Every analog channel the config lists (OP25 only ever used the first)."""
+    out = []
+    for raw in (config or {}).get("channels") or []:
+        if isinstance(raw, dict) and raw.get("freq") not in (None, ""):
+            out.append({"frequency": to_hz(raw["freq"]), "name": str(raw.get("name") or "").strip()})
+    if not out:
+        freq = analog_frequency(config)
+        if freq:
+            out.append({"frequency": freq, "name": ""})
+    return out
+
+
+def _analog_squelch(config: Dict[str, Any], default_db: float) -> float:
+    """TR's squelch is an absolute power level — OP25's "power" mode. Its
+    squelch_threshold carries over; noise/voice squelch_db has no TR
+    equivalent, so those fall back to the default (which needs site tuning)."""
+    value = (config or {}).get("squelch_threshold")
+    if value is None or (isinstance(value, str) and not value.strip()):
+        return float(default_db)
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        raise ValueError(f"squelch_threshold {value!r} isn't a number.")
+
+
+def build_tr_config(
+    system_type: str,
+    name: str,
+    config: Dict[str, Any],
+    *,
+    device_serial: Optional[str],
+    ppm: Optional[float],
+    sample_rate: int,
+    gain: float,
+    digital_recorders: int,
+    min_duration: float,
+    squelch_db: float,
+    live_port: int,
+    center_override: Optional[int] = None,
+) -> Tuple[Dict[str, Any], Dict[str, str]]:
+    """SystemConfig fields -> (config.json dict, {filename: contents} to write beside it).
+
+    ValueError when the system can't be expressed (DMR, no channels, too wide).
+    """
+    config = config or {}
+    mode = system_mode(system_type, config)
+    files: Dict[str, str] = {}
+    # Kept per system, not globally: audioArchive/callLog false makes TR delete
+    # the wav/json after uploadScript returns, which is why drb-upload.sh
+    # copies them out before it exits.
+    system: Dict[str, Any] = {
+        "shortName": SHORT_NAME,
+        "uploadScript": TR_UPLOAD_SCRIPT,
+        "audioArchive": False,
+        "callLog": False,
+        # WAV, not TR's 32k AAC: the recorder's own rule is one lossless copy
+        # for Whisper (call_recorder AUDIO_* notes). 8 kHz mono is ~1 MB/min.
+        "compressWav": False,
+        "minDuration": float(min_duration),
+        "maxDuration": MAX_DURATION_SECONDS,
+        "hideEncrypted": False,
+    }
+    source: Dict[str, Any] = {
+        "driver": "osmosdr",
+        # By serial, like OP25 (sdr_settings.pin_op25_device): "rtl=0" means
+        # whichever dongle enumerated first, which a decoder can take.
+        "device": f"rtl={device_serial}" if device_serial else "rtl=0",
+        "rate": int(sample_rate),
+        "gain": float(gain),
+        "ppm": float(ppm or 0.0),
+    }
+
+    if mode == DMR:
+        raise ValueError("DMR isn't supported by the edge node.")
+
+    if mode == TRUNKED:
+        ccs = [to_hz(c) for c in config.get("control_channels") or []]
+        if not ccs:
+            raise ValueError(f"System {name!r} has no control channels.")
+        talkgroups = [tg for tg in config.get("talkgroups") or [] if tg.get("id") is not None]
+        system.update({
+            "type": "p25",
+            "control_channels": ccs,
+            # TR's default; "fsk4" is accepted too. Overridable from C2.
+            "modulation": str(config.get("modulation") or "qpsk"),
+        })
+        # Same rule as OP25's whitelist: a talkgroup list means record only
+        # those; no list means record everything.
+        if talkgroups:
+            files[TALKGROUPS_FILE] = talkgroups_csv(talkgroups)
+            system["talkgroupsFile"] = f"{TR_APP_DIR}/{TALKGROUPS_FILE}"
+            system["recordUnknown"] = False
+        else:
+            system["recordUnknown"] = True
+        # Voice channels are wherever the system grants them; only the control
+        # channels are known here, so the SDR is centred on those. A system
+        # whose voice channels sit outside that ~2 MHz loses those calls (TR
+        # logs "no source covering" for each) — needs a second source.
+        if center_override:
+            half = int(sample_rate) * USABLE_BANDWIDTH_SHARE / 2
+            covered = [c for c in ccs if abs(c - center_override) <= half]
+            if not covered:
+                raise ValueError(f"TR_CENTER_HZ {center_override} covers none of {name!r}'s control channels.")
+        else:
+            covered = widest_cover(ccs, int(sample_rate))
+        if len(covered) < len(ccs):
+            system["control_channels"] = covered
+            logger.warning(
+                f"Control channels span {(max(ccs) - min(ccs)) / 1e6:.3f} MHz, wider than one SDR; "
+                f"using the {len(covered)} of {len(ccs)} that fit "
+                f"({min(covered) / 1e6:.5f}-{max(covered) / 1e6:.5f} MHz)."
+            )
+        freqs = covered
+        source["digitalRecorders"] = int(digital_recorders)
+    elif mode == CONVENTIONAL:
+        chans = conventional_channels(config)
+        if not chans:
+            raise ValueError(f"System {name!r} has no conventional channels.")
+        # No NAC filter: TR's conventionalP25 records whatever it decodes, so a
+        # mistyped NAC can no longer silently drop every call (node-26#33).
+        system.update({"type": "conventionalP25", "channelFile": f"{TR_APP_DIR}/{CHANNELS_FILE}",
+                       "squelch": float(squelch_db)})
+        files[CHANNELS_FILE] = channels_csv(chans)
+        freqs = [c["frequency"] for c in chans]
+    else:  # ANALOG
+        chans = _analog_channels(config)
+        if not chans:
+            raise ValueError(f"System {name!r} has no analog frequency.")
+        squelch = _analog_squelch(config, squelch_db)
+        system.update({"type": "conventional", "channelFile": f"{TR_APP_DIR}/{CHANNELS_FILE}",
+                       "squelch": squelch})
+        files[CHANNELS_FILE] = channels_csv(chans)
+        freqs = [c["frequency"] for c in chans]
+
+    source["center"] = choose_center(freqs, int(sample_rate), center_override)
+
+    tr = {
+        "ver": 2,
+        "sources": [source],
+        "systems": [system],
+        "tempDir": TR_TEMP_DIR,
+        "captureDir": TR_CAPTURE_DIR,
+        "callTimeout": CALL_TIMEOUT_SECONDS,
+        # File log too (stdout still goes to docker logs): trunk_recorder.py
+        # reads the control-channel decode rate from it for the health check.
+        "logFile": True,
+        "logDir": f"{TR_APP_DIR}/logs",
+        "logColor": "none",  # TR 5.x: a string ("console"|"none"), not a bool
+        # Live audio for Discord: simplestream sends every call's PCM to the
+        # edge-node, which follows one talkgroup at a time into PulseAudio.
+        "audioStreaming": True,
+        "plugins": [{
+            "name": "simplestream",
+            "library": "libsimplestream.so",
+            "streams": [{"TGID": 0, "address": "127.0.0.1", "port": int(live_port), "sendTGID": True}],
+        }],
+    }
+    return tr, files
+
+
+def live_sample_rate(system_type: str, config: Dict[str, Any]) -> int:
+    """simplestream sends analog audio at 16 kHz and digital at 8 kHz (Plugins.md)."""
+    return 16000 if system_mode(system_type, config or {}) == ANALOG else 8000

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

@@ -0,0 +1,575 @@
+"""
+Trunk Recorder backend, edge-node side (node-26#45). Active only when
+RECORDER_BACKEND=trunk-recorder.
+
+    C2 SystemConfig --apply_config--> <tr_config_dir>/config.json (+ CSVs)
+                                      trunk-recorder/run.sh restarts TR on change
+    TR finishes a call --drb-upload.sh--> <tr_inbox_dir>/<stem>.wav, then <stem>.json
+    ingest loop --> MQTT call_start --> /upload (call_recorder.upload_recording,
+                    i.e. the node-26#44 spool) --> MQTT call_end
+    TR simplestream (UDP) --LiveAudioBridge--> pacat --> drb_sink --> Discord
+
+Why an inbox directory and not an HTTP endpoint: TR deletes the WAV/JSON as
+soon as uploadScript returns (audioArchive/callLog false), so the script must
+copy the files out synchronously anyway, and a file on the host disk survives
+an edge-node restart where a POST to a down edge-node would not. The JSON is
+moved in LAST, so its presence means the WAV is complete.
+
+Restart safety: a call_id is a pure function of the JSON (tr_calls.parse_call),
+and the WAV is renamed to `<YYYYmmdd_HHMMSS>_<call_id>.wav` (the name the spool
+recovers a call_id from) before anything is sent. A crash at any step re-runs
+the same call with the same id, so C2 overwrites instead of duplicating; a JSON
+whose WAV is already gone was handed to the uploader and is just removed.
+"""
+import asyncio
+import json
+import math
+import os
+import re
+import time
+import wave
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set
+
+from app.config import settings
+from app.internal.logger import logger
+from app.internal.tr_calls import TrCall, parse_call
+from app.models import SystemConfig
+
+CONFIG_FILE = "config.json"
+STATUS_FILE = "drb-status.json"       # written by trunk-recorder/run.sh every few seconds
+RESTART_REQUEST = "restart-request"   # run.sh restarts TR when this appears
+STOP_REQUEST = "stop-request"         # run.sh keeps TR stopped while this exists
+
+INBOX_POLL_SECONDS = 1.0
+INGEST_CONCURRENCY = 4
+# Hand-off is wav-then-json; a JSON whose WAV never shows is dropped after this.
+ORPHAN_JSON_SECONDS = 60.0
+# C2 creates the call doc from MQTT call_start, and only that message carries
+# the talkgroup and start time. With the broker down, call_start is dropped
+# (mqtt_manager buffers only call_end), so a call waits in the inbox this long
+# for MQTT to come back before it is sent without one.
+MQTT_WAIT_SECONDS = 600.0
+# call_start must be written before /upload stamps audio_gcs_uri: C2's
+# _on_call_start writes the doc with merge=False and would erase it
+# (server-26#303; drop this lead once C2 makes call_start create-only).
+CALL_START_LEAD_SECONDS = 1.0
+# A WAV with no JSON this old is a hand-off drb-upload.sh never finished.
+STRAY_WAV_SECONDS = 3600.0
+# How long radio_backend waits for run.sh to (re)start TR after a request: it
+# polls every 5 s and may spend 15 s stopping the old process.
+RESTART_WAIT_SECONDS = 30.0
+RESTART_POLL_SECONDS = 1.0
+# The name process() renames a claimed WAV to; never swept (its JSON is what
+# drives it, and os.replace keeps TR's old mtime).
+CANONICAL_WAV_RE = re.compile(r"^\d{8}_\d{6}_[0-9a-f-]{36}\.wav$")
+# run.sh rewrites its status every 5 s; past this the supervisor is gone.
+STATUS_STALE_SECONDS = 20.0
+LOG_TAIL_BYTES = 64 * 1024
+# TR logs this for every trunked system every few seconds (main.cpp
+# check_message_count). Exact wording unverified against a live log.
+DECODE_RATE_RE = re.compile(r"Decode Rate:\s*([0-9.]+)\s*/\s*sec", re.IGNORECASE)
+
+
+def enabled() -> bool:
+    return settings.recorder_backend == "trunk-recorder"
+
+
+def config_dir() -> Path:
+    return Path(settings.tr_config_dir or (Path(settings.config_path) / "trunk-recorder"))
+
+
+def inbox_dir() -> Path:
+    return Path(settings.tr_inbox_dir or (Path(settings.recordings_path) / "tr-inbox"))
+
+
+def _atomic_write(path: Path, text: str) -> None:
+    tmp = path.with_name(path.name + ".tmp")
+    tmp.write_text(text)
+    os.replace(tmp, path)
+
+
+def wav_seconds(path: Path) -> Optional[float]:
+    """Length of a WAV from its header; None if it can't be read."""
+    try:
+        with wave.open(str(path), "rb") as w:
+            rate = w.getframerate()
+            return w.getnframes() / rate if rate else None
+    except (OSError, EOFError, wave.Error):
+        return None
+
+
+def canonical_wav_name(call: TrCall) -> str:
+    stamp = datetime.fromtimestamp(call.start_epoch).strftime("%Y%m%d_%H%M%S")
+    return f"{stamp}_{call.call_id}.wav"
+
+
+class TrunkRecorder:
+    def __init__(self):
+        self.config_error: Optional[str] = None
+        self.calls_ingested = 0
+        self.encrypted_skipped = 0
+        self.last_call_epoch: Optional[float] = None
+        self.last_call: Optional[Dict[str, Any]] = None
+        self._task: Optional[asyncio.Task] = None
+        self._inflight: Set[str] = set()
+        self._sem = asyncio.Semaphore(INGEST_CONCURRENCY)
+        self.live = LiveAudioBridge()
+
+    # ------------------------------------------------------------------
+    # Config + supervisor requests
+    # ------------------------------------------------------------------
+
+    async def apply_config(self, config: SystemConfig) -> bool:
+        """Write TR's config for `config`. run.sh restarts TR when config.json
+        changes; a restart request covers a CSV-only change."""
+        from app.internal.config_manager import load_node_config
+        from app.internal.sdr_settings import op25_serial
+        from app.internal.secondary_sdr_client import secondary_sdr_client
+        from app.internal.tr_config import build_tr_config, live_sample_rate
+
+        node_cfg = load_node_config()
+        # TR takes the dongle OP25 would have (pin or first detected), so the
+        # secondary decoders' reservation logic is unchanged.
+        devs = await secondary_sdr_client.devices()
+        serial = op25_serial(node_cfg, devs)
+        if serial and devs and sum(d["serial"] == serial for d in devs) > 1:
+            serial = None  # two dongles share it: by-serial would be ambiguous
+        try:
+            tr, files = build_tr_config(
+                config.type, config.name, config.config,
+                device_serial=serial, ppm=node_cfg.ppm_override,
+                sample_rate=settings.tr_sample_rate, gain=settings.tr_gain,
+                digital_recorders=settings.tr_digital_recorders, min_duration=settings.tr_min_duration,
+                squelch_db=settings.tr_squelch_db, live_port=settings.tr_live_port,
+                center_override=settings.tr_center_hz,
+            )
+        except ValueError as e:
+            self.config_error = f"System {config.name!r}: {e}"
+            logger.error(f"Trunk Recorder config not written — {self.config_error}")
+            return False
+        try:
+            d = config_dir()
+            d.mkdir(parents=True, exist_ok=True)
+            (d / "logs").mkdir(exist_ok=True)
+            # CSVs first: run.sh restarts on config.json, which must never
+            # point at a file that isn't there yet.
+            for name, text in files.items():
+                _atomic_write(d / name, text)
+            _atomic_write(d / CONFIG_FILE, json.dumps(tr, indent=2))
+        except OSError as e:
+            self.config_error = f"Couldn't write {config_dir()}: {e}"
+            logger.error(self.config_error)
+            return False
+        self.config_error = None
+        self.live.sample_rate = live_sample_rate(config.type, config.config)
+        logger.info(f"Trunk Recorder config written for {config.name} (SDR {serial or 'rtl=0'}).")
+        return True
+
+    def request_restart(self) -> bool:
+        try:
+            d = config_dir()
+            d.mkdir(parents=True, exist_ok=True)
+            (d / STOP_REQUEST).unlink(missing_ok=True)
+            (d / RESTART_REQUEST).write_text(str(time.time()))
+            return True
+        except OSError as e:
+            logger.error(f"Couldn't ask Trunk Recorder to restart: {e}")
+            return False
+
+    async def wait_started_since(self, since: float, timeout: Optional[float] = None) -> Optional[str]:
+        """Wait until run.sh reports TR running with started_at >= `since`.
+
+        Callers hand the spare dongles to the secondary decoders right after a
+        start; returning before TR has actually reopened its SDR would let a
+        decoder take it. None when started, else why not.
+        """
+        timeout = RESTART_WAIT_SECONDS if timeout is None else timeout
+        deadline = time.time() + timeout
+        while True:
+            st = self.supervisor_status()
+            if st and not st["stale"] and st.get("state") == "running" \
+                    and float(st.get("started_at") or 0) >= math.floor(since):
+                return None
+            if time.time() >= deadline:
+                break
+            await asyncio.sleep(RESTART_POLL_SECONDS)
+        if st is None:
+            return "The trunk-recorder service isn't running (is COMPOSE_PROFILES=trunk-recorder set?)."
+        return f"Trunk Recorder didn't restart within {timeout:.0f}s (state: {st.get('state')})."
+
+    def request_stop(self) -> bool:
+        try:
+            d = config_dir()
+            d.mkdir(parents=True, exist_ok=True)
+            (d / STOP_REQUEST).write_text(str(time.time()))
+            return True
+        except OSError as e:
+            logger.error(f"Couldn't ask Trunk Recorder to stop: {e}")
+            return False
+
+    # ------------------------------------------------------------------
+    # Health
+    # ------------------------------------------------------------------
+
+    def supervisor_status(self, now: Optional[float] = None) -> Optional[Dict[str, Any]]:
+        now = time.time() if now is None else now
+        try:
+            data = json.loads((config_dir() / STATUS_FILE).read_text())
+        except (OSError, ValueError):
+            return None
+        data["age_seconds"] = round(now - float(data.get("epoch") or 0), 1)
+        data["stale"] = data["age_seconds"] > STATUS_STALE_SECONDS
+        return data
+
+    def log_facts(self, now: Optional[float] = None) -> Dict[str, Any]:
+        now = time.time() if now is None else now
+        logs = config_dir() / "logs"
+        try:
+            files = [p for p in logs.iterdir() if p.is_file()]
+        except OSError:
+            files = []
+        if not files:
+            return {"log_age_seconds": None, "decode_rate": None}
+        newest = max(files, key=lambda p: p.stat().st_mtime)
+        age = now - newest.stat().st_mtime
+        rate = None
+        try:
+            with open(newest, "rb") as f:
+                f.seek(max(0, newest.stat().st_size - LOG_TAIL_BYTES))
+                tail = f.read().decode(errors="replace")
+            found = DECODE_RATE_RE.findall(tail)
+            rate = float(found[-1]) if found else None
+        except OSError:
+            pass
+        return {"log_age_seconds": round(age, 1), "decode_rate": rate}
+
+    def log_tail(self, lines: int = 80) -> List[str]:
+        """Last lines of TR's newest log file, for the dashboard's log panel."""
+        try:
+            files = [p for p in (config_dir() / "logs").iterdir() if p.is_file()]
+            if not files:
+                return []
+            newest = max(files, key=lambda p: p.stat().st_mtime)
+            with open(newest, "rb") as f:
+                f.seek(max(0, newest.stat().st_size - LOG_TAIL_BYTES))
+                text = f.read().decode(errors="replace")
+        except OSError:
+            return []
+        return text.splitlines()[-max(1, min(int(lines), 500)):]
+
+    def sweep_stray_wavs(self, now: Optional[float] = None) -> int:
+        """Delete WAVs whose JSON never arrived (drb-upload.sh died between
+        its two moves). Canonical names (already claimed by a JSON)
+        are never touched, so only true strays go."""
+        now = time.time() if now is None else now
+        removed = 0
+        try:
+            jsons = {p.stem for p in inbox_dir().glob("*.json")}
+            for wav in inbox_dir().glob("*.wav"):
+                if wav.stem in jsons or CANONICAL_WAV_RE.match(wav.name) \
+                        or now - wav.stat().st_mtime < STRAY_WAV_SECONDS:
+                    continue
+                logger.error(f"Trunk Recorder WAV {wav.name} has had no call JSON for an hour — deleting it.")
+                wav.unlink(missing_ok=True)
+                removed += 1
+        except OSError:
+            pass
+        return removed
+
+    def inbox_backlog(self) -> int:
+        try:
+            return sum(1 for _ in inbox_dir().glob("*.json"))
+        except OSError:
+            return 0
+
+    def health_facts(self) -> Dict[str, Any]:
+        return {
+            "status": self.supervisor_status(),
+            **self.log_facts(),
+            "config_error": self.config_error,
+            "inbox_backlog": self.inbox_backlog(),
+            "calls_ingested": self.calls_ingested,
+            "encrypted_skipped": self.encrypted_skipped,
+            "last_call_epoch": self.last_call_epoch,
+            "last_call": self.last_call,
+            "live": self.live.stats(),
+        }
+
+    # ------------------------------------------------------------------
+    # Ingest
+    # ------------------------------------------------------------------
+
+    async def start(self) -> None:
+        inbox_dir().mkdir(parents=True, exist_ok=True)
+        if self._task is None or self._task.done():
+            self._task = asyncio.create_task(self._run())
+        await self.live.start()
+
+    async def stop(self) -> None:
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+            self._task = None
+        await self.live.stop()
+
+    async def _run(self) -> None:
+        backlog = self.inbox_backlog()
+        if backlog:
+            logger.info(f"Trunk Recorder inbox has {backlog} call(s) from before this start — sending.")
+        last_sweep = 0.0
+        while True:
+            try:
+                await self.scan_once()
+                if time.time() - last_sweep > 300:
+                    last_sweep = time.time()
+                    self.sweep_stray_wavs()
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                logger.error(f"Trunk Recorder inbox pass failed: {e}")
+            await asyncio.sleep(INBOX_POLL_SECONDS)
+
+    async def scan_once(self, wait: bool = False) -> int:
+        """Start a task per ready call. Returns how many were started; with
+        `wait`, also waits for them (tests)."""
+        from app.internal.mqtt_manager import mqtt_manager
+        try:
+            ready = sorted(inbox_dir().glob("*.json"), key=lambda p: p.stat().st_mtime)
+        except OSError:
+            return 0
+        started: List[asyncio.Task] = []
+        now = time.time()
+        for path in ready:
+            if path.name in self._inflight:
+                continue
+            try:
+                age = now - path.stat().st_mtime
+            except OSError:
+                continue
+            if not mqtt_manager.is_connected and age < MQTT_WAIT_SECONDS:
+                continue
+            self._inflight.add(path.name)
+            started.append(asyncio.create_task(self._guarded(path)))
+        if wait and started:
+            await asyncio.gather(*started)
+        return len(started)
+
+    async def _guarded(self, path: Path) -> None:
+        try:
+            async with self._sem:
+                await self.process(path)
+        except Exception as e:
+            logger.error(f"Trunk Recorder call {path.name} failed: {e}")
+        finally:
+            self._inflight.discard(path.name)
+
+    async def process(self, json_path: Path) -> None:
+        from app.internal.call_recorder import call_recorder
+        from app.internal.config_manager import load_node_config
+        from app.internal.mqtt_manager import mqtt_manager
+
+        try:
+            data = json.loads(json_path.read_text())
+            call = parse_call(data, settings.node_id, json_path.stem)
+        except (OSError, ValueError) as e:
+            logger.error(f"Trunk Recorder call {json_path.name} is unreadable ({e}) — dropping it.")
+            json_path.with_suffix(".wav").unlink(missing_ok=True)
+            json_path.unlink(missing_ok=True)
+            return
+
+        raw_wav = json_path.with_suffix(".wav")
+        wav = json_path.parent / canonical_wav_name(call)
+        if raw_wav.exists():
+            os.replace(raw_wav, wav)
+        if not wav.exists():
+            age = time.time() - json_path.stat().st_mtime
+            if age > ORPHAN_JSON_SECONDS:
+                # Either TR's hand-off broke, or a previous pass already handed
+                # this WAV to the uploader (which deletes or spools it) and
+                # died before removing the JSON. Either way nothing to send.
+                # Still close it in C2: if a previous pass uploaded and died
+                # before this point, the doc would otherwise stay "active".
+                # _on_call_end merges, so a repeat is harmless.
+                logger.warning(f"Trunk Recorder call {call.call_id}: no audio beside {json_path.name} — removing it.")
+                await mqtt_manager.publish_metadata("call_end", {**call.end_payload(), "audio_skipped": "no_audio"})
+                json_path.unlink(missing_ok=True)
+            return
+
+        await mqtt_manager.publish_metadata("call_start", call.start_payload())
+        seconds = wav_seconds(wav)
+        end = call.end_payload(seconds)
+
+        if call.encrypted:
+            # Encrypted P25 voice decodes to noise: nothing for Whisper. The
+            # call is still reported so C2 sees the activity.
+            wav.unlink(missing_ok=True)
+            end["audio_skipped"] = "encrypted"
+            self.encrypted_skipped += 1
+        elif settings.c2_url:
+            await asyncio.sleep(CALL_START_LEAD_SECONDS)
+            node_cfg = load_node_config()
+            url = await call_recorder.upload_recording(
+                wav, call.call_id,
+                talkgroup_id=call.tgid, talkgroup_name=call.tgid_name or None,
+                system_id=node_cfg.assigned_system_id,
+                audio_start_epoch=call.start_epoch, audio_end_epoch=call.audio_end_epoch(seconds),
+                extra=call.upload_extra(),
+            )
+            if url:
+                end["audio_url"] = url
+            else:
+                kind = (call_recorder.spool.last_result or {}).get("kind")
+                end["audio_upload"] = {"retry": "spooled", "lost": "lost"}.get(kind, "rejected")
+        else:
+            wav.unlink(missing_ok=True)
+
+        await mqtt_manager.publish_metadata("call_end", end)
+        json_path.unlink(missing_ok=True)
+        self.calls_ingested += 1
+        self.last_call_epoch = time.time()
+        self.last_call = {"call_id": call.call_id, "tgid": call.tgid, "tgid_name": call.tgid_name,
+                          "units": call.units, "duration": round(call.stop_epoch - call.start_epoch, 1),
+                          "encrypted": call.encrypted, "emergency": call.emergency}
+        logger.info(
+            f"TR call {call.call_id}: TG {call.tgid} {call.tgid_name!r}, "
+            f"{call.stop_epoch - call.start_epoch:.1f}s, units {call.units or '-'}"
+            + (" [EMERGENCY]" if call.emergency else "") + (" [encrypted]" if call.encrypted else "")
+        )
+
+
+class LiveAudioBridge:
+    """TR simplestream (UDP, s16le mono, 4-byte little-endian TGID prefix) ->
+    `pacat` into drb_sink, so Discord keeps reading drb_sink.monitor unchanged.
+
+    TR records every talkgroup at once and streams them all to one port;
+    interleaving two calls' packets would garble both. So this follows ONE
+    talkgroup — the first heard — until it has been quiet for HOLD_SECONDS,
+    like the one-voice-channel scanner OP25 was. Recording is unaffected:
+    every call is still recorded and uploaded in full.
+    """
+
+    HOLD_SECONDS = 2.0
+    RESPAWN_SECONDS = 5.0
+    MAX_BUFFERED_BYTES = 64 * 1024  # ~4 s at 8 kHz; beyond this pacat is stuck, drop
+
+    def __init__(self):
+        self.sample_rate = 8000
+        self.current_tgid: Optional[int] = None
+        self.last_packet_epoch: Optional[float] = None
+        self._last_current_epoch = 0.0
+        self.packets = 0
+        self.dropped = 0
+        self._transport = None
+        self._proc: Optional[asyncio.subprocess.Process] = None
+        self._proc_rate: Optional[int] = None
+        self._spawn_after = 0.0
+        self._spawning = False
+
+    def stats(self) -> Dict[str, Any]:
+        return {"current_tgid": self.current_tgid, "last_packet_epoch": self.last_packet_epoch,
+                "packets": self.packets, "dropped": self.dropped,
+                "pacat_running": self._proc is not None and self._proc.returncode is None}
+
+    async def start(self) -> None:
+        loop = asyncio.get_running_loop()
+        bridge = self
+
+        class _Proto(asyncio.DatagramProtocol):
+            def datagram_received(self, data, addr):
+                bridge.feed(data)
+
+        try:
+            self._transport, _ = await loop.create_datagram_endpoint(
+                _Proto, local_addr=("127.0.0.1", settings.tr_live_port),
+            )
+            logger.info(f"Trunk Recorder live audio: listening on 127.0.0.1:{settings.tr_live_port}.")
+        except OSError as e:
+            logger.error(f"Trunk Recorder live audio disabled — can't bind UDP {settings.tr_live_port}: {e}")
+
+    async def stop(self) -> None:
+        if self._transport:
+            self._transport.close()
+            self._transport = None
+        if self._proc and self._proc.returncode is None:
+            self._proc.terminate()
+        self._proc = None
+
+    def select(self, tgid: int, now: float) -> bool:
+        """Follow-one-talkgroup policy. True if this packet should be played."""
+        if self.current_tgid is None or now - self._last_current_epoch > self.HOLD_SECONDS:
+            self.current_tgid = tgid
+        if tgid != self.current_tgid:
+            return False
+        self._last_current_epoch = now
+        return True
+
+    def feed(self, data: bytes, now: Optional[float] = None) -> Optional[bytes]:
+        """One datagram. Returns the PCM that was (or would be) played."""
+        if len(data) <= 4:
+            return None
+        now = time.time() if now is None else now
+        tgid = int.from_bytes(data[:4], "little")
+        self.packets += 1
+        self.last_packet_epoch = now
+        if not self.select(tgid, now):
+            return None
+        pcm = data[4:]
+        self._write(pcm)
+        return pcm
+
+    def _write(self, pcm: bytes) -> None:
+        proc = self._proc
+        if proc is None or proc.returncode is not None:
+            self._ensure_pacat()
+            return
+        if self._proc_rate != self.sample_rate:
+            # A config change, not a failure: no respawn throttle.
+            self._spawn_after = 0.0
+            self._ensure_pacat()
+            return
+        stdin = proc.stdin
+        if stdin is None or stdin.is_closing():
+            return
+        try:
+            if stdin.transport.get_write_buffer_size() > self.MAX_BUFFERED_BYTES:
+                self.dropped += 1
+                return
+            stdin.write(pcm)
+        except (BrokenPipeError, ConnectionResetError, RuntimeError):
+            self._proc = None  # pacat died; respawned on the next packet
+
+    def _ensure_pacat(self) -> None:
+        if self._spawning or time.time() < self._spawn_after:
+            return
+        self._spawning = True
+        self._spawn_after = time.time() + self.RESPAWN_SECONDS
+        try:
+            asyncio.get_running_loop().create_task(self._spawn())
+        except RuntimeError:
+            self._spawning = False
+
+    async def _spawn(self) -> None:
+        try:
+            if self._proc and self._proc.returncode is None:
+                self._proc.terminate()
+            sink = settings.pulse_source.removesuffix(".monitor")
+            self._proc = await asyncio.create_subprocess_exec(
+                "pacat", "--playback", "--raw", "--format=s16le", f"--rate={self.sample_rate}",
+                "--channels=1", f"--device={sink}", "--latency-msec=250", "--client-name=drb-trunk-recorder",
+                stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.DEVNULL,
+                stderr=asyncio.subprocess.DEVNULL,
+            )
+            self._proc_rate = self.sample_rate
+        except Exception as e:
+            logger.error(f"Trunk Recorder live audio: pacat failed to start ({e}); retrying.")
+            self._proc = None
+        finally:
+            self._spawning = False
+
+
+trunk_recorder = TrunkRecorder()

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

@@ -0,0 +1,433 @@
+"""
+On-disk spool for recordings whose upload to C2 failed (node-26#44).
+
+Before this, `upload_recording` deleted the file in a `finally` whether or not
+the POST succeeded, so a C2 outage, a network blip, a 5xx or a timeout lost the
+call's audio for good — and with it the transcript and the incident link.
+
+Now a recording is deleted only after a 2xx that carries a storage URL. A
+failure that can succeed later is MOVED into this spool with a JSON sidecar
+holding everything needed to re-POST it, and a background task retries with
+exponential backoff. The spool is drained on startup, so a reboot or container
+restart does not lose it either.
+
+Failure classes (see `classify`):
+
+  retry    network error, timeout, 5xx, 2xx with no URL (C2 accepted the
+           request but storage failed), 401 (a reissued key is on its way
+           over MQTT/key pickup), 403 (not approved yet), 408, 429
+  reject   any other 4xx — 400 empty file, 413 too large.
+           Retrying the same bytes cannot change the answer, so the file is
+           deleted with an ERROR log naming the status.
+
+BOUNDED on purpose: /recordings is a bind mount of the host disk, and a full
+disk already stopped node-002 recording for 17 days (node-26#41). Entries older
+than `upload_spool_max_age_hours` are dropped, and past
+`upload_spool_max_bytes` the OLDEST entries go first — the newest audio is the
+most likely to still matter to a live incident.
+
+File layout, one pair per call:
+
+    <spool>/<recording name>.flac
+    <spool>/<recording name>.flac.json     sidecar (written FIRST, atomically)
+
+The sidecar is written before the audio is moved in, and deleted after the
+audio is removed, so a crash at any point leaves at worst a sidecar with no
+audio — which the next scan cleans up — and never audio that cannot be
+re-uploaded.
+"""
+import asyncio
+import json
+import os
+import shutil
+import time
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple
+
+import httpx
+
+from app.config import settings
+from app.internal import credentials
+from app.internal.logger import logger
+
+SIDECAR_SUFFIX = ".json"
+UPLOAD_TIMEOUT_SECONDS = 120.0
+# Statuses in the 4xx range that are worth retrying unchanged.
+# 403 = node not approved yet: approval can arrive later, and the spool caps
+# bound what a never-approved node keeps.
+RETRYABLE_4XX = {401, 403, 408, 429}
+
+OK, RETRY, REJECT = "ok", "retry", "reject"
+
+
+@dataclass
+class UploadMeta:
+    """Everything /upload needs besides the file itself."""
+
+    call_id: str
+    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
+    # Further form fields, sent as-is (Trunk Recorder calls carry units,
+    # emergency, encryption — node-26#45). Kept in the sidecar like the rest.
+    extra: Optional[Dict[str, str]] = None
+
+
+# By extension, so a spooled file is re-sent with the type it was first sent
+# with — the PulseAudio recorder writes FLAC, Trunk Recorder WAV.
+AUDIO_MIME_BY_SUFFIX = {".flac": "audio/flac", ".wav": "audio/wav", ".mp3": "audio/mpeg", ".m4a": "audio/mp4"}
+
+
+def mime_for(path: Path, default: str = "application/octet-stream") -> str:
+    return AUDIO_MIME_BY_SUFFIX.get(path.suffix.lower(), default)
+
+
+@dataclass
+class UploadResult:
+    kind: str                     # OK | RETRY | REJECT
+    url: Optional[str] = None
+    status: Optional[int] = None  # HTTP status, None for a network error
+    detail: str = ""
+
+
+def classify(status: int, url: Optional[str]) -> str:
+    if 200 <= status < 300:
+        # C2 answers 200 {"url": null} when its GCS write failed: nothing was
+        # stored, so this is a server-side failure, not a success.
+        return OK if url else RETRY
+    if status in RETRYABLE_4XX or status >= 500:
+        return RETRY
+    if 400 <= status < 500:
+        return REJECT
+    return RETRY
+
+
+def _form(meta: UploadMeta) -> Dict[str, str]:
+    form: Dict[str, str] = {"call_id": meta.call_id, "node_id": settings.node_id}
+    if meta.talkgroup_id is not None:
+        form["talkgroup_id"] = str(meta.talkgroup_id)
+    if meta.talkgroup_name:
+        form["talkgroup_name"] = meta.talkgroup_name
+    if meta.system_id:
+        form["system_id"] = meta.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 meta.audio_start_epoch is not None:
+        form["audio_start_epoch"] = f"{meta.audio_start_epoch:.3f}"
+    if meta.audio_end_epoch is not None:
+        form["audio_end_epoch"] = f"{meta.audio_end_epoch:.3f}"
+    for key, value in (meta.extra or {}).items():
+        if value is not None and key not in form:
+            form[key] = str(value)
+    return form
+
+
+async def post_recording(
+    file_path: Path, meta: UploadMeta, mime: str, timeout: float = UPLOAD_TIMEOUT_SECONDS,
+) -> UploadResult:
+    """One POST to C2 /upload. Never raises; never touches the file on disk."""
+    upload_url = f"{settings.c2_url}/upload"
+    api_key = credentials.get_api_key()
+    headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
+    try:
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            with open(file_path, "rb") as f:
+                r = await client.post(
+                    upload_url,
+                    files={"file": (file_path.name, f, mime)},
+                    data=_form(meta),
+                    headers=headers,
+                )
+    except Exception as e:
+        return UploadResult(RETRY, detail=f"{type(e).__name__}: {e}")
+
+    url: Optional[str] = None
+    if 200 <= r.status_code < 300:
+        try:
+            url = r.json().get("url")
+        except Exception:
+            url = None
+    kind = classify(r.status_code, url)
+    detail = f"HTTP {r.status_code}"
+    if kind != OK:
+        detail += f": {r.text[:200]}"
+    return UploadResult(kind, url=url, status=r.status_code, detail=detail)
+
+
+def _fsync_dir(directory: Path) -> None:
+    try:
+        fd = os.open(directory, os.O_RDONLY)
+    except OSError:
+        return
+    try:
+        os.fsync(fd)
+    except OSError:
+        pass
+    finally:
+        os.close(fd)
+
+
+Uploader = Callable[[Path, UploadMeta], Awaitable[UploadResult]]
+
+
+class UploadSpool:
+    def __init__(
+        self,
+        directory: Path,
+        uploader: Uploader,
+        max_bytes: int,
+        max_age_seconds: float,
+        backoff_min: float,
+        backoff_max: float,
+        poll_seconds: float = 5.0,
+        clock: Callable[[], float] = time.time,
+    ):
+        self.directory = Path(directory)
+        self._upload = uploader
+        self.max_bytes = max_bytes
+        self.max_age_seconds = max_age_seconds
+        self.backoff_min = backoff_min
+        self.backoff_max = backoff_max
+        self.poll_seconds = poll_seconds
+        self._clock = clock
+        self._task: Optional[asyncio.Task] = None
+        self._wake = asyncio.Event()
+        self.last_result: Optional[Dict[str, Any]] = None
+
+    # ------------------------------------------------------------------
+    # Entries
+    # ------------------------------------------------------------------
+
+    def _backoff(self, attempts: int) -> float:
+        return min(self.backoff_min * (2 ** max(0, attempts - 1)), self.backoff_max)
+
+    @staticmethod
+    def _write_sidecar(path: Path, data: Dict[str, Any]) -> None:
+        # fsync file AND directory: on a Pi that loses power, an un-synced
+        # rename can come back as a zero-length sidecar after the audio move
+        # was already journaled.
+        tmp = path.with_name(path.name + ".tmp")
+        with open(tmp, "w") as f:
+            f.write(json.dumps(data))
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(tmp, path)
+        _fsync_dir(path.parent)
+
+    def _recover_sidecar(self, audio: Path, sidecar: Path, why: str) -> Optional[Dict[str, Any]]:
+        """Rebuild minimal metadata from the recording's own name
+        (`YYYYMMDD_HHMMSS_<call_id>.flac`) rather than delete good audio. C2
+        already has the talkgroup from call_start; only call_id is required."""
+        parts = audio.name.split("_", 2)
+        call_id = parts[2].split(".", 1)[0] if len(parts) == 3 else ""
+        if not call_id:
+            logger.error(f"Spooled audio {audio.name}: {why}, and no call_id in its name — dropping it.")
+            self._remove(audio, sidecar)
+            return None
+        logger.error(f"Spooled audio {audio.name}: {why} — rebuilt metadata from the filename (call {call_id}).")
+        try:
+            spooled_at = audio.stat().st_mtime
+        except OSError:
+            spooled_at = self._clock()
+        data = {**asdict(UploadMeta(call_id=call_id)), "spooled_at": spooled_at, "attempts": 1,
+                "next_attempt_at": self._clock(), "last_error": why, "last_status": None}
+        self._write_sidecar(sidecar, data)
+        return data
+
+    def add(self, file_path: Path, meta: UploadMeta, result: UploadResult) -> Optional[Path]:
+        """Move a recording whose upload failed into the spool. Returns its new path."""
+        try:
+            self.directory.mkdir(parents=True, exist_ok=True)
+            dest = self.directory / file_path.name
+            sidecar = dest.with_name(dest.name + SIDECAR_SUFFIX)
+            now = self._clock()
+            self._write_sidecar(sidecar, {
+                **asdict(meta),
+                "spooled_at": now,
+                "attempts": 1,
+                "next_attempt_at": now + self._backoff(1),
+                "last_error": result.detail,
+                "last_status": result.status,
+            })
+            shutil.move(str(file_path), str(dest))
+            _fsync_dir(self.directory)
+        except Exception as e:
+            logger.error(f"Could not spool recording for call {meta.call_id} ({e}) — audio is lost.")
+            return None
+        logger.warning(
+            f"Upload of call {meta.call_id} failed ({result.detail}) — spooled for retry "
+            f"({self.stats()['files']} waiting)."
+        )
+        self._wake.set()
+        return dest
+
+    def _entries(self) -> List[Tuple[Path, Path, Dict[str, Any]]]:
+        """(audio, sidecar, data) oldest first. Cleans up half-written pairs."""
+        if not self.directory.is_dir():
+            return []
+        entries = []
+        sidecars = set()
+        for sidecar in self.directory.glob(f"*{SIDECAR_SUFFIX}"):
+            audio = sidecar.with_name(sidecar.name[: -len(SIDECAR_SUFFIX)])
+            sidecars.add(audio.name)
+            if not audio.exists():
+                sidecar.unlink(missing_ok=True)
+                continue
+            try:
+                data = json.loads(sidecar.read_text())
+                if not data.get("call_id"):
+                    raise ValueError("no call_id")
+            except Exception as e:
+                data = self._recover_sidecar(audio, sidecar, f"unreadable sidecar ({e})")
+                if data is None:
+                    continue
+            entries.append((audio, sidecar, data))
+        for audio in list(self.directory.iterdir()):
+            if audio.is_file() and not audio.name.endswith((SIDECAR_SUFFIX, ".tmp")) and audio.name not in sidecars:
+                sidecar = audio.with_name(audio.name + SIDECAR_SUFFIX)
+                data = self._recover_sidecar(audio, sidecar, "no metadata sidecar")
+                if data is not None:
+                    entries.append((audio, sidecar, data))
+        entries.sort(key=lambda e: (e[2].get("spooled_at", 0.0), e[0].name))
+        return entries
+
+    @staticmethod
+    def _remove(audio: Path, sidecar: Path) -> None:
+        # Audio first: a crash in between leaves a sidecar, which _entries() cleans.
+        audio.unlink(missing_ok=True)
+        sidecar.unlink(missing_ok=True)
+
+    def stats(self) -> Dict[str, Any]:
+        entries = self._entries()
+        size = 0
+        for audio, _, _ in entries:
+            try:
+                size += audio.stat().st_size
+            except OSError:
+                pass
+        oldest = entries[0][2].get("spooled_at") if entries else None
+        return {
+            "files": len(entries),
+            "bytes": size,
+            "oldest_age_seconds": round(self._clock() - oldest, 1) if oldest else None,
+            "last_result": self.last_result,
+        }
+
+    def enforce_caps(self) -> int:
+        """Drop expired entries, then the oldest until under max_bytes. Returns how many went."""
+        entries = self._entries()
+        now = self._clock()
+        dropped = 0
+        kept = []
+        for audio, sidecar, data in entries:
+            if now - data.get("spooled_at", now) > self.max_age_seconds:
+                logger.error(
+                    f"Spooled call {data['call_id']} was never uploaded within "
+                    f"{self.max_age_seconds / 3600:.0f}h ({data.get('attempts')} attempts, "
+                    f"last: {data.get('last_error')}) — deleting its audio."
+                )
+                self._remove(audio, sidecar)
+                dropped += 1
+            else:
+                kept.append((audio, sidecar, data))
+        sizes = []
+        for audio, sidecar, data in kept:
+            try:
+                sizes.append(audio.stat().st_size)
+            except OSError:
+                sizes.append(0)
+        total = sum(sizes)
+        i = 0
+        while total > self.max_bytes and i < len(kept):
+            audio, sidecar, data = kept[i]
+            logger.error(
+                f"Upload spool over {self.max_bytes} bytes — evicting the oldest, call {data['call_id']}."
+            )
+            self._remove(audio, sidecar)
+            total -= sizes[i]
+            dropped += 1
+            i += 1
+        return dropped
+
+    # ------------------------------------------------------------------
+    # Retry
+    # ------------------------------------------------------------------
+
+    async def drain_once(self, force: bool = False) -> int:
+        """Attempt every due entry (every entry if `force`). Returns how many were uploaded."""
+        self.enforce_caps()
+        uploaded = 0
+        for audio, sidecar, data in self._entries():
+            if not force and data.get("next_attempt_at", 0.0) > self._clock():
+                continue
+            meta = UploadMeta(**{k: data.get(k) for k in UploadMeta.__dataclass_fields__})
+            result = await self._upload(audio, meta)
+            self.last_result = {"call_id": meta.call_id, "kind": result.kind, "status": result.status,
+                                "detail": result.detail, "epoch": self._clock()}
+            if result.kind == OK:
+                logger.info(f"Spooled call {meta.call_id} uploaded after {data.get('attempts')} failed attempt(s).")
+                self._remove(audio, sidecar)
+                uploaded += 1
+            elif result.kind == REJECT:
+                logger.error(
+                    f"C2 rejected spooled call {meta.call_id} ({result.detail}) — not retryable, deleting it."
+                )
+                self._remove(audio, sidecar)
+            else:
+                attempts = int(data.get("attempts", 1)) + 1
+                data.update(attempts=attempts, last_error=result.detail, last_status=result.status,
+                            next_attempt_at=self._clock() + self._backoff(attempts))
+                self._write_sidecar(sidecar, data)
+                logger.warning(
+                    f"Retry {attempts} of spooled call {meta.call_id} failed ({result.detail}); next in "
+                    f"{self._backoff(attempts):.0f}s."
+                )
+                # C2 is still down: the rest would fail the same way. Leave them
+                # for their own timers rather than hammering it in one pass.
+                break
+        return uploaded
+
+    def wake(self) -> None:
+        """Retry everything now (e.g. a new API key just arrived)."""
+        for audio, sidecar, data in self._entries():
+            if data.get("next_attempt_at", 0.0) > self._clock():
+                data["next_attempt_at"] = self._clock()
+                self._write_sidecar(sidecar, data)
+        self._wake.set()
+
+    async def _run(self) -> None:
+        # Startup drain: whatever survived a restart is attempted immediately.
+        force = True
+        while True:
+            try:
+                await self.drain_once(force=force)
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                logger.error(f"Upload spool pass failed: {e}")
+            force = False
+            self._wake.clear()
+            try:
+                await asyncio.wait_for(self._wake.wait(), timeout=self.poll_seconds)
+            except asyncio.TimeoutError:
+                pass
+
+    def start(self) -> None:
+        if self._task is None or self._task.done():
+            pending = len(self._entries())
+            if pending:
+                logger.info(f"Upload spool has {pending} recording(s) from before this start — draining.")
+            self._task = asyncio.create_task(self._run())
+
+    async def stop(self) -> None:
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+            self._task = None

drb-edge-node/app/main.py

@@ -111,8 +111,20 @@ async def on_call_end(data: dict):
         )
         if audio_url:
             data["audio_url"] = audio_url
+        elif settings.c2_url:
+            last = call_recorder.spool.last_result or {}
+            if last.get("kind") == "lost":
+                data["audio_upload"] = "lost"
+            elif last.get("kind") == "retry":
+                # The file is in the upload spool and will be retried; C2 attaches
+                # it to this call_id whenever it lands (node-26#44).
+                data["audio_upload"] = "spooled"
             else:
-            logger.error(f"Audio upload failed for call {data['call_id']}. Verify C2_URL and Node API Key.")
+                data["audio_upload"] = "rejected"
+                logger.error(
+                    f"Audio upload for call {data['call_id']} was rejected ({last.get('detail')}). "
+                    "C2 refused this recording (malformed request or file too large); it was not kept."
+                )
     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.
@@ -156,10 +168,12 @@ async def on_command(payload: dict):
         except ValueError as e:
             logger.error(f"Rejected SDR settings from C2: {e}")
     elif action == "op25_restart":
-        from app.internal.op25_client import op25_client
+        # Restarts whichever recorder is active (node-26#45): C2 still calls
-        await op25_client.stop()
+        # it "Restart OP25".
+        from app.internal import radio_backend
+        await radio_backend.stop()
         await asyncio.sleep(2)
-        await op25_client.start()
+        await radio_backend.start()
     elif action == "node_update":
         # node-26#15: the container cannot (and must not) update itself. It
         # drops a sentinel in /configs; the host's drb-update.path unit
@@ -179,6 +193,8 @@ async def on_api_key(payload: dict):
     if key:
         credentials.save_api_key(key)
         mqtt_manager.update_api_key(key)
+        # Uploads spooled on a 401 can go now rather than at their next backoff.
+        call_recorder.spool.wake()
         logger.info("Node API key received and saved.")
 
 
@@ -187,7 +203,14 @@ async def _generate_op25_config(config: SystemConfig) -> bool:
 
     Trunked / conventional / analog are told apart in radio_config (node-26#24);
     a system saved before that has no `mode` and is trunked, as it always was.
+
+    With RECORDER_BACKEND=trunk-recorder this writes Trunk Recorder's
+    config.json instead (node-26#45) — the one place every caller (C2 push,
+    dashboard, startup resume) goes through.
     """
+    from app.internal import trunk_recorder as tr
+    if tr.enabled():
+        return await tr.trunk_recorder.apply_config(config)
     from app.internal.op25_client import op25_client
     from app.internal.radio_config import build_generate_payload
     node_cfg = load_node_config()
@@ -250,19 +273,20 @@ async def on_config_push(payload: dict):
             logger.error(f"Ignoring invalid sdr_pins in config push: {e}")
     save_node_config(node_cfg)
 
-    from app.internal.op25_client import op25_client
+    from app.internal import radio_backend
     if not await _generate_op25_config(config):
-        logger.error(f"Failed to generate OP25 config for {config.name}")
+        logger.error(f"Failed to generate {radio_backend.name()} config for {config.name}")
         return
 
     # OP25's (pinned) dongle may be one a decoder holds right now: free the
     # spares, restart OP25 on its own dongle, then hand the rest back out.
+    # (Trunk Recorder takes the same dongle — node-26#45.)
     from app.internal.secondary_sdr_client import secondary_sdr_client
     from app.internal.sdr_settings import apply_secondaries
     await secondary_sdr_client.apply([], {}, [])
-    await op25_client.stop()
+    await radio_backend.stop()
     await asyncio.sleep(2)
-    await op25_client.start()
+    await radio_backend.start()
     logger.info(f"Config push applied: {config.name}")
     await apply_secondaries()
 
@@ -295,6 +319,19 @@ async def lifespan(app: FastAPI):
 
     # Start services (radio_bot starts when C2's retained discord/desired arrives)
     await mqtt_manager.connect()
+    from app.internal import radio_backend
+    from app.internal import trunk_recorder as tr
+    if tr.enabled():
+        # node-26#45: Trunk Recorder records; the OP25 watcher and the
+        # PulseAudio recorder stay off so no call is uploaded twice. Only the
+        # upload spool runs. OP25 is stopped in case it was running before the
+        # backend was switched — it would hold the SDR Trunk Recorder needs.
+        logger.info("Recorder backend: Trunk Recorder (OP25 and the PulseAudio recorder are off).")
+        from app.internal.op25_client import op25_client
+        await op25_client.stop()
+        await call_recorder.start(capture=False)
+        await tr.trunk_recorder.start()
+    else:
         await metadata_watcher.start()
         await call_recorder.start()  # persistent PulseAudio ring buffer
 
@@ -316,13 +353,14 @@ async def lifespan(app: FastAPI):
     # not-yet-approved node listening to a system chosen on the local
     # dashboard (an override) must come back on it after a reboot too.
     if active_config:
-        from app.internal.op25_client import op25_client
+        logger.info(f"Node has a system to listen to — generating the {radio_backend.name()} config.")
-        logger.info("Node has a system to listen to — waiting for OP25 API then generating config.")
         for attempt in range(10):
             if await _generate_op25_config(active_config):
-                await op25_client.start()
+                # No wait at startup: the trunk-recorder container may still be
+                # coming up, and the app must start serving regardless.
+                await radio_backend.start(wait=False)
                 break
-            logger.warning(f"OP25 not ready yet (attempt {attempt + 1}/10), retrying in 3s…")
+            logger.warning(f"{radio_backend.name()} not ready yet (attempt {attempt + 1}/10), retrying in 3s…")
             await asyncio.sleep(3)
 
     # After op25 has claimed its dongle, so the decoders only get the spares.
@@ -341,6 +379,7 @@ async def lifespan(app: FastAPI):
     heartbeat_task.cancel()
     telemetry_task.cancel()
     await metadata_watcher.stop()
+    await tr.trunk_recorder.stop()
     await call_recorder.stop()
     await radio_bot.stop()
     await mqtt_manager.disconnect()

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

@@ -82,6 +82,9 @@ async def get_status():
         # it was discarded rather than uploaded. Non-zero means either the
         # console is not decoding or something else is feeding drb_sink.
         "unattributed_segments": metadata_watcher.unattributed_segments,
+        # Recordings whose upload to C2 failed and are waiting for a retry
+        # (node-26#44), plus the outcome of the most recent upload attempt.
+        "upload_spool": call_recorder.spool.stats(),
         "active_tgid": active_tgid,
         "active_tgid_name": active_tgid_name,
         "active_call_id": metadata_watcher.active_call_id,
@@ -94,9 +97,18 @@ async def get_status():
         "icecast_port": settings.icecast_port,
         "icecast_mount": settings.icecast_mount,
         "op25": op25_status,
+        # Which recorder is live (node-26#45) and, for Trunk Recorder, its
+        # supervisor state / decode rate / handed-over calls.
+        "recorder_backend": settings.recorder_backend,
+        "trunk_recorder": _tr_facts(),
     }
 
 
+def _tr_facts():
+    from app.internal import trunk_recorder as tr
+    return tr.trunk_recorder.health_facts() if tr.enabled() else None
+
+
 @router.get("/health")
 async def get_health():
     """Plain-language setup/status checklist for the dashboard (see internal/health.py)."""
@@ -107,7 +119,9 @@ async def get_health():
 
 @router.post("/op25/start")
 async def start_op25():
-    err = await op25_client.start_with_reason()
+    # The active recorder (node-26#45) — the dashboard's Start button.
+    from app.internal import radio_backend
+    err = await radio_backend.start_with_reason()
     if err:
         raise HTTPException(502, err)
     return {"ok": True}
@@ -115,14 +129,18 @@ async def start_op25():
 
 @router.post("/op25/stop")
 async def stop_op25():
-    ok = await op25_client.stop()
+    from app.internal import radio_backend
+    ok = await radio_backend.stop()
     if not ok:
-        raise HTTPException(502, "The op25 container did not answer the stop request.")
+        raise HTTPException(502, f"{radio_backend.name()} did not accept the stop request.")
     return {"ok": True}
 
 
 @router.get("/op25/log")
 async def op25_log(lines: int = 80):
+    from app.internal import trunk_recorder as tr
+    if tr.enabled():
+        return {"lines": tr.trunk_recorder.log_tail(lines)}
     out = await op25_client.log(lines)
     if out is None:
         raise HTTPException(502, "The op25 container isn't answering.")
@@ -135,17 +153,22 @@ async def get_config():
 
 
 async def _restart_op25_with(config: SystemConfig) -> None:
-    """Generate OP25's config for `config` and restart OP25 on it."""
+    """Generate the active recorder's config for `config` and restart it on it."""
     from app.main import _generate_op25_config
+    from app.internal import radio_backend
     if not await _generate_op25_config(config):
+        from app.internal import trunk_recorder as tr
+        if tr.enabled():
+            detail = tr.trunk_recorder.config_error or f"Trunk Recorder config for {config.name} failed."
+            raise HTTPException(502, detail)
         raise HTTPException(
             502, f"The op25 container did not accept the config for {config.name} — is it running?",
         )
-    await op25_client.stop()
+    await radio_backend.stop()
     await asyncio.sleep(2)
-    err = await op25_client.start_with_reason()
+    err = await radio_backend.start_with_reason()
     if err:
-        raise HTTPException(502, f"Config saved, but OP25 did not start: {err}")
+        raise HTTPException(502, f"Config saved, but {radio_backend.name()} did not start: {err}")
 
 
 async def _apply_override(config: SystemConfig) -> None:
@@ -205,6 +228,12 @@ class ListenBody(BaseModel):
     frequencies: List[float] = []
     talkgroups: List[Talkgroup] = []       # p25 only; empty = every talkgroup
     nac: Optional[str] = None              # p25conv only, hex; empty = any NAC
+    # analog only (node-26#46): OP25 squelch. noise (default) needs no
+    # calibration; squelch_db = dB of quieting to open (OP25 default 8);
+    # squelch_threshold = absolute dB for power mode (OP25 default -60).
+    squelch_mode: Optional[Literal["noise", "voice", "power"]] = None
+    squelch_db: Optional[float] = None
+    squelch_threshold: Optional[float] = None
 
 
 # RTL-SDR (R820T/R828D) tuning range.
@@ -239,14 +268,21 @@ async def listen(body: ListenBody):
         else:
             is_p25 = body.mode == "p25"
             label = (body.name or "").strip() or (f"P25 {first} MHz" if is_p25 else f"Analog {first} MHz")
+            sys_config = {
+                "control_channels": freqs,
+                "talkgroups": [tg.model_dump() for tg in body.talkgroups] if is_p25 else [],
+            }
+            if not is_p25:
+                sys_config["squelch_mode"] = body.squelch_mode or "noise"
+                if body.squelch_db is not None:
+                    sys_config["squelch_db"] = body.squelch_db
+                if body.squelch_threshold is not None:
+                    sys_config["squelch_threshold"] = body.squelch_threshold
             config = SystemConfig(
                 system_id=f"local-{body.mode}",
                 name=label,
                 type="P25" if is_p25 else "NBFM",
-                config={
+                config=sys_config,
-                    "control_channels": freqs,
-                    "talkgroups": [tg.model_dump() for tg in body.talkgroups] if is_p25 else [],
-                },
             )
     await _apply_override(config)
     return {"ok": True, "name": config.name}
@@ -334,7 +370,8 @@ async def revert_config():
     else:
         # Nothing from C2 to go back to: stop rather than keep decoding a
         # system the dashboard now says is cleared.
-        await op25_client.stop()
+        from app.internal import radio_backend
+        await radio_backend.stop()
 
     if mqtt_manager.is_connected:
         await mqtt_manager.publish_checkin()

drb-edge-node/app/templates/index.html

@@ -142,6 +142,7 @@
     <audio id="player" controls preload="none">Your browser can't play the stream — use the direct link.</audio>
     <dl class="kv">
       <dt>Listening to</dt><dd id="now-system">—</dd>
+      <dt>Recorder</dt><dd id="now-backend">—</dd>
       <dt>Talkgroup</dt><dd id="now-tg">—</dd>
       <dt>Last heard</dt><dd id="now-heard">—</dd>
       <dt>Stream link</dt><dd><a id="stream-link" class="mono" href="#" target="_blank" rel="noopener">—</a></dd>
@@ -216,6 +217,20 @@
         <label for="an-name">Name <span class="hint">(optional)</span></label>
         <input type="text" id="an-name" placeholder="Fire dispatch">
       </div>
+      <div class="field">
+        <label for="an-sq-mode">Squelch</label>
+        <select id="an-sq-mode">
+          <option value="noise" selected>Noise (recommended)</option>
+          <option value="voice">Voice (noise + speech detector)</option>
+          <option value="power">Power (old, needs calibration)</option>
+        </select>
+        <span class="hint">Noise squelch opens on a real carrier and needs no tuning per site.</span>
+      </div>
+      <div class="field">
+        <label for="an-sq-level">Squelch level (dB) <span class="hint">(optional)</span></label>
+        <input type="text" id="an-sq-level" class="mono" placeholder="8" inputmode="decimal">
+        <span class="hint" id="an-sq-hint">Noise/voice: dB of quieting to open — 8 default, 12-15 for strong signals only, ~5 for weak ones.</span>
+      </div>
     </div>
 
     <div class="pane" data-pane="system" hidden>
@@ -304,7 +319,8 @@
         ? `<span class="mono">TG ${esc(heard.tgid)}</span>${heard.name ? ' ' + esc(heard.name) : ''}` +
           ` · ${ago(heard.epoch)}${heard.freq ? ` · <span class="mono">${esc(mhz(heard.freq))}</span>` : ''}`
         : 'Nothing heard since the node started.';
-      const op = d.checks.find(c => c.id === 'op25');
+      // 'recorder' is the Trunk Recorder backend's check (node-26#45).
+      const op = d.checks.find(c => c.id === 'op25' || c.id === 'recorder');
       const src = d.checks.find(c => c.id === 'source');
       const running = op && op.state === 'ok';
       $('op25-start').disabled = running || (src && src.state !== 'ok');
@@ -326,6 +342,15 @@
       $('node-id').textContent = d.node_id || '';
       document.title = `${d.node_name || 'DRB Edge Node'} — DRB`;
       $('now-system').textContent = d.system_name ? `${d.system_name} (${d.system_kind || d.system_type})` : 'Nothing chosen yet';
+      const isTR = d.recorder_backend === 'trunk-recorder';
+      const trState = isTR && d.trunk_recorder && d.trunk_recorder.status ? d.trunk_recorder.status.state : null;
+      $('now-backend').textContent = isTR
+        ? `Trunk Recorder${trState ? ` (${trState.replace(/_/g, ' ')})` : ' (not running)'} — every talkgroup recorded at once`
+        : 'OP25 — one call at a time';
+      recorderName = isTR ? 'Trunk Recorder' : 'OP25';
+      $('op25-start').textContent = `Start ${recorderName}`;
+      $('op25-stop').textContent = `Stop ${recorderName}`;
+      if ($('op25-log').hidden) $('log-toggle').textContent = `Show ${recorderName} log`;
       $('now-tg').innerHTML = d.active_tgid
         ? `<span class="live">${esc(d.active_tgid_name || 'Talkgroup')}</span> <span class="mono">TG ${esc(d.active_tgid)}</span>`
         : 'No call right now';
@@ -360,7 +385,8 @@
     $('ack-btn').hidden = !(d.local_choice && d.has_c2_system && d.enforce_override_timeout);
   }
 
-  // ── OP25 controls + log ────────────────────────────────────────────────
+  // ── OP25 / Trunk Recorder controls + log ───────────────────────────────
+  let recorderName = 'OP25';
   async function op25(action) {
     $('op25-msg').className = 'msg';
     $('op25-msg').textContent = action === 'start' ? 'Starting…' : 'Stopping…';
@@ -377,25 +403,32 @@
     const pre = $('op25-log');
     try {
       const d = await api('/api/op25/log?lines=120');
-      pre.textContent = d.lines.length ? d.lines.join('\n') : 'OP25 has printed nothing since it last started.';
+      pre.textContent = d.lines.length ? d.lines.join('\n') : `${recorderName} has printed nothing since it last started.`;
     } catch (e) {
       pre.textContent = `Couldn't load the log: ${e.message}`;
     }
     pre.hidden = false;
-    $('log-toggle').textContent = 'Hide OP25 log';
+    $('log-toggle').textContent = `Hide ${recorderName} log`;
     $('log-toggle').setAttribute('aria-expanded', 'true');
   }
   function toggleLog() {
     const pre = $('op25-log');
     if (pre.hidden) return loadLog();
     pre.hidden = true;
-    $('log-toggle').textContent = 'Show OP25 log';
+    $('log-toggle').textContent = `Show ${recorderName} log`;
     $('log-toggle').setAttribute('aria-expanded', 'false');
   }
 
   // ── Listen locally ─────────────────────────────────────────────────────
   let systems = [];
   function mode() { return document.querySelector('input[name=mode]:checked').value; }
+  $('an-sq-mode').addEventListener('change', () => {
+    const power = $('an-sq-mode').value === 'power';
+    $('an-sq-level').placeholder = power ? '-60' : '8';
+    $('an-sq-hint').textContent = power
+      ? 'Power: absolute signal level in dB to open (-60 default); depends on this SDR, gain and antenna.'
+      : 'Noise/voice: dB of quieting to open — 8 default, 12-15 for strong signals only, ~5 for weak ones.';
+  });
   document.querySelectorAll('input[name=mode]').forEach(r => r.addEventListener('change', () => {
     document.querySelectorAll('.pane').forEach(p => { p.hidden = p.dataset.pane !== mode(); });
     $('listen-msg').textContent = '';
@@ -443,7 +476,12 @@
     } else {
       const freqs = parseFreqs($('an-freq').value);
       if (freqs.length !== 1 || isNaN(freqs[0])) { msg.className = 'msg err'; msg.textContent = 'Enter one frequency in MHz, e.g. 154.4300'; return; }
-      body = { mode: 'analog', frequencies: freqs, name: $('an-name').value };
+      const sqMode = $('an-sq-mode').value;
+      const sqRaw = $('an-sq-level').value.trim();
+      const sqLevel = sqRaw === '' ? null : Number(sqRaw);
+      if (sqLevel !== null && isNaN(sqLevel)) { msg.className = 'msg err'; msg.textContent = 'Squelch level must be a number of dB, e.g. 8 (or -60 for power mode).'; return; }
+      body = { mode: 'analog', frequencies: freqs, name: $('an-name').value, squelch_mode: sqMode };
+      if (sqLevel !== null) body[sqMode === 'power' ? 'squelch_threshold' : 'squelch_db'] = sqLevel;
     }
     $('listen-btn').disabled = true;
     msg.className = 'msg';

drb-edge-node/tests/test_conventional.py

@@ -78,6 +78,17 @@ def test_analog_payload_reads_channels_or_legacy(cfg):
     p = rc.build_generate_payload("NBFM", "Fire", cfg, ICE, "rtl-sdr-v3", None)
     assert p["channels"] == [154_430_000]
     assert "p25_mode" not in p
+    assert p["nbfm_squelch_mode"] == "noise"   # node-26#46 default
+
+
+def test_analog_squelch_is_plumbed_and_validated():
+    cfg = {"channels": [{"freq": 154.43}], "squelch_mode": "Power", "squelch_threshold": "-55", "squelch_db": 12}
+    p = rc.build_generate_payload("NBFM", "Fire", cfg, ICE, "rtl-sdr-v3", None)
+    assert p["nbfm_squelch_mode"] == "power" and p["nbfm_squelch"] == -55.0 and p["nbfm_noise_squelch_db"] == 12.0
+    with pytest.raises(ValueError):
+        rc.analog_squelch({"squelch_mode": "loud"})
+    with pytest.raises(ValueError):
+        rc.analog_squelch({"squelch_db": "lots"})
 
 
 def test_system_mode():

drb-edge-node/tests/test_discord_radio.py

@@ -263,6 +263,18 @@ async def test_local_pause_logs_out_until_next_desired(env):
     assert env.mgr.is_connected
 
 
+async def test_retained_redelivery_does_not_undo_local_pause(env):
+    """node-26#34: the broker redelivers the retained desired message (same
+    seq) on every reconnect; that must not log the bots back in."""
+    await env.mgr.apply_desired(_desired(4, (1, 11, "t1")))
+    await env.mgr.pause_locally()
+    await env.mgr.apply_desired(_desired(4, (1, 11, "t1")))
+    assert not env.mgr.is_connected
+    assert env.published[-1]["streams"]["1:11"]["state"] == "paused_locally"
+    await env.mgr.apply_desired(_desired(5, (1, 11, "t1")))
+    assert env.mgr.is_connected
+
+
 async def test_no_message_content_intent_on_real_client():
     client = discord_radio._default_client_factory()
     try:

drb-edge-node/tests/test_local_dashboard.py

@@ -100,6 +100,17 @@ def test_key_on_disk_but_mqtt_down_says_restart():
     assert "restart the edge-node container" in c["mqtt"]["action"]
 
 
+def test_mqtt_refused_vs_unreachable_are_told_apart():
+    """node-26#21: 'not authorised' and 'can't reach the broker' need different fixes."""
+    refused = _by_id(health.build_checks(_facts(
+        mqtt_connected=False, mqtt_last_error={"kind": "refused", "detail": "Not authorized", "epoch": 1.0})))
+    assert refused["mqtt"]["state"] == "error"
+    assert "refused" in refused["mqtt"]["summary"] and "install.sh" in refused["mqtt"]["action"]
+    unreachable = _by_id(health.build_checks(_facts(
+        mqtt_connected=False, mqtt_last_error={"kind": "unreachable", "detail": "timed out", "epoch": 1.0})))
+    assert "Can't reach" in unreachable["mqtt"]["summary"] and "firewall" in unreachable["mqtt"]["action"]
+
+
 def test_standalone_node_is_not_an_error():
     c = _by_id(health.build_checks(_facts(c2_url=None, has_api_key=False, mqtt_connected=False)))
     assert c["c2"]["state"] == "off"
@@ -218,6 +229,17 @@ def test_listen_analog_single_frequency(client, op25):
     assert r.status_code == 200, r.text
     cfg = config_manager.load_node_config().override_config
     assert (cfg.type, cfg.name, cfg.config["control_channels"]) == ("NBFM", "Fire", [154.43])
+    assert cfg.config["squelch_mode"] == "noise"   # node-26#46 default
+
+
+def test_listen_analog_squelch_is_saved(client, op25):
+    r = client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43],
+                                         "squelch_mode": "power", "squelch_threshold": -55})
+    assert r.status_code == 200, r.text
+    cfg = config_manager.load_node_config().override_config.config
+    assert (cfg["squelch_mode"], cfg["squelch_threshold"]) == ("power", -55)
+    assert client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43],
+                                            "squelch_mode": "loud"}).status_code == 422
 
 
 def test_listen_p25_conventional(client, op25):

drb-edge-node/tests/test_mqtt_manager.py

@@ -140,3 +140,41 @@ def test_build_client_does_not_touch_offline_buffer(monkeypatch):
 
     assert list(mqtt_manager._offline_buffer) == [("nodes/test/metadata", {"call_id": "sentinel"})]
     mqtt_manager._offline_buffer.clear()
+
+
+def test_disk_report_reads_free_and_total(tmp_path):
+    from app.internal.mqtt_manager import disk_report
+    rep = disk_report(str(tmp_path))
+    assert rep["disk_total_bytes"] > 0
+    assert 0 <= rep["disk_free_bytes"] <= rep["disk_total_bytes"]
+
+
+def test_disk_report_nulls_when_path_unreadable(tmp_path):
+    # node-26#41: nulls (not omitted keys) so C2 never keeps a stale reading.
+    from app.internal.mqtt_manager import disk_report
+    assert disk_report(str(tmp_path / "missing")) == {"disk_free_bytes": None, "disk_total_bytes": None}
+
+
+def test_last_error_records_refusal_and_clears_on_connect(monkeypatch):
+    """node-26#21: the refusal reason survives the disconnect that follows it."""
+    from unittest.mock import MagicMock
+    mqtt_manager._last_error = None
+    mqtt_manager._loop = None
+    mqtt_manager._on_connect(MagicMock(), None, None, 135, None)
+    mqtt_manager._on_disconnect(MagicMock(), None, None, 135, None)
+    assert mqtt_manager.last_error["kind"] == "refused"
+    monkeypatch.setattr(mqtt_manager, "_flush_offline_buffer", MagicMock())
+    monkeypatch.setattr(mqtt_manager, "_publish_checkin", MagicMock())
+    monkeypatch.setattr(mqtt_manager, "_maybe_request_key", MagicMock())
+    monkeypatch.setattr("asyncio.run_coroutine_threadsafe", lambda *a, **k: None)
+    mqtt_manager._on_connect(MagicMock(), None, None, 0, None)
+    assert mqtt_manager.last_error is None
+    mqtt_manager._connected = False
+
+
+def test_last_error_unreachable_from_connect_fail():
+    mqtt_manager._last_error = None
+    mqtt_manager._connected = False
+    mqtt_manager._on_connect_fail(None, None)
+    assert mqtt_manager.last_error["kind"] == "unreachable"
+    mqtt_manager._last_error = None

drb-edge-node/tests/test_trunk_recorder.py

@@ -0,0 +1,508 @@
+"""
+Trunk Recorder backend (node-26#45): config generation from the C2 system
+config, the TR call JSON -> MQTT + /upload mapping, the inbox hand-off, the
+live-audio follow policy and the health checklist.
+
+No radio here: nothing in this file has run against a real Trunk Recorder.
+The sample call JSON below is built from `create_call_json` in
+trunk-recorder/call_concluder/call_concluder.cc at tag v5.2.1 (key names and
+order as written there); the values are invented.
+"""
+import json
+from pathlib import Path
+
+import pytest
+
+from app.config import settings
+from app.internal import health, radio_backend, tr_config, upload_spool
+from app.internal import trunk_recorder as trmod
+from app.internal.conventional import channel_tgid
+from app.internal.tr_calls import parse_call
+from app.internal.upload_spool import UploadMeta
+
+# Shape: call_concluder.cc v5.2.1 create_call_json. Values invented.
+SAMPLE_CALL = {
+    "freq": 851012500,
+    "freq_error": -112,
+    "signal": -48,
+    "noise": -98,
+    "source_num": 0,
+    "recorder_num": 2,
+    "tdma_slot": 0,
+    "phase2_tdma": 0,
+    "start_time": 1793145600,
+    "stop_time": 1793145612,
+    "start_time_ms": 1793145600250,
+    "stop_time_ms": 1793145612730,
+    "emergency": 0,
+    "priority": 1,
+    "mode": 0,
+    "duplex": 0,
+    "encrypted": 0,
+    "call_length": 12,
+    "call_length_ms": 12480,
+    "talkgroup": 101,
+    "talkgroup_tag": "PD Dispatch",
+    "talkgroup_description": "Police Dispatch",
+    "talkgroup_group_tag": "Law Dispatch",
+    "talkgroup_group": "Police",
+    "color_code": -1,
+    "audio_type": "digital",
+    "short_name": "drb",
+    "freqList": [
+        {"freq": 851012500, "time": 1793145600, "pos": 0.0, "len": 5.2, "error_count": 3, "spike_count": 1},
+        {"freq": 851012500, "time": 1793145606, "pos": 5.2, "len": 7.28, "error_count": 2, "spike_count": 0},
+    ],
+    "srcList": [
+        {"src": 1234567, "time": 1793145600, "pos": 0.0, "emergency": 0, "signal_system": "", "tag": "Car 12",
+         "tag_ota": ""},
+        {"src": 7654321, "time": 1793145606, "pos": 5.2, "emergency": 1, "signal_system": "", "tag": "",
+         "tag_ota": ""},
+        {"src": 1234567, "time": 1793145610, "pos": 9.9, "emergency": 0, "signal_system": "", "tag": "Car 12",
+         "tag_ota": ""},
+        {"src": -1, "time": 1793145611, "pos": 11.0, "emergency": 0, "signal_system": "", "tag": "", "tag_ota": ""},
+    ],
+}
+
+TR_KW = dict(device_serial="00000101", ppm=1.5, sample_rate=2_400_000, gain=34.0, digital_recorders=4,
+             min_duration=1.0, squelch_db=-50.0, live_port=9123)
+
+
+# ---------------------------------------------------------------------------
+# Config generation
+# ---------------------------------------------------------------------------
+
+def test_trunked_config_from_c2_system():
+    cfg = {"control_channels": [851.0125, 851.2625], "talkgroups": [{"id": 101, "name": "PD Dispatch, Main"},
+                                                                    {"id": 202, "name": "Fire"}]}
+    tr, files = tr_config.build_tr_config("P25", "County", cfg, **TR_KW)
+    assert tr["ver"] == 2
+    src, sys_ = tr["sources"][0], tr["systems"][0]
+    assert src["driver"] == "osmosdr" and src["device"] == "rtl=00000101"
+    assert src["ppm"] == 1.5 and src["gain"] == 34.0 and src["rate"] == 2_400_000
+    assert src["digitalRecorders"] == 4
+    assert all(abs(f - src["center"]) <= 960_000 for f in (851_012_500, 851_262_500))
+    assert all(abs(f - src["center"]) >= tr_config.DC_SPIKE_GUARD_HZ for f in (851_012_500, 851_262_500))
+    assert sys_["type"] == "p25" and sys_["control_channels"] == [851_012_500, 851_262_500]
+    assert sys_["talkgroupsFile"] == "/app/talkgroups.csv" and sys_["recordUnknown"] is False
+    assert sys_["uploadScript"] == "/drb/drb-upload.sh"
+    assert sys_["audioArchive"] is False and sys_["callLog"] is False and sys_["compressWav"] is False
+    assert tr["tempDir"] == "/dev/shm" and tr["audioStreaming"] is True
+    stream = tr["plugins"][0]["streams"][0]
+    assert tr["plugins"][0]["library"] == "libsimplestream.so"
+    assert stream == {"TGID": 0, "address": "127.0.0.1", "port": 9123, "sendTGID": True}
+    lines = files["talkgroups.csv"].splitlines()
+    assert lines[0].startswith("Decimal,")       # TR: header required, Decimal first
+    assert lines[1] == '101,D,"PD Dispatch, Main","PD Dispatch, Main",,'
+
+
+def test_trunked_without_talkgroups_records_everything():
+    tr, files = tr_config.build_tr_config("P25", "County", {"control_channels": [851.0125]}, **TR_KW)
+    assert tr["systems"][0]["recordUnknown"] is True
+    assert "talkgroupsFile" not in tr["systems"][0] and not files
+
+
+def test_conventional_p25_uses_synthetic_tgids():
+    cfg = {"mode": "conventional", "channels": [{"freq": 460.400, "name": "PD1", "nac": "293"},
+                                                {"freq": 460.425, "name": "PD3"}]}
+    tr, files = tr_config.build_tr_config("P25", "WPPD", cfg, **TR_KW)
+    sys_ = tr["systems"][0]
+    assert sys_["type"] == "conventionalP25" and sys_["channelFile"] == "/app/channels.csv"
+    assert "digitalRecorders" not in tr["sources"][0]
+    rows = files["channels.csv"].splitlines()
+    assert rows[0].startswith("TG Number,Frequency")
+    # Same id the OP25 path reports for the channel, so C2 names both alike.
+    assert rows[1] == f"{channel_tgid(460_400_000)},460400000,PD1,PD1"
+
+
+def test_analog_squelch_threshold_carries_over_else_default():
+    cfg = {"channels": [{"freq": 154.25, "name": "FD"}], "squelch_mode": "power", "squelch_threshold": -62}
+    tr, files = tr_config.build_tr_config("NBFM", "FD", cfg, **TR_KW)
+    assert tr["systems"][0]["type"] == "conventional" and tr["systems"][0]["squelch"] == -62.0
+    tr, _ = tr_config.build_tr_config("NBFM", "FD", {"channels": [{"freq": 154.25}]}, **TR_KW)
+    assert tr["systems"][0]["squelch"] == -50.0
+    # Single channel: centre moved off it (DC spike).
+    assert abs(tr["sources"][0]["center"] - 154_250_000) >= tr_config.DC_SPIKE_GUARD_HZ
+    assert tr_config.live_sample_rate("NBFM", {}) == 16000 and tr_config.live_sample_rate("P25", {}) == 8000
+
+
+def test_legacy_analog_and_unpinned_sdr():
+    kw = {**TR_KW, "device_serial": None}
+    tr, _ = tr_config.build_tr_config("NBFM", "Old", {"control_channels": [155.1]}, **kw)
+    assert tr["sources"][0]["device"] == "rtl=0"
+    assert "155100000" in _["channels.csv"]
+
+
+@pytest.mark.parametrize("stype,cfg", [
+    ("DMR", {"control_channels": [450.0]}),
+    ("P25", {"control_channels": []}),
+    ("P25", {"mode": "conventional", "channels": [{"freq": 851.0}, {"freq": 855.0}]}),  # 4 MHz: every channel must fit
+    ("P25", {"mode": "conventional", "channels": []}),
+])
+def test_unbuildable_systems_raise(stype, cfg):
+    with pytest.raises(ValueError):
+        tr_config.build_tr_config(stype, "x", cfg, **TR_KW)
+
+
+def test_center_override_is_validated():
+    assert tr_config.choose_center([851_000_000], 2_400_000, override=851_500_000) == 851_500_000
+    with pytest.raises(ValueError):
+        tr_config.choose_center([851_000_000], 2_400_000, override=853_000_000)
+
+
+# ---------------------------------------------------------------------------
+# Call JSON -> MQTT + /upload
+# ---------------------------------------------------------------------------
+
+def test_parse_call_maps_tr_json():
+    call = parse_call(SAMPLE_CALL, "node-001", "101-1793145600.250_851012500-call_7")
+    assert call.tgid == 101 and call.tgid_name == "PD Dispatch" and call.freq == 851_012_500
+    assert call.start_epoch == pytest.approx(1793145600.25) and call.stop_epoch == pytest.approx(1793145612.73)
+    assert call.units == [1234567, 7654321]          # unique, in order, undecoded (-1) dropped
+    assert call.emergency is True                     # a unit's emergency flag counts
+    assert call.encrypted is False
+    assert call.decode_errors == 5 and call.spikes == 1 and call.call_length == pytest.approx(12.48)
+    # Deterministic: same JSON -> same id (restart safety), different node -> different id.
+    assert parse_call(SAMPLE_CALL, "node-001", "101-1793145600.250_851012500-call_7").call_id == call.call_id
+    assert parse_call(SAMPLE_CALL, "node-002", "101-1793145600.250_851012500-call_7").call_id != call.call_id
+
+    start = call.start_payload()
+    assert start["tgid"] == 101 and start["tgid_name"] == "PD Dispatch" and start["srcaddr"] == 1234567
+    assert start["attributed"] is True and start["driver"] == "trunk-recorder"
+    assert start["started_at"].startswith("2026-")
+    end = call.end_payload()
+    assert end["ended_at_epoch"] == pytest.approx(1793145612.73)
+    assert end["audio_start_epoch"] == call.start_epoch and end["units"] == [1234567, 7654321]
+    assert end["emergency"] is True and end["encrypted"] is False
+
+
+def test_upload_form_carries_tr_fields():
+    call = parse_call(SAMPLE_CALL, "node-001")
+    meta = UploadMeta(call_id=call.call_id, talkgroup_id=call.tgid, talkgroup_name=call.tgid_name,
+                      system_id="sys1", audio_start_epoch=call.start_epoch, audio_end_epoch=call.stop_epoch,
+                      extra=call.upload_extra())
+    form = upload_spool._form(meta)
+    # Fields C2's /upload declares today:
+    assert form["call_id"] == call.call_id and form["talkgroup_id"] == "101"
+    assert form["talkgroup_name"] == "PD Dispatch" and form["system_id"] == "sys1"
+    # Extra fields (ignored by C2 until it declares them):
+    assert form["units"] == "1234567,7654321" and form["emergency"] == "1" and form["encrypted"] == "0"
+    assert form["freq"] == "851012500" and form["recorder_backend"] == "trunk-recorder"
+    assert json.loads(form["src_list"])[1] == {"src": 7654321, "time": 1793145606, "pos": 5.2, "emergency": True}
+    # Extra can never overwrite a core field.
+    meta.extra = {"call_id": "evil"}
+    assert upload_spool._form(meta)["call_id"] == call.call_id
+
+
+def test_parse_call_old_tr_without_ms_and_bad_json():
+    old = {k: v for k, v in SAMPLE_CALL.items() if not k.endswith("_ms")}
+    call = parse_call(old, "n")
+    assert call.start_epoch == 1793145600.0 and call.stop_epoch == 1793145612.0
+    with pytest.raises(ValueError):
+        parse_call({**SAMPLE_CALL, "talkgroup": 0}, "n")
+    with pytest.raises(ValueError):
+        parse_call({k: v for k, v in old.items() if k != "start_time"}, "n")
+
+
+def test_mime_follows_suffix():
+    assert upload_spool.mime_for(Path("a.wav")) == "audio/wav"
+    assert upload_spool.mime_for(Path("a.flac")) == "audio/flac"
+
+
+# ---------------------------------------------------------------------------
+# Inbox hand-off
+# ---------------------------------------------------------------------------
+
+class FakeMQTT:
+    is_connected = True
+
+    def __init__(self):
+        self.events = []
+
+    async def publish_metadata(self, event, data):
+        self.events.append((event, dict(data)))
+
+
+class FakeRecorder:
+    def __init__(self, url="gs://b/calls/x.wav"):
+        self.url = url
+        self.uploads = []
+
+        class _Spool:
+            last_result = {"kind": "retry"}
+        self.spool = _Spool()
+
+    async def upload_recording(self, path, call_id, **kw):
+        assert path.exists()
+        self.uploads.append((path.name, call_id, kw))
+        path.unlink()
+        return self.url
+
+
+@pytest.fixture
+def tr_env(tmp_path, monkeypatch):
+    monkeypatch.setattr(settings, "recorder_backend", "trunk-recorder")
+    monkeypatch.setattr(settings, "tr_inbox_dir", str(tmp_path / "inbox"))
+    monkeypatch.setattr(settings, "tr_config_dir", str(tmp_path / "tr"))
+    monkeypatch.setattr(settings, "c2_url", "https://c2.example")
+    monkeypatch.setattr(trmod, "CALL_START_LEAD_SECONDS", 0)
+    mqtt, rec = FakeMQTT(), FakeRecorder()
+    import app.internal.call_recorder as cr
+    import app.internal.mqtt_manager as mm
+    monkeypatch.setattr(mm, "mqtt_manager", mqtt)
+    monkeypatch.setattr(cr, "call_recorder", rec)
+    (tmp_path / "inbox").mkdir()
+    return tmp_path / "inbox", mqtt, rec
+
+
+def _wav(path: Path, seconds: float = 2.0, rate: int = 8000) -> None:
+    import wave
+    with wave.open(str(path), "wb") as w:
+        w.setnchannels(1)
+        w.setsampwidth(2)
+        w.setframerate(rate)
+        w.writeframes(b"\x00\x00" * int(seconds * rate))
+
+
+def _drop(inbox: Path, data=SAMPLE_CALL, stem="101-1793145600.250_851012500-call_7", wav=True, seconds=11.68):
+    if wav:
+        _wav(inbox / f"{stem}.wav", seconds)
+    (inbox / f"{stem}.json").write_text(json.dumps(data))
+
+
+async def test_inbox_call_is_published_and_uploaded_once(tr_env):
+    inbox, mqtt, rec = tr_env
+    _drop(inbox)
+    t = trmod.TrunkRecorder()
+    assert await t.scan_once(wait=True) == 1
+    assert [e for e, _ in mqtt.events] == ["call_start", "call_end"]
+    start, end = mqtt.events[0][1], mqtt.events[1][1]
+    assert len(rec.uploads) == 1
+    name, call_id, kw = rec.uploads[0]
+    assert call_id == start["call_id"] == end["call_id"]
+    # Renamed to the spool's recoverable <date>_<time>_<call_id> form.
+    assert name.endswith(f"_{call_id}.wav") and name.count("_") == 2
+    assert kw["talkgroup_id"] == 101 and kw["extra"]["units"] == "1234567,7654321"
+    assert end["audio_url"] == "gs://b/calls/x.wav"
+    # TR drops the gaps between transmissions: the audio ends at start + WAV
+    # length (11.68 s), not at stop_time (12.48 s after start).
+    assert end["audio_end_epoch"] == pytest.approx(1793145600.25 + 11.68)
+    assert kw["audio_end_epoch"] == pytest.approx(1793145600.25 + 11.68)
+    assert end["ended_at_epoch"] == pytest.approx(1793145612.73)
+    assert json.loads(kw["extra"]["freq_list"])[1]["pos"] == 5.2
+    assert list(inbox.iterdir()) == []
+    assert await t.scan_once(wait=True) == 0
+
+
+async def test_encrypted_call_is_reported_not_uploaded(tr_env):
+    inbox, mqtt, rec = tr_env
+    _drop(inbox, {**SAMPLE_CALL, "encrypted": 1})
+    t = trmod.TrunkRecorder()
+    await t.scan_once(wait=True)
+    assert rec.uploads == []
+    assert mqtt.events[-1][1]["audio_skipped"] == "encrypted" and t.encrypted_skipped == 1
+    assert list(inbox.iterdir()) == []
+
+
+async def test_spooled_upload_is_reported(tr_env):
+    inbox, mqtt, rec = tr_env
+    rec.url = None
+    _drop(inbox)
+    await trmod.TrunkRecorder().scan_once(wait=True)
+    assert mqtt.events[-1][1]["audio_upload"] == "spooled"
+
+
+async def test_json_waits_for_wav_then_orphan_is_dropped(tr_env, monkeypatch):
+    inbox, mqtt, rec = tr_env
+    _drop(inbox, wav=False)
+    t = trmod.TrunkRecorder()
+    await t.scan_once(wait=True)
+    assert mqtt.events == [] and len(list(inbox.glob("*.json"))) == 1
+    monkeypatch.setattr(trmod, "ORPHAN_JSON_SECONDS", -1)
+    await t.scan_once(wait=True)
+    # Closed in C2 anyway: a previous pass may have uploaded it and died.
+    assert [e for e, _ in mqtt.events] == ["call_end"] and mqtt.events[0][1]["audio_skipped"] == "no_audio"
+    assert list(inbox.glob("*.json")) == []
+
+
+def test_stray_wavs_are_swept_but_claimed_ones_kept(tr_env):
+    inbox, _, _ = tr_env
+    (inbox / "stray-call_1.wav").write_bytes(b"x")
+    (inbox / "waiting-call_2.wav").write_bytes(b"x")
+    (inbox / "waiting-call_2.json").write_text("{}")
+    claimed = "20261027_120000_" + "0" * 8 + "-0000-0000-0000-" + "0" * 12 + ".wav"
+    (inbox / claimed).write_bytes(b"x")
+    t = trmod.TrunkRecorder()
+    assert t.sweep_stray_wavs() == 0                       # all younger than an hour
+    assert t.sweep_stray_wavs(now=4_000_000_000) == 1
+    assert sorted(p.name for p in inbox.iterdir()) == sorted([claimed, "waiting-call_2.json", "waiting-call_2.wav"])
+
+
+async def test_waits_for_mqtt_so_call_start_isnt_lost(tr_env):
+    inbox, mqtt, rec = tr_env
+    mqtt.is_connected = False
+    _drop(inbox)
+    assert await trmod.TrunkRecorder().scan_once(wait=True) == 0
+    assert rec.uploads == []
+
+
+async def test_unreadable_json_is_dropped(tr_env):
+    inbox, mqtt, rec = tr_env
+    (inbox / "bad.wav").write_bytes(b"x")
+    (inbox / "bad.json").write_text("{not json")
+    await trmod.TrunkRecorder().scan_once(wait=True)
+    assert list(inbox.iterdir()) == [] and mqtt.events == []
+
+
+# ---------------------------------------------------------------------------
+# Config files on disk, supervisor requests, backend dispatch
+# ---------------------------------------------------------------------------
+
+async def test_apply_config_writes_files_and_restart(tr_env, monkeypatch, tmp_path):
+    from app.models import SystemConfig
+    import app.internal.secondary_sdr_client as ssc
+
+    async def devices():
+        return [{"serial": "00000101", "index": 0}]
+    monkeypatch.setattr(ssc.secondary_sdr_client, "devices", devices)
+    t = trmod.TrunkRecorder()
+    ok = await t.apply_config(SystemConfig(system_id="s1", name="County", type="P25",
+                                           config={"control_channels": [851.0125], "talkgroups": [{"id": 1}]}))
+    assert ok and t.config_error is None
+    d = tmp_path / "tr"
+    cfg = json.loads((d / "config.json").read_text())
+    assert cfg["sources"][0]["device"] == "rtl=00000101" and (d / "talkgroups.csv").exists()
+
+    bad = await t.apply_config(SystemConfig(system_id="s2", name="Wide", type="P25",
+                                            config={"mode": "conventional",
+                                                    "channels": [{"freq": 851.0}, {"freq": 856.0}]}))
+    assert not bad and "MHz" in t.config_error
+    # radio_backend reports the config error instead of pretending to start.
+    monkeypatch.setattr(trmod, "trunk_recorder", t)
+    assert "MHz" in await radio_backend.start_with_reason()
+    t.config_error = None
+    monkeypatch.setattr(trmod, "RESTART_POLL_SECONDS", 0.01)
+    # start() waits for run.sh to report a restart, so decoders aren't handed
+    # the dongle before TR has it.
+    monkeypatch.setattr(trmod, "RESTART_WAIT_SECONDS", 0.05)
+    assert "isn't running" in await radio_backend.start_with_reason()
+    assert (d / "restart-request").exists()
+    import time as _t
+    (d / "drb-status.json").write_text(json.dumps({"state": "running", "epoch": _t.time(),
+                                                   "started_at": _t.time() - 600}))
+    assert "didn't restart" in await radio_backend.start_with_reason()   # still the old process
+    (d / "drb-status.json").write_text(json.dumps({"state": "running", "epoch": _t.time() + 5,
+                                                   "started_at": _t.time() + 5}))
+    assert await radio_backend.start_with_reason() is None
+    assert await radio_backend.start_with_reason(wait=False) is None
+    assert await radio_backend.stop() and (d / "stop-request").exists()
+    assert await radio_backend.start() and not (d / "stop-request").exists()
+
+
+async def test_op25_backend_still_drives_op25(monkeypatch):
+    monkeypatch.setattr(settings, "recorder_backend", "op25")
+    calls = []
+
+    async def start_with_reason():
+        calls.append("start")
+        return None
+    monkeypatch.setattr(radio_backend.op25_client, "start_with_reason", start_with_reason)
+    assert await radio_backend.start() and calls == ["start"]
+
+
+def test_supervisor_status_and_decode_rate(tr_env, tmp_path):
+    d = tmp_path / "tr"
+    (d / "logs").mkdir(parents=True)
+    t = trmod.TrunkRecorder()
+    assert t.supervisor_status() is None
+    (d / "drb-status.json").write_text(json.dumps({"state": "running", "epoch": 1000, "started_at": 900}))
+    st = t.supervisor_status(now=1005)
+    assert st["state"] == "running" and not st["stale"]
+    assert t.supervisor_status(now=1100)["stale"]
+    (d / "logs" / "10-27-2026_1200_00.log").write_text(
+        "[2026-10-27 12:00:03.1] (info)   [drb]\tControl Channel Message Decode Rate: 38/sec, count:  114\n"
+        "[2026-10-27 12:00:06.1] (error)  [drb]\tControl Channel Message Decode Rate: 0/sec, count:  0\n")
+    assert t.log_facts()["decode_rate"] == 0.0
+    assert t.log_tail(1)[0].endswith("count:  0")
+
+
+# ---------------------------------------------------------------------------
+# Live audio: follow one talkgroup
+# ---------------------------------------------------------------------------
+
+def test_live_bridge_follows_one_talkgroup():
+    b = trmod.LiveAudioBridge()
+    b._write = lambda pcm: None
+    pkt = lambda tg: tg.to_bytes(4, "little") + b"\x01\x00" * 80   # noqa: E731
+    assert b.feed(pkt(101), now=0.0) is not None
+    assert b.feed(pkt(202), now=0.5) is None                      # simultaneous call: not mixed in
+    assert b.feed(pkt(101), now=1.0) is not None
+    assert b.feed(pkt(202), now=1.0 + b.HOLD_SECONDS + 0.1) is not None   # 101 went quiet: switch
+    assert b.current_tgid == 202 and b.packets == 4
+    assert b.feed(b"\x00\x00", now=9.0) is None
+
+
+# ---------------------------------------------------------------------------
+# Health checklist
+# ---------------------------------------------------------------------------
+
+def _facts(tr, mode="trunked", **over):
+    now = 1_000_000.0
+    f = {
+        "now": now, "node_id": "n", "default_password": False, "c2_url": "https://c2", "has_api_key": True,
+        "mqtt_connected": True, "mqtt_broker": "m", "mqtt_port": 8883, "sdr_count": 1,
+        "source": {"name": "County", "type": "P25", "system_id": "s1", "local": False, "mode": mode},
+        "op25_status": "stopped", "console": {}, "audio_capturing": False, "last_voice_epoch": None,
+        "icecast_live": False, "icecast_port": 8000, "backend": "trunk-recorder", "tr": tr,
+    }
+    f.update(over)
+    return {c["id"]: c for c in health.build_checks(f)}
+
+
+def _tr(**over):
+    now = 1_000_000.0
+    t = {"status": {"state": "running", "started_at": now - 600, "restarts": 0, "stale": False, "age_seconds": 2},
+         "log_age_seconds": 3, "decode_rate": 38.0, "config_error": None, "inbox_backlog": 0,
+         "calls_ingested": 5, "encrypted_skipped": 0, "last_call_epoch": now - 30, "last_call": None,
+         "live": {"last_packet_epoch": now - 30}}
+    t.update(over)
+    return t
+
+
+def test_health_tr_running_and_decoding():
+    c = _facts(_tr())
+    assert "op25" not in c and c["recorder"]["state"] == "ok"
+    assert c["decode"]["state"] == "ok" and "38" in c["decode"]["summary"]
+    assert c["audio"]["state"] == "ok" and c["stream"]["state"] == "ok"
+
+
+@pytest.mark.parametrize("tr,cid,state", [
+    (None, "recorder", "error"),
+    ({"status": None}, "recorder", "error"),                               # profile not enabled
+    ({"status": {"state": "crashed", "last_exit": 1, "stale": False}}, "recorder", "error"),
+    ({"status": {"state": "running", "stale": True, "age_seconds": 60}}, "recorder", "error"),
+    ({"config_error": "too wide"}, "recorder", "error"),
+    ({"decode_rate": 0.0}, "decode", "error"),
+    ({"decode_rate": None}, "decode", "warn"),
+    ({"inbox_backlog": 50}, "audio", "warn"),
+])
+def test_health_tr_failures(tr, cid, state):
+    facts = None if tr is None else _tr(**tr)
+    assert _facts(facts)[cid]["state"] == state
+
+
+def test_health_tr_conventional_has_no_decode_rate():
+    assert _facts(_tr(decode_rate=None), mode="conventional")["decode"]["state"] == "ok"
+
+
+def test_wide_control_channel_list_is_cut_to_what_one_sdr_covers():
+    # MTA Zone 2 (node-002): five control channels spanning 4.15 MHz.
+    ccs = [770_156_250, 770_431_250, 773_293_750, 773_843_750, 774_306_250]
+    assert tr_config.widest_cover(ccs, 2_400_000) == ccs[2:]
+
+
+def test_center_override_keeps_only_the_control_channels_it_covers():
+    tr, _ = tr_config.build_tr_config(
+        "P25", "MTA", {"control_channels": [770.15625, 770.43125, 773.29375]},
+        **{**TR_KW, "center_override": 770_900_000})
+    assert tr["systems"][0]["control_channels"] == [770_156_250, 770_431_250]

drb-edge-node/tests/test_upload_spool.py

@@ -0,0 +1,230 @@
+"""
+Upload spool (node-26#44): a failed upload must never delete the recording.
+
+Before the spool, upload_recording unlinked the file in a `finally`, so any C2
+outage, 5xx or timeout lost that call's audio for good.
+"""
+import asyncio
+import json
+from pathlib import Path
+
+import pytest
+
+from app.config import settings
+from app.internal import upload_spool
+from app.internal.call_recorder import CallRecorder
+from app.internal.upload_spool import OK, REJECT, RETRY, UploadMeta, UploadResult, UploadSpool, classify
+
+
+class Clock:
+    def __init__(self, t: float = 1_000_000.0):
+        self.t = t
+
+    def __call__(self) -> float:
+        return self.t
+
+
+class FakeUploader:
+    def __init__(self, *results: UploadResult):
+        self.results = list(results)
+        self.calls = []
+
+    async def __call__(self, path: Path, meta: UploadMeta) -> UploadResult:
+        assert path.exists(), "the uploader must be handed a file that exists"
+        self.calls.append((path.name, meta))
+        return self.results.pop(0) if self.results else UploadResult(OK, url="gs://x")
+
+
+def _recording(tmp_path: Path, name: str = "20261027_call-1.flac", size: int = 100) -> Path:
+    p = tmp_path / "rec" / name
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(b"f" * size)
+    return p
+
+
+def _spool(tmp_path: Path, uploader, clock=None, **kw) -> UploadSpool:
+    args = dict(max_bytes=10_000, max_age_seconds=3600, backoff_min=10, backoff_max=80)
+    args.update(kw)
+    return UploadSpool(tmp_path / "spool", uploader, clock=clock or Clock(), **args)
+
+
+# ---------------------------------------------------------------------------
+# Classification
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("status,url,kind", [
+    (200, "gs://b/o", OK),
+    (200, None, RETRY),      # C2's GCS write failed: nothing stored
+    (401, None, RETRY),      # a reissued key is on its way
+    (408, None, RETRY),
+    (429, None, RETRY),
+    (500, None, RETRY),
+    (503, None, RETRY),
+    (400, None, REJECT),
+    (403, None, RETRY),      # not approved yet: approval may still come
+    (413, None, REJECT),
+])
+def test_classify(status, url, kind):
+    assert classify(status, url) == kind
+
+
+# ---------------------------------------------------------------------------
+# upload_recording: delete only on success
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def recorder(tmp_path, monkeypatch):
+    monkeypatch.setattr(settings, "c2_url", "https://c2.example")
+    rec = CallRecorder()
+    rec.spool = _spool(tmp_path, rec._post)
+    return rec
+
+
+def _patch_post(monkeypatch, result: UploadResult):
+    async def fake(path, meta, mime, timeout=None):
+        return result
+    monkeypatch.setattr(upload_spool, "post_recording", fake)
+
+
+async def test_failed_upload_keeps_the_file_in_the_spool(tmp_path, recorder, monkeypatch):
+    _patch_post(monkeypatch, UploadResult(RETRY, detail="ConnectError: refused"))
+    path = _recording(tmp_path)
+    url = await recorder.upload_recording(path, "call-1", talkgroup_id=101, talkgroup_name="Dispatch",
+                                          system_id="sys-1", audio_start_epoch=1.5, audio_end_epoch=9.25)
+    assert url is None
+    assert not path.exists()
+    spooled = tmp_path / "spool" / path.name
+    assert spooled.read_bytes() == b"f" * 100
+    meta = json.loads((tmp_path / "spool" / (path.name + ".json")).read_text())
+    assert meta["call_id"] == "call-1" and meta["talkgroup_id"] == 101 and meta["system_id"] == "sys-1"
+    assert meta["audio_start_epoch"] == 1.5 and meta["audio_end_epoch"] == 9.25
+    assert meta["last_error"] == "ConnectError: refused"
+    assert recorder.spool.stats()["files"] == 1
+
+
+async def test_successful_upload_deletes_the_file(tmp_path, recorder, monkeypatch):
+    _patch_post(monkeypatch, UploadResult(OK, url="gs://b/call-1.flac", status=200))
+    path = _recording(tmp_path)
+    assert await recorder.upload_recording(path, "call-1") == "gs://b/call-1.flac"
+    assert not path.exists()
+    assert recorder.spool.stats()["files"] == 0
+
+
+async def test_rejected_upload_is_not_spooled(tmp_path, recorder, monkeypatch):
+    _patch_post(monkeypatch, UploadResult(REJECT, status=413, detail="HTTP 413: File too large"))
+    path = _recording(tmp_path)
+    assert await recorder.upload_recording(path, "call-1") is None
+    assert not path.exists()
+    assert recorder.spool.stats()["files"] == 0
+    assert recorder.spool.last_result["kind"] == REJECT
+
+
+async def test_real_post_network_error_is_retryable(tmp_path, monkeypatch):
+    monkeypatch.setattr(settings, "c2_url", "http://127.0.0.1:9")  # discard port: refused
+    path = _recording(tmp_path)
+    result = await upload_spool.post_recording(path, UploadMeta(call_id="c"), "audio/flac")
+    assert result.kind == RETRY and result.status is None
+    assert path.exists(), "post_recording must never touch the file"
+
+
+# ---------------------------------------------------------------------------
+# Retry, startup drain, caps
+# ---------------------------------------------------------------------------
+
+async def test_retry_backs_off_then_uploads(tmp_path):
+    clock = Clock()
+    up = FakeUploader(UploadResult(RETRY, status=503, detail="HTTP 503"), UploadResult(OK, url="gs://x"))
+    sp = _spool(tmp_path, up, clock=clock)
+    sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY, detail="first"))
+
+    assert await sp.drain_once() == 0 and up.calls == []      # not due yet
+    clock.t += 10
+    assert await sp.drain_once() == 0 and len(up.calls) == 1  # retry 2 fails
+    data = json.loads(next((tmp_path / "spool").glob("*.json")).read_text())
+    assert data["attempts"] == 2 and data["next_attempt_at"] == clock.t + 20
+    clock.t += 20
+    assert await sp.drain_once() == 1
+    assert list((tmp_path / "spool").iterdir()) == []
+
+
+async def test_startup_drain_uploads_what_survived_a_restart(tmp_path):
+    clock = Clock()
+    first = _spool(tmp_path, FakeUploader(), clock=clock)
+    first.add(_recording(tmp_path, "a_c1.flac"), UploadMeta(call_id="c1", talkgroup_id=7), UploadResult(RETRY))
+    first.add(_recording(tmp_path, "b_c2.flac"), UploadMeta(call_id="c2"), UploadResult(RETRY))
+
+    # A new process: same directory, nothing due by the sidecar timers.
+    up = FakeUploader()
+    restarted = _spool(tmp_path, up, clock=clock, poll_seconds=0.01)
+    restarted.start()
+    for _ in range(100):
+        if len(up.calls) == 2:
+            break
+        await asyncio.sleep(0.01)
+    await restarted.stop()
+    assert [m.call_id for _, m in up.calls] == ["c1", "c2"]
+    assert up.calls[0][1].talkgroup_id == 7
+    assert list((tmp_path / "spool").iterdir()) == []
+
+
+async def test_rejected_on_retry_is_deleted(tmp_path):
+    clock = Clock()
+    sp = _spool(tmp_path, FakeUploader(UploadResult(REJECT, status=413)), clock=clock)
+    sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY))
+    await sp.drain_once(force=True)
+    assert sp.stats()["files"] == 0
+
+
+async def test_age_cap_drops_expired_entries(tmp_path):
+    clock = Clock()
+    sp = _spool(tmp_path, FakeUploader(), clock=clock, max_age_seconds=3600)
+    sp.add(_recording(tmp_path, "old_c1.flac"), UploadMeta(call_id="c1"), UploadResult(RETRY))
+    clock.t += 1800
+    sp.add(_recording(tmp_path, "new_c2.flac"), UploadMeta(call_id="c2"), UploadResult(RETRY))
+    clock.t += 1801
+    assert sp.enforce_caps() == 1
+    assert [p.name for p in (tmp_path / "spool").glob("*.flac")] == ["new_c2.flac"]
+
+
+async def test_size_cap_evicts_oldest_first(tmp_path):
+    clock = Clock()
+    sp = _spool(tmp_path, FakeUploader(), clock=clock, max_bytes=250)
+    for i in range(3):
+        sp.add(_recording(tmp_path, f"{i}_c{i}.flac", size=100), UploadMeta(call_id=f"c{i}"), UploadResult(RETRY))
+        clock.t += 1
+    assert sp.enforce_caps() == 1
+    assert sorted(p.name for p in (tmp_path / "spool").glob("*.flac")) == ["1_c1.flac", "2_c2.flac"]
+    assert sp.stats()["bytes"] == 200
+
+
+async def test_half_written_pairs_are_cleaned(tmp_path):
+    sp = _spool(tmp_path, FakeUploader())
+    d = tmp_path / "spool"
+    d.mkdir()
+    (d / "lost.flac.json").write_text(json.dumps({"call_id": "x", "spooled_at": 0}))  # no audio
+    (d / "orphan.flac").write_bytes(b"a")                       # no sidecar, no call_id in name
+    assert sp.stats()["files"] == 0
+    assert list(d.iterdir()) == []
+
+
+async def test_lost_or_corrupt_sidecar_is_rebuilt_from_the_filename(tmp_path):
+    """A power cut can zero a sidecar; the audio must survive it."""
+    up = FakeUploader()
+    sp = _spool(tmp_path, up)
+    d = tmp_path / "spool"
+    d.mkdir()
+    (d / "20261027_120000_call-a.flac").write_bytes(b"a")
+    (d / "20261027_120001_call_b.flac").write_bytes(b"b")
+    (d / "20261027_120001_call_b.flac.json").write_text("")
+    assert sp.stats()["files"] == 2
+    await sp.drain_once()
+    assert sorted(m.call_id for _, m in up.calls) == ["call-a", "call_b"]
+
+
+async def test_wake_makes_everything_due_now(tmp_path):
+    clock = Clock()
+    up = FakeUploader()
+    sp = _spool(tmp_path, up, clock=clock)
+    sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY, status=401))
+    sp.wake()
+    assert await sp.drain_once() == 1

op25-container/Dockerfile

@@ -17,8 +17,16 @@ RUN apt-get update && \
 # Install custom PulseAudio system config (enables anonymous access for edge-node)
 COPY system.pa /etc/pulse/system.pa
 
-# Clone the boatbod op25 repository
+# Clone the boatbod op25 repository, PINNED (node-26#29, #46). An unpinned
-RUN git clone -b gr310 https://github.com/boatbod/op25 /op25
+# clone plus CI's layer cache meant the image held whatever gr310 HEAD was at
+# the last cache miss. 5b58335 (gr310, 2026-08-23) is the commit tk_drb_conv
+# was verified against for node-26#29 and the first gr310 merge that carries
+# the 2026-08-03 NBFM noise/voice squelch (PR #296, op25_nbfm.py
+# nbfm_squelch_mode). Bump deliberately, re-checking tk_drb_conv against
+# tk_p25.py at the new commit.
+ARG OP25_COMMIT=5b583357240e0172bfae905f0c966474e0d0f094
+RUN git clone -b gr310 https://github.com/boatbod/op25 /op25 && \
+    git -C /op25 checkout --detach "${OP25_COMMIT}"
 
 # Set the working directory
 WORKDIR /op25

op25-container/app/models.py

@@ -54,6 +54,13 @@ class P25Mode(str, Enum):
     CONVENTIONAL = "conventional"
 
 
+class NbfmSquelchMode(str, Enum):
+    """Values op25_nbfm.py accepts for nbfm_squelch_mode (gr310, 2026-08-03)."""
+    POWER = "power"
+    NOISE = "noise"
+    VOICE = "voice"
+
+
 class ConventionalChannel(BaseModel):
     """One fixed P25 conventional voice channel (node-26#24)."""
     frequency: int                 # Hz
@@ -77,7 +84,16 @@ class ConfigGenerator(BaseModel):
     hardware_preset: str = "rtl-sdr-v3"
     ppm_override: Optional[float] = None  # when set, overrides the preset's default PPM
     # NBFM only: power squelch threshold in dB (op25_nbfm default is -60).
+    # Used only in "power" squelch mode.
     nbfm_squelch: Optional[float] = None
+    # NBFM only (node-26#46): OP25 gr310's squelch algorithm — "power" (the
+    # absolute-level squelch, needs per-site calibration), "noise" (FM
+    # quieting, no calibration) or "voice" (noise + speech detector).
+    # Defaults to noise: power mode left analog open on static.
+    nbfm_squelch_mode: NbfmSquelchMode = NbfmSquelchMode.NOISE
+    # dB of quieting needed to open in noise/voice mode (OP25 default 8;
+    # 12-15 opens only on solid signals, ~5 chases weak ones).
+    nbfm_noise_squelch_db: Optional[float] = None
 
 class DemodType(str, Enum):
     CQPSK = "cqpsk"
@@ -99,6 +115,9 @@ class ChannelConfig(BaseModel):
     frequency: Optional[int] = None
     # Key name op25_nbfm.py actually reads (from_dict(config, 'nbfm_squelch_threshold', -60)).
     nbfm_squelch_threshold: Optional[float] = None
+    # Also read by op25_nbfm.py from the channel dict (node-26#46).
+    nbfm_squelch_mode: Optional[str] = None
+    nbfm_noise_squelch_db: Optional[float] = None
     destination: Optional[str] = "udp://127.0.0.1:23456"
     # P25 conventional only, read by tk_drb_conv.py (not by multi_rx): the NAC
     # to lock the frame assembler to. multi_rx passes the whole channel dict

op25-container/app/routers/op25_controller.py

@@ -262,6 +262,8 @@ def create_op25_router():
                     frequency=freq,
                     meta_stream_name="",
                     nbfm_squelch_threshold=generator.nbfm_squelch,
+                    nbfm_squelch_mode=generator.nbfm_squelch_mode.value,
+                    nbfm_noise_squelch_db=generator.nbfm_noise_squelch_db,
                 )]
                 devices = [DeviceConfig(
                     gains=preset["gains"], ppm=ppm, tunable=False,

op25-container/tests/test_conventional.py

@@ -149,6 +149,23 @@ def test_generate_config_route(tmp_path, monkeypatch):
     cfg = json.loads((tmp_path / "active.cfg.json").read_text())
     assert cfg["trunking"]["module"] == "tk_drb_conv.py"
     assert cfg["channels"][0]["frequency"] == 154430000
+    # node-26#46: analog defaults to OP25's noise squelch, not power.
+    assert cfg["channels"][0]["nbfm_squelch_mode"] == "noise"
+    assert "nbfm_noise_squelch_db" not in cfg["channels"][0]
+
+    r = client.post("/op25/generate-config", json={
+        "type": "NBFM", "systemName": "Fire", "channels": [154430000], "icecastConfig": ice,
+        "nbfm_squelch_mode": "power", "nbfm_squelch": -55, "nbfm_noise_squelch_db": 12,
+    })
+    assert r.status_code == 200, r.text
+    ch = json.loads((tmp_path / "active.cfg.json").read_text())["channels"][0]
+    assert ch["nbfm_squelch_mode"] == "power" and ch["nbfm_squelch_threshold"] == -55
+    assert ch["nbfm_noise_squelch_db"] == 12
+
+    r = client.post("/op25/generate-config", json={
+        "type": "NBFM", "systemName": "Fire", "channels": [154430000], "nbfm_squelch_mode": "loud",
+    })
+    assert r.status_code == 422
 
 
 # ---------------------------------------------------------------------------

trunk-recorder/README.md

@@ -0,0 +1,34 @@
+# Trunk Recorder on a DRB node (node-26#45)
+
+Opt-in alternative to the OP25 recorder. On a node, set both in `.env`, then
+`docker compose up -d`:
+
+    RECORDER_BACKEND=trunk-recorder
+    COMPOSE_PROFILES=trunk-recorder
+
+Set both back (`RECORDER_BACKEND=op25`, remove `COMPOSE_PROFILES`) to return
+to OP25.
+
+- `run.sh`: supervisor that replaces the image's CMD. It restarts TR when the
+  edge-node rewrites `configs/trunk-recorder/config.json`, and writes
+  `drb-status.json` for the dashboard health check.
+- `drb-upload.sh`: TR's `uploadScript`. It copies each finished call (WAV +
+  JSON) into `recordings/tr-inbox`, where the edge-node uploads it through the
+  normal spool.
+
+## License (GPL-3.0)
+
+Trunk Recorder is GPL-3.0. DRB runs the unmodified upstream image
+`robotastic/trunk-recorder:5.2.1` (digest pinned in `docker-compose.yml`) as a
+separate program. It talks to DRB only through files, a script argument list
+and a local UDP socket. That makes it mere aggregation, the same position we
+already hold for OP25.
+
+Anyone we distribute a node image or device to must still be able to get the
+corresponding source. Upstream source for this exact version is at
+<https://github.com/TrunkRecorder/trunk-recorder/tree/v5.2.1>. If we ever ship
+the TR image ourselves (mirrored to our registry, preinstalled on a sold
+device), we must also offer that source directly, in writing, for three years,
+or ship it alongside. Pointing to upstream alone is not enough at that point.
+Any patch to TR itself makes DRB's copy a modified version, which must then be
+published under GPL-3.0.

trunk-recorder/drb-upload.sh

@@ -0,0 +1,28 @@
+#!/bin/sh
+# Trunk Recorder uploadScript (node-26#45). TR runs this after every call as
+#     drb-upload.sh <call.wav> <call.json> <call.m4a>
+# (run_upload_script_argv, call_concluder.cc v5.2.1; the m4a only exists with
+# compressWav, which DRB leaves off). TR deletes the wav/json as soon as this
+# exits (audioArchive/callLog false), so it copies them into the edge-node's
+# inbox first. Order matters: the WAV lands first and the JSON last, so the
+# JSON's presence tells the edge-node the WAV is complete. Exit non-zero = TR
+# marks the call failed.
+set -u
+WAV="${1:?wav path}"
+JSON="${2:?json path}"
+INBOX="${DRB_TR_INBOX:-/recordings/tr-inbox}"
+TMP="$INBOX/.incoming"
+
+[ -f "$WAV" ] && [ -f "$JSON" ] || { echo "drb-upload: missing $WAV or $JSON" >&2; exit 1; }
+mkdir -p "$TMP" || exit 1
+
+stem="$(basename "$JSON" .json)"
+# Hard link when on the same filesystem (no copy, TR's unlink leaves ours);
+# copy otherwise.
+put() { ln -f "$1" "$2" 2>/dev/null || cp -f "$1" "$2"; }
+
+put "$WAV" "$TMP/$stem.wav" || exit 1
+put "$JSON" "$TMP/$stem.json" || exit 1
+mv -f "$TMP/$stem.wav" "$INBOX/$stem.wav" || exit 1
+mv -f "$TMP/$stem.json" "$INBOX/$stem.json" || exit 1
+exit 0

trunk-recorder/run.sh

@@ -0,0 +1,87 @@
+#!/bin/sh
+# Supervisor for Trunk Recorder inside its container (node-26#45).
+#
+# The edge-node writes /app/config.json from the C2 system config but has no
+# Docker access, so it cannot restart this container. This loop does it:
+#   - waits for config.json, then runs trunk-recorder on it
+#   - restarts it when config.json changes or /app/restart-request appears
+#   - keeps it stopped while /app/stop-request exists (dashboard Stop)
+#   - restarts it after a crash, backing off 5 s -> 60 s (`restarts` counts
+#     crashes only, not config restarts)
+#   - writes /app/drb-status.json every few seconds for the edge-node's health
+#     check (state, pid, started_at, restarts, last_exit, epoch)
+set -u
+APP="${DRB_TR_APP:-/app}"
+CFG="$APP/config.json"
+STATUS="$APP/drb-status.json"
+RESTART="$APP/restart-request"
+STOP="$APP/stop-request"
+REC="${DRB_RECORDINGS:-/recordings}"
+mkdir -p "$APP/logs" "$REC/tr-capture" "$REC/tr-inbox"
+
+pid=""
+started=0
+restarts=0
+last_exit=null
+backoff=5
+cfgsum=""
+
+write_status() {
+  printf '{"state":"%s","pid":%s,"started_at":%s,"restarts":%s,"last_exit":%s,"epoch":%s}\n' \
+    "$1" "${pid:-null}" "$started" "$restarts" "$last_exit" "$(date +%s)" > "$STATUS.tmp" \
+    && mv -f "$STATUS.tmp" "$STATUS"
+}
+
+sum() { md5sum "$CFG" 2>/dev/null | cut -d' ' -f1; }
+
+stop_tr() {
+  if [ -n "$pid" ] && kill -0 "$pid" 2>/dev/null; then
+    kill -TERM "$pid"
+    i=0
+    while kill -0 "$pid" 2>/dev/null && [ "$i" -lt 30 ]; do sleep 0.5; i=$((i + 1)); done
+    kill -KILL "$pid" 2>/dev/null
+    wait "$pid" 2>/dev/null
+  fi
+  pid=""
+}
+
+trap 'stop_tr; write_status stopped; exit 0' TERM INT
+
+nap() { sleep "$1" & wait $!; }
+
+while :; do
+  if [ -f "$STOP" ]; then
+    [ -n "$pid" ] && { echo "drb: stop requested"; stop_tr; }
+    write_status stopped; nap 2; continue
+  fi
+  if [ ! -f "$CFG" ]; then
+    write_status waiting_for_config; nap 5; continue
+  fi
+  if [ -z "$pid" ]; then
+    rm -f "$RESTART"
+    cfgsum="$(sum)"
+    echo "drb: starting trunk-recorder"
+    trunk-recorder --config="$CFG" &
+    pid=$!
+    started="$(date +%s)"
+  fi
+  write_status running
+  nap 5
+  if ! kill -0 "$pid" 2>/dev/null; then
+    wait "$pid" 2>/dev/null
+    last_exit=$?
+    pid=""
+    restarts=$((restarts + 1))
+    # A run that lasted a while resets the backoff.
+    [ $(( $(date +%s) - started )) -gt 120 ] && backoff=5
+    echo "drb: trunk-recorder exited ($last_exit); restarting in ${backoff}s"
+    write_status crashed
+    nap "$backoff"
+    backoff=$((backoff * 2)); [ "$backoff" -gt 60 ] && backoff=60
+    continue
+  fi
+  if [ -f "$RESTART" ] || [ "$(sum)" != "$cfgsum" ]; then
+    echo "drb: config changed or restart requested — restarting trunk-recorder"
+    stop_tr
+  fi
+done