Releases: VERSION + automatic patch tags + promote; updater installs newer versions (node-26#49)

VERSION holds MAJOR.MINOR (1.1), edited by hand. Every push to main is
tagged vMAJOR.MINOR.PATCH by the "Tag release" workflow, with PATCH counted
as an integer by scripts/next-version.sh. "Promote release" moves the major
tag (v1) to a tested release and refuses to move it backwards.

drb-update.sh: a newer release on the node's channel is now a reason to
update by itself. REF=latest follows the highest vX.Y.Z tag (test nodes);
REF=v1 follows the promoted tag (customer nodes). Versions compare per
field as integers (1.10.0 > 1.9.3, never string or float order). It never
downgrades, and an untagged target still needs a request. After an update
it writes DRB_VERSION to .env; the edge node reports it in checkin.

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

.gitea/workflows/build-edge-node.yml

@@ -38,15 +38,20 @@ jobs:
         run: |
           echo "REPO_NAME=$(echo ${GITHUB_REPOSITORY} | awk -F'/' '{print $2}')" >> $GITHUB_OUTPUT
           echo "VERSION=$(git describe --tags --always | sed 's/^v//')" >> $GITHUB_OUTPUT
+          # Source pin (node-26#19): the git tree hash of the build context.
+          # A node pulls src-<tree of its own checkout> — scripts/image-tags.sh
+          # computes the identical value — so its code and image always match.
+          echo "SRC_TAG=src-$(git rev-parse HEAD:drb-edge-node)" >> $GITHUB_OUTPUT
 
       - uses: docker/build-push-action@v6
         with:
           context: ./drb-edge-node
           file: ./drb-edge-node/Dockerfile
-          platforms: linux/arm64
+          platforms: linux/amd64,linux/arm64
           push: true
           tags: |
             git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.VERSION }}
+            git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.SRC_TAG }}
             git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:latest
           cache-from: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache
           cache-to: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache,mode=max

.gitea/workflows/build-icecast.yml

@@ -38,15 +38,20 @@ jobs:
         run: |
           echo "REPO_NAME=$(echo ${GITHUB_REPOSITORY} | awk -F'/' '{print $2}')" >> $GITHUB_OUTPUT
           echo "VERSION=$(git describe --tags --always | sed 's/^v//')" >> $GITHUB_OUTPUT
+          # Source pin (node-26#19): the git tree hash of the build context.
+          # A node pulls src-<tree of its own checkout> — scripts/image-tags.sh
+          # computes the identical value — so its code and image always match.
+          echo "SRC_TAG=src-$(git rev-parse HEAD:icecast)" >> $GITHUB_OUTPUT
 
       - uses: docker/build-push-action@v6
         with:
           context: ./icecast
           file: ./icecast/Dockerfile
-          platforms: linux/arm64
+          platforms: linux/amd64,linux/arm64
           push: true
           tags: |
             git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.VERSION }}
+            git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.SRC_TAG }}
             git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:latest
           cache-from: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache
           cache-to: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache,mode=max

.gitea/workflows/build-op25.yml

@@ -38,15 +38,20 @@ jobs:
         run: |
           echo "REPO_NAME=$(echo ${GITHUB_REPOSITORY} | awk -F'/' '{print $2}')" >> $GITHUB_OUTPUT
           echo "VERSION=$(git describe --tags --always | sed 's/^v//')" >> $GITHUB_OUTPUT
+          # Source pin (node-26#19): the git tree hash of the build context.
+          # A node pulls src-<tree of its own checkout> — scripts/image-tags.sh
+          # computes the identical value — so its code and image always match.
+          echo "SRC_TAG=src-$(git rev-parse HEAD:op25-container)" >> $GITHUB_OUTPUT
 
       - uses: docker/build-push-action@v6
         with:
           context: ./op25-container
           file: ./op25-container/Dockerfile
-          platforms: linux/arm64
+          platforms: linux/amd64,linux/arm64
           push: true
           tags: |
             git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.VERSION }}
+            git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.SRC_TAG }}
             git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:stable
           # Registry cache — avoids full recompile of GnuRadio/OP25 when only app code changes
           cache-from: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache

.gitea/workflows/build-secondary-sdr.yml

@@ -38,15 +38,20 @@ jobs:
         run: |
           echo "REPO_NAME=$(echo ${GITHUB_REPOSITORY} | awk -F'/' '{print $2}')" >> $GITHUB_OUTPUT
           echo "VERSION=$(git describe --tags --always | sed 's/^v//')" >> $GITHUB_OUTPUT
+          # Source pin (node-26#19): the git tree hash of the build context.
+          # A node pulls src-<tree of its own checkout> — scripts/image-tags.sh
+          # computes the identical value — so its code and image always match.
+          echo "SRC_TAG=src-$(git rev-parse HEAD:secondary-sdr-container)" >> $GITHUB_OUTPUT
 
       - uses: docker/build-push-action@v6
         with:
           context: ./secondary-sdr-container
           file: ./secondary-sdr-container/Dockerfile
-          platforms: linux/arm64
+          platforms: linux/amd64,linux/arm64
           push: true
           tags: |
             git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.VERSION }}
+            git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.SRC_TAG }}
             git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:latest
           cache-from: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache
           cache-to: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache,mode=max

.gitea/workflows/promote.yml

@@ -0,0 +1,39 @@
+name: Promote release
+
+# node-26#49: move the major tag (v1) to a tested release. Customer nodes
+# (updater REF=v1) install it on their next timer run, because its version is
+# numerically higher than what they run. Never moves it backwards.
+
+on:
+  workflow_dispatch:
+    inputs:
+      version:
+        description: "Release to promote, e.g. 1.1.3 (must already be tagged v1.1.3)"
+        required: true
+
+jobs:
+  promote:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Move the major tag
+        env:
+          VERSION: ${{ inputs.version }}
+        run: |
+          [[ "$VERSION" =~ ^([0-9]+)\.[0-9]+\.[0-9]+$ ]] || { echo "version must be MAJOR.MINOR.PATCH, got '$VERSION'"; exit 1; }
+          MAJOR="${BASH_REMATCH[1]}"
+          git fetch --force --tags origin
+          git rev-parse -q --verify "refs/tags/v$VERSION" >/dev/null || { echo "tag v$VERSION does not exist"; exit 1; }
+          # Refuse to go backwards: compare with the release v$MAJOR points at now.
+          CUR="$(git tag --points-at "v$MAJOR" 2>/dev/null | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$' \
+                 | sed 's/^v//' | sort -t. -k1,1n -k2,2n -k3,3n | tail -1 || true)"
+          if [ -n "$CUR" ]; then
+            HIGHEST="$(printf '%s\n%s\n' "$CUR" "$VERSION" | sort -t. -k1,1n -k2,2n -k3,3n | tail -1)"
+            [ "$HIGHEST" = "$VERSION" ] || { echo "v$MAJOR is at $CUR, newer than $VERSION — refusing to roll back here"; exit 1; }
+          fi
+          git tag -f "v$MAJOR" "v$VERSION^{commit}"
+          git push --force origin "refs/tags/v$MAJOR"
+          echo "v$MAJOR -> $VERSION (was ${CUR:-untagged})"

.gitea/workflows/publish-public.yml

@@ -95,7 +95,7 @@ jobs:
         with:
           context: ${{ matrix.context }}
           file: ${{ matrix.file }}
-          platforms: linux/arm64
+          platforms: linux/amd64,linux/arm64
           push: true
           tags: |
             ${{ vars.PUBLIC_REGISTRY }}/${{ vars.PUBLIC_NAMESPACE }}/drb-${{ matrix.name }}:${{ steps.meta.outputs.VERSION }}

.gitea/workflows/release-images.yml

@@ -0,0 +1,91 @@
+name: Release images
+
+# Make sure every node image a release ref needs exists in the registry
+# (node-26#19). A node on `v1` pulls <image>:src-<tree hash of that image's
+# directory at v1>. The per-image build workflows only run when their own
+# directory changes on main, and anything built before #19 has no src- tag at
+# all, so a release tag can point at trees nobody ever built. This fills the
+# gaps: for each image, if src-<tree> is already in the registry it does
+# nothing; otherwise it builds and pushes it (registry cache keeps op25 off a
+# full recompile when only its app code changed).
+#
+# Runs on every v* tag push, including a re-cut `v1`. Also runnable by hand
+# (workflow_dispatch) on any ref — run it once on the ref `v1` points at
+# before re-cutting `v1` onto a commit that contains #19, so nodes never see
+# a moment where their pinned images don't exist yet.
+#
+# Pushes ONLY src- tags. :latest / :stable keep tracking main, as before.
+
+on:
+  workflow_dispatch:
+  push:
+    tags:
+      - "v*"
+
+jobs:
+  ensure:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - name: edge-node
+            context: drb-edge-node
+          - name: icecast
+            context: icecast
+          - name: op25-client
+            context: op25-container
+          - name: secondary-sdr
+            context: secondary-sdr-container
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: docker/setup-qemu-action@v3
+
+      - uses: docker/setup-buildx-action@v3
+        with:
+          config-inline: |
+            [registry."git.vpn.cusano.net"]
+              http = false
+              insecure = false
+
+      - uses: docker/login-action@v3
+        with:
+          registry: git.vpn.cusano.net
+          username: ${{ gitea.actor }}
+          password: ${{ secrets.BUILD_TOKEN }}
+
+      - name: Resolve source tag
+        id: meta
+        run: |
+          REPO_NAME="$(echo ${GITHUB_REPOSITORY} | awk -F'/' '{print $2}')"
+          # Must match scripts/image-tags.sh and the build-*.yml workflows.
+          IMAGE="git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${REPO_NAME}/${{ matrix.name }}"
+          TAG="src-$(git rev-parse HEAD:${{ matrix.context }})"
+          echo "IMAGE=$IMAGE" >> $GITHUB_OUTPUT
+          echo "TAG=$TAG" >> $GITHUB_OUTPUT
+          # Present only counts if BOTH platforms are in it: src- tags pushed
+          # before node-26#28 are arm64-only, and an amd64 node can't pull them.
+          RAW="$(docker buildx imagetools inspect --raw "$IMAGE:$TAG" 2>/dev/null | tr -d ' \n' || true)"
+          if printf '%s' "$RAW" | grep -q '"architecture":"amd64"' \
+             && printf '%s' "$RAW" | grep -q '"architecture":"arm64"'; then
+            echo "EXISTS=true" >> $GITHUB_OUTPUT
+            echo "$IMAGE:$TAG already published for amd64 + arm64 — nothing to do"
+          else
+            echo "EXISTS=false" >> $GITHUB_OUTPUT
+            echo "$IMAGE:$TAG missing or not multi-arch — building it"
+          fi
+
+      - if: steps.meta.outputs.EXISTS != 'true'
+        uses: docker/build-push-action@v6
+        with:
+          context: ./${{ matrix.context }}
+          file: ./${{ matrix.context }}/Dockerfile
+          # amd64 + arm64, same as the build-*.yml workflows (node-26#28).
+          platforms: linux/amd64,linux/arm64
+          push: true
+          tags: ${{ steps.meta.outputs.IMAGE }}:${{ steps.meta.outputs.TAG }}
+          cache-from: type=registry,ref=${{ steps.meta.outputs.IMAGE }}:buildcache

.gitea/workflows/version-tag.yml

@@ -0,0 +1,31 @@
+name: Tag release
+
+# node-26#49: every push to main is a release, vMAJOR.MINOR.PATCH (see
+# scripts/next-version.sh). Test nodes (updater REF=latest) install the
+# highest tag on their next timer run; customer nodes (REF=v1) wait for the
+# Promote workflow.
+
+on:
+  push:
+    branches: [main]
+
+concurrency:
+  group: version-tag
+  cancel-in-progress: false
+
+jobs:
+  tag:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Tag this commit
+        run: |
+          git fetch --force --tags origin
+          TAG="$(bash scripts/next-version.sh)"
+          if [ -z "$TAG" ]; then echo "already tagged"; exit 0; fi
+          git tag "$TAG"
+          git push origin "refs/tags/$TAG"
+          echo "released $TAG"

.gitignore

@@ -13,6 +13,7 @@ venv/
 # Runtime-generated configs and recordings
 configs/
 recordings/
+.drb-update-compose.yml
 
 # Logs
 *.log

README.md

@@ -142,13 +142,17 @@ this repo at the `v1` tag, enrols with C2, pulls the prebuilt images and starts:
 
 ```bash
 curl -fsSL https://git.vpn.cusano.net/logan/node-26/raw/tag/v1/install.sh \
-  | sudo bash -s -- --token DRB-xxxx --node-id node-003 \
+  | sudo bash -s -- --node-id node-003 \
       --c2-url https://api.<domain> --mqtt-broker mqtt.<domain>
 ```
 
-Mint the `--token` at **Settings → Nodes** in the web app (the panel prints the
+It prompts (hidden input) for an enrollment token, minted at **Settings → Nodes**
-whole command). Run `install.sh --help` for every flag; each also has a
+in the web app. The token never goes on the command line — argv lands in shell
-`DRB_*` env var. `--build` compiles op25 on the Pi (~1h) instead of pulling.
+history and `ps` (node-26#18). Unattended installs use `--token-file PATH` (mode
+0600) or an exported `DRB_ENROLLMENT_TOKEN` (pass it through with
+`sudo --preserve-env=DRB_ENROLLMENT_TOKEN`; plain `sudo` drops it); the old `--token` flag still works
+but warns. Run `install.sh --help` for every flag; each also has a `DRB_*` env
+var. `--build` compiles op25 on the Pi (~1h) instead of pulling.
 
 ### Local dev / manual
 

VERSION

@@ -0,0 +1 @@
+1.1

docker-compose.yml

@@ -1,6 +1,10 @@
+# Image tags: a provisioned node pins every image to the source tree it has
+# checked out (DRB_TAG_* = src-<git tree hash>, written into .env by install.sh
+# and scripts/drb-update.sh via scripts/image-tags.sh — node-26#19). Unset, as
+# in local dev, they fall back to :latest / :stable, which CI builds from main.
 services:
   icecast:
-    image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/icecast:latest
+    image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/icecast:${DRB_TAG_ICECAST:-latest}
     build: ./icecast
     restart: unless-stopped
     network_mode: host
@@ -16,7 +20,7 @@ services:
   # rather than being exposed. edge-node still reaches both over localhost
   # because it shares this host network namespace.
   op25:
-    image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/op25-client:stable
+    image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/op25-client:${DRB_TAG_OP25:-stable}
     build: ./op25-container
     restart: unless-stopped
     privileged: true
@@ -38,7 +42,7 @@ services:
   # node-26#9. Same network/device access as op25 for the same reason: it
   # needs the raw USB device, not a virtualized one.
   secondary-sdr:
-    image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/secondary-sdr:latest
+    image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/secondary-sdr:${DRB_TAG_SECONDARY_SDR:-latest}
     build: ./secondary-sdr-container
     restart: unless-stopped
     privileged: true
@@ -48,7 +52,7 @@ services:
       - /dev:/dev
 
   edge-node:
-    image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/edge-node:latest
+    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
     restart: unless-stopped
     network_mode: host

drb-edge-node/app/config.py

@@ -6,6 +6,9 @@ class Settings(BaseSettings):
     # Node identity
     node_id: str
     node_name: str = "Unnamed Node"
+    # The release installed on this node (node-26#49). The host updater writes
+    # DRB_VERSION to .env after each update; "untagged" before the first one.
+    drb_version: str = "untagged"
     node_lat: float = 0.0
     node_lon: float = 0.0
 
@@ -54,6 +57,13 @@ class Settings(BaseSettings):
     # Bounded wait for the shared PulseAudio socket before launching FFmpeg.
     pulse_wait_timeout: float = 30.0
 
+    # Discord radio (node-26#25): with the broker unreachable this long, log
+    # every radio bot out. MUST stay below C2's
+    # DISCORD_LEASE_OFFLINE_RELEASE_SECONDS (default 600), after which C2 hands
+    # this node's bot tokens to other nodes — so a token is never logged in on
+    # two nodes at once.
+    discord_mqtt_loss_stop_seconds: int = 300
+
     # ------------------------------------------------------------------
     # Call segmentation
     #
@@ -146,6 +156,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,
@@ -760,51 +785,40 @@ class CallRecorder:
         system_id: Optional[str] = None,
         audio_start_epoch: Optional[float] = None,
         audio_end_epoch: Optional[float] = None,
+        radios: Optional[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,
-
+            radios=radios,
-        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

@@ -1,299 +1,568 @@
+"""
+Discord radio streams on the edge, driven by desired state (node-26#25,
+server-26#203/#204).
+
+C2 retains ``nodes/{id}/discord/desired`` = the FULL list of streams this node
+should carry: ``{seq, streams: [{stream_id, guild_id, channel_id, token_id,
+system_name, ...}]}``. This module reconciles toward that list and publishes
+``nodes/{id}/discord/actual`` = ``{applied_seq, sessions: [token_id],
+streams: {stream_id: {state, error, code}}}`` after every change.
+
+Rules this enforces by construction:
+
+* One gateway session per leased token; a token no stream needs any more is
+  logged OUT (``client.close()``), not just disconnected from voice. C2 frees
+  the token only after an ``actual`` with ``applied_seq`` >= the drop and the
+  token absent from ``sessions`` — so one token is never logged in on two
+  nodes (node-26#25 item 1).
+* One voice connection per (token, guild); one token can serve many guilds.
+* The watchdog is a separate reconcile loop that never closes itself: closing
+  a session cannot cancel it (node-26#25 item 2).
+* No message-content intent, no ``@bot leave/join`` mention controls — voice
+  is controlled only through C2 (node-26#25 item 3).
+* MQTT-loss fail-safe: if the broker has been unreachable for
+  ``discord_mqtt_loss_stop_seconds`` every radio bot is logged out, because C2
+  returns this node's tokens to the pool after its own (longer) grace.
+
+Bot token strings never travel over MQTT: they are fetched over HTTPS from C2
+(``GET /discord/node/{node}/tokens/{id}``, node API key), which hands a token
+only to the node holding its lease, and are kept in memory only.
+
+Audio: each voice connection plays the PulseAudio monitor (``-f pulse`` on
+``settings.pulse_source``) — the accuracy path, never Icecast.
+"""
+from __future__ import annotations
+
 import asyncio
-from typing import Optional
+import time
+from dataclasses import dataclass
+from typing import Awaitable, Callable, Optional
+
 import discord
-from discord.ext import commands
+
 from app.config import settings
 from app.internal import pulse
 from app.internal.logger import logger
 
-BOT_READY_TIMEOUT = 15   # seconds to wait for Discord bot to become ready
+BOT_READY_TIMEOUT = 15     # seconds to wait for a Discord login to become ready
-WATCHDOG_INTERVAL = 30   # seconds between voice-connection health checks
+WATCHDOG_INTERVAL = 15     # seconds between reconcile passes
-REJOIN_DELAY = 5  # seconds to wait before attempting a rejoin
+STREAM_RETRY_DELAY = 5     # seconds to back off before re-arming an audio source
-STREAM_RETRY_DELAY = 5   # seconds to back off before re-arming the audio source
+REPORT_EVERY_TICKS = 4     # unconditional actual-state report every ~60s
+_VOICE_TYPES = (discord.ChannelType.voice, discord.ChannelType.stage_voice)
 
 
-class RadioBot:
+@dataclass
-    def __init__(self):
+class _Session:
-        self._bot: Optional[commands.Bot] = None
+    token_id: str
-        self._voice_client: Optional[discord.VoiceClient] = None
+    client: object
-        self._task: Optional[asyncio.Task] = None
+    task: Optional[asyncio.Task] = None
-        self._watchdog_task: Optional[asyncio.Task] = None
+    presence: Optional[str] = None
-        self._ready_event: Optional[asyncio.Event] = None
+
-        self._current_token: Optional[str] = None
+    def alive(self) -> bool:
+        return self.task is not None and not self.task.done() and not self.client.is_closed()
+
+
+@dataclass
+class _Voice:
+    stream_id: str
+    token_id: str
+    guild_id: int
+    channel_id: int
+    vc: Optional[object] = None
+    state: str = "connecting"
+    error: Optional[str] = None
+    code: Optional[str] = None
+
+
+def _join_refusal(guild, channel) -> Optional[tuple[str, str]]:
+    """Why Discord would refuse this bot's voice join, or None.
+
+    Discord does not answer a join the bot lacks permission for — the gateway
+    just never sends the voice-server update, discord.py times out after 30s,
+    and the admin page used to read "(TimeoutError: )". Checking first turns
+    that into an instruction. Best effort: if the bot's own member or the
+    permission data is not cached, let the join go ahead.
+    """
+    me = getattr(guild, "me", None)
+    permissions_for = getattr(channel, "permissions_for", None)
+    if me is None or permissions_for is None:
+        return None
+    try:
+        perms = permissions_for(me)
+    except Exception:
+        return None
+    missing = [label for attr, label in (("view_channel", "View Channel"), ("connect", "Connect"))
+               if not getattr(perms, attr, True)]
+    if getattr(channel, "type", None) == discord.ChannelType.voice and not getattr(perms, "speak", True):
+        missing.append("Speak")
+    if missing:
+        return ("missing_permissions",
+                f"The radio bot lacks {', '.join(missing)} on that voice channel — "
+                "allow it in the channel's permission settings in Discord.")
+    limit = getattr(channel, "user_limit", 0) or 0
+    # The bot's own lingering voice state (a dropped connection) is not a
+    # slot it needs; counting it would block the rejoin of a full channel.
+    me_id = getattr(me, "id", None)
+    members = [m for m in (getattr(channel, "members", None) or [])
+               if me_id is None or getattr(m, "id", None) != me_id]
+    if limit and len(members) >= limit and not getattr(perms, "move_members", False):
+        return ("channel_full", f"That voice channel is full ({len(members)}/{limit}) — "
+                                "raise its user limit or free a slot.")
+    return None
+
+
+def _default_client_factory():
+    intents = discord.Intents.none()
+    intents.guilds = True
+    intents.voice_states = True
+    return discord.Client(intents=intents)
+
+
+def _default_audio_factory():
+    # before_options land ahead of -i:  ffmpeg -f pulse -i drb_sink.monitor …
+    source = discord.FFmpegPCMAudio(settings.pulse_source, before_options="-f pulse")
+    return discord.PCMVolumeTransformer(source, volume=1.0)
+
+
+async def _default_token_fetcher(token_id: str) -> Optional[str]:
+    import httpx
+    from app.internal import credentials
+    api_key = credentials.get_api_key()
+    if not settings.c2_url or not api_key:
+        logger.error("Cannot fetch a radio-bot token: C2_URL or the node API key is missing.")
+        return None
+    url = f"{settings.c2_url.rstrip('/')}/discord/node/{settings.node_id}/tokens/{token_id}"
+    try:
+        async with httpx.AsyncClient(timeout=10) as client:
+            r = await client.get(url, headers={"Authorization": f"Bearer {api_key}"})
+            if r.status_code != 200:
+                logger.error(f"C2 refused radio-bot token {token_id}: HTTP {r.status_code}")
+                return None
+            return r.json().get("token")
+    except Exception as e:
+        logger.error(f"Fetching radio-bot token {token_id} failed: {e}")
+        return None
+
+
+async def _default_publisher(payload: dict) -> None:
+    from app.internal.mqtt_manager import mqtt_manager
+    mqtt_manager.publish_discord_actual(payload)
+
+
+def _default_mqtt_up() -> bool:
+    from app.internal.mqtt_manager import mqtt_manager
+    return mqtt_manager.is_connected
+
+
+class RadioManager:
+    def __init__(
+        self,
+        client_factory: Callable[[], object] = _default_client_factory,
+        token_fetcher: Callable[[str], Awaitable[Optional[str]]] = _default_token_fetcher,
+        publisher: Callable[[dict], Awaitable[None]] = _default_publisher,
+        audio_factory: Callable[[], object] = _default_audio_factory,
+        mqtt_up: Callable[[], bool] = _default_mqtt_up,
+        ready_timeout: float = BOT_READY_TIMEOUT,
+    ):
+        self._client_factory = client_factory
+        self._fetch_token = token_fetcher
+        self._publish = publisher
+        self._audio_factory = audio_factory
+        self._mqtt_up = mqtt_up
+        self._ready_timeout = ready_timeout
+
+        self._desired: dict[str, dict] = {}
+        self._applied_seq: Optional[int] = None
+        self._sessions: dict[str, _Session] = {}
+        self._voices: dict[str, _Voice] = {}
+        self._session_errors: dict[str, str] = {}
+        self._tokens: dict[str, str] = {}          # token_id -> string, memory only
+        self._lock = asyncio.Lock()
         self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self._watchdog: Optional[asyncio.Task] = None
+        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
 
-        # Remembered so we can rejoin after an unexpected disconnect
+    # ------------------------------------------------------------------
-        self._guild_id: Optional[int] = None
+    # Inputs
-        self._channel_id: Optional[int] = None
+    # ------------------------------------------------------------------
-        self._was_streaming: bool = False
-
-    async def join(self, guild_id: int, channel_id: int, token: str,
-                   call_active: bool = False, system_name: str = None) -> bool:
-        # (Re)start the bot if the token changed or the bot isn't running
-        if self._current_token != token or not self._is_bot_running():
-            if not await self._start_bot(token):
-                return False
-
-        guild = self._bot.get_guild(guild_id)
-        if not guild:
-            logger.error(f"Guild {guild_id} not found — bot may not be a member.")
-            return False
-
-        channel = guild.get_channel(channel_id)
-        if not isinstance(channel, discord.VoiceChannel):
-            logger.error(f"Channel {channel_id} is not a voice channel.")
-            return False
 
+    async def apply_desired(self, payload: dict) -> None:
+        """Handle a (retained) desired-state message from C2."""
+        if not isinstance(payload, dict):
+            return
+        seq = payload.get("seq")
+        seq = seq if isinstance(seq, int) and not isinstance(seq, bool) else None
+        if seq is not None and self._applied_seq is not None and seq < self._applied_seq:
+            logger.warning(f"Ignoring stale Discord desired state seq {seq} (applied {self._applied_seq})")
+            return
+        wanted = {}
+        for s in payload.get("streams") or []:
             try:
-            if self._voice_client and self._voice_client.is_connected():
+                wanted[str(s["stream_id"])] = {
-                await self._voice_client.disconnect(force=True)
+                    "stream_id": str(s["stream_id"]), "guild_id": int(s["guild_id"]),
-            self._voice_client = await channel.connect()
+                    "channel_id": int(s["channel_id"]), "token_id": str(s["token_id"]),
-            # Remember where we are so the watchdog can rejoin if we drop
+                    "system_name": s.get("system_name"),
-            self._guild_id = guild_id
+                }
-            self._channel_id = channel_id
+            except (KeyError, TypeError, ValueError):
-            # Bounded wait for the shared PulseAudio socket. Historically FFmpeg was
+                logger.error(f"Malformed Discord desired stream entry ignored: {s!r}")
-            # launched before the op25 container had created it, failed instantly,
+        self._ensure_watchdog()
-            # and the bot sat silently connected forever.
+        self._desired = wanted
-            await pulse.wait_until_ready()
+        if self._paused and self._redelivery_of_paused_state(seq):
-            self._play_stream()
+            logger.info(f"Discord desired seq {seq} is a redelivery — staying paused locally.")
-            if system_name:
+        else:
-                await self._bot.change_presence(
+            self._paused = False
-                    activity=discord.Activity(
+            self._paused_at_seq = None
-                        type=discord.ActivityType.listening,
+        await self.reconcile()
-                        name=system_name,
+        if seq is not None:
-                    )
+            self._applied_seq = seq
-                )
+        await self.publish_actual()
-            logger.info(f"Joined #{channel.name} in {guild.name}")
-            return True
-        except Exception as e:
-            logger.error(f"Failed to join voice channel: {e}")
-            return False
 
-    async def leave(self) -> bool:
+    def _redelivery_of_paused_state(self, seq: Optional[int]) -> bool:
-        # Clear remembered channel so the watchdog doesn't rejoin
+        """True when `seq` is not newer than what was applied at pause time."""
-        self._guild_id = None
+        return seq is not None and self._paused_at_seq is not None and seq <= self._paused_at_seq
-        self._channel_id = None
-        self._was_streaming = False
 
-        if self._voice_client and self._voice_client.is_connected():
+    async def pause_locally(self) -> None:
+        """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()
+
+    # ------------------------------------------------------------------
+    # Reconcile
+    # ------------------------------------------------------------------
+
+    def _target(self) -> dict[str, dict]:
+        return {} if (self._paused or self._failsafe) else self._desired
+
+    async def reconcile(self) -> None:
+        async with self._lock:
+            target = self._target()
+            # 1) voice connections no longer wanted (or re-pointed)
+            for sid, v in list(self._voices.items()):
+                d = target.get(sid)
+                if not d or (d["token_id"], d["guild_id"], d["channel_id"]) != (v.token_id, v.guild_id, v.channel_id):
+                    await self._disconnect(v)
+                    del self._voices[sid]
+            # 2) gateway sessions no stream needs: LOG OUT, not just leave voice
+            needed = {d["token_id"] for d in target.values()}
+            for tid in list(self._sessions):
+                if tid not in needed:
+                    await self._close_session(tid)
+            for tid in list(self._tokens):
+                if tid not in needed:
+                    self._tokens.pop(tid, None)
+            # 3) sessions + voice for everything wanted
+            for sid, d in sorted(target.items()):
+                v = self._voices.get(sid)
+                if v is None:
+                    v = self._voices[sid] = _Voice(sid, d["token_id"], d["guild_id"], d["channel_id"])
+                sess = await self._ensure_session(d["token_id"])
+                if sess is None:
+                    v.state, v.code = "error", "login_failed"
+                    v.error = self._session_errors.get(d["token_id"], "The radio bot could not log in to Discord.")
+                    continue
+                await self._ensure_voice(sess, v, d)
+            # 4) presence follows voice: "Listening to <system>" only while a
+            #    voice connection for that bot is actually up.
+            for sess in list(self._sessions.values()):
+                await self._sync_presence(sess, target)
+
+    async def _ensure_session(self, token_id: str) -> Optional[_Session]:
+        sess = self._sessions.get(token_id)
+        if sess and sess.alive():
+            return sess
+        if sess:
+            await self._close_session(token_id)
+        token = self._tokens.get(token_id) or await self._fetch_token(token_id)
+        if not token:
+            self._session_errors[token_id] = "DRB would not hand this node the radio bot's token."
+            return None
+        client = self._client_factory()
+        sess = _Session(token_id, client)
+        ready = asyncio.Event()
+
+        async def on_ready():
+            ready.set()
+        client.event(on_ready)
+        sess.task = asyncio.create_task(client.start(token))
+        waiter = asyncio.create_task(ready.wait())
+        done, _ = await asyncio.wait({waiter, sess.task}, timeout=self._ready_timeout,
+                                     return_when=asyncio.FIRST_COMPLETED)
+        if waiter not in done:
+            waiter.cancel()
+            err = sess.task.exception() if sess.task.done() and not sess.task.cancelled() else None
+            msg = f"{type(err).__name__}: {err}" if err else f"not ready after {self._ready_timeout:.0f}s"
+            logger.error(f"Radio bot {token_id} failed to log in: {msg}")
+            self._session_errors[token_id] = f"The radio bot could not log in to Discord ({msg[:120]})."
+            await self._shutdown_client(sess)
+            if isinstance(err, discord.LoginFailure):
+                self._tokens.pop(token_id, None)
+            return None
+        self._tokens[token_id] = token
+        self._sessions[token_id] = sess
+        self._session_errors.pop(token_id, None)
+        logger.info(f"Radio bot {token_id} logged in as {getattr(client, 'user', None)}")
+        return sess
+
+    async def _ensure_voice(self, sess: _Session, v: _Voice, d: dict) -> None:
+        client = sess.client
+        guild = client.get_guild(v.guild_id)
+        if guild is None:
+            v.state, v.code = "error", "not_in_guild"
+            v.error = "This radio bot is not a member of that Discord server — invite it first."
+            return
+        channel = guild.get_channel(v.channel_id)
+        if channel is None or getattr(channel, "type", None) not in _VOICE_TYPES:
+            v.state, v.code = "error", "bad_channel"
+            v.error = "That channel is not a voice channel the radio bot can see."
+            return
+        vc = v.vc
+        if vc is not None and vc.is_connected() and getattr(vc.channel, "id", None) == v.channel_id:
+            if not vc.is_playing():
+                self._play(v)
+            v.state, v.error, v.code = "connected", None, None
+            return
+        refusal = _join_refusal(guild, channel)
+        if refusal:
+            if v.error != refusal[1]:   # once per change, not every watchdog pass
+                logger.error(f"Stream {v.stream_id}: {refusal[1]}")
+            v.state, v.code, v.error = "error", refusal[0], refusal[1]
+            return
+        v.state = "connecting"
         try:
-                self._stop_stream()
+            existing = guild.voice_client
-                await self._voice_client.disconnect(force=True)
+            if existing is not None and existing.is_connected():
-                self._voice_client = None
+                await existing.move_to(channel)
-                await self._bot.change_presence(activity=None)
+                v.vc = existing
-                logger.info("Disconnected from voice channel.")
+            else:
-                return True
+                if existing is not None:
+                    await existing.disconnect(force=True)
+                v.vc = await channel.connect()
+        except asyncio.TimeoutError:
+            # discord.py's TimeoutError has no message, so the old text read
+            # "(TimeoutError: )" — useless on the admin page.
+            v.vc = None
+            v.state, v.code = "error", "connect_timeout"
+            # Permissions were checked above, so this is usually the voice
+            # handshake itself (e.g. a pre-2.7 discord.py refused with 4017 for
+            # lacking DAVE E2EE, then retried until the 30s timeout) or UDP egress.
+            v.error = ("Discord never completed the voice join (timed out). Permissions look fine, so "
+                       "check the node's discord.py voice log (DAVE close 4017, UDP blocked).")
+            logger.error(f"Stream {v.stream_id}: voice connect timed out")
+            return
         except Exception as e:
-                logger.error(f"Failed to disconnect: {e}")
+            v.vc = None
-        return False
+            v.state, v.code = "error", "connect_failed"
+            v.error = f"Could not connect to the voice channel ({type(e).__name__}: {str(e)[:120]})."
+            logger.error(f"Stream {v.stream_id}: voice connect failed: {type(e).__name__}: {e}")
+            return
+        # Bounded wait for the shared PulseAudio socket (op25 container creates it).
+        if not await pulse.wait_until_ready():
+            v.state, v.code = "error", "no_audio"
+            v.error = "The node's audio path (PulseAudio) is not up."
+            return
+        self._play(v, pulse_checked=True)
+        v.state, v.error, v.code = "connected", None, None
+        logger.info(f"Stream {v.stream_id} connected (bot {v.token_id})")
 
+    async def _sync_presence(self, sess: _Session, target: dict[str, dict]) -> None:
+        """Show "Listening to <system>" only while this bot is really in voice.
+
+        It used to be set once on the first successful join and never cleared,
+        so a bot whose voice later dropped (or kept failing to rejoin) still
+        looked healthy in Discord's member list.
+        """
+        name = None
+        for sid, v in sorted(self._voices.items()):
+            if (v.token_id == sess.token_id and v.state == "connected"
+                    and v.vc is not None and v.vc.is_connected()):
+                name = (target.get(sid) or {}).get("system_name") or name
+                if name:
+                    break
+        if sess.presence == name:
+            return
+        try:
+            activity = discord.Activity(type=discord.ActivityType.listening, name=name) if name else None
+            await sess.client.change_presence(activity=activity)
+            sess.presence = name
+        except Exception as e:
+            logger.warning(f"Presence update failed: {e}")
+
+    # ------------------------------------------------------------------
+    # Audio
+    # ------------------------------------------------------------------
+
+    def _play(self, v: _Voice, pulse_checked: bool = False) -> None:
+        vc = v.vc
+        if vc is None or not vc.is_connected():
+            return
+        if not pulse_checked and not pulse.is_ready():
+            logger.error(f"PulseAudio socket {pulse.socket_path()} missing — retrying audio in {STREAM_RETRY_DELAY}s.")
+            self._schedule_replay(v.stream_id)
+            return
+        try:
+            vc.play(self._audio_factory(), after=lambda err, sid=v.stream_id: self._on_audio_end(sid, err))
+        except Exception as e:
+            logger.error(f"Stream {v.stream_id}: could not start audio: {e}")
+            self._schedule_replay(v.stream_id)
+
+    def _on_audio_end(self, stream_id: str, error) -> None:
+        """discord.py calls this from its player thread."""
+        if error:
+            logger.error(f"Stream {stream_id} audio ended with error: {error}")
+        self._schedule_replay(stream_id, STREAM_RETRY_DELAY if error else 0.5)
+
+    def _schedule_replay(self, stream_id: str, delay: float = STREAM_RETRY_DELAY) -> None:
+        loop = self._loop
+        if loop is None or loop.is_closed():
+            return
+
+        async def _later():
+            await asyncio.sleep(delay)
+            v = self._voices.get(stream_id)
+            if v and v.vc and v.vc.is_connected() and not v.vc.is_playing():
+                self._play(v)
+
+        loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_later()))
+
+    # ------------------------------------------------------------------
+    # Teardown
+    # ------------------------------------------------------------------
+
+    async def _disconnect(self, v: _Voice) -> None:
+        vc, v.vc = v.vc, None
+        if vc is None:
+            return
+        try:
+            if vc.is_playing():
+                vc.stop()
+            await vc.disconnect(force=True)
+        except Exception as e:
+            logger.warning(f"Stream {v.stream_id}: voice disconnect error: {e}")
+
+    async def _close_session(self, token_id: str) -> None:
+        sess = self._sessions.pop(token_id, None)
+        if sess is None:
+            return
+        for v in self._voices.values():
+            if v.token_id == token_id:
+                await self._disconnect(v)
+        await self._shutdown_client(sess)
+        logger.info(f"Radio bot {token_id} logged out")
+
+    async def _shutdown_client(self, sess: _Session) -> None:
+        try:
+            await sess.client.close()
+        except Exception as e:
+            logger.warning(f"Radio bot {sess.token_id} close error: {e}")
+        if sess.task and not sess.task.done():
+            sess.task.cancel()
+            try:
+                await sess.task
+            except BaseException:
+                pass
+
+    async def stop(self) -> None:
+        """Node shutdown: log every radio bot out."""
+        if self._watchdog:
+            self._watchdog.cancel()
+            self._watchdog = None
+        async with self._lock:
+            for v in list(self._voices.values()):
+                await self._disconnect(v)
+            self._voices.clear()
+            for tid in list(self._sessions):
+                await self._close_session(tid)
+            self._tokens.clear()
+
+    # ------------------------------------------------------------------
+    # Watchdog + MQTT-loss fail-safe
+    # ------------------------------------------------------------------
+
+    def _ensure_watchdog(self) -> None:
+        self._loop = asyncio.get_running_loop()
+        if self._watchdog is None or self._watchdog.done():
+            self._watchdog = asyncio.create_task(self._watchdog_loop())
+
+    async def _watchdog_loop(self) -> None:
+        while True:
+            await asyncio.sleep(WATCHDOG_INTERVAL)
+            try:
+                await self.tick()
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                logger.error(f"Discord watchdog error: {e}")
+
+    async def tick(self, now: Optional[float] = None) -> None:
+        """One watchdog pass: fail-safe check, then reconcile + report if needed."""
+        now = time.monotonic() if now is None else now
+        before = self.snapshot()
+        if self._mqtt_up():
+            self._mqtt_down_since = None
+            if self._failsafe:
+                logger.info("MQTT back — resuming Discord radio streams.")
+                self._failsafe = False
+        else:
+            if self._mqtt_down_since is None:
+                self._mqtt_down_since = now
+            elif (not self._failsafe and self._sessions
+                  and now - self._mqtt_down_since >= settings.discord_mqtt_loss_stop_seconds):
+                logger.warning("MQTT unreachable past the fail-safe — logging every radio bot out "
+                               "(C2 will hand these tokens to other nodes).")
+                self._failsafe = True
+        await self.reconcile()
+        self._ticks += 1
+        # On change, and every few passes regardless: a periodic report is what
+        # lets C2 settle a release whose first ack was lost.
+        if self.snapshot() != before or self._ticks % REPORT_EVERY_TICKS == 0:
+            await self.publish_actual()
+
+    # ------------------------------------------------------------------
+    # Reporting
+    # ------------------------------------------------------------------
+
+    def snapshot(self) -> dict:
+        streams = {}
+        for sid in self._desired:
+            v = self._voices.get(sid)
+            if self._paused:
+                streams[sid] = {"state": "paused_locally", "error": None, "code": None}
+            elif v is None:
+                streams[sid] = {"state": "connecting", "error": None, "code": None}
+            else:
+                streams[sid] = {"state": v.state, "error": v.error, "code": v.code}
+        return {
+            "applied_seq": self._applied_seq,
+            "sessions": sorted(tid for tid, s in self._sessions.items() if s.alive()),
+            "streams": streams,
+        }
+
+    async def publish_actual(self) -> None:
+        try:
+            await self._publish(self.snapshot())
+        except Exception as e:
+            logger.warning(f"Publishing Discord actual state failed: {e}")
+
+    # metadata_watcher hooks — audio plays continuously, nothing to do per call.
     def start_stream(self):
-        """Called when an OP25 call starts — stream plays continuously, nothing to do."""
         pass
 
     def stop_stream(self):
-        """Called when an OP25 call ends — stream plays continuously, nothing to do."""
         pass
 
-    async def stop(self):
-        self._guild_id = None
-        self._channel_id = None
-        self._was_streaming = False
-
-        if self._watchdog_task:
-            self._watchdog_task.cancel()
-            self._watchdog_task = None
-
-        await self.leave()
-
-        if self._task:
-            self._task.cancel()
-        if self._bot:
-            await self._bot.close()
-        self._bot = None
-        self._task = None
-        self._current_token = None
-        self._ready_event = None
-
-    def _play_stream(self):
-        """
-        Feed Discord voice straight from the PulseAudio monitor.
-
-        Icecast is NOT used here: it lags ~1 s at connect and drifts to 100 s+,
-        which is unusable for live listening. Icecast remains the frontend/mobile
-        listening path only.
-        """
-        if not self._voice_client:
-            return
-
-        if not pulse.is_ready():
-            logger.error(
-                f"PulseAudio socket {pulse.socket_path()} missing — "
-                f"cannot start Discord audio; retrying in {STREAM_RETRY_DELAY}s."
-            )
-            self._schedule_restart()
-            return
-
-        # before_options land ahead of -i, so this becomes:
-        #   ffmpeg -f pulse -i drb_sink.monitor …
-        source = discord.FFmpegPCMAudio(
-            settings.pulse_source,
-            before_options="-f pulse",
-        )
-        self._voice_client.play(
-            discord.PCMVolumeTransformer(source, volume=1.0),
-            after=self._on_stream_end,
-        )
-
-    def _schedule_restart(self, delay: float = STREAM_RETRY_DELAY):
-        """Re-arm the audio source after a delay — safe to call from any thread."""
-        if not self._loop:
-            return
-
-        async def _delayed_restart():
-            await asyncio.sleep(delay)
-            vc = self._voice_client
-            if vc and vc.is_connected() and not vc.is_playing():
-                self._play_stream()
-
-        self._loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_delayed_restart()))
-
-    def _on_stream_end(self, error):
-        if error:
-            logger.error(f"Stream ended with error: {error}")
-        vc = self._voice_client
-        if not (self._loop and vc and vc.is_connected() and not vc.is_playing()):
-            return
-        if error:
-            # Back off before retrying — prevents a tight loop when PulseAudio is
-            # unavailable (FFmpeg exits immediately in that case).
-            self._schedule_restart()
-        else:
-            self._loop.call_soon_threadsafe(self._play_stream)
-
-    def _stop_stream(self):
-        if self._voice_client and self._voice_client.is_playing():
-            self._voice_client.stop()
-
-    async def _watchdog_loop(self):
-        """Periodically verify the voice connection is alive; rejoin silently if not."""
-        await asyncio.sleep(WATCHDOG_INTERVAL)  # give initial join time to settle
-        while True:
-            try:
-                await asyncio.sleep(WATCHDOG_INTERVAL)
-
-                # Only act if we're supposed to be in a channel
-                if not self._guild_id or not self._channel_id:
-                    continue
-
-                connected = self._voice_client is not None and self._voice_client.is_connected()
-                if not connected:
-                    logger.warning("Watchdog: voice connection lost — attempting rejoin.")
-                    await asyncio.sleep(REJOIN_DELAY)
-                    rejoined = await self.join(
-                        self._guild_id,
-                        self._channel_id,
-                        self._current_token,
-                    )
-                    if rejoined:
-                        logger.info("Watchdog: successfully rejoined voice channel.")
-                    else:
-                        logger.error("Watchdog: rejoin failed — will retry next cycle.")
-
-            except asyncio.CancelledError:
-                return
-            except Exception as e:
-                logger.error(f"Watchdog error: {e}")
-
-    async def _start_bot(self, token: str) -> bool:
-        await self.stop()  # clean up any previous instance
-
-        intents = discord.Intents.default()
-        intents.voice_states = True
-        intents.message_content = True
-        intents.messages = True
-        self._bot = commands.Bot(command_prefix="!", intents=intents)
-        self._ready_event = asyncio.Event()
-        self._current_token = token
-        self._loop = asyncio.get_event_loop()
-
-        @self._bot.event
-        async def on_ready():
-            logger.info(f"Discord bot ready: {self._bot.user} ({self._bot.user.id})")
-            self._ready_event.set()
-
-        @self._bot.event
-        async def on_voice_state_update(
-            member: discord.Member,
-            before: discord.VoiceState,
-            after: discord.VoiceState,
-        ):
-            """Detect when our own bot gets disconnected from a voice channel."""
-            if self._bot.user and member.id != self._bot.user.id:
-                return
-            if before.channel is not None and after.channel is None:
-                # Bot was disconnected (kicked or server drop).
-                # Don't null _voice_client here — Discord.py reconnects voice
-                # transparently and would leave _voice_client stale as None.
-                # The watchdog detects real disconnects via is_connected().
-                logger.warning("Bot was disconnected from voice channel — watchdog will rejoin if needed.")
-
-        @self._bot.event
-        async def on_message(message: discord.Message):
-            if message.author.bot:
-                return
-            if self._bot.user not in message.mentions:
-                return
-            content = message.content.lower()
-            if "leave" in content:
-                await self.leave()
-                try:
-                    await message.reply("Disconnected.")
-                except Exception:
-                    pass
-            elif "joinme" in content or "join" in content:
-                member = message.guild.get_member(message.author.id) if message.guild else None
-                vc = member.voice.channel if member and member.voice else None
-                if not vc:
-                    try:
-                        await message.reply("You're not in a voice channel.")
-                    except Exception:
-                        pass
-                    return
-                try:
-                    if self._voice_client and self._voice_client.is_connected():
-                        await self._voice_client.move_to(vc)
-                    else:
-                        self._voice_client = await vc.connect()
-                    self._channel_id = vc.id
-                    # A fresh connect() has no audio source attached yet.
-                    if not self._voice_client.is_playing():
-                        self._play_stream()
-                    await message.reply(f"Joined {vc.name}.")
-                except Exception as e:
-                    logger.error(f"joinme failed: {e}")
-
-        self._task = asyncio.create_task(self._bot.start(token))
-
-        try:
-            await asyncio.wait_for(self._ready_event.wait(), timeout=BOT_READY_TIMEOUT)
-            self._watchdog_task = asyncio.create_task(self._watchdog_loop())
-            return True
-        except asyncio.TimeoutError:
-            logger.error("Timed out waiting for Discord bot to become ready.")
-            await self.stop()
-            return False
-
-    def _is_bot_running(self) -> bool:
-        return (
-            self._bot is not None
-            and self._task is not None
-            and not self._task.done()
-        )
-
     @property
     def is_connected(self) -> bool:
-        return self._voice_client is not None and self._voice_client.is_connected()
+        return any(v.vc is not None and v.vc.is_connected() for v in self._voices.values())
 
 
-radio_bot = RadioBot()
+radio_bot = RadioManager()

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

@@ -50,7 +50,7 @@ async def fetch_api_key() -> Optional[str]:
     if r.status_code == 401:
         logger.error(
             "Key pickup: C2 rejected the saved pickup secret (401) — it was rotated by a re-enroll elsewhere. "
-            "Re-run install.sh with --token to recover."
+            "Re-run install.sh and give it a fresh enrollment token when it asks (node-26#18)."
         )
         return None
     if r.status_code == 404:

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

@@ -132,6 +132,11 @@ ATTRIBUTION_LOOKAHEAD_SECONDS = 2.0
 CONSOLE_HISTORY_SECONDS = 180.0
 CONSOLE_HISTORY_MAX = 1200
 
+# Sightings of the same radio closer together than this are one transmission.
+# The radio ID shows up in only ~44% of 0.5 s polls during a burst (server-26#304
+# P0 probe, 2026-11-08), so gaps of a few polls inside one key-up are normal.
+TX_MERGE_GAP_SECONDS = 2.0
+
 # How far back to look for a duplicate before appending a grant. OP25's call_log
 # deque drains on read so repeats should not happen, but a re-delivered entry
 # would otherwise inflate the transmission count and the attribution score.
@@ -207,6 +212,21 @@ def _talker(channel: Dict[str, Any]) -> Optional[tuple]:
     return tgid, rid
 
 
+def rebase_radios(radios: List[Dict[str, Any]], shift: float, span: float) -> List[Dict[str, Any]]:
+    """Move call-relative radio offsets into another frame (the trimmed audio file).
+
+    `shift` is call start minus the new origin; offsets land clamped to [0, span]
+    so a talker before the file's first sample sits at 0, not before it.
+    """
+    span = max(0.0, span)
+    return [
+        {**r,
+         "start": round(min(max(0.0, r["start"] + shift), span), 2),
+         "end": round(min(max(0.0, r["end"] + shift), span), 2)}
+        for r in radios
+    ]
+
+
 def _iso(epoch: Optional[float]) -> Optional[str]:
     if epoch is None:
         return None
@@ -254,6 +274,10 @@ class MetadataWatcher:
         self._current_srcaddr: Optional[int] = None
         self._started_at: Optional[float] = None   # audio onset, or grant epoch in fallback mode
         self._transmissions: int = 0
+        # Every radio heard on the open segment, in order: [rid, first_epoch,
+        # last_epoch]. srcaddr above keeps only the latest one; one recording
+        # often holds a dispatcher and a unit (server-26#304).
+        self._tx: List[List[Any]] = []
         # True when the open segment is governed by audio, False for the
         # console fallback. Fixed at open so capture flapping cannot switch the
         # rules underneath a live segment.
@@ -559,6 +583,7 @@ class MetadataWatcher:
             self._current_tgid_name = entry.get("tgtag") or ""
             self._current_freq = entry.get("freq")
             self._current_srcaddr = _as_int(entry.get("rid"))
+            self._note_tx(self._current_srcaddr, started_at)
             self._transmissions += 1
             logger.info(
                 f"Late attribution: segment {self._active_call_id} adopted tgid {tgid} from a grant "
@@ -571,11 +596,13 @@ class MetadataWatcher:
             # of a single conversation lands in one file.
             self._transmissions += 1
             self._refresh_meta_from_log(entry)
+            self._note_tx(_as_int(entry.get("rid")), started_at)
             return
 
         if self._same_channel(tgid, entry.get("freq"), entry.get("tgtag") or ""):
             self._transmissions += 1
             self._refresh_meta_from_log(entry)
+            self._note_tx(_as_int(entry.get("rid")), started_at)
             return
 
         # FORCED SPLIT. Two talkgroups can be back to back with no silence
@@ -626,6 +653,7 @@ class MetadataWatcher:
                 ours = True
                 if srcaddr is not None:
                     self._current_srcaddr = srcaddr
+                    self._note_tx(srcaddr, now)
                 self._last_activity = now
                 self._refresh_meta_from_channel(channel)
 
@@ -638,6 +666,7 @@ class MetadataWatcher:
                 self._current_tgid_name = active[0].get("tag") or ""
                 self._current_freq = active[0].get("freq")
                 self._current_srcaddr = _as_int(active[0].get("srcaddr"))
+                self._note_tx(self._current_srcaddr, now)
                 logger.info(f"Late attribution: segment {self._active_call_id} adopted tgid {tgid} from channel state.")
             return
 
@@ -709,6 +738,7 @@ class MetadataWatcher:
             self._last_tx_end = None
             self._last_activity = now
             self._refresh_meta_from_log(entry)
+            self._note_tx(_as_int(entry.get("rid")), started_at)
             return
 
         await self._close_segment(started_at + _tail_pad(), reason="tgid_change")
@@ -730,6 +760,7 @@ class MetadataWatcher:
                 tx_active = True
                 if srcaddr is not None:
                     self._current_srcaddr = srcaddr
+                    self._note_tx(srcaddr, now)
                 self._refresh_meta_from_channel(channel)
             elif chan_tgid is not None:
                 foreign_active_tgid = chan_tgid
@@ -792,6 +823,28 @@ class MetadataWatcher:
     # Segment open / close
     # ------------------------------------------------------------------
 
+    def _note_tx(self, rid: Optional[int], epoch: float) -> None:
+        """Record a sighting of `rid` talking on the open segment."""
+        if rid is None or self._active_call_id is None:
+            return
+        last = self._tx[-1] if self._tx else None
+        if last is not None and last[0] == rid and epoch - last[2] <= TX_MERGE_GAP_SECONDS:
+            last[2] = max(last[2], epoch)
+            return
+        self._tx.append([rid, epoch, epoch])
+
+    def _radios(self, start: float, end: float) -> List[Dict[str, Any]]:
+        """The segment's transmissions as offsets (seconds) from its start, clamped to it."""
+        span = max(0.0, end - start)
+        return [
+            {
+                "radio": rid,
+                "start": round(min(max(0.0, first - start), span), 2),
+                "end": round(min(max(0.0, last - start), span), 2),
+            }
+            for rid, first, last in self._tx
+        ]
+
     def _refresh_meta_from_log(self, entry: Dict[str, Any]) -> None:
         if not self._current_tgid_name:
             self._current_tgid_name = entry.get("tgtag") or ""
@@ -826,6 +879,8 @@ class MetadataWatcher:
         self._started_at = started_at
         self._transmissions = max(1, transmissions)
         self._audio_driven = audio_driven
+        self._tx = []
+        self._note_tx(srcaddr, started_at)
 
         # Console fallback assumes the transmission is still up; it learns
         # otherwise from the next channel scan.
@@ -876,6 +931,7 @@ class MetadataWatcher:
             "ended_at": _iso(end_epoch),
             "ended_at_epoch": end_epoch,
             "transmissions": self._transmissions,
+            "radios": self._radios(started_at if started_at is not None else end_epoch, end_epoch),
             "end_reason": reason,
             "attributed": attributed,
             "driver": "audio" if self._audio_driven else "console",
@@ -910,6 +966,7 @@ class MetadataWatcher:
         self._current_srcaddr = None
         self._started_at = None
         self._transmissions = 0
+        self._tx = []
         self._tx_active = False
         self._last_tx_end = None
         self._audio_driven = False
@@ -949,6 +1006,10 @@ class MetadataWatcher:
                 self._current_freq = found.freq
             if found.rid is not None and self._current_srcaddr is None:
                 self._current_srcaddr = found.rid
+                if not self._tx:
+                    # Only as the sole entry: the window's latest radio has no
+                    # timing of its own, so it must not reorder real sightings.
+                    self._note_tx(found.rid, start)
             self._transmissions = max(self._transmissions, found.grants)
             return
 

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,9 +10,26 @@ 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]]
+DesiredCallback = Callable[[Dict[str, Any]], Awaitable[None]]
 
 # CONNACK codes meaning "your credentials are wrong", as paho reports them:
 # MQTT 3.1.1 return codes 4/5, and the MQTT 5 reason codes paho maps them to
@@ -42,10 +61,17 @@ 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
         self.on_api_key: Optional[ApiKeyCallback] = None
+        self.on_discord_desired: Optional[DesiredCallback] = None
 
         self._offline_buffer = deque(maxlen=settings.offline_call_buffer_size)
 
@@ -55,6 +81,10 @@ class MQTTManager:
         self._t_metadata = f"nodes/{nid}/metadata"
         self._t_commands = f"nodes/{nid}/commands"
         self._t_config = f"nodes/{nid}/config"
+        # Discord radio desired state (retained by C2) and our actual state —
+        # node-26#25. Token IDs only; the strings come over HTTPS.
+        self._t_discord_desired = f"nodes/{nid}/discord/desired"
+        self._t_discord_actual = f"nodes/{nid}/discord/actual"
         # TODO(mqtt-cutover): dead once enrollment lands client-side. This
         # was the pre-dynsec key-delivery path (server retain-publishes the
         # api_key here after admin approval; node asks for redelivery via
@@ -123,14 +153,24 @@ 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)
             client.subscribe(self._t_api_key, qos=2)  # TODO(mqtt-cutover): see _t_api_key comment above
             client.subscribe(self._t_discovery, qos=0)
             logger.info("MQTT connected.")
@@ -140,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.
@@ -148,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:
@@ -159,6 +203,8 @@ class MQTTManager:
             asyncio.run_coroutine_threadsafe(self.on_command(payload), self._loop)
         elif msg.topic == self._t_config and self.on_config_push:
             asyncio.run_coroutine_threadsafe(self.on_config_push(payload), self._loop)
+        elif msg.topic == self._t_discord_desired and self.on_discord_desired:
+            asyncio.run_coroutine_threadsafe(self.on_discord_desired(payload), self._loop)
         elif msg.topic == self._t_api_key and self.on_api_key:
             asyncio.run_coroutine_threadsafe(self.on_api_key(payload), self._loop)
         elif msg.topic == self._t_discovery:
@@ -247,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)
 
@@ -285,6 +332,12 @@ class MQTTManager:
         else:
             self._publish(self._t_metadata, payload, qos=1)
 
+    def publish_discord_actual(self, payload: dict) -> None:
+        """What the Discord radio manager is actually doing. Not buffered while
+        offline: a fresh report is sent after every change and periodically,
+        and only the newest one means anything."""
+        self._publish(self._t_discord_actual, payload, qos=1)
+
     async def _flush_offline_buffer(self):
         if not self._offline_buffer:
             return
@@ -325,6 +378,7 @@ class MQTTManager:
             "enforce_override_timeout": config.enforce_override_timeout,
             "secondary_sdr_mode": config.secondary_sdr_mode,
             "secondary_sdr_priority": config.secondary_sdr_priority,
+            "version": settings.drb_version,
         }
         payload["sdr_pins"] = config.sdr_pins
         # Conventional channels (node-26#24): name/freq/NAC and whether the
@@ -357,6 +411,15 @@ 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")
         self._publish(self._t_checkin, payload, qos=1)
 
     def _publish(self, topic: str, payload: dict, qos: int = 0, retain: bool = False):
@@ -375,5 +438,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,424 @@
+"""
+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
+    # JSON list of {radio, start, end} — every radio heard on the call
+    # (server-26#304). Sent here as well as on call_end because C2 starts the
+    # pipeline from /upload, before call_end arrives.
+    radios: Optional[str] = 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}"
+    if meta.radios:
+        form["radios"] = meta.radios
+    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

@@ -1,4 +1,5 @@
 import asyncio
+import json
 from contextlib import asynccontextmanager
 from datetime import datetime, timezone
 from typing import Optional
@@ -9,7 +10,7 @@ from app.models import SystemConfig, normalize_sdr_pins, normalize_secondary_pri
 from app.internal.logger import logger
 from app.internal.mqtt_manager import mqtt_manager
 from app.internal import credentials
-from app.internal.metadata_watcher import metadata_watcher
+from app.internal.metadata_watcher import metadata_watcher, rebase_radios
 from app.internal.call_recorder import call_recorder
 from app.internal.discord_radio import radio_bot
 from app.internal.config_manager import (
@@ -93,6 +94,15 @@ async def on_call_end(data: dict):
         data["audio_end_at"] = _iso(recording.audio_end_epoch)
         data["audio_start_epoch"] = recording.audio_start_epoch
         data["audio_end_epoch"] = recording.audio_end_epoch
+        # Radio offsets were measured from the call start; C2 maps an offset
+        # to wall clock as audio_start_epoch + t (above), so put them in the
+        # audio file's frame (server-26#304).
+        if data.get("radios") and data.get("started_at_epoch") is not None:
+            data["radios"] = rebase_radios(
+                data["radios"],
+                shift=data["started_at_epoch"] - recording.audio_start_epoch,
+                span=recording.audio_end_epoch - recording.audio_start_epoch,
+            )
         data["audio_lead_trimmed"] = round(recording.lead_trimmed, 3)
         data["audio_tail_trimmed"] = round(recording.tail_trimmed, 3)
         if recording.clamped_seconds:
@@ -108,11 +118,24 @@ async def on_call_end(data: dict):
             system_id=node_cfg.assigned_system_id,
             audio_start_epoch=recording.audio_start_epoch,
             audio_end_epoch=recording.audio_end_epoch,
+            radios=json.dumps(data["radios"]) if data.get("radios") else None,
         )
         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.
@@ -144,20 +167,11 @@ async def on_command(payload: dict):
     action = payload.get("action")
     logger.info(f"Command received: {action}")
 
-    if action == "discord_join":
+    if action in ("discord_join", "discord_leave"):
-        token = payload.get("token")
+        # Pre-desired-state C2. Discord streams now arrive on
-        if not token:
+        # nodes/{id}/discord/desired (node-26#25); a raw-token join is refused
-            logger.error("discord_join command missing token — ignoring.")
+        # so a bot token is never taken from an MQTT command again.
-            return
+        logger.warning(f"Ignoring legacy {action} command — Discord radio is driven by desired state now.")
-        await radio_bot.join(
-            guild_id=int(payload["guild_id"]),
-            channel_id=int(payload["channel_id"]),
-            token=token,
-            call_active=metadata_watcher.is_active,
-            system_name=payload.get("system_name"),
-        )
-    elif action == "discord_leave":
-        await radio_bot.leave()
     elif action in ("set_sdr_config", "set_secondary_priority"):
         from app.internal.sdr_settings import set_sdr_settings
         try:
@@ -188,6 +202,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.")
 
 
@@ -300,8 +316,9 @@ async def lifespan(app: FastAPI):
     mqtt_manager.on_command = on_command
     mqtt_manager.on_config_push = on_config_push
     mqtt_manager.on_api_key = on_api_key
+    mqtt_manager.on_discord_desired = radio_bot.apply_desired
 
-    # Start services (radio_bot starts on-demand when a discord_join command arrives)
+    # Start services (radio_bot starts when C2's retained discord/desired arrives)
     await mqtt_manager.connect()
     await metadata_watcher.start()
     await call_recorder.start()  # persistent PulseAudio ring buffer

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={
+                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}
@@ -400,15 +416,25 @@ async def set_sdr(body: SdrSettingsBody):
     return {"ok": True, "running": running}
 
 
+@router.get("/discord")
+async def discord_status():
+    """What the node's radio bots are doing (desired state from C2, actual here)."""
+    return radio_bot.snapshot()
+
+
 @router.post("/discord/join")
-async def discord_join(guild_id: int, channel_id: int):
+async def discord_join():
-    ok = await radio_bot.join(guild_id, channel_id)
+    # node-26#25 item 4: this used to call join() without a token and 500 every
-    if not ok:
+    # time. A node cannot lease a radio bot by itself — C2 owns the pool.
-        raise HTTPException(500, "Failed to join voice channel")
+    raise HTTPException(
-    return {"ok": True}
+        409,
+        "Discord radio channels are started from Discord (/join) or the DRB web app; "
+        "the node cannot pick a radio bot on its own.",
+    )
 
 
 @router.post("/discord/leave")
 async def discord_leave():
-    await radio_bot.leave()
+    """Local stop: log every radio bot out until DRB sends a new channel list."""
+    await radio_bot.pause_locally()
     return {"ok": True}

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/requirements.txt

@@ -3,7 +3,9 @@ uvicorn[standard]
 pydantic-settings
 paho-mqtt>=2.0.0
 httpx
-discord.py[voice]
+# >=2.7: Discord requires DAVE (E2EE) voice since 2026-03-01; 2.7 is the first
+# discord.py with it (the [voice] extra pulls `davey`). Older releases cannot join voice.
+discord.py[voice]>=2.7
 PyNaCl
 python-multipart
 pytest

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

@@ -0,0 +1,345 @@
+"""
+Edge Discord radio manager (node-26#25): desired-state reconcile, logout on
+release, watchdog that survives a dead session, MQTT-loss fail-safe.
+
+Discord itself is faked: FakeClient logs "in" by firing on_ready, guilds and
+channels are plain objects, voice connections record play/disconnect.
+"""
+import asyncio
+
+import discord
+import pytest
+
+from app.internal import discord_radio
+from app.internal.discord_radio import RadioManager
+
+
+class FakeVC:
+    def __init__(self, channel):
+        self.channel = channel
+        self.connected = True
+        self.playing = False
+        self.plays = 0
+
+    def is_connected(self):
+        return self.connected
+
+    def is_playing(self):
+        return self.playing
+
+    def play(self, source, after=None):
+        self.playing = True
+        self.plays += 1
+
+    def stop(self):
+        self.playing = False
+
+    async def disconnect(self, force=False):
+        self.connected = False
+        self.playing = False
+        self.channel.guild.voice_client = None
+
+    async def move_to(self, channel):
+        self.channel = channel
+        channel.guild.voice_client = self
+
+
+class FakeChannel:
+    def __init__(self, guild, cid, kind=discord.ChannelType.voice):
+        self.guild, self.id, self.type = guild, cid, kind
+        self.perms = discord.Permissions(view_channel=True, connect=True, speak=True)
+        self.user_limit = 0
+        self.members = []
+        self.connect_error = None
+
+    def permissions_for(self, member):
+        assert member is self.guild.me
+        return self.perms
+
+    async def connect(self):
+        if self.connect_error:
+            raise self.connect_error
+        vc = FakeVC(self)
+        self.guild.voice_client = vc
+        return vc
+
+
+class FakeGuild:
+    def __init__(self, gid, channel_ids):
+        self.id = gid
+        self.me = object()
+        self.voice_client = None
+        self.channels = {c: FakeChannel(self, c) for c in channel_ids}
+
+    def get_channel(self, cid):
+        return self.channels.get(cid)
+
+
+class FakeClient:
+    instances: list = []
+
+    def __init__(self, guilds, fail_login=False):
+        self.guilds = guilds
+        self.fail_login = fail_login
+        self.closed = False
+        self.handlers = {}
+        self.token = None
+        self.user = "radio-bot"
+        FakeClient.instances.append(self)
+
+    def event(self, coro):
+        self.handlers[coro.__name__] = coro
+        return coro
+
+    async def start(self, token):
+        self.token = token
+        if self.fail_login:
+            raise discord.LoginFailure("Improper token has been passed.")
+        await self.handlers["on_ready"]()
+        while not self.closed:
+            await asyncio.sleep(0.01)
+
+    async def close(self):
+        self.closed = True
+
+    def is_closed(self):
+        return self.closed
+
+    def get_guild(self, gid):
+        return self.guilds.get(gid)
+
+    async def change_presence(self, activity=None):
+        self.presence = activity
+
+
+@pytest.fixture
+def env(monkeypatch):
+    FakeClient.instances = []
+    guilds = {1: FakeGuild(1, [11, 12]), 2: FakeGuild(2, [21])}
+    published = []
+    fetched = []
+    state = {"mqtt": True, "fail_login": False}
+
+    async def fetch(token_id):
+        fetched.append(token_id)
+        return f"secret-{token_id}"
+
+    async def publish(payload):
+        published.append(payload)
+
+    async def pulse_ready(timeout=None):
+        return True
+    monkeypatch.setattr(discord_radio.pulse, "wait_until_ready", pulse_ready)
+    monkeypatch.setattr(discord_radio.pulse, "is_ready", lambda: True)
+
+    mgr = RadioManager(
+        client_factory=lambda: FakeClient(guilds, fail_login=state["fail_login"]),
+        token_fetcher=fetch, publisher=publish, audio_factory=lambda: object(),
+        mqtt_up=lambda: state["mqtt"], ready_timeout=1,
+    )
+    yield type("Env", (), {"mgr": mgr, "guilds": guilds, "published": published,
+                           "fetched": fetched, "state": state})
+    # Leave no background tasks behind.
+    if mgr._watchdog:
+        mgr._watchdog.cancel()
+
+
+def _desired(seq, *streams):
+    return {"seq": seq, "streams": [
+        {"stream_id": f"{g}:{c}", "guild_id": str(g), "channel_id": str(c), "token_id": t, "system_name": "County"}
+        for g, c, t in streams
+    ]}
+
+
+async def test_apply_connects_and_reports(env):
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    actual = env.published[-1]
+    assert actual["applied_seq"] == 1
+    assert actual["sessions"] == ["t1"]
+    assert actual["streams"] == {"1:11": {"state": "connected", "error": None, "code": None}}
+    assert env.mgr.is_connected
+    vc = env.guilds[1].voice_client
+    assert vc.plays == 1 and vc.channel.id == 11
+    assert FakeClient.instances[0].presence.name == "County"
+
+
+async def test_one_session_serves_two_guilds(env):
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1"), (2, 21, "t1")))
+    assert len(FakeClient.instances) == 1
+    assert env.published[-1]["sessions"] == ["t1"]
+    assert env.guilds[1].voice_client.is_connected() and env.guilds[2].voice_client.is_connected()
+
+
+async def test_dropping_the_last_stream_logs_the_bot_out(env):
+    """node-26#25 item 1: leave() used to keep the gateway session alive."""
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1"), (2, 21, "t1")))
+    client = FakeClient.instances[0]
+    await env.mgr.apply_desired(_desired(2, (2, 21, "t1")))
+    assert not client.closed                      # still needed for guild 2
+    assert env.guilds[1].voice_client is None
+    await env.mgr.apply_desired(_desired(3))
+    assert client.closed                          # logged OUT, not just out of voice
+    assert env.published[-1] == {"applied_seq": 3, "sessions": [], "streams": {}}
+
+
+async def test_stale_desired_is_ignored(env):
+    await env.mgr.apply_desired(_desired(5, (1, 11, "t1")))
+    await env.mgr.apply_desired(_desired(4))
+    assert env.mgr.snapshot()["applied_seq"] == 5 and env.mgr.is_connected
+
+
+async def test_dead_session_is_recovered_by_the_watchdog(env):
+    """node-26#25 item 2: the old watchdog cancelled itself on rejoin."""
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    first = FakeClient.instances[0]
+    first.closed = True                           # bot task dies
+    await asyncio.sleep(0.05)
+    env.guilds[1].voice_client.connected = False
+    await env.mgr.tick()
+    assert len(FakeClient.instances) == 2 and not FakeClient.instances[1].closed
+    assert env.mgr.snapshot()["streams"]["1:11"]["state"] == "connected"
+    # The watchdog task itself is still alive after the recovery.
+    assert env.mgr._watchdog is not None and not env.mgr._watchdog.done()
+
+
+async def test_dropped_voice_is_reconnected(env):
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    env.guilds[1].voice_client.connected = False
+    env.guilds[1].voice_client = None
+    await env.mgr.tick()
+    assert env.guilds[1].voice_client.is_connected()
+
+
+async def test_bot_not_in_guild_is_reported_not_swallowed(env):
+    await env.mgr.apply_desired(_desired(1, (9, 91, "t1")))
+    st = env.published[-1]["streams"]["9:91"]
+    assert st["state"] == "error" and st["code"] == "not_in_guild"
+
+
+async def test_login_failure_is_reported(env):
+    env.state["fail_login"] = True
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    st = env.published[-1]["streams"]["1:11"]
+    assert st["state"] == "error" and st["code"] == "login_failed"
+    assert env.published[-1]["sessions"] == []
+
+
+async def test_mqtt_loss_failsafe_logs_out_then_resumes(env, monkeypatch):
+    monkeypatch.setattr(discord_radio.settings, "discord_mqtt_loss_stop_seconds", 300)
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    client = FakeClient.instances[0]
+    env.state["mqtt"] = False
+    await env.mgr.tick(now=1000.0)
+    assert not client.closed                       # inside the window
+    await env.mgr.tick(now=1299.0)
+    assert not client.closed
+    await env.mgr.tick(now=1300.0)
+    assert client.closed and env.mgr.snapshot()["sessions"] == []
+    # Token strings are not kept across a fail-safe logout: re-login refetches
+    # from C2, which refuses if the lease has moved on.
+    assert env.mgr._tokens == {}
+    env.state["mqtt"] = True
+    before = len(env.fetched)
+    await env.mgr.tick(now=1400.0)
+    assert len(env.fetched) == before + 1
+    assert env.mgr.snapshot()["streams"]["1:11"]["state"] == "connected"
+
+
+async def test_token_fetch_refused_reports_error(env):
+    async def refuse(token_id):
+        return None
+    env.mgr._fetch_token = refuse
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    st = env.published[-1]["streams"]["1:11"]
+    assert st["state"] == "error" and FakeClient.instances == []
+
+
+async def test_local_pause_logs_out_until_next_desired(env):
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    await env.mgr.pause_locally()
+    assert FakeClient.instances[0].closed
+    assert env.published[-1]["streams"]["1:11"]["state"] == "paused_locally"
+    await env.mgr.apply_desired(_desired(2, (1, 11, "t1")))
+    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:
+        assert client.intents.message_content is False
+        assert client.intents.voice_states and client.intents.guilds
+    finally:
+        await client.close()
+
+
+def _presence(client):
+    activity = getattr(client, "presence", None)
+    return activity.name if activity else None
+
+
+async def test_missing_connect_permission_is_named_not_a_timeout(env):
+    """Discord ignores a join the bot may not make; say which permission is missing."""
+    env.guilds[1].channels[11].perms = discord.Permissions(view_channel=True, speak=True)
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    st = env.published[-1]["streams"]["1:11"]
+    assert st["state"] == "error" and st["code"] == "missing_permissions"
+    assert "Connect" in st["error"] and "View Channel" not in st["error"]
+    assert env.guilds[1].voice_client is None
+    assert _presence(FakeClient.instances[0]) is None   # not "Listening to …" while out of voice
+
+
+async def test_full_channel_is_reported(env):
+    ch = env.guilds[1].channels[11]
+    ch.user_limit, ch.members = 2, [object(), object()]
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    st = env.published[-1]["streams"]["1:11"]
+    assert st["code"] == "channel_full" and "2/2" in st["error"]
+
+
+async def test_bots_own_lingering_voice_state_does_not_fill_the_channel(env):
+    """A dropped connection can leave the bot listed in the channel; that is
+    not a slot it needs to rejoin."""
+    guild = env.guilds[1]
+    guild.me = type("Me", (), {"id": 999})()
+    ch = guild.channels[11]
+    ch.user_limit, ch.members = 2, [object(), guild.me]
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    assert env.published[-1]["streams"]["1:11"]["state"] == "connected"
+
+
+async def test_connect_timeout_has_a_readable_error(env):
+    env.guilds[1].channels[11].connect_error = asyncio.TimeoutError()
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    st = env.published[-1]["streams"]["1:11"]
+    assert st["state"] == "error" and st["code"] == "connect_timeout"
+    assert "timed out" in st["error"] and "TimeoutError: )" not in st["error"]
+
+
+async def test_presence_is_cleared_when_voice_is_lost_and_rejoin_fails(env):
+    """The owner saw "Listening to <system>" on a bot that was not in voice:
+    presence was set once on join and never cleared."""
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    client = FakeClient.instances[0]
+    assert _presence(client) == "County"
+    env.guilds[1].voice_client.connected = False
+    env.guilds[1].voice_client = None
+    env.guilds[1].channels[11].connect_error = RuntimeError("voice handshake failed")
+    await env.mgr.tick()
+    st = env.mgr.snapshot()["streams"]["1:11"]
+    assert st["state"] == "error" and st["code"] == "connect_failed"
+    assert _presence(client) is None
+    env.guilds[1].channels[11].connect_error = None
+    await env.mgr.tick()
+    assert _presence(client) == "County"

drb-edge-node/tests/test_host_updater.py

@@ -24,21 +24,42 @@ pytestmark = pytest.mark.skipif(
 )
 
 REPO = "https://git.example.net/logan/node-26.git"
-OLD, NEW = "a" * 40, "b" * 40
+OLD, NEW, TREE = "a" * 40, "b" * 40, "c" * 40
+REPO_ROOT = SCRIPT.parents[1]
 
 STUB_GIT = r'''#!/usr/bin/env bash
 echo "git $*" >> "$STUB/calls"
-while [ "$1" = -c ]; do shift 2; done
+while [ "$1" = -c ] || [ "$1" = -C ]; do shift 2; done
 case "$1" in
-  rev-parse) if [ "$2" = HEAD ]; then cat "$STUB/head"; else echo "$STUB_NEW"; fi ;;
+  rev-parse)
+    if [ "$2" = HEAD ]; then cat "$STUB/head"
+    elif [ "$2" = --verify ]; then echo "$STUB_TREE"   # tree hash of <commit>:<dir>
+    else echo "$STUB_NEW"; fi ;;
   remote) echo "$STUB_ORIGIN" ;;
   fetch) exit 0 ;;
   show-ref) [ "$4" = "refs/tags/v1" ] ;;
+  cat-file) [ -z "${STUB_NO_PIN:-}" ] ;;   # target has scripts/image-tags.sh
+  show) case "$2" in
+          *:scripts/image-tags.sh) cat "$STUB_REPO/scripts/image-tags.sh" ;;
+          *:docker-compose.yml) cat "$STUB_REPO/docker-compose.yml" ;;
+        esac ;;
   checkout) echo "${@: -1}" > "$STUB/head" ;;
+  tag)
+    if [ "$2" = --points-at ]; then
+      if [ "$3" = HEAD ]; then cat "$STUB/tags_head" 2>/dev/null; else cat "$STUB/tags_new" 2>/dev/null; fi
+    else cat "$STUB/tags_all" 2>/dev/null; fi
+    exit 0 ;;
   *) exit 0 ;;
 esac
 '''
-STUB_DOCKER = '#!/usr/bin/env bash\necho "docker $*" >> "$STUB/calls"\n'
+STUB_DOCKER = r'''#!/usr/bin/env bash
+echo "docker $*" >> "$STUB/calls"
+case "$*" in
+  *"config --images"*) echo "reg.example/edge-node:$STUB_PULLED_TAG" ;;
+  "image inspect"*) [ -z "${STUB_MISSING_IMAGE:-}" ] || exit 1 ;;
+esac
+exit 0
+'''
 STUB_CURL = r'''#!/usr/bin/env bash
 echo "curl $*" >> "$STUB/calls"
 for a in "$@"; do case "$a" in @*) cat "${a#@}" >> "$STUB/headers" ;; esac; done
@@ -79,6 +100,7 @@ def env(tmp_path):
     e.update({
         "PATH": f"{bindir}:{os.environ['PATH']}",
         "STUB": str(stub), "STUB_NEW": NEW, "STUB_ORIGIN": REPO,
+        "STUB_TREE": TREE, "STUB_REPO": str(REPO_ROOT), "STUB_PULLED_TAG": f"src-{TREE}",
         "DRB_UPDATER_CONF": str(conf),
         "DRB_UPDATER_STATE": str(tmp_path / "state"),
         "DRB_UPDATER_LOCK": str(tmp_path / "lock"),
@@ -94,13 +116,17 @@ def _calls(env):
     return (env["stub"] / "calls").read_text()
 
 
+def _pulled(calls: str) -> bool:
+    return any(line.startswith("docker compose ") and line.endswith(" pull") for line in calls.splitlines())
+
+
 def test_no_request_means_checkin_only(env):
     r = _run(env)
     assert r.returncode == 0, r.stderr
     calls = _calls(env)
     assert "update-check" in calls
     assert "docker" not in calls
-    assert "fetch" not in calls
+    assert "checkout" not in calls
 
 
 def test_c2_request_updates_once_and_reports(env):
@@ -108,7 +134,7 @@ def test_c2_request_updates_once_and_reports(env):
     r = _run(env)
     assert r.returncode == 0, r.stderr + r.stdout
     calls = _calls(env)
-    assert "docker compose pull" in calls
+    assert _pulled(calls)
     assert "docker compose up -d --no-build --force-recreate --remove-orphans" in calls
     assert (env["stub"] / "head").read_text().strip() == NEW
     assert '"applied_request_at":"2026-10-04T12:00:00+00:00"' in calls
@@ -126,10 +152,21 @@ def test_mqtt_sentinel_triggers_update_and_is_consumed(env):
     assert r.returncode == 0, r.stderr
     assert not sentinel.exists()
     calls = _calls(env)
-    assert "docker compose pull" in calls
+    assert _pulled(calls)
     assert "evil" not in calls and "rm -rf" not in calls
 
 
+def test_update_prunes_build_cache_and_stale_images_after_up(env):
+    # node-26#41: the build cache was never pruned and filled node-002's disk.
+    (env["stub"] / "response").write_text(json.dumps({"update_requested_at": "2026-10-04T12:00:00+00:00"}))
+    assert _run(env).returncode == 0
+    lines = _calls(env).splitlines()
+    up = next(i for i, ln in enumerate(lines) if ln.startswith("docker compose up -d"))
+    builder = next(i for i, ln in enumerate(lines) if ln.startswith("docker builder prune -af --filter until="))
+    images = next(i for i, ln in enumerate(lines) if ln.startswith("docker image prune -af --filter until="))
+    assert up < builder and up < images  # only after the new containers are running
+
+
 def test_c2_response_is_never_used_as_a_ref_or_command(env):
     (env["stub"] / "response").write_text(json.dumps({"update_requested_at": "$(touch /tmp/pwned); main"}))
     r = _run(env)
@@ -153,7 +190,7 @@ def test_refuses_when_env_registry_differs(env):
     (env["inst"] / "configs" / "update-requested").write_text("")
     r = _run(env)
     assert r.returncode != 0
-    assert "docker compose pull" not in _calls(env)
+    assert not _pulled(_calls(env))
 
 
 def test_non_https_repo_in_conf_is_rejected(env):
@@ -177,3 +214,113 @@ def test_plain_http_c2_skips_checkin(env):
     r = _run(env)
     assert r.returncode == 0
     assert "curl" not in _calls(env)
+
+
+# ── node-26#19: images pinned to the target's source tree ─────────────────
+
+def test_pins_images_and_pulls_before_checkout(env):
+    (env["inst"] / "configs" / "update-requested").write_text("")
+    r = _run(env)
+    assert r.returncode == 0, r.stderr + r.stdout
+    lines = _calls(env).splitlines()
+    pull = next(i for i, ln in enumerate(lines) if ln.startswith("docker compose ") and ln.endswith(" pull"))
+    checkout = next(i for i, ln in enumerate(lines) if ln.startswith("git") and " checkout " in ln)
+    assert pull < checkout, "images must be pulled before the new code is checked out"
+    dotenv = (env["inst"] / ".env").read_text()
+    for var in ("DRB_TAG_EDGE_NODE", "DRB_TAG_OP25", "DRB_TAG_ICECAST", "DRB_TAG_SECONDARY_SDR"):
+        assert f"{var}=src-{TREE}\n" in dotenv
+    assert dotenv.startswith("NODE_ID=node-7\n")  # existing lines kept
+    assert not (env["inst"] / ".drb-update-compose.yml").exists()
+
+
+def test_missing_pinned_image_changes_nothing(env):
+    env["env"]["STUB_MISSING_IMAGE"] = "1"
+    (env["inst"] / "configs" / "update-requested").write_text("")
+    r = _run(env)
+    assert r.returncode != 0
+    assert (env["stub"] / "head").read_text().strip() == OLD
+    calls = _calls(env)
+    assert " checkout " not in calls
+    assert "up -d" not in calls
+    assert "DRB_TAG_" not in (env["inst"] / ".env").read_text()
+
+
+def test_target_without_pinning_still_updates(env):
+    env["env"]["STUB_NO_PIN"] = "1"
+    (env["inst"] / "configs" / "update-requested").write_text("")
+    r = _run(env)
+    assert r.returncode == 0, r.stderr + r.stdout
+    assert (env["stub"] / "head").read_text().strip() == NEW
+    assert _pulled(_calls(env))
+    assert "DRB_TAG_" not in (env["inst"] / ".env").read_text()
+
+
+# ---------------------------------------------------------------------------
+# Releases (node-26#49): a newer version on the node's channel is a reason to
+# update by itself, and versions compare per field as integers.
+# ---------------------------------------------------------------------------
+
+def _channel(env, ref, all_tags, target_tags, head_tags):
+    env["conf"].write_text(env["conf"].read_text().replace("REF=v1", f"REF={ref}"))
+    (env["stub"] / "tags_all").write_text("\n".join(all_tags) + "\n")
+    (env["stub"] / "tags_new").write_text("\n".join(target_tags) + "\n")
+    (env["stub"] / "tags_head").write_text("\n".join(head_tags) + "\n")
+
+
+def test_latest_channel_installs_a_newer_release_unasked(env):
+    # 1.1.10 is newer than 1.1.9: compared as integers, not strings or floats.
+    _channel(env, "latest", ["v1", "v1.1.9", "v1.1.10", "v1.1.2"], ["v1.1.10"], ["v1.1.9"])
+    r = _run(env)
+    assert r.returncode == 0, r.stderr + r.stdout
+    assert "release 1.1.10 on latest (installed 1.1.9)" in r.stdout
+    assert (env["stub"] / "head").read_text().strip() == NEW
+    assert "DRB_VERSION=1.1.10" in (env["inst"] / ".env").read_text()
+
+
+def test_never_downgrades(env):
+    _channel(env, "latest", ["v1.1.9", "v1.2.0"], ["v1.1.9"], ["v1.2.0"])
+    r = _run(env)
+    assert r.returncode == 0, r.stderr
+    assert "docker" not in _calls(env)
+
+
+def test_same_release_is_not_reinstalled(env):
+    _channel(env, "v1", ["v1", "v1.1.3"], ["v1", "v1.1.3"], ["v1", "v1.1.3"])
+    assert _run(env).returncode == 0
+    assert "docker" not in _calls(env)
+
+
+def test_promoted_major_tag_installs_on_stable_nodes(env):
+    _channel(env, "v1", ["v1", "v1.1.0", "v1.1.4"], ["v1", "v1.1.4"], ["v1.1.0"])
+    r = _run(env)
+    assert r.returncode == 0, r.stderr + r.stdout
+    assert "release 1.1.4 on v1" in r.stdout
+    assert (env["stub"] / "head").read_text().strip() == NEW
+
+
+def test_untagged_target_still_needs_a_request(env):
+    """v1 before its first promotion points at a commit with no release tag."""
+    _channel(env, "v1", ["v1"], ["v1"], [])
+    assert _run(env).returncode == 0
+    assert "docker" not in _calls(env)
+
+
+def test_next_version_counts_patches_as_integers(tmp_path):
+    script = REPO_ROOT / "scripts" / "next-version.sh"
+    g = ["git", "-c", "user.email=t@t", "-c", "user.name=t", "-C", str(tmp_path)]
+    subprocess.run(["git", "init", "-q", str(tmp_path)], check=True)
+    (tmp_path / "VERSION").write_text("1.1\n")
+    subprocess.run(g + ["commit", "-q", "--allow-empty", "-m", "a"], check=True)
+
+    def nxt():
+        return subprocess.run(["bash", str(script)], cwd=tmp_path, capture_output=True, text=True,
+                              check=True).stdout.strip()
+
+    assert nxt() == "v1.1.0"
+    for t in ("v1.1.0", "v1.1.9", "v1.0.30"):
+        subprocess.run(g + ["tag", t], check=True)
+    assert nxt() == "", "a commit that already has a release gets no second one"
+    subprocess.run(g + ["commit", "-q", "--allow-empty", "-m", "b"], check=True)
+    assert nxt() == "v1.1.10"
+    (tmp_path / "VERSION").write_text("1.2\n")
+    assert nxt() == "v1.2.0"

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_metadata_watcher.py

@@ -27,6 +27,7 @@ from app.internal.metadata_watcher import (
     MAX_SEGMENT_SECONDS,
     MetadataWatcher,
     OP25_OFFLINE_GRACE,
+    rebase_radios,
 )
 from app.internal.op25_client import TerminalUpdate, parse_terminal_messages
 
@@ -1147,3 +1148,86 @@ async def test_console_history_is_bounded(hearing, clock, audio):
     clock.advance(CONSOLE_HISTORY_SECONDS + 1)
     await tick(hearing, update())
     assert all(e.epoch >= clock.now - CONSOLE_HISTORY_SECONDS for e in hearing._console)
+
+
+# ---------------------------------------------------------------------------
+# Every radio on the call (server-26#304) — not just the last srcaddr
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_call_end_lists_every_radio_in_order(watcher, clock):
+    """Dispatcher, unit, dispatcher on one talkgroup: three transmissions."""
+    start = clock.now
+    await tick(watcher, update(call_log=[grant(1234, start, rid=555)],
+                               channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=777)]))
+    clock.advance(0.5)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=777)]))
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(0.5)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
+    clock.advance(settings.call_idle_timeout + 0.5)
+    await tick(watcher, update(channels=[channel()]))
+
+    payload = watcher.on_call_end.call_args[0][0]
+    assert [r["radio"] for r in payload["radios"]] == [555, 777, 555]
+    assert payload["radios"][0] == {"radio": 555, "start": 0.0, "end": 1.0}
+    assert payload["radios"][1] == {"radio": 777, "start": 2.0, "end": 2.5}
+    # srcaddr keeps its old meaning (the latest talker) for existing readers.
+    assert payload["srcaddr"] == 555
+
+
+@pytest.mark.asyncio
+async def test_missed_polls_inside_one_keyup_stay_one_transmission(watcher, clock):
+    """The radio ID drops out of some polls mid-transmission; that is not a new key-up."""
+    await tick(watcher, update(call_log=[grant(1234, clock.now, rid=555)],
+                               channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(0.5)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(0.5)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
+    clock.advance(settings.call_idle_timeout + 0.5)
+    await tick(watcher, update(channels=[channel()]))
+
+    radios = watcher.on_call_end.call_args[0][0]["radios"]
+    assert radios == [{"radio": 555, "start": 0.0, "end": 1.5}]
+
+
+@pytest.mark.asyncio
+async def test_radio_list_resets_between_calls(watcher, clock):
+    await tick(watcher, update(call_log=[grant(1234, clock.now, rid=555)]))
+    clock.advance(0.2)
+    await tick(watcher, update(call_log=[grant(4321, clock.now, rid=888)]))
+
+    first = watcher.on_call_end.call_args[0][0]
+    assert [r["radio"] for r in first["radios"]] == [555]
+    assert watcher._tx == [[888, clock.now, clock.now]]
+
+
+@pytest.mark.asyncio
+async def test_audio_mode_records_a_radio_seen_only_in_a_grant(hearing, clock, audio):
+    """A short key-up between channel polls is only visible as a call_log grant."""
+    audio.speak()
+    await tick(hearing, update(call_log=[grant(1234, clock.now, rid=555)],
+                               channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(1.0)
+    audio.speak()
+    await tick(hearing, update(call_log=[grant(1234, clock.now, rid=777)]))
+    clock.advance(settings.call_silence_timeout + 0.5)
+    await tick(hearing, update(channels=[channel()]))
+
+    radios = hearing.on_call_end.call_args[0][0]["radios"]
+    assert [r["radio"] for r in radios] == [555, 777]
+
+
+def test_rebase_radios_moves_offsets_into_the_audio_frame():
+    """The file starts 0.25 s of pre-roll before the call, then 1.0 s of lead was trimmed."""
+    radios = [{"radio": 555, "start": 0.0, "end": 2.0}, {"radio": 777, "start": 3.0, "end": 9.0}]
+    out = rebase_radios(radios, shift=-0.75, span=6.0)
+    assert out == [{"radio": 555, "start": 0.0, "end": 1.25}, {"radio": 777, "start": 2.25, "end": 6.0}]

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_upload_spool.py

@@ -0,0 +1,237 @@
+"""
+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
+
+
+def test_form_carries_the_radio_list_and_omits_it_when_empty():
+    """C2 starts correlation from /upload, so the radio list rides on it (server-26#304)."""
+    radios = json.dumps([{"radio": 555, "start": 0.0, "end": 1.0}])
+    assert upload_spool._form(UploadMeta(call_id="c1", radios=radios))["radios"] == radios
+    assert "radios" not in upload_spool._form(UploadMeta(call_id="c1"))

install.sh

@@ -1,16 +1,21 @@
 #!/usr/bin/env bash
 #
-# DRB edge node — one-shot bootstrap for a clean Raspberry Pi OS (arm64).
+# DRB edge node — one-shot bootstrap for a clean Raspberry Pi OS (arm64) or Debian (amd64).
 #
 #   curl -fsSL https://git.vpn.cusano.net/logan/node-26/raw/tag/v1/install.sh \
-#     | sudo bash -s -- --token DRB-xxxx --node-id node-003 \
+#     | sudo bash -s -- --node-id node-003 \
 #         --c2-url https://api.<domain> --mqtt-broker mqtt.<domain>
 #
+# It prompts (hidden input) for the enrollment token. The token is never
+# taken from the command line by design (node-26#18, CISO condition from
+# server-26#61): argv lands in shell history and in `ps`. Unattended installs
+# use --token-file PATH (a 0600 file) or an exported DRB_ENROLLMENT_TOKEN.
+#
 # (fetch install.sh from the same tag it installs; use raw/branch/main with
 #  --track-main for the owner's own always-latest nodes)
 #
 # Hosting, the pinned ref, and setup.sh's retirement are settled (owner
-# decisions, 2026-09-06). One standing hazard remains — D3 below.
+# decisions, 2026-09-06). D3 below is mitigated by image pinning (node-26#19).
 #
 # What it does, in order:
 #   1. Preflight (root, arch, apt, SDR present)
@@ -48,15 +53,18 @@
 #       the mint panel / node-26#4). DEFAULT_REF below is `v1`. `--track-main`
 #       stays as an opt-in for the owner's own nodes.
 #
-# STANDING HAZARD
+# MITIGATED HAZARD
 #
 #   D3  SOURCE-OVERLAY. docker-compose.yml bind-mounts ./drb-edge-node/app and
-#       ./op25-container/app OVER the image's /app. So even in prebuilt mode
+#       ./op25-container/app OVER the image's /app, so the running Python is
-#       the running Python is the CLONED REF's code against the pulled image's
+#       the CLONED REF's code against the pulled image's dependencies. Those
-#       dependencies. `v1` == the commit CI built the current :latest/:stable
+#       used to be :latest/:stable, built from main — silently a different
-#       from, so today they match — but the moment `v1` and the image tags
+#       commit from `v1`. Since node-26#19 every image is pinned to
-#       diverge this silently mixes them. Fix: drop those two mounts from the
+#       src-<git tree hash of its own directory at the checked-out ref>
-#       prod compose path, or always retag images from the same ref as `v1`.
+#       (scripts/image-tags.sh, written into .env as DRB_TAG_*), which CI tags
+#       identically. The mounts now overlay the same code the image was built
+#       from. A pinned image that was never built fails the pull loudly; the
+#       "Release images" workflow builds any a release tag is missing.
 # ---------------------------------------------------------------------------
 set -euo pipefail
 
@@ -78,6 +86,7 @@ MQTT_BROKER="${DRB_MQTT_BROKER:-}"
 MQTT_PORT="${DRB_MQTT_PORT:-8883}"
 MQTT_TLS="${DRB_MQTT_TLS:-true}"
 ENROLLMENT_TOKEN="${DRB_ENROLLMENT_TOKEN:-}"
+unset DRB_ENROLLMENT_TOKEN   # read once; don't leak it into every child's env
 DASHBOARD_USER="${DRB_DASHBOARD_USERNAME:-admin}"
 DASHBOARD_PASS="${DRB_DASHBOARD_PASSWORD:-}"
 REGISTRY="${DRB_IMAGE_REGISTRY:-git.vpn.cusano.net}"   # D1 (public host)
@@ -108,7 +117,10 @@ usage() {
   cat <<'USAGE'
 Usage: install.sh [options]
 
-  --token TOKEN        Enrollment token (Settings -> Nodes -> New token)
+  (enrollment token)   Prompted for on the terminal, input hidden. Mint one at
+                       Settings -> Nodes -> New token.
+  --token-file PATH    read the enrollment token from PATH (unattended installs;
+                       keep it mode 0600 and delete it afterwards)
   --node-id ID         Unique node id, e.g. node-003
   --name NAME          Display name (default: node id)
   --lat N --lon N      Decimal degrees for the map
@@ -132,13 +144,19 @@ Usage: install.sh [options]
 
 Every option also has an env var: DRB_NODE_ID, DRB_C2_URL, DRB_ENROLLMENT_TOKEN,
 DRB_MQTT_BROKER, DRB_REF, DRB_INSTALL_DIR, DRB_REGISTRY_USER/PASS, ...
-Secrets are read from the environment or prompted on the TTY, never from a pipe.
+Secrets are read from a file, the environment or a TTY prompt, never from a
+pipe. Never put the token on a command line: `--token X` and
+`sudo DRB_ENROLLMENT_TOKEN=X bash` both expose it in shell history and `ps`.
+An exported DRB_ENROLLMENT_TOKEN only reaches the installer through sudo with
+`sudo --preserve-env=DRB_ENROLLMENT_TOKEN bash` (plain sudo drops it).
+(`--token` still works for old copied commands but is deprecated, see node-26#18.)
 USAGE
 }
 
 while [ $# -gt 0 ]; do
   case "$1" in
-    --token)          ENROLLMENT_TOKEN="$2"; shift 2 ;;
+    --token)          ENROLLMENT_TOKEN="$2"; TOKEN_ON_ARGV=1; shift 2 ;;
+    --token-file)     TOKEN_FILE="$2"; shift 2 ;;
     --node-id)        NODE_ID="$2"; shift 2 ;;
     --name)           NODE_NAME="$2"; shift 2 ;;
     --lat)            NODE_LAT="$2"; shift 2 ;;
@@ -162,6 +180,18 @@ while [ $# -gt 0 ]; do
   esac
 done
 
+if [ "${TOKEN_ON_ARGV:-0}" = 1 ]; then
+  warn "--token is deprecated: the token is now in your shell history and visible in"
+  warn "\`ps\` for this whole install (node-26#18). Omit it to be prompted, or use --token-file."
+fi
+if [ -n "${TOKEN_FILE:-}" ]; then
+  [ -r "$TOKEN_FILE" ] || die "--token-file: cannot read $TOKEN_FILE"
+  _mode="$(stat -c %a "$TOKEN_FILE" 2>/dev/null || echo 600)"
+  (( 8#$_mode & 8#044 )) && warn "--token-file $TOKEN_FILE is readable by other users — chmod 600 it"
+  ENROLLMENT_TOKEN="$(tr -d '[:space:]' < "$TOKEN_FILE")"
+  [ -n "$ENROLLMENT_TOKEN" ] || die "--token-file: $TOKEN_FILE is empty"
+fi
+
 [ "$ENROLL_WAIT" = forever ] || [[ "$ENROLL_WAIT" =~ ^[0-9]+$ ]] \
   || die "--wait-approval takes a number of seconds (or omit it to wait until approved)"
 
@@ -181,6 +211,14 @@ ask_secret() {
   read -rsp "$__p: " __r </dev/tty; echo >/dev/tty
   printf -v "$__v" '%s' "$__r"
 }
+# Make sure ENROLLMENT_TOKEN is set: already given, else prompt on the TTY.
+# Returns 1 if there is still no token (non-interactive, or Enter at the prompt).
+need_token() {
+  [ -n "$ENROLLMENT_TOKEN" ] || ask_secret ENROLLMENT_TOKEN "Enrollment token (input hidden)"
+  [ -n "$ENROLLMENT_TOKEN" ]
+}
+NO_TOKEN_HELP="mint one at Settings -> Nodes and re-run the install command; it prompts for
+  the token (unattended: --token-file PATH, or an exported DRB_ENROLLMENT_TOKEN)"
 genpw() { head -c 24 /dev/urandom | base64 | tr -d '\n=' ; }
 
 # ── 1. Preflight ────────────────────────────────────────────────────────────
@@ -190,8 +228,8 @@ command -v apt-get >/dev/null || die "apt-get not found — this script targets
 
 ARCH="$(dpkg --print-architecture)"
 case "$ARCH" in
-  arm64|aarch64) ok "arch $ARCH" ;;
+  arm64|aarch64|amd64|x86_64) ok "arch $ARCH" ;;
-  *) warn "arch is $ARCH — CI only builds linux/arm64 images. Prebuilt pull will fail; use --build." ;;
+  *) warn "arch is $ARCH — CI only builds linux/amd64 + linux/arm64 images. Prebuilt pull will fail; use --build." ;;
 esac
 ok "running as root"
 
@@ -334,6 +372,20 @@ EOF
   ok ".env written for '$NODE_ID'"
 fi
 
+# Pin every image to the source just checked out (node-26#19, D3). Written
+# into .env — fresh or preserved — so later manual `docker compose` runs on
+# this node use the same pins.
+# A --ref older than #19 has no helper; its compose file hardcodes its tags.
+if [ -f "$INSTALL_DIR/scripts/image-tags.sh" ]; then
+  # shellcheck source=scripts/image-tags.sh
+  . "$INSTALL_DIR/scripts/image-tags.sh"
+  drb_pin_image_tags "$INSTALL_DIR/.env" "$INSTALL_DIR" "$RESOLVED_SHA" || die "could not write image pins to .env"
+  ok "images pinned to the source at ${RESOLVED_SHA:0:12} (DRB_TAG_* in .env)"
+else
+  warn "$REF predates image pinning (node-26#19) — using its compose file's default image tags"
+  drb_images_present() { return 0; }
+fi
+
 # ── 5. Enrollment ───────────────────────────────────────────────────────────
 # Client half of Server/drb-c2-core/app/routers/enrollment.py. It does NOT
 # exist in the edge-node app today (mqtt_manager.py:73-85 says so explicitly),
@@ -416,8 +468,8 @@ wait_for_key() {  # wait_for_key PICKUP_SECRET
   while :; do
     case "$CRED_HTTP" in
       200) [ -n "$(cred_field '.api_key')" ] && { trap - INT TERM; return 0; } ;;
-      401) die "C2 rejected the saved pickup secret (401) while waiting — re-run with --token DRB-… to re-enroll." ;;
+      401) die "C2 rejected the saved pickup secret (401) while waiting — re-run the installer to re-enroll (it asks for a fresh enrollment token)." ;;
-      404) die "C2 no longer knows node '$NODE_ID' (404) — it was deleted. Re-run with --token DRB-… to enroll again." ;;
+      404) die "C2 no longer knows node '$NODE_ID' (404) — it was deleted. Re-run the installer to enroll again (it asks for an enrollment token)." ;;
       *)   : ;;  # 000/5xx: transient, keep waiting
     esac
     if [ "$_shown" = 0 ]; then
@@ -435,8 +487,7 @@ wait_for_key() {  # wait_for_key PICKUP_SECRET
 
 do_fresh_enroll() {
   say "Enrolling '$NODE_ID' with $C2_URL"
-  [ -n "$ENROLLMENT_TOKEN" ] || ask_secret ENROLLMENT_TOKEN "Enrollment token"
+  need_token || die "no enrollment token — $NO_TOKEN_HELP"
-  [ -n "$ENROLLMENT_TOKEN" ] || die "no enrollment token — mint one at Settings -> Nodes, then re-run with --token"
 
   local BODY RESP PICKUP STATUS KEY
   BODY="$(jq -nc --arg id "$NODE_ID" --arg n "${NODE_NAME:-$NODE_ID}" \
@@ -459,7 +510,7 @@ do_fresh_enroll() {
   or have an admin Reissue key (Settings -> Nodes) and write the key into
   $CREDS by hand (see node-26#6)." ;;
       *401*) die "enroll refused (401): enrollment token missing/invalid/revoked —
-  mint a fresh one at Settings -> Nodes and re-run with --token." ;;
+  $NO_TOKEN_HELP." ;;
       *429*) die "enroll refused (429): rate limited. Wait ~1 minute, then re-run." ;;
       *)     die "enroll failed: $RESP" ;;
     esac
@@ -522,24 +573,26 @@ elif [ -s "$PICKUP_FILE" ]; then
       fi
       ;;
     401)
-      if [ -n "$ENROLLMENT_TOKEN" ]; then
+      warn "saved pickup secret rejected (401) — likely rotated by a re-enroll elsewhere."
-        warn "saved pickup secret rejected (401) — likely rotated by a re-enroll elsewhere. Re-enrolling with --token."
+      if need_token; then
+        warn "re-enrolling with the enrollment token."
         rm -f "$PICKUP_FILE"
         do_fresh_enroll
       else
-        die "saved pickup secret is stale (401) and no --token was given. Re-run with
+        die "saved pickup secret is stale (401) and no enrollment token was given.
-  --token DRB-… to re-enroll (only works while the node is still pending), or
+  Re-enrolling only works while the node is still pending: $NO_TOKEN_HELP.
-  have an admin Reissue key for an approved node and write $CREDS by hand."
+  For an approved node, have an admin Reissue key and write $CREDS by hand."
       fi
       ;;
     404)
-      if [ -n "$ENROLLMENT_TOKEN" ]; then
+      warn "C2 does not know node '$NODE_ID' (404) — deleted server-side or never fully enrolled."
-        warn "C2 does not know node '$NODE_ID' (404) — deleted server-side or never fully enrolled. Re-enrolling with --token."
+      if need_token; then
+        warn "re-enrolling with the enrollment token."
         rm -f "$PICKUP_FILE"
         do_fresh_enroll
       else
-        die "C2 does not know node '$NODE_ID' (404) and no --token was given.
+        die "C2 does not know node '$NODE_ID' (404) and no enrollment token was given:
-  Re-run with --token DRB-… to enroll it again."
+  $NO_TOKEN_HELP."
       fi
       ;;
     000)
@@ -570,6 +623,11 @@ if [ "$DO_START" = 1 ]; then
   requirement or a transient network error: re-run with DRB_REGISTRY_USER /
   DRB_REGISTRY_PASS set, or with --build to compile on the Pi (~1h for op25)."
     fi
+    # compose pull only WARNS for a missing image that also has a build:
+    # section, so check the pinned images really arrived.
+    drb_images_present || die "the images pinned to $REF (${RESOLVED_SHA:0:12}) are not in
+  $REGISTRY yet. CI may still be building them — wait and re-run; the owner can
+  run the 'Release images' workflow on that ref; or re-run with --build."
     docker compose up --no-build -d
   fi
   # ALWAYS recreate edge-node (node-26#14): `up -d` leaves an already-running
@@ -618,6 +676,6 @@ if [ -z "$(jq -r '.api_key // empty' "$CREDS" 2>/dev/null)" ]; then
   if [ -s "$PICKUP_FILE" ]; then
     printf "${Y}This node has no api_key yet. Once an admin approves it at %s/settings/nodes\nthe node fetches the key and comes online by itself (no re-run needed).${N}\n\n" "$APP_URL"
   else
-    printf "${Y}This node has no api_key and no saved pickup secret, so a plain re-run cannot\nfix it. Re-run with --token DRB-… to enroll; or, if the node is already\napproved, have an admin Reissue key and write it into\n%s by hand.${N}\n\n" "$CREDS"
+    printf "${Y}This node has no api_key and no saved pickup secret. Re-run the installer\nand give it an enrollment token when it asks; or, if the node is already\napproved, have an admin Reissue key and write it into\n%s by hand.${N}\n\n" "$CREDS"
   fi
 fi

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
 
 
 # ---------------------------------------------------------------------------

scripts/drb-update.sh

@@ -1,7 +1,10 @@
 #!/usr/bin/env bash
 #
 # DRB node host updater (node-26#15). Runs as root from systemd:
-#   drb-update.timer  -> every ~15 min: HTTPS check-in with C2; update if asked
+#   drb-update.timer  -> every ~15 min: HTTPS check-in with C2; update if asked,
+#                        or if the configured channel has a newer release
+#                        (node-26#49: REF=latest follows the highest vX.Y.Z
+#                        tag, REF=v1 follows the promoted major tag)
 #   drb-update.path   -> fires when the edge-node container drops
 #                        <install>/configs/update-requested (MQTT node_update)
 # Both start drb-update.service, which runs this script. install-updater.sh
@@ -17,13 +20,51 @@
 #   * Node credentials go to curl via a 0600 header file, never argv.
 #   * C2 must be https.
 # What this does NOT defend against: a compromised git host or registry —
-# refs and images are not signed (see the issue filed with node-26#15).
+# refs and images are not signed (node-26#17).
+#   * Images are pinned to the checked-out source (node-26#19): each one is
+#     pulled as src-<git tree hash of its build directory>, never :latest.
 #
 # The whole body is inside main(), called on the last line: bash parses the
 # function completely before running it, so `git checkout` replacing this
 # very file mid-run cannot change what is executing.
 set -euo pipefail
 
+# ── Release versions (node-26#49) ─────────────────────────────────────────
+# Versions are MAJOR.MINOR.PATCH and every comparison is per field, as
+# integers: 1.10.0 > 1.9.3. Never string or float order ("1.10" < "1.9" as a
+# float, "1.10.0" < "1.9.3" as a string).
+semver_sort() { sort -t. -k1,1n -k2,2n -k3,3n; }
+
+semver_gt() {  # semver_gt A B -> true when A is newer than B
+  local a b i
+  IFS=. read -r -a a <<< "$1"; IFS=. read -r -a b <<< "$2"
+  for i in 0 1 2; do
+    (( 10#${a[i]:-0} > 10#${b[i]:-0} )) && return 0
+    (( 10#${a[i]:-0} < 10#${b[i]:-0} )) && return 1
+  done
+  return 1
+}
+
+version_at() {  # highest X.Y.Z release tagged on a commit, or nothing
+  git tag --points-at "$1" 2>/dev/null | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$' \
+    | sed 's/^v//' | semver_sort | tail -1 || true
+}
+
+resolve_target() {  # REF -> a ref name git can resolve, or fail
+  local ref="$1" top
+  if [ "$ref" = latest ]; then
+    top="$(git tag -l 'v*' | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$' | sed 's/^v//' | semver_sort | tail -1 || true)"
+    [ -n "$top" ] || return 1
+    echo "refs/tags/v$top"
+  elif git show-ref --verify -q "refs/remotes/origin/$ref"; then
+    echo "refs/remotes/origin/$ref"
+  elif git show-ref --verify -q "refs/tags/$ref"; then
+    echo "refs/tags/$ref"
+  else
+    return 1
+  fi
+}
+
 main() {
   local CONF="${DRB_UPDATER_CONF:-/etc/drb/updater.conf}"
   local STATE_DIR="${DRB_UPDATER_STATE:-/var/lib/drb-update}"
@@ -49,6 +90,12 @@ main() {
   flock -n 9 || { log "another update is already running"; return 0; }
 
   cd "$INSTALL_DIR"
+  # This runs as root, but a pre-install.sh checkout (e.g. /home/pi/node-26)
+  # is owned by the login user, and git >= 2.35.2 then refuses every command
+  # with "dubious ownership" — which surfaced as origin '' != REPO_URL. Trust
+  # exactly the configured INSTALL_DIR, for this process only (env config is
+  # command-line scope, which safe.directory honours; no global config change).
+  export GIT_CONFIG_COUNT=1 GIT_CONFIG_KEY_0=safe.directory GIT_CONFIG_VALUE_0="$INSTALL_DIR"
   envget() { grep -E "^$1=" .env 2>/dev/null | head -1 | cut -d= -f2- | tr -d '"' || true; }
   local NODE_ID C2_URL
   NODE_ID="$(envget NODE_ID)"; C2_URL="$(envget C2_URL)"; C2_URL="${C2_URL%/}"
@@ -110,6 +157,22 @@ main() {
     new_request="$requested"
     reason="${reason:-C2 request $requested}"
   fi
+  # A newer release on this node's channel is a reason on its own (node-26#49).
+  # Only ever forward, and only between tagged releases: an untagged target
+  # (a branch REF, or v1 before its first promotion) still needs a request.
+  if [ -z "$reason" ]; then
+    local origin_url target tv iv
+    origin_url="$(git remote get-url origin 2>/dev/null || true)"
+    if [ "$origin_url" = "$REPO_URL" ] \
+       && git -c protocol.allow=never -c protocol.https.allow=always fetch -q --force --tags --prune origin \
+       && target="$(resolve_target "$REF")"; then
+      tv="$(version_at "$(git rev-parse "$target^{commit}")")"
+      iv="$(version_at HEAD)"
+      if [ -n "$tv" ] && { [ -z "$iv" ] || semver_gt "$tv" "$iv"; }; then
+        reason="release $tv on $REF (installed ${iv:-untagged})"
+      fi
+    fi
+  fi
   [ -n "$reason" ] || { log "no update requested"; return 0; }
 
   # ── Update ─────────────────────────────────────────────────────────────
@@ -137,15 +200,50 @@ do_update() {
   git -c protocol.allow=never -c protocol.https.allow=always fetch --force --tags --prune origin || return 1
 
   local target
-  if git show-ref --verify -q "refs/remotes/origin/$REF"; then
+  target="$(resolve_target "$REF")" || { err "ref '$REF' not found on origin"; return 1; }
-    target="refs/remotes/origin/$REF"
-  elif git show-ref --verify -q "refs/tags/$REF"; then
-    target="refs/tags/$REF"
-  else
-    err "ref '$REF' not found on origin"; return 1
-  fi
   local old new
   old="$(git rev-parse HEAD)"; new="$(git rev-parse "$target^{commit}")"
+
+  local env_registry
+  env_registry="$(grep -E '^IMAGE_REGISTRY=' .env | head -1 | cut -d= -f2- | tr -d '"' || true)"
+  [ "${env_registry:-git.vpn.cusano.net}" = "$REGISTRY" ] \
+    || { err ".env IMAGE_REGISTRY '$env_registry' != configured '$REGISTRY' — refusing"; return 1; }
+
+  # Image pins for the TARGET commit (node-26#19): each image is tagged with
+  # the git tree hash of its own source directory, so the images always match
+  # the code being checked out. The helper is read from the target commit so
+  # its service list matches the target's compose file.
+  # Read into a variable, not `source <(...)`: a failed `git show` must leave
+  # pin=0 rather than "succeed" with no functions defined.
+  local pin=0 helper
+  if git cat-file -e "$new:scripts/image-tags.sh" 2>/dev/null; then
+    helper="$(git show "$new:scripts/image-tags.sh")" && [ -n "$helper" ] && eval "$helper" \
+      && declare -F drb_image_tags drb_images_present drb_pin_image_tags >/dev/null && pin=1
+  fi
+  [ "$pin" = 1 ] || log "target ${new:0:12} predates image pinning — using its compose file's default tags"
+
+  if [ "$BUILD" != 1 ]; then
+    # Pull BEFORE checkout. The app code is bind-mounted from the checkout
+    # (install.sh D3) and edge-node runs uvicorn --reload, so checking out first
+    # would run the new code on the old images for as long as the pull takes —
+    # or for good, if the pinned images don't exist. Pulling against the
+    # target's own compose file means a failed pull changes nothing. It is
+    # written next to the real one so its relative paths (.env, env_file,
+    # build contexts) resolve exactly the same way.
+    local tfile="$INSTALL_DIR/.drb-update-compose.yml" pulled=0
+    git show "$new:docker-compose.yml" > "$tfile" || { rm -f "$tfile"; return 1; }
+    local -a tc=(-f "$tfile" --project-directory "$INSTALL_DIR")
+    (
+      if [ "$pin" = 1 ]; then
+        while IFS= read -r l; do export "$l"; done < <(drb_image_tags "$INSTALL_DIR" "$new")
+      fi
+      docker compose "${tc[@]}" pull || exit 1
+      if [ "$pin" = 1 ]; then drb_images_present "${tc[@]}" || exit 1; fi
+    ) && pulled=1
+    rm -f "$tfile"
+    [ "$pulled" = 1 ] || { err "images for ${new:0:12} are not available — nothing changed (CI may still be building them; the 'Release images' workflow publishes any that are missing)"; return 1; }
+  fi
+
   if [ "$old" != "$new" ]; then
     git -c advice.detachedHead=false checkout --force --detach "$new" || return 1
     log "checked out $REF: ${old:0:12} -> ${new:0:12}"
@@ -153,20 +251,42 @@ do_update() {
     bash scripts/install-updater.sh --refresh || log "unit refresh failed (continuing with the old units)"
   fi
 
-  local env_registry
+  if [ "$pin" = 1 ]; then
-  env_registry="$(grep -E '^IMAGE_REGISTRY=' .env | head -1 | cut -d= -f2- | tr -d '"' || true)"
+    drb_pin_image_tags "$INSTALL_DIR/.env" "$INSTALL_DIR" "$new" || { err "could not write image pins to .env"; return 1; }
-  [ "${env_registry:-git.vpn.cusano.net}" = "$REGISTRY" ] \
+  fi
-    || { err ".env IMAGE_REGISTRY '$env_registry' != configured '$REGISTRY' — refusing"; return 1; }
+
+  # The release this node now runs, for its checkin (node-26#49).
+  local ver; ver="$(version_at "$new")"
+  sed -i '/^DRB_VERSION=/d' "$INSTALL_DIR/.env"
+  printf 'DRB_VERSION=%s\n' "${ver:-untagged}" >> "$INSTALL_DIR/.env"
 
   if [ "$BUILD" = 1 ]; then
-    docker compose build || return 1
+    docker compose build || return 1   # tags the local builds with the same pins
-  else
-    docker compose pull || return 1
   fi
   # --force-recreate: app code is bind-mounted from the checkout (install.sh
   # D3), so a code-only change also needs every container restarted.
   docker compose up -d --no-build --force-recreate --remove-orphans || return 1
+
+  # Drop pinned images this node no longer runs. `image prune` only removes
+  # dangling images, and every previous src- tag is still tagged. `image rm`
+  # without -f refuses anything a container still uses.
+  if [ "$pin" = 1 ]; then
+    local keep ref
+    keep="$(drb_image_tags "$INSTALL_DIR" "$new" | cut -d= -f2)"
+    docker images --format '{{.Repository}}:{{.Tag}}' | grep -E ':src-[0-9a-f]{40}$' \
+      | while IFS= read -r ref; do
+          grep -qxF "${ref##*:}" <<< "$keep" || docker image rm "$ref" >/dev/null 2>&1 || true
+        done || true
+  fi
   docker image prune -f >/dev/null 2>&1 || true
+
+  # node-26#41: neither of the above ever freed BuildKit cache or untagged-
+  # but-unused images, so a BUILD=1 node grew ~40 GB of build cache plus
+  # 16 GB of stale images until the root disk hit 100% and recording silently
+  # stopped (node-002, 17 days). Both prunes skip anything a container uses;
+  # the age filters keep the cache warm for the next build.
+  docker builder prune -af --filter until=72h >/dev/null 2>&1 || true
+  docker image prune -af --filter until=168h >/dev/null 2>&1 || true
 }
 
 main "$@"

scripts/image-tags.sh

@@ -0,0 +1,65 @@
+# shellcheck shell=bash
+#
+# Pin each node image to the exact source it was built from (node-26#19).
+# Sourced by install.sh and scripts/drb-update.sh; not run directly.
+#
+# The tag is the git TREE hash of the image's build context, `src-<tree>`.
+# CI computes the same value (`git rev-parse HEAD:<dir>` in the build
+# workflows), so a node asks for the image built from the very directory it
+# has checked out. That is exact even though each image is only rebuilt when
+# its own directory changes: an unchanged directory keeps the same tree hash,
+# and so the same image, across commits. :latest / :stable still exist for
+# local dev; a provisioned node never uses them.
+#
+# docker-compose.yml reads these as ${DRB_TAG_*:-latest}. The values live in
+# .env so every manual `docker compose ...` on the node sees the same pins.
+
+# VAR:build-context-dir, one per compose service with an image.
+DRB_IMAGE_SOURCES="DRB_TAG_ICECAST:icecast DRB_TAG_OP25:op25-container \
+DRB_TAG_SECONDARY_SDR:secondary-sdr-container DRB_TAG_EDGE_NODE:drb-edge-node"
+
+# drb_image_tags REPO_DIR COMMIT -> "VAR=src-<tree>" lines. A directory that
+# does not exist at COMMIT is skipped (its compose file won't reference it).
+drb_image_tags() {
+  local _p _var _dir _tree
+  for _p in $DRB_IMAGE_SOURCES; do
+    _var="${_p%%:*}"; _dir="${_p#*:}"
+    _tree="$(git -C "$1" rev-parse --verify -q "$2:$_dir" 2>/dev/null)" || continue
+    printf '%s=src-%s\n' "$_var" "$_tree"
+  done
+}
+
+# drb_env_set ENV_FILE KEY VALUE — replace KEY's line in place, or append it.
+# Rewrites through the same inode so .env keeps its 0600 mode and owner.
+drb_env_set() {
+  local _tmp; _tmp="$(mktemp)"
+  awk -v k="$2" -v v="$3" '
+    index($0, k "=") == 1 { if (!done) print k "=" v; done = 1; next }
+    { print }
+    END { if (!done) print k "=" v }' "$1" > "$_tmp" && cat "$_tmp" > "$1"
+  local _rc=$?; rm -f "$_tmp"; return $_rc
+}
+
+# drb_pin_image_tags ENV_FILE REPO_DIR COMMIT — write the pins into .env and
+# export them, so compose sees them whether or not it re-reads .env.
+drb_pin_image_tags() {
+  local _line
+  while IFS= read -r _line; do
+    drb_env_set "$1" "${_line%%=*}" "${_line#*=}" || return 1
+    export "$_line"
+  done < <(drb_image_tags "$2" "$3")
+}
+
+# drb_images_present COMPOSE_ARGS... — 0 if every image the compose project
+# names is in the local image store. `compose pull` only WARNS when a service
+# that also has a build: section fails to pull, so its exit code alone does
+# not prove the pinned images exist.
+drb_images_present() {
+  local _imgs _img _missing=0
+  _imgs="$(docker compose "$@" config --images)" || return 1
+  while IFS= read -r _img; do
+    [ -n "$_img" ] || continue
+    docker image inspect "$_img" >/dev/null 2>&1 || { printf 'missing image: %s\n' "$_img" >&2; _missing=1; }
+  done <<< "$_imgs"
+  return $_missing
+}

scripts/next-version.sh

@@ -0,0 +1,20 @@
+#!/usr/bin/env bash
+# node-26#49 — the release tag this commit should get, or nothing if it has one.
+#
+# VERSION holds MAJOR.MINOR and is edited by hand. PATCH is automatic: one
+# more than the highest existing vMAJOR.MINOR.PATCH tag, compared as an
+# integer (so .10 follows .9), starting at 0 for a new MAJOR.MINOR.
+#
+# Usage: scripts/next-version.sh   (from the repo root, tags fetched)
+set -euo pipefail
+
+base="$(tr -d ' \r\n' < VERSION)"
+[[ "$base" =~ ^[0-9]+\.[0-9]+$ ]] || { echo "VERSION must be MAJOR.MINOR, got '$base'" >&2; exit 1; }
+pattern="^v${base//./\\.}\\.[0-9]+$"
+
+if git tag --points-at HEAD | grep -qE "$pattern"; then
+  exit 0   # this commit is already released
+fi
+
+last="$(git tag -l "v${base}.*" | grep -E "$pattern" | sed "s/^v${base}\.//" | sort -n | tail -1 || true)"
+echo "v${base}.$(( ${last:--1} + 1 ))"