Fix stale log text for rejected uploads: 403 is now retried, so only 400/413 reach it (node-26#44)

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

.env.example

@@ -99,6 +99,15 @@ 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
+
 # OP25 container (usually no need to change)
 OP25_API_URL=http://localhost:8001
 OP25_TERMINAL_URL=http://localhost:8081

drb-edge-node/app/config.py

@@ -153,6 +153,23 @@ 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
+
     # ------------------------------------------------------------------
     # Local dashboard / API authentication
     #

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

@@ -66,10 +66,10 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import 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,6 +320,17 @@ 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
     # ------------------------------------------------------------------
@@ -328,8 +339,11 @@ class CallRecorder:
         """Start the persistent capture. Call once from app lifespan."""
         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 +765,17 @@ class CallRecorder:
     # Upload (unchanged interface)
     # ------------------------------------------------------------------
 
+    async def _post(self, file_path: Path, meta: UploadMeta) -> UploadResult:
+        return await upload_spool.post_recording(file_path, meta, 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, AUDIO_MIME, timeout=settings.upload_inline_timeout_seconds,
+        )
+
     async def upload_recording(
         self,
         file_path: Path,
@@ -761,50 +786,37 @@ class CallRecorder:
         audio_start_epoch: Optional[float] = None,
         audio_end_epoch: Optional[float] = 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"
-        api_key = credentials.get_api_key()
-        headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
-
-        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,
+        meta = UploadMeta(
+            call_id=call_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name,
+            system_id=system_id, audio_start_epoch=audio_start_epoch, audio_end_epoch=audio_end_epoch,
         )
-                    r.raise_for_status()
-                    audio_url = r.json().get("url")
-                    logger.info(f"Upload complete: {audio_url}")
-                    return audio_url
-        except Exception as e:
-            logger.error(f"Upload failed: {e}")
+        result = await self._post_inline(file_path, meta)
+        self.spool.last_result = {"call_id": call_id, "kind": result.kind, "status": result.status,
+                                  "detail": result.detail, "epoch": time.time()}
+        if result.kind == upload_spool.OK:
+            logger.info(f"Upload complete: {result.url}")
+            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

@@ -154,6 +154,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,
@@ -333,6 +349,7 @@ async def gather_facts() -> Dict[str, Any]:
         "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,6 +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
@@ -60,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
@@ -146,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)
@@ -164,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.
@@ -172,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:
@@ -273,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)
 
@@ -390,6 +411,14 @@ class MQTTManager:
         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")
         self._publish(self._t_checkin, payload, qos=1)
 
     def _publish(self, topic: str, payload: dict, qos: int = 0, retain: bool = False):
@@ -408,5 +437,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_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/upload_spool.py

@@ -0,0 +1,418 @@
+"""
+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
+
+
+@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}"
+    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.
@@ -179,6 +191,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.")
 
 

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,
@@ -205,6 +208,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 +248,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={
-                    "control_channels": freqs,
-                    "talkgroups": [tg.model_dump() for tg in body.talkgroups] if is_p25 else [],
-                },
+                config=sys_config,
             )
     await _apply_override(config)
     return {"ok": True, "name": config.name}

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

@@ -216,6 +216,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>
@@ -396,6 +410,13 @@
   // ── 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 +464,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

@@ -153,3 +153,28 @@ 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_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
-RUN git clone -b gr310 https://github.com/boatbod/op25 /op25
+# Clone the boatbod op25 repository, PINNED (node-26#29, #46). An unpinned
+# 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
 
 
 # ---------------------------------------------------------------------------