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

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

.env.example

@@ -99,6 +99,15 @@ TRIM_SILENCE_THRESHOLD_DB=-40
 # Seconds of audio kept either side of detected speech.
 TRIM_SILENCE_GUARD_SECONDS=0.25
 
+# Upload spool: a recording whose upload to C2 fails is kept on disk and
+# retried, never deleted (default dir: <recordings>/spool). Bounded so a long
+# C2 outage cannot fill the host disk; the oldest recordings go first.
+# UPLOAD_SPOOL_DIR=/recordings/spool
+UPLOAD_SPOOL_MAX_BYTES=536870912
+UPLOAD_SPOOL_MAX_AGE_HOURS=72
+UPLOAD_RETRY_MIN_SECONDS=15
+UPLOAD_RETRY_MAX_SECONDS=600
+
 # OP25 container (usually no need to change)
 OP25_API_URL=http://localhost:8001
 OP25_TERMINAL_URL=http://localhost:8081

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

.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
 

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

@@ -153,6 +153,23 @@ class Settings(BaseSettings):
     # Offline call buffer — how many call_end events to keep while disconnected
     offline_call_buffer_size: int = 35
 
+    # Upload spool (node-26#44) — recordings whose upload to C2 failed are kept
+    # here and retried instead of deleted. Defaults to <recordings_path>/spool,
+    # i.e. on the host disk via the /recordings bind mount, so it survives a
+    # restart. Bounded both ways because a full host disk stops recording
+    # outright (node-26#41): FLAC is ~1.3 MB per minute of audio, so 512 MiB is
+    # roughly 6 h of continuous traffic. Oldest entries are evicted first.
+    upload_spool_dir: Optional[str] = None
+    upload_spool_max_bytes: int = 512 * 1024 * 1024
+    upload_spool_max_age_hours: float = 72.0
+    # Exponential backoff between retries of one spooled recording.
+    upload_retry_min_seconds: float = 15.0
+    upload_retry_max_seconds: float = 600.0
+    # Timeout for the FIRST, inline upload attempt. metadata_watcher awaits the
+    # call close, so a black-holed C2 used to stall segmentation for the full
+    # 120 s; a failure here just lands in the spool.
+    upload_inline_timeout_seconds: float = 30.0
+
     # ------------------------------------------------------------------
     # Local dashboard / API authentication
     #

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

@@ -66,10 +66,10 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import List, Optional, Tuple
 
-import httpx
 from app.config import settings
-from app.internal import audio_trim, credentials, pcm, pulse
+from app.internal import audio_trim, pcm, pulse, upload_spool
 from app.internal.logger import logger
+from app.internal.upload_spool import UploadMeta, UploadResult, UploadSpool
 
 # Safety cap on a single recording; mirrors MAX_SEGMENT_SECONDS in metadata_watcher.
 MAX_RECORDING_SECONDS = 600
@@ -320,6 +320,17 @@ class CallRecorder:
         # why a capture process exited (see _classify_capture_exit).
         self._last_stderr_lines: deque[str] = deque(maxlen=10)
 
+        # Failed uploads wait here for a retry instead of being deleted
+        # (node-26#44). See internal/upload_spool.py.
+        self.spool = UploadSpool(
+            Path(settings.upload_spool_dir or (Path(settings.recordings_path) / "spool")),
+            uploader=self._post,
+            max_bytes=settings.upload_spool_max_bytes,
+            max_age_seconds=settings.upload_spool_max_age_hours * 3600,
+            backoff_min=settings.upload_retry_min_seconds,
+            backoff_max=settings.upload_retry_max_seconds,
+        )
+
     # ------------------------------------------------------------------
     # Lifecycle
     # ------------------------------------------------------------------
@@ -328,8 +339,11 @@ class CallRecorder:
         """Start the persistent capture. Call once from app lifespan."""
         self._stream_task = asyncio.create_task(self._capture_loop())
         logger.info("PulseAudio ring-buffer starting.")
+        if settings.c2_url:
+            self.spool.start()
 
     async def stop(self) -> None:
+        await self.spool.stop()
         if self._stream_task:
             self._stream_task.cancel()
             try:
@@ -751,6 +765,17 @@ class CallRecorder:
     # Upload (unchanged interface)
     # ------------------------------------------------------------------
 
+    async def _post(self, file_path: Path, meta: UploadMeta) -> UploadResult:
+        return await upload_spool.post_recording(file_path, meta, AUDIO_MIME)
+
+    async def _post_inline(self, file_path: Path, meta: UploadMeta) -> UploadResult:
+        # metadata_watcher awaits on_call_end, so this attempt blocks the poll
+        # loop. With the spool behind it, it can afford a short timeout; the
+        # background retries get the full UPLOAD_TIMEOUT_SECONDS.
+        return await upload_spool.post_recording(
+            file_path, meta, AUDIO_MIME, timeout=settings.upload_inline_timeout_seconds,
+        )
+
     async def upload_recording(
         self,
         file_path: Path,
@@ -761,50 +786,37 @@ class CallRecorder:
         audio_start_epoch: Optional[float] = None,
         audio_end_epoch: Optional[float] = None,
     ) -> Optional[str]:
+        """
+        Upload once, inline. Returns the storage URL on success.
+
+        The file is deleted ONLY after a 2xx with a URL (node-26#44). A
+        retryable failure moves it into the spool, which keeps retrying in the
+        background; C2's /upload writes `audio_gcs_uri` onto the call doc by
+        call_id whenever the upload lands, so a late upload still attaches.
+        """
         if not settings.c2_url:
             logger.info("No C2_URL configured — skipping upload.")
+            file_path.unlink(missing_ok=True)
             return None
 
-        upload_url = f"{settings.c2_url}/upload"
+        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,
-
-        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

@@ -74,6 +74,43 @@ class _Voice:
     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
@@ -146,6 +183,10 @@ class RadioManager:
         self._mqtt_down_since: Optional[float] = None
         self._failsafe = False
         self._paused = False
+        # The applied seq at the moment of a local pause (node-26#34). The
+        # broker redelivers the retained desired message on every reconnect
+        # with the SAME seq; only a strictly newer seq from C2 lifts the pause.
+        self._paused_at_seq: Optional[int] = None
         self._ticks = 0
 
     # ------------------------------------------------------------------
@@ -173,15 +214,25 @@ class RadioManager:
                 logger.error(f"Malformed Discord desired stream entry ignored: {s!r}")
         self._ensure_watchdog()
         self._desired = wanted
+        if self._paused and self._redelivery_of_paused_state(seq):
+            logger.info(f"Discord desired seq {seq} is a redelivery — staying paused locally.")
+        else:
             self._paused = False
+            self._paused_at_seq = None
         await self.reconcile()
         if seq is not None:
             self._applied_seq = seq
         await self.publish_actual()
 
+    def _redelivery_of_paused_state(self, seq: Optional[int]) -> bool:
+        """True when `seq` is not newer than what was applied at pause time."""
+        return seq is not None and self._paused_at_seq is not None and seq <= self._paused_at_seq
+
     async def pause_locally(self) -> None:
-        """Local-dashboard stop: log every bot out until C2 sends a new list."""
+        """Local-dashboard stop: log every bot out until C2 sends a NEW list
+        (a seq strictly greater than the one applied now — node-26#34)."""
         self._paused = True
+        self._paused_at_seq = self._applied_seq
         await self.reconcile()
         await self.publish_actual()
 
@@ -220,6 +271,10 @@ class RadioManager:
                     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)
@@ -275,7 +330,12 @@ class RadioManager:
             if not vc.is_playing():
                 self._play(v)
             v.state, v.error, v.code = "connected", None, None
-            await self._set_presence(sess, d.get("system_name"))
+            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:
@@ -287,11 +347,23 @@ class RadioManager:
                 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:
             v.vc = None
             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: {e}")
+            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():
@@ -300,15 +372,27 @@ class RadioManager:
             return
         self._play(v, pulse_checked=True)
         v.state, v.error, v.code = "connected", None, None
-        await self._set_presence(sess, d.get("system_name"))
         logger.info(f"Stream {v.stream_id} connected (bot {v.token_id})")
 
-    async def _set_presence(self, sess: _Session, name: Optional[str]) -> None:
+    async def _sync_presence(self, sess: _Session, target: dict[str, dict]) -> None:
-        if not name or sess.presence == name:
+        """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:
-            await sess.client.change_presence(
+            activity = discord.Activity(type=discord.ActivityType.listening, name=name) if name else None
-                activity=discord.Activity(type=discord.ActivityType.listening, name=name))
+            await sess.client.change_presence(activity=activity)
             sess.presence = name
         except Exception as e:
             logger.warning(f"Presence update failed: {e}")

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

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

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

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

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

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

drb-edge-node/app/main.py

@@ -111,8 +111,20 @@ async def on_call_end(data: dict):
         )
         if audio_url:
             data["audio_url"] = audio_url
+        elif settings.c2_url:
+            last = call_recorder.spool.last_result or {}
+            if last.get("kind") == "lost":
+                data["audio_upload"] = "lost"
+            elif last.get("kind") == "retry":
+                # The file is in the upload spool and will be retried; C2 attaches
+                # it to this call_id whenever it lands (node-26#44).
+                data["audio_upload"] = "spooled"
             else:
-            logger.error(f"Audio upload failed for call {data['call_id']}. Verify C2_URL and Node API Key.")
+                data["audio_upload"] = "rejected"
+                logger.error(
+                    f"Audio upload for call {data['call_id']} was rejected ({last.get('detail')}). "
+                    "C2 refused this recording (malformed request or file too large); it was not kept."
+                )
     elif recording is not None and recording.all_silence:
         # Explicit policy: an all-silence recording is not uploaded. It has no
         # transcript value and silence is what makes Whisper invent text.
@@ -179,6 +191,8 @@ async def on_api_key(payload: dict):
     if key:
         credentials.save_api_key(key)
         mqtt_manager.update_api_key(key)
+        # Uploads spooled on a 401 can go now rather than at their next backoff.
+        call_recorder.spool.wake()
         logger.info("Node API key received and saved.")
 
 

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

@@ -82,6 +82,9 @@ async def get_status():
         # it was discarded rather than uploaded. Non-zero means either the
         # console is not decoding or something else is feeding drb_sink.
         "unattributed_segments": metadata_watcher.unattributed_segments,
+        # Recordings whose upload to C2 failed and are waiting for a retry
+        # (node-26#44), plus the outcome of the most recent upload attempt.
+        "upload_spool": call_recorder.spool.stats(),
         "active_tgid": active_tgid,
         "active_tgid_name": active_tgid_name,
         "active_call_id": metadata_watcher.active_call_id,
@@ -205,6 +208,12 @@ class ListenBody(BaseModel):
     frequencies: List[float] = []
     talkgroups: List[Talkgroup] = []       # p25 only; empty = every talkgroup
     nac: Optional[str] = None              # p25conv only, hex; empty = any NAC
+    # analog only (node-26#46): OP25 squelch. noise (default) needs no
+    # calibration; squelch_db = dB of quieting to open (OP25 default 8);
+    # squelch_threshold = absolute dB for power mode (OP25 default -60).
+    squelch_mode: Optional[Literal["noise", "voice", "power"]] = None
+    squelch_db: Optional[float] = None
+    squelch_threshold: Optional[float] = None
 
 
 # RTL-SDR (R820T/R828D) tuning range.
@@ -239,14 +248,21 @@ async def listen(body: ListenBody):
         else:
             is_p25 = body.mode == "p25"
             label = (body.name or "").strip() or (f"P25 {first} MHz" if is_p25 else f"Analog {first} MHz")
+            sys_config = {
+                "control_channels": freqs,
+                "talkgroups": [tg.model_dump() for tg in body.talkgroups] if is_p25 else [],
+            }
+            if not is_p25:
+                sys_config["squelch_mode"] = body.squelch_mode or "noise"
+                if body.squelch_db is not None:
+                    sys_config["squelch_db"] = body.squelch_db
+                if body.squelch_threshold is not None:
+                    sys_config["squelch_threshold"] = body.squelch_threshold
             config = SystemConfig(
                 system_id=f"local-{body.mode}",
                 name=label,
                 type="P25" if is_p25 else "NBFM",
-                config={
+                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}

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

@@ -47,8 +47,18 @@ class FakeVC:
 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
@@ -57,6 +67,7 @@ class FakeChannel:
 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}
 
@@ -149,6 +160,7 @@ async def test_apply_connects_and_reports(env):
     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):
@@ -251,6 +263,18 @@ async def test_local_pause_logs_out_until_next_desired(env):
     assert env.mgr.is_connected
 
 
+async def test_retained_redelivery_does_not_undo_local_pause(env):
+    """node-26#34: the broker redelivers the retained desired message (same
+    seq) on every reconnect; that must not log the bots back in."""
+    await env.mgr.apply_desired(_desired(4, (1, 11, "t1")))
+    await env.mgr.pause_locally()
+    await env.mgr.apply_desired(_desired(4, (1, 11, "t1")))
+    assert not env.mgr.is_connected
+    assert env.published[-1]["streams"]["1:11"]["state"] == "paused_locally"
+    await env.mgr.apply_desired(_desired(5, (1, 11, "t1")))
+    assert env.mgr.is_connected
+
+
 async def test_no_message_content_intent_on_real_client():
     client = discord_radio._default_client_factory()
     try:
@@ -258,3 +282,64 @@ async def test_no_message_content_intent_on_real_client():
         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,37 @@ 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" ;;
   *) 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 +95,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,6 +111,10 @@ 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
@@ -108,7 +129,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 +147,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 +185,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 +209,42 @@ 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()

drb-edge-node/tests/test_local_dashboard.py

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

drb-edge-node/tests/test_mqtt_manager.py

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

drb-edge-node/tests/test_upload_spool.py

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

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

@@ -17,7 +17,9 @@
 #   * 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
@@ -49,6 +51,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%/}"
@@ -146,6 +154,47 @@ do_update() {
   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 +202,37 @@ 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; }
 
   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
+}