Merge pull request 'checkin: report host disk free/total (#41)' (#43) from feat/41-disk-reporting into main

/recordings is a bind mount of the host checkout, so shutil.disk_usage on it
is the disk the recorder writes to and Docker fills. Explicit nulls on error
so C2 never keeps a stale reading. Server half: server-26 feat/41-node-disk.

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

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

@@ -19,6 +19,22 @@ jobs:
       - name: Run tests
         run: python3 -m pytest tests -v
 
+  test-op25:
+    # Config generation + tk_drb_conv state machine only (node-26#24) — no
+    # GNU Radio here; tests/test_conventional.py stubs OP25's own modules.
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: op25-container
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install dependencies
+        run: python3 -m pip install --break-system-packages -r requirements.txt pytest httpx
+
+      - name: Run tests
+        run: python3 -m pytest tests -v
+
   lint:
     runs-on: ubuntu-latest
     steps:

.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
-whole command). Run `install.sh --help` for every flag; each also has a
-`DRB_*` env var. `--build` compiles op25 on the Pi (~1h) instead of pulling.
+It prompts (hidden input) for an enrollment token, minted at **Settings → Nodes**
+in the web app. The token never goes on the command line — argv lands in shell
+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

@@ -54,6 +54,13 @@ class Settings(BaseSettings):
     # Bounded wait for the shared PulseAudio socket before launching FFmpeg.
     pulse_wait_timeout: float = 30.0
 
+    # Discord radio (node-26#25): with the broker unreachable this long, log
+    # every radio bot out. MUST stay below C2's
+    # DISCORD_LEASE_OFFLINE_RELEASE_SECONDS (default 600), after which C2 hands
+    # this node's bot tokens to other nodes — so a token is never logged in on
+    # two nodes at once.
+    discord_mqtt_loss_stop_seconds: int = 300
+
     # ------------------------------------------------------------------
     # Call segmentation
     #

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

@@ -0,0 +1,243 @@
+"""
+Attribution for CONVENTIONAL channels — P25 conventional (node-26#24) and
+analog NBFM (node-26#20) share this one path.
+
+metadata_watcher labels audio with whatever the OP25 console says was talking.
+On a trunked system that is a talkgroup from a control-channel grant. On a
+conventional channel there is no control channel, so:
+
+  * P25 conventional: link control in each LDU1 usually carries the talkgroup
+    and unit id, and our tk_drb_conv module adds `voice` (frames arriving) and
+    encryption. The talkgroup is often the P25 default (1) or "everyone"
+    (0xFFFF) and identical on every channel, which says nothing about WHICH
+    channel was keyed, and sometimes it is missing altogether.
+  * Analog: OP25 reports the channel with tgid None and nothing else — the
+    console never says anyone is talking.
+
+So a conventional channel gets a stable SYNTHETIC talkgroup id derived from its
+frequency (channel_tgid), used whenever link control gives nothing that
+distinguishes the channel. A real, distinct link-control talkgroup is still
+used when present. normalize() rewrites the console update so the rest of the
+watcher sees ordinary "talkgroup X is active" rows and grants and needs no
+conventional special cases, except one: with exactly ONE conventional channel
+configured, every sound OP25 puts on drb_sink came from that channel, so audio
+the console said nothing about is attributed to it (sole_channel) instead of
+being discarded as orphan audio. With two channels there is no such certainty
+and orphan audio is still discarded (DEFERRED.md, metadata_watcher row).
+
+Encrypted conventional traffic is never a talker: the decoder mutes it, so no
+audio follows, and letting it compete for attribution could mislabel a clear
+call on the other channel. It is reported instead (encryption_report).
+"""
+import time
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+# P25 talkgroups that identify no particular group: TIA-102 default (1) and
+# "everyone" (0xFFFF). 0 is "none" and already rejected by _as_int.
+NON_DISTINCT_TGIDS = frozenset({1, 0xFFFF})
+
+# Synthetic ids are frequency in 100 Hz units: 460.400 MHz -> 4_604_000. Any
+# RTL-SDR frequency (>= 24 MHz) gives >= 240_000, clear of the 16-bit P25
+# talkgroup space, and 6.25 kHz channel steps stay distinct. C2 resolves the
+# same number back to the channel name (server talkgroups.name_from_system).
+SYNTHETIC_TGID_MIN = 240_000
+
+# How long a channel's last real talkgroup is remembered for a transmission
+# whose link control hasn't decoded (yet). Covers the back-and-forth of one
+# conversation; past it, an undecoded call is labelled with the channel.
+REMEMBER_TGID_SECONDS = 30.0
+
+
+def channel_tgid(freq: Any) -> Optional[int]:
+    """Stable synthetic talkgroup id for a conventional channel frequency (Hz or MHz)."""
+    try:
+        f = float(freq)
+    except (TypeError, ValueError):
+        return None
+    if f <= 0:
+        return None
+    hz = f * 1_000_000 if f < 1_000_000 else f
+    return int(round(hz / 100))
+
+
+def _as_int(value: Any) -> Optional[int]:
+    try:
+        n = int(value)
+    except (TypeError, ValueError):
+        return None
+    return n if n > 0 else None
+
+
+def _truthy(value: Any) -> bool:
+    return bool(value) and value not in ("0", "false", "False")
+
+
+ALGID_NAMES = {
+    0x81: "DES-OFB", 0x83: "Triple DES", 0x84: "AES-256", 0x85: "AES-128",
+    0x9F: "DES-XL", 0xA0: "DVI-XL", 0xA1: "DVP-XL", 0xAA: "ADP (RC4)",
+}
+
+
+def algid_name(algid: Any) -> str:
+    a = _as_int(algid)
+    if a is None:
+        return "unknown algorithm"
+    return f"{ALGID_NAMES.get(a, 'algorithm')} (ALGID 0x{a:02X})"
+
+
+class ConventionalChannels:
+    """Per-poll normalizer plus what it has learned about the channels."""
+
+    def __init__(self) -> None:
+        # msgqid -> latest raw conventional row. channel_update is not on every
+        # poll, so this outlives a single update.
+        self._rows: Dict[str, Dict[str, Any]] = {}
+        self._known = False   # have we seen a channel_update at all
+        # synthetic channel id -> the talkgroup that channel last carried. A
+        # transmission's header frame sets `voice` ~180 ms before its first
+        # link control names the talkgroup; without this, every such poll
+        # would label the call with the synthetic id and the real talkgroup
+        # a poll later would force a bogus split (drb-audio-review, #24).
+        self._last_identity: Dict[int, Tuple[int, float]] = {}
+        self._clock: Callable[[], float] = time.time
+
+    # -- identity --------------------------------------------------------
+
+    @staticmethod
+    def _name(row: Dict[str, Any]) -> str:
+        name = str(row.get("name") or "").strip()
+        if name and not name.startswith("["):
+            return name
+        f = row.get("freq")
+        try:
+            return f"{float(f) / 1e6:.4f} MHz"
+        except (TypeError, ValueError):
+            return "Conventional"
+
+    def identity(self, row: Dict[str, Any], lc_tgid: Optional[int]) -> Optional[int]:
+        """Distinctive link-control talkgroup; else (no link control decoded
+        yet) what this channel last carried; else the channel's synthetic id."""
+        synthetic = channel_tgid(row.get("freq"))
+        now = self._clock()
+        if lc_tgid is not None and lc_tgid not in NON_DISTINCT_TGIDS:
+            if synthetic is not None:
+                self._last_identity[synthetic] = (lc_tgid, now)
+            return lc_tgid
+        remembered = self._last_identity.get(synthetic) if synthetic is not None else None
+        if lc_tgid is None and remembered and now - remembered[1] <= REMEMBER_TGID_SECONDS:
+            self._last_identity[synthetic] = (remembered[0], now)   # still the same conversation
+            return remembered[0]
+        if synthetic is not None and lc_tgid is not None:
+            self._last_identity[synthetic] = (synthetic, now)   # explicit default TG: channel identity
+        return synthetic
+
+    # -- normalization ---------------------------------------------------
+
+    def normalize(self, update: Any) -> Any:
+        """Rewrite conventional rows and grants in place; trunked ones untouched."""
+        if update.channels:
+            self._known = True
+            self._rows = {}
+        for i, row in enumerate(update.channels):
+            if not _truthy(row.get("conventional")):
+                continue
+            key = str(row.get("msgqid", i))
+            self._rows[key] = row
+            update.channels[i] = self._normalize_row(row)
+
+        for i, entry in enumerate(update.call_log):
+            row = self._rows.get(str(entry.get("rcvr")))
+            if row is None:
+                continue
+            lc = _as_int(entry.get("tgid"))
+            tgid = self.identity(row, lc)
+            fixed = dict(entry)
+            fixed["lc_tgid"] = lc
+            fixed["tgid"] = tgid
+            fixed["tgtag"] = self._label(row, lc, entry.get("tgtag"))
+            fixed["freq"] = entry.get("freq") or row.get("freq")
+            update.call_log[i] = fixed
+        return update
+
+    def _label(self, row: Dict[str, Any], lc: Optional[int], tag: Any) -> str:
+        name = self._name(row)
+        tag = str(tag or "").strip()
+        if lc is not None and lc not in NON_DISTINCT_TGIDS and tag and tag not in ("Conventional", name):
+            return tag
+        return name
+
+    def _normalize_row(self, row: Dict[str, Any]) -> Dict[str, Any]:
+        lc = _as_int(row.get("tgid"))
+        encrypted = _truthy(row.get("encrypted"))
+        # tgid is only ever set while a call is up (tk_p25 / tk_drb_conv clear
+        # it on TDU and sync timeout); `voice` is tk_drb_conv's own signal.
+        active = (lc is not None or _truthy(row.get("voice"))) and not encrypted
+        out = dict(row)
+        out["lc_tgid"] = lc
+        out["tgid"] = self.identity(row, lc) if active else None
+        out["tag"] = self._label(row, lc, row.get("tag"))
+        out["drb_conventional"] = True
+        out["drb_active"] = active
+        return out
+
+    # -- what the watcher asks -------------------------------------------
+
+    def same_channel_relabel(self, current_tgid: Optional[int], current_freq: Any,
+                             new_tgid: Optional[int], new_freq: Any) -> Optional[str]:
+        """Two labels for ONE conventional channel are not a talkgroup change.
+
+        "adopt": the open segment holds the channel's synthetic id and the
+                 console now names the real talkgroup — relabel, don't split.
+        "keep":  the segment holds a real talkgroup and the console offers the
+                 synthetic id (link control not decoded on this frame) — ignore.
+        None:    not the same conventional channel; the normal split rules apply.
+        """
+        if current_tgid is None or new_tgid is None or current_tgid == new_tgid:
+            return None
+        synthetic = channel_tgid(new_freq)
+        if synthetic is None or channel_tgid(current_freq) != synthetic:
+            return None
+        if not any(channel_tgid(r.get("freq")) == synthetic for r in self._rows.values()):
+            return None   # trunked: freq says nothing about identity
+        if current_tgid == synthetic:
+            return "adopt"
+        if new_tgid == synthetic:
+            return "keep"
+        return None
+
+    def sole_channel(self) -> Optional[Dict[str, Any]]:
+        """The one conventional channel this node listens to, if exactly one
+        (and it is not encrypted). None on any trunked or multi-channel setup."""
+        if len(self._rows) != 1:
+            return None
+        row = next(iter(self._rows.values()))
+        tgid = channel_tgid(row.get("freq"))
+        if tgid is None:
+            return None
+        if _truthy(row.get("encrypted")):
+            return None
+        return {"tgid": tgid, "name": self._name(row), "freq": row.get("freq")}
+
+    def encryption_report(self) -> List[Dict[str, Any]]:
+        """Per conventional channel: is it (or was its last call) encrypted."""
+        out = []
+        for row in self._rows.values():
+            enc_calls = _as_int(row.get("enc_calls")) or 0
+            clear_calls = _as_int(row.get("clear_calls")) or 0
+            now_enc = _truthy(row.get("encrypted"))
+            last_enc = row.get("last_enc_time") or 0
+            last_clear = row.get("last_clear_time") or 0
+            out.append({
+                "name": self._name(row),
+                "freq": row.get("freq"),
+                "nac": _as_int(row.get("nac")),
+                "encrypted_now": now_enc,
+                "enc_calls": enc_calls,
+                "clear_calls": clear_calls,
+                # The channel "is encrypted" when its most recent call was.
+                "encrypted": now_enc or (enc_calls > 0 and float(last_enc) >= float(last_clear)),
+                "algid": _as_int(row.get("algid")) or _as_int(row.get("last_enc_algid")),
+                "keyid": _as_int(row.get("keyid")) or _as_int(row.get("last_enc_keyid")),
+                "voice": _truthy(row.get("voice")),
+            })
+        return out

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

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

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

@@ -0,0 +1,351 @@
+"""
+Plain-language health checklist for the local dashboard (/api/health).
+
+A first-time owner opens http://<node>/ and needs three answers: is this node
+working, if not what is wrong, and what do I do about it. Each check below
+answers one of those for one link in the chain
+
+    password -> C2 enrollment -> MQTT -> SDR -> what to listen to -> OP25
+             -> decoding -> audio (PulseAudio) -> local stream (Icecast)
+
+`gather_facts()` does the I/O; `build_checks()` is a pure function of those
+facts so every state (unapproved, no SDR, OP25 crashed, …) is unit-testable.
+
+Read-only: nothing here changes node state, and it never polls the OP25
+terminal itself (that drains OP25's call_log, which metadata_watcher owns) —
+it reads metadata_watcher.console_state instead.
+"""
+import time
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from app.config import settings
+from app.internal.logger import logger
+
+OK, WARN, ERROR, OFF = "ok", "warn", "error", "off"
+
+# OP25's console answers within a second or two of a healthy multi_rx; past
+# this the console is considered silent.
+CONSOLE_STALE_SECONDS = 5.0
+# P25 control channels send several messages a second; this long without one
+# means OP25 is not decoding.
+TSBK_STALE_SECONDS = 15.0
+
+
+def _check(cid: str, label: str, state: str, summary: str, action: str = "") -> Dict[str, str]:
+    return {"id": cid, "label": label, "state": state, "summary": summary, "action": action}
+
+
+def _mhz(freq: Any) -> Optional[str]:
+    try:
+        f = float(freq)
+    except (TypeError, ValueError):
+        return None
+    if f <= 0:
+        return None
+    return f"{f / 1e6:.4f} MHz" if f > 1e5 else f"{f:.4f} MHz"
+
+
+def _ago(seconds: float) -> str:
+    seconds = max(0, int(seconds))
+    if seconds < 60:
+        return f"{seconds}s ago"
+    if seconds < 3600:
+        return f"{seconds // 60} min ago"
+    return f"{seconds // 3600} h ago"
+
+
+def _channel_label(ch: Dict[str, Any]) -> str:
+    freq = _mhz(ch.get("freq"))
+    name = ch.get("name") or ""
+    if name and freq and freq not in name:
+        return f"{name} ({freq})"
+    return name or freq or "a channel"
+
+
+def _conventional_decode_check(chans: List[Dict[str, Any]]) -> Dict[str, str]:
+    """P25 conventional (node-26#24): no control channel to lock on to, so the
+    proof of decoding is P25 frames (a NAC) heard on the channels — and the one
+    failure the owner can't fix is encryption."""
+    if not chans:
+        return _check("decode", "Decoding", WARN, "Waiting for OP25 to report its channels…")
+    encrypted = [c for c in chans if c.get("encrypted")]
+    if encrypted:
+        from app.internal.conventional import algid_name
+        parts = []
+        for c in encrypted:
+            key = f", key 0x{c['keyid']:04X}" if c.get("keyid") is not None else ""
+            parts.append(f"{_channel_label(c)}: {algid_name(c.get('algid'))}{key}")
+        # "Everything is encrypted" only when no clear call was ever heard —
+        # `encrypted` alone means a channel's most recent call was.
+        every = len(encrypted) == len(chans) and not any(c.get("clear_calls") for c in chans)
+        return _check(
+            "decode", "Decoding", ERROR if every else WARN,
+            ("Encrypted — " if every else "Partly encrypted — ") + "; ".join(parts)
+            + ". Encrypted calls can't be decoded, so nothing from "
+            + ("these channels" if every else "those channels") + " is recorded.",
+            "Nothing to fix on the node: the agency encrypts this traffic. "
+            + ("Point the node at a different system." if every else "Clear calls are still recorded."),
+        )
+    heard = [c for c in chans if c.get("nac")]
+    listening = ", ".join(_channel_label(c) for c in chans)
+    if not heard:
+        return _check(
+            "decode", "Decoding", OK,
+            f"Listening on {listening} (P25 conventional) — no P25 traffic heard yet.",
+            "Normal on a quiet channel: conventional channels are silent until someone talks. If it stays "
+            "silent through a busy period, check the frequency, antenna and the SDR's PPM.",
+        )
+    nacs = ", ".join(sorted({f"0x{c['nac']:03X}" for c in heard}))
+    return _check("decode", "Decoding", OK, f"Listening on {listening} (P25 conventional) — decoding, NAC {nacs}.")
+
+
+def active_config(node_cfg) -> Any:
+    if node_cfg.override_system_id and node_cfg.override_config:
+        return node_cfg.override_config
+    return node_cfg.system_config
+
+
+def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]:
+    """Facts -> ordered checklist. See gather_facts() for the keys."""
+    now = f.get("now", time.time())
+    checks: List[Dict[str, str]] = []
+
+    # 1. Dashboard password
+    if f["default_password"]:
+        checks.append(_check(
+            "password", "Dashboard password", WARN,
+            "Still the factory default — anyone on this network can log in.",
+            "Set DASHBOARD_PASSWORD in the node's .env file, then restart the edge-node container.",
+        ))
+    else:
+        checks.append(_check("password", "Dashboard password", OK, "Changed from the default."))
+
+    # 2. C2 enrollment / approval
+    c2_url, has_key = f["c2_url"], f["has_api_key"]
+    if not c2_url:
+        checks.append(_check(
+            "c2", "Server (C2) link", OFF,
+            "Standalone — no C2 server is set. Local listening below still works.",
+            "To join a DRB server, re-run install.sh with an enrollment token.",
+        ))
+    elif not has_key:
+        checks.append(_check(
+            "c2", "Server (C2) link", WARN,
+            f"Waiting for approval — this node has no API key from {c2_url} yet, so nothing is uploaded.",
+            f"Ask a DRB admin to approve node \"{f['node_id']}\" on the Nodes page, then re-run "
+            "install.sh (or restart the edge-node container). Local listening below works meanwhile.",
+        ))
+    else:
+        checks.append(_check("c2", "Server (C2) link", OK, f"Approved — API key on file for {c2_url}."))
+
+    # 3. MQTT
+    broker = f"{f['mqtt_broker']}:{f['mqtt_port']}"
+    if f["mqtt_connected"]:
+        checks.append(_check("mqtt", "Live link (MQTT)", OK, f"Connected to {broker}."))
+    elif not c2_url and not has_key:
+        checks.append(_check(
+            "mqtt", "Live link (MQTT)", OFF, f"Not connected to {broker} (expected for a standalone node).",
+        ))
+    elif not has_key:
+        checks.append(_check(
+            "mqtt", "Live link (MQTT)", WARN,
+            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).",
+        ))
+    else:
+        checks.append(_check(
+            "mqtt", "Live link (MQTT)", ERROR,
+            f"Not connected to {broker} even though an API key is on file.",
+            "If the key was just installed or the node was re-approved, restart the edge-node container "
+            "so it loads the key. If it still fails, the key may be stale (every re-approval issues a new "
+            "one) — re-run install.sh, and check the node can reach the broker.",
+        ))
+
+    # 4. SDR
+    sdr_count = f.get("sdr_count")
+    if sdr_count is None:
+        checks.append(_check(
+            "sdr", "SDR receiver", WARN, "Couldn't check — the SDR service isn't answering.",
+            "Check the secondary-sdr container is running.",
+        ))
+    elif sdr_count == 0:
+        checks.append(_check(
+            "sdr", "SDR receiver", ERROR, "No RTL-SDR dongle detected.",
+            "Plug the SDR into a USB port (avoid unpowered hubs), then restart the stack.",
+        ))
+    else:
+        checks.append(_check(
+            "sdr", "SDR receiver", OK, f"{sdr_count} SDR{'s' if sdr_count != 1 else ''} detected.",
+        ))
+
+    # 5. What to listen to
+    src = f.get("source")
+    if not src:
+        checks.append(_check(
+            "source", "What to listen to", WARN, "Nothing chosen yet.",
+            "Pick a system or enter a frequency under \"Listen locally\" below.",
+        ))
+    else:
+        origin = "chosen on this dashboard" if src["local"] else "assigned by the server"
+        kind = "P25 conventional" if src.get("mode") == "conventional" else src["type"]
+        checks.append(_check("source", "What to listen to", OK, f"{src['name']} ({kind}) — {origin}."))
+
+    # 6. OP25
+    op25 = f.get("op25_status")
+    running = op25 == "running"
+    if op25 is None:
+        checks.append(_check(
+            "op25", "Radio decoder (OP25)", ERROR, "The op25 container isn't answering.",
+            "Check the op25 container is running; its log usually says why it stopped.",
+        ))
+    elif running:
+        checks.append(_check("op25", "Radio decoder (OP25)", OK, "Running."))
+    elif op25 == "crashed":
+        checks.append(_check(
+            "op25", "Radio decoder (OP25)", ERROR, "Stopped unexpectedly.",
+            "Open \"OP25 log\" below for the reason — usually the SDR is unplugged or in use. Then press Start.",
+        ))
+    elif not src:
+        checks.append(_check("op25", "Radio decoder (OP25)", OFF, "Stopped — choose what to listen to first."))
+    else:
+        checks.append(_check("op25", "Radio decoder (OP25)", WARN, "Stopped.", "Press Start."))
+
+    # 7. Decoding
+    console = f.get("console") or {}
+    last_poll = console.get("last_poll_epoch")
+    is_analog = bool(src) and str(src.get("type", "")).upper() in ("NBFM", "ANALOG")
+    is_conventional = bool(src) and src.get("mode") == "conventional"
+    if not running:
+        checks.append(_check("decode", "Decoding", OFF, "Starts when OP25 runs."))
+    elif last_poll is None or now - last_poll > CONSOLE_STALE_SECONDS:
+        checks.append(_check(
+            "decode", "Decoding", WARN, "OP25 is running but hasn't reported in yet.",
+            "Give it 20 seconds after a start. If this stays, open the OP25 log below.",
+        ))
+    elif is_analog:
+        chans = console.get("channels") or []
+        freq = _mhz(chans[0].get("freq")) if chans else None
+        checks.append(_check(
+            "decode", "Decoding", OK, f"Tuned to {freq or 'the analog frequency'} (analog FM — no decoding needed).",
+        ))
+    elif is_conventional:
+        checks.append(_conventional_decode_check(f.get("conventional") or []))
+    else:
+        trunk = console.get("trunk") or []
+        tsbk = [float(t.get("last_tsbk") or 0) for t in trunk]
+        if not trunk:
+            checks.append(_check("decode", "Decoding", WARN, "Waiting for OP25 to report control-channel status…"))
+        elif max(tsbk) > 0 and now - max(tsbk) < TSBK_STALE_SECONDS:
+            sys_ = trunk[tsbk.index(max(tsbk))]
+            cc = _mhz(sys_.get("rxchan"))
+            checks.append(_check(
+                "decode", "Decoding", OK, f"Locked on the control channel{' ' + cc if cc else ''}.",
+            ))
+        else:
+            checks.append(_check(
+                "decode", "Decoding", ERROR, "Not decoding the control channel.",
+                "Usually a wrong control-channel frequency, a weak signal/antenna, or SDR frequency error (PPM). "
+                "Try another control channel for the same system.",
+            ))
+
+    # 8. Audio (PulseAudio — recording + Discord path)
+    if not f["audio_capturing"]:
+        checks.append(_check(
+            "audio", "Audio", ERROR if running else OFF,
+            "No audio reaching the recorder." if running else "Nothing to hear until OP25 runs.",
+            "The op25 container owns the audio path (PulseAudio); restart it if this persists." if running else "",
+        ))
+    else:
+        last_voice = f.get("last_voice_epoch")
+        if last_voice is None:
+            checks.append(_check("audio", "Audio", OK, "Audio path up — no voice heard yet."))
+        elif now - last_voice < 3:
+            checks.append(_check("audio", "Audio", OK, "Voice right now."))
+        else:
+            checks.append(_check("audio", "Audio", OK, f"Audio path up — last voice {_ago(now - last_voice)}."))
+
+    # 9. Local stream (Icecast)
+    ice = f.get("icecast_live")
+    if ice is None:
+        checks.append(_check(
+            "stream", "Local stream", ERROR, f"Icecast isn't answering on port {f['icecast_port']}.",
+            "Check the icecast container is running.",
+        ))
+    elif ice:
+        checks.append(_check("stream", "Local stream", OK, "Live — press play in the player below."))
+    elif running:
+        checks.append(_check(
+            "stream", "Local stream", WARN, "OP25 is running but nothing is feeding the stream yet.",
+            "It normally appears within ~20 seconds of a start; if not, check the OP25 log below.",
+        ))
+    else:
+        checks.append(_check("stream", "Local stream", OFF, "Goes live when OP25 runs."))
+
+    return checks
+
+
+async def icecast_live() -> Optional[bool]:
+    """True if the configured mount has a source, False if not, None if Icecast is down."""
+    url = f"http://{settings.icecast_host}:{settings.icecast_port}/status-json.xsl"
+    try:
+        async with httpx.AsyncClient(timeout=3) as client:
+            r = await client.get(url)
+            r.raise_for_status()
+            sources = (r.json().get("icestats") or {}).get("source") or []
+    except Exception as e:
+        logger.debug(f"Icecast status check failed: {e}")
+        return None
+    if isinstance(sources, dict):
+        sources = [sources]
+    return any(str(s.get("listenurl", "")).endswith(settings.icecast_mount) for s in sources)
+
+
+async def gather_facts() -> Dict[str, Any]:
+    from app.internal import auth, credentials
+    from app.internal.call_recorder import call_recorder
+    from app.internal.config_manager import load_node_config
+    from app.internal.metadata_watcher import metadata_watcher
+    from app.internal.mqtt_manager import mqtt_manager
+    from app.internal.op25_client import op25_client
+    from app.internal.radio_config import system_mode
+    from app.internal.secondary_sdr_client import secondary_sdr_client
+
+    node_cfg = load_node_config()
+    cfg = active_config(node_cfg)
+    op25 = await op25_client.status()
+
+    devs = await secondary_sdr_client.devices()
+    sdr_count: Optional[int] = len(devs) if devs is not None else None
+    if not sdr_count:
+        # op25's own lsusb count — its :stable image may lack lsusb and say 0,
+        # so only trust a positive answer from it.
+        op25_devs = await op25_client.devices()
+        if op25_devs and op25_devs.get("count"):
+            sdr_count = op25_devs["count"]
+
+    activity = call_recorder.audio_activity()
+    return {
+        "now": time.time(),
+        "node_id": settings.node_id,
+        "default_password": auth.is_using_default_password(),
+        "c2_url": settings.c2_url,
+        "has_api_key": bool(credentials.get_api_key()),
+        "mqtt_connected": mqtt_manager.is_connected,
+        "mqtt_broker": settings.mqtt_broker,
+        "mqtt_port": settings.mqtt_port,
+        "sdr_count": sdr_count,
+        "source": None if cfg is None else {
+            "name": cfg.name, "type": cfg.type, "system_id": cfg.system_id,
+            "mode": system_mode(cfg.type, cfg.config),
+            "local": bool(node_cfg.override_system_id and node_cfg.override_config),
+        },
+        "conventional": metadata_watcher.conventional_channels,
+        "op25_status": (op25 or {}).get("status") if op25 is not None else None,
+        "console": metadata_watcher.console_state,
+        "audio_capturing": call_recorder.is_capturing,
+        "last_voice_epoch": activity.last_voice_epoch,
+        "icecast_live": await icecast_live(),
+        "icecast_port": settings.icecast_port,
+    }

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

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

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

@@ -36,6 +36,19 @@ have seen the whole window and ask the rolling console history "what was active
 during this audio, give or take a few seconds" — see _attribute and the
 ATTRIBUTION_* constants.
 
+CONVENTIONAL CHANNELS (node-26#24 P25 conventional, node-26#20 analog). There
+is no control channel, so internal/conventional.py rewrites each console update
+first: a conventional channel that is keyed becomes an ordinary active row whose
+talkgroup is the link-control TGID when it is distinctive, else a stable
+synthetic id derived from the channel frequency. Such a row counts as a talker
+without a radio id (`drb_active`) — link control does not always carry one and
+analog never does. One more rule, applied only at CLOSE: with exactly one
+conventional channel configured, audio the console said nothing about can only
+have come from that channel, so it is attributed to it rather than discarded.
+Not at open: on a channel whose link control carries a real talkgroup, the
+grant may arrive a poll after the audio, and a premature synthetic label would
+force a bogus split when it does.
+
 ORPHAN AUDIO. If nothing in the console history overlaps the window, the audio
 is unattributed: Liquidsoap fallback, a test tone, stray noise, or a dropped
 `call_log`. Policy is DISCARD AND SHOUT — the recording is not uploaded and no
@@ -74,6 +87,7 @@ from typing import Optional, Callable, Awaitable, Any, List, Dict
 
 from app.config import settings
 from app.internal.call_recorder import AudioActivity
+from app.internal.conventional import ConventionalChannels
 from app.internal.op25_client import op25_client
 from app.internal.logger import logger
 
@@ -177,6 +191,22 @@ def _as_float(value: Any) -> Optional[float]:
         return None
 
 
+def _talker(channel: Dict[str, Any]) -> Optional[tuple]:
+    """(tgid, rid) if this channel row says someone is transmitting, else None.
+
+    Trunked rows need both a talkgroup and a source address — an idle channel
+    can keep its last tgid. A normalized conventional row (`drb_active`) only
+    has a tgid while a call is up, and may legitimately have no radio id.
+    """
+    tgid = _as_int(channel.get("tgid"))
+    if tgid is None:
+        return None
+    rid = _as_int(channel.get("srcaddr"))
+    if rid is None and not channel.get("drb_active"):
+        return None
+    return tgid, rid
+
+
 def _iso(epoch: Optional[float]) -> Optional[str]:
     if epoch is None:
         return None
@@ -235,6 +265,13 @@ class MetadataWatcher:
         self._last_tx_end: Optional[float] = None  # epoch of the srcaddr 1→0 edge
         self._last_ok_poll: float = 0.0
 
+        # Dashboard-only view of what the OP25 console last said. Read by
+        # /api/health; never consulted by segmentation or attribution.
+        self._terminal_epoch: Optional[float] = None
+        self._terminal_channels: List[Dict[str, Any]] = []
+        self._terminal_trunk: List[Dict[str, Any]] = []
+        self._last_heard: Optional[ConsoleEvent] = None
+
         # Rolling console history for close-time attribution.
         self._console: deque[ConsoleEvent] = deque(maxlen=CONSOLE_HISTORY_MAX)
 
@@ -245,6 +282,10 @@ class MetadataWatcher:
         # Field-visible counter of discarded orphan audio.
         self._unattributed_segments: int = 0
 
+        # Conventional-channel normalizer (node-26#24/#20); also remembers the
+        # channel list and their encryption state for the dashboard + checkin.
+        self._conventional = ConventionalChannels()
+
         # Injectable for tests; production is always the host wall clock.
         self._clock: Callable[[], float] = time.time
 
@@ -293,7 +334,9 @@ class MetadataWatcher:
             return
 
         self._last_ok_poll = now
+        update = self._conventional.normalize(update)
         self._record_console(update, now)
+        self._note_terminal(update, now)
 
         activity = self._snapshot()
         if activity is None or not activity.capturing:
@@ -333,10 +376,10 @@ class MetadataWatcher:
                 self._console.append(event)
 
         for channel in update.channels:
-            tgid = _as_int(channel.get("tgid"))
-            srcaddr = _as_int(channel.get("srcaddr"))
-            if tgid is None or srcaddr is None:
+            talker = _talker(channel)
+            if talker is None:
                 continue  # idle channel says nothing about who is talking
+            tgid, srcaddr = talker
             self._console.append(ConsoleEvent(
                 epoch=now,
                 tgid=tgid,
@@ -350,6 +393,18 @@ class MetadataWatcher:
         while self._console and self._console[0].epoch < cutoff:
             self._console.popleft()
 
+    def _note_terminal(self, update: Any, now: float) -> None:
+        """Keep the latest console snapshot for the dashboard. No side effects
+        on segment state — trunk_update/channel_update are not sent on every
+        poll, so the last non-empty one is kept."""
+        self._terminal_epoch = now
+        if update.channels:
+            self._terminal_channels = update.channels
+        if getattr(update, "trunk", None):
+            self._terminal_trunk = update.trunk
+        if self._console:
+            self._last_heard = self._console[-1]
+
     def _is_duplicate_grant(self, event: ConsoleEvent) -> bool:
         for index in range(len(self._console) - 1, -1, -1):
             if len(self._console) - index > _GRANT_DEDUPE_DEPTH:
@@ -359,7 +414,7 @@ class MetadataWatcher:
                 return True
         return False
 
-    def _attribute(self, start: float, end: float) -> Optional[Attribution]:
+    def _attribute(self, start: float, end: float, allow_sole_channel: bool = False) -> Optional[Attribution]:
         """
         Resolve which talkgroup a stretch of audio belongs to.
 
@@ -367,7 +422,8 @@ class MetadataWatcher:
         how well its console activity overlaps the audio itself, preferring
         real overlap over merely being nearby, and grants over channel rows.
         Returns None only when NOTHING was observed in that band at all — the
-        orphan-audio case.
+        orphan-audio case — unless allow_sole_channel and the node listens to
+        exactly one conventional channel (see module docstring).
         """
         low = start - ATTRIBUTION_LOOKBACK_SECONDS
         high = end + ATTRIBUTION_LOOKAHEAD_SECONDS
@@ -395,7 +451,14 @@ class MetadataWatcher:
                 found.rid = event.rid  # most recent wins
 
         if not candidates:
-            return None
+            sole = self._conventional.sole_channel() if allow_sole_channel else None
+            if sole is None:
+                return None
+            logger.info(
+                f"No console activity near {start:.3f}-{end:.3f}; attributing to the only "
+                f"conventional channel, {sole['name']} (tgid {sole['tgid']})."
+            )
+            return Attribution(tgid=sole["tgid"], name=sole["name"], freq=sole["freq"])
 
         best = max(
             candidates.values(),
@@ -510,6 +573,11 @@ class MetadataWatcher:
             self._refresh_meta_from_log(entry)
             return
 
+        if self._same_channel(tgid, entry.get("freq"), entry.get("tgtag") or ""):
+            self._transmissions += 1
+            self._refresh_meta_from_log(entry)
+            return
+
         # FORCED SPLIT. Two talkgroups can be back to back with no silence
         # between them; pure audio segmentation would merge them into one file
         # under one label, which is exactly the kind of wrong that corrupts
@@ -526,6 +594,21 @@ class MetadataWatcher:
             transmissions=1,
         )
 
+    def _same_channel(self, tgid: int, freq: Any, name: str) -> bool:
+        """A conventional channel seen under its synthetic id and its real
+        talkgroup is one channel, not a split (see ConventionalChannels.
+        same_channel_relabel). Adopts the real talkgroup when offered."""
+        decision = self._conventional.same_channel_relabel(self._current_tgid, self._current_freq, tgid, freq)
+        if decision == "adopt":
+            logger.info(
+                f"Segment {self._active_call_id}: link control names tgid {tgid} for the channel "
+                f"labelled {self._current_tgid} — relabelling, not splitting."
+            )
+            self._current_tgid = tgid
+            if name:
+                self._current_tgid_name = name
+        return decision is not None
+
     async def _scan_channels(self, channels: List[Dict[str, Any]], now: float) -> None:
         """Channel rows in audio mode: refresh metadata, catch an unlogged split."""
         if self._active_call_id is None:
@@ -534,14 +617,15 @@ class MetadataWatcher:
         active: List[Dict[str, Any]] = []
         ours = False
         for channel in channels:
-            tgid = _as_int(channel.get("tgid"))
-            srcaddr = _as_int(channel.get("srcaddr"))
-            if tgid is None or srcaddr is None:
+            talker = _talker(channel)
+            if talker is None:
                 continue
+            tgid, srcaddr = talker
             active.append(channel)
-            if tgid == self._current_tgid:
+            if tgid == self._current_tgid or self._same_channel(tgid, channel.get("freq"), channel.get("tag") or ""):
                 ours = True
-                self._current_srcaddr = srcaddr
+                if srcaddr is not None:
+                    self._current_srcaddr = srcaddr
                 self._last_activity = now
                 self._refresh_meta_from_channel(channel)
 
@@ -638,13 +722,14 @@ class MetadataWatcher:
         foreign_active_tgid: Optional[int] = None
 
         for channel in channels:
-            srcaddr = _as_int(channel.get("srcaddr"))
-            chan_tgid = _as_int(channel.get("tgid"))
-            if srcaddr is None:
+            talker = _talker(channel)
+            if talker is None:
                 continue
+            chan_tgid, srcaddr = talker
             if chan_tgid == self._current_tgid:
                 tx_active = True
-                self._current_srcaddr = srcaddr
+                if srcaddr is not None:
+                    self._current_srcaddr = srcaddr
                 self._refresh_meta_from_channel(channel)
             elif chan_tgid is not None:
                 foreign_active_tgid = chan_tgid
@@ -848,7 +933,7 @@ class MetadataWatcher:
         two talkgroups' console activity overlaps one recording, i.e. the split
         logic should have fired and did not.
         """
-        found = self._attribute(start, end)
+        found = self._attribute(start, end, allow_sole_channel=self._current_tgid is None)
         if found is None:
             return
 
@@ -898,6 +983,25 @@ class MetadataWatcher:
     def is_active(self) -> bool:
         return self._active_call_id is not None
 
+    @property
+    def console_state(self) -> Dict[str, Any]:
+        """Last thing the OP25 console reported, for /api/health only."""
+        heard = self._last_heard
+        return {
+            "last_poll_epoch": self._terminal_epoch,
+            "channels": self._terminal_channels,
+            "trunk": self._terminal_trunk,
+            "last_heard": None if heard is None else {
+                "epoch": heard.epoch, "tgid": heard.tgid, "name": heard.name, "freq": heard.freq,
+            },
+        }
+
+    @property
+    def conventional_channels(self) -> List[Dict[str, Any]]:
+        """Per conventional channel: name, freq, NAC heard, encryption. Empty on
+        a trunked system. Read by /api/health and the checkin."""
+        return self._conventional.encryption_report()
+
     @property
     def unattributed_segments(self) -> int:
         """Orphan-audio segments discarded since start. Surfaced on /api/status."""

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

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import shutil
 from datetime import datetime, timezone
 from collections import deque
 from typing import Optional, Callable, Awaitable, Dict, Any
@@ -8,9 +9,26 @@ from app.config import settings
 from app.internal.logger import logger
 from app.internal import credentials, key_pickup
 
+
+def disk_report(path: str) -> Dict[str, Optional[int]]:
+    """Free/total bytes of the filesystem holding `path` (node-26#41).
+
+    /recordings is a bind mount of the host checkout, so this is the host disk
+    the recorder writes to and Docker's images/build cache live on. A full disk
+    stopped node-002 recording for 17 days while it still showed Online. Nulls
+    on error, so C2 overwrites a stale reading instead of keeping it.
+    """
+    try:
+        usage = shutil.disk_usage(path)
+    except OSError:
+        return {"disk_free_bytes": None, "disk_total_bytes": None}
+    return {"disk_free_bytes": usage.free, "disk_total_bytes": usage.total}
+
+
 CommandCallback = Callable[[Dict[str, Any]], Awaitable[None]]
 ConfigCallback = Callable[[Dict[str, Any]], Awaitable[None]]
 ApiKeyCallback = Callable[[Dict[str, Any]], Awaitable[None]]
+DesiredCallback = Callable[[Dict[str, Any]], Awaitable[None]]
 
 # CONNACK codes meaning "your credentials are wrong", as paho reports them:
 # MQTT 3.1.1 return codes 4/5, and the MQTT 5 reason codes paho maps them to
@@ -46,6 +64,7 @@ class MQTTManager:
         self.on_command: Optional[CommandCallback] = None
         self.on_config_push: Optional[ConfigCallback] = None
         self.on_api_key: Optional[ApiKeyCallback] = None
+        self.on_discord_desired: Optional[DesiredCallback] = None
 
         self._offline_buffer = deque(maxlen=settings.offline_call_buffer_size)
 
@@ -55,6 +74,10 @@ class MQTTManager:
         self._t_metadata = f"nodes/{nid}/metadata"
         self._t_commands = f"nodes/{nid}/commands"
         self._t_config = f"nodes/{nid}/config"
+        # Discord radio desired state (retained by C2) and our actual state —
+        # node-26#25. Token IDs only; the strings come over HTTPS.
+        self._t_discord_desired = f"nodes/{nid}/discord/desired"
+        self._t_discord_actual = f"nodes/{nid}/discord/actual"
         # TODO(mqtt-cutover): dead once enrollment lands client-side. This
         # was the pre-dynsec key-delivery path (server retain-publishes the
         # api_key here after admin approval; node asks for redelivery via
@@ -131,6 +154,7 @@ class MQTTManager:
             self._connected = True
             client.subscribe(self._t_commands, qos=1)
             client.subscribe(self._t_config, qos=1)
+            client.subscribe(self._t_discord_desired, qos=1)
             client.subscribe(self._t_api_key, qos=2)  # TODO(mqtt-cutover): see _t_api_key comment above
             client.subscribe(self._t_discovery, qos=0)
             logger.info("MQTT connected.")
@@ -159,6 +183,8 @@ class MQTTManager:
             asyncio.run_coroutine_threadsafe(self.on_command(payload), self._loop)
         elif msg.topic == self._t_config and self.on_config_push:
             asyncio.run_coroutine_threadsafe(self.on_config_push(payload), self._loop)
+        elif msg.topic == self._t_discord_desired and self.on_discord_desired:
+            asyncio.run_coroutine_threadsafe(self.on_discord_desired(payload), self._loop)
         elif msg.topic == self._t_api_key and self.on_api_key:
             asyncio.run_coroutine_threadsafe(self.on_api_key(payload), self._loop)
         elif msg.topic == self._t_discovery:
@@ -285,6 +311,12 @@ class MQTTManager:
         else:
             self._publish(self._t_metadata, payload, qos=1)
 
+    def publish_discord_actual(self, payload: dict) -> None:
+        """What the Discord radio manager is actually doing. Not buffered while
+        offline: a fresh report is sent after every change and periodically,
+        and only the newest one means anything."""
+        self._publish(self._t_discord_actual, payload, qos=1)
+
     async def _flush_offline_buffer(self):
         if not self._offline_buffer:
             return
@@ -327,6 +359,11 @@ class MQTTManager:
             "secondary_sdr_priority": config.secondary_sdr_priority,
         }
         payload["sdr_pins"] = config.sdr_pins
+        # Conventional channels (node-26#24): name/freq/NAC and whether the
+        # agency encrypts them — the one decode failure nobody can fix on the
+        # node, so C2 has to be told. Empty list on a trunked system.
+        from app.internal.metadata_watcher import metadata_watcher
+        payload["radio_channels"] = metadata_watcher.conventional_channels
         secondary = await secondary_sdr_client.status()
         # Unreachable secondary-sdr: say so with explicit nulls. C2 only
         # overwrites keys present in the checkin, so omitting them would leave
@@ -352,6 +389,7 @@ class MQTTManager:
             count = devices.get("count") or None
         if count is not None:
             payload["sdr_count"] = count
+        payload.update(disk_report(settings.recordings_path))
         self._publish(self._t_checkin, payload, qos=1)
 
     def _publish(self, topic: str, payload: dict, qos: int = 0, retain: bool = False):

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

@@ -11,6 +11,11 @@ from app.internal.logger import logger
 #                    each id is also a top-level key holding that channel's dict
 #                    (freq/tgid/tag/srcaddr/svcopts/hold_tgid/…).
 #
+#   trunk_update   — per-system trunking state keyed "0", "1", …; each carries
+#                    `last_tsbk` (OP25's time.time() of the last control-channel
+#                    message decoded) — the only direct proof OP25 is DECODING,
+#                    not merely running. Surfaced on the dashboard only.
+#
 #   call_log       — an EVENT QUEUE, not a snapshot. `log` holds entries appended
 #                    by tk_p25.log_call() at channel-grant time, each stamped with
 #                    OP25's own time.time(). get_call_log() DRAINS the deque, so
@@ -18,7 +23,7 @@ from app.internal.logger import logger
 #                    it forever. The deque is capped at CALL_LOG_MAX_LEN = 10, so
 #                    the consumer must keep up.
 #
-# Everything else (trunk_update, rx_update, terminal_config, …) is ignored.
+# Everything else (rx_update, terminal_config, …) is ignored.
 TERMINAL_UPDATE_COMMAND = [{"command": "update", "arg1": 0, "arg2": 0}]
 
 
@@ -28,6 +33,7 @@ class TerminalUpdate:
 
     channels: List[Dict[str, Any]] = field(default_factory=list)
     call_log: List[Dict[str, Any]] = field(default_factory=list)
+    trunk: List[Dict[str, Any]] = field(default_factory=list)
 
 
 class OP25Client:
@@ -45,6 +51,23 @@ class OP25Client:
             logger.error(f"OP25 start failed: {e}")
             return False
 
+    async def start_with_reason(self) -> Optional[str]:
+        """Start OP25; None on success, else a human-readable reason."""
+        try:
+            async with httpx.AsyncClient(timeout=10) as client:
+                r = await client.post(f"{self.api_url}/op25/start")
+        except Exception as e:
+            logger.error(f"OP25 start failed: {e}")
+            return "The op25 container isn't answering."
+        if r.is_success:
+            return None
+        try:
+            detail = r.json().get("detail")
+        except Exception:
+            detail = None
+        logger.error(f"OP25 start failed: HTTP {r.status_code} {detail or r.text[:200]}")
+        return detail or f"op25 container returned HTTP {r.status_code}."
+
     async def stop(self) -> bool:
         try:
             async with httpx.AsyncClient(timeout=10) as client:
@@ -65,6 +88,17 @@ class OP25Client:
             logger.error(f"OP25 status failed: {e}")
             return None
 
+    async def log(self, lines: int = 80) -> Optional[List[str]]:
+        """Tail of multi_rx/liquidsoap output, kept by the op25 container."""
+        try:
+            async with httpx.AsyncClient(timeout=5) as client:
+                r = await client.get(f"{self.api_url}/op25/log", params={"lines": lines})
+                r.raise_for_status()
+                return r.json().get("lines", [])
+        except Exception as e:
+            logger.error(f"OP25 log fetch failed: {e}")
+            return None
+
     async def devices(self) -> Optional[Dict[str, Any]]:
         try:
             async with httpx.AsyncClient(timeout=5) as client:
@@ -144,6 +178,11 @@ def parse_terminal_messages(messages: Any) -> TerminalUpdate:
                 if isinstance(channel, dict):
                     update.channels.append(channel)
 
+        elif json_type == "trunk_update":
+            for key, system in msg.items():
+                if key != "json_type" and isinstance(system, dict):
+                    update.trunk.append(system)
+
         elif json_type == "call_log":
             for entry in msg.get("log") or []:
                 if isinstance(entry, dict):

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

@@ -0,0 +1,129 @@
+"""
+What kind of radio system a SystemConfig describes, and the op25-container
+generate-config payload for it.
+
+C2 system documents (the `config` blob is passed through untouched):
+
+    P25 trunked       type "P25", config.mode absent or "trunked",
+                      config.control_channels = [MHz, ...]
+    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]
+
+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.
+"""
+from typing import Any, Dict, List, Optional
+
+TRUNKED, CONVENTIONAL, ANALOG, DMR = "trunked", "conventional", "analog", "dmr"
+
+# Mirrors op25-container's MAX_CONVENTIONAL_CHANNELS: one audio.py carries two.
+MAX_CONVENTIONAL_CHANNELS = 2
+
+
+def to_hz(freq: Any) -> int:
+    """MHz floats (< 1e6) or Hz ints -> Hz."""
+    f = float(freq)
+    return int(round(f * 1_000_000)) if f < 1_000_000 else int(f)
+
+
+def parse_nac(value: Any) -> Optional[int]:
+    """NAC as typed by an admin ("293", "0x293") or stored as an int. Hex, like
+    RadioReference prints it. None when blank; ValueError when nonsense."""
+    if value is None or (isinstance(value, str) and not value.strip()):
+        return None
+    if isinstance(value, bool):
+        raise ValueError(f"NAC {value!r} isn't a number.")
+    if isinstance(value, int):
+        nac = value
+    else:
+        text = str(value).strip().lower()
+        text = text[2:] if text.startswith("0x") else text
+        try:
+            nac = int(text, 16)
+        except ValueError:
+            raise ValueError(f"NAC {value!r} isn't hex (e.g. 293 or 0x293).")
+    if not 0 < nac <= 0xFFF:
+        raise ValueError(f"NAC {value!r} is outside 001-FFF.")
+    return nac
+
+
+def system_mode(system_type: str, config: Dict[str, Any]) -> str:
+    t = str(system_type or "").upper()
+    if t in ("NBFM", "ANALOG"):
+        return ANALOG
+    if t == "DMR":
+        return DMR
+    mode = str((config or {}).get("mode") or TRUNKED).lower()
+    return CONVENTIONAL if mode == CONVENTIONAL else TRUNKED
+
+
+def system_kind(system_type: str, config: Dict[str, Any]) -> str:
+    """Label for dashboards: "P25 trunked", "P25 conventional", "Analog FM", "DMR"."""
+    return {
+        TRUNKED: "P25 trunked", CONVENTIONAL: "P25 conventional", ANALOG: "Analog FM", DMR: "DMR",
+    }[system_mode(system_type, config)]
+
+
+def conventional_channels(config: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """config.channels -> [{frequency: Hz, name, nac}] (ValueError if malformed)."""
+    out = []
+    for raw in (config or {}).get("channels") or []:
+        if not isinstance(raw, dict) or raw.get("freq") in (None, ""):
+            raise ValueError("Every conventional channel needs a frequency.")
+        out.append({
+            "frequency": to_hz(raw["freq"]),
+            "name": str(raw.get("name") or "").strip(),
+            "nac": parse_nac(raw.get("nac")),
+        })
+    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, ""):
+        return to_hz(chans[0]["freq"])
+    legacy = (config or {}).get("control_channels") or []
+    return to_hz(legacy[0]) if legacy else None
+
+
+def build_generate_payload(
+    system_type: str,
+    name: str,
+    config: Dict[str, Any],
+    icecast: Dict[str, Any],
+    hardware_preset: str,
+    ppm_override: Optional[float],
+) -> Dict[str, Any]:
+    """SystemConfig fields -> op25-container /generate-config body."""
+    mode = system_mode(system_type, config)
+    payload: Dict[str, Any] = {
+        "type": system_type,
+        "systemName": name,
+        "icecastConfig": icecast,
+        "hardware_preset": hardware_preset,
+    }
+    if ppm_override is not None:
+        payload["ppm_override"] = ppm_override
+
+    if mode == CONVENTIONAL:
+        payload["type"] = "P25"
+        payload["p25_mode"] = CONVENTIONAL
+        payload["channels"] = []
+        payload["conv_channels"] = conventional_channels(config)
+        return payload
+
+    if mode == ANALOG:
+        freq = analog_frequency(config)
+        payload["channels"] = [freq] if freq else []
+        return payload
+
+    talkgroups = (config or {}).get("talkgroups", []) or []
+    payload["channels"] = [to_hz(ch) for ch in (config or {}).get("control_channels", []) or []]
+    payload["tags"] = [
+        {"talkgroup": str(tg.get("name", "")), "tagDec": int(tg["id"])}
+        for tg in talkgroups if tg.get("id") is not None
+    ]
+    payload["whitelist"] = [int(tg["id"]) for tg in talkgroups if tg.get("id") is not None]
+    return payload

drb-edge-node/app/main.py

@@ -144,20 +144,11 @@ async def on_command(payload: dict):
     action = payload.get("action")
     logger.info(f"Command received: {action}")
 
-    if action == "discord_join":
-        token = payload.get("token")
-        if not token:
-            logger.error("discord_join command missing token — ignoring.")
-            return
-        await radio_bot.join(
-            guild_id=int(payload["guild_id"]),
-            channel_id=int(payload["channel_id"]),
-            token=token,
-            call_active=metadata_watcher.is_active,
-            system_name=payload.get("system_name"),
-        )
-    elif action == "discord_leave":
-        await radio_bot.leave()
+    if action in ("discord_join", "discord_leave"):
+        # Pre-desired-state C2. Discord streams now arrive on
+        # nodes/{id}/discord/desired (node-26#25); a raw-token join is refused
+        # so a bot token is never taken from an MQTT command again.
+        logger.warning(f"Ignoring legacy {action} command — Discord radio is driven by desired state now.")
     elif action in ("set_sdr_config", "set_secondary_priority"):
         from app.internal.sdr_settings import set_sdr_settings
         try:
@@ -191,36 +182,30 @@ async def on_api_key(payload: dict):
         logger.info("Node API key received and saved.")
 
 
-def _to_hz(freq) -> int:
-    """Convert a frequency to Hz. Accepts MHz floats (< 1e6) or Hz ints."""
-    f = float(freq)
-    return int(f * 1_000_000) if f < 1_000_000 else int(f)
-
-
 async def _generate_op25_config(config: SystemConfig) -> bool:
-    """Translate a SystemConfig (Firestore format) into OP25 active.cfg.json + op25.liq."""
+    """Translate a SystemConfig (Firestore format) into OP25 active.cfg.json + op25.liq.
+
+    Trunked / conventional / analog are told apart in radio_config (node-26#24);
+    a system saved before that has no `mode` and is trunked, as it always was.
+    """
     from app.internal.op25_client import op25_client
+    from app.internal.radio_config import build_generate_payload
     node_cfg = load_node_config()
-    raw = config.config
-    payload = {
-        "type": config.type,
-        "systemName": config.name,
-        "channels": [_to_hz(ch) for ch in raw.get("control_channels", [])],
-        "tags": [
-            {"talkgroup": str(tg.get("name", "")), "tagDec": int(tg["id"])}
-            for tg in raw.get("talkgroups", [])
-            if tg.get("id") is not None
-        ],
-        "whitelist": [int(tg["id"]) for tg in raw.get("talkgroups", []) if tg.get("id") is not None],
-        "icecastConfig": {
-            "icecast_host": settings.icecast_host,
-            "icecast_port": settings.icecast_port,
-            "icecast_mountpoint": settings.icecast_mount,
-            "icecast_password": settings.icecast_source_password,
-        },
-        "hardware_preset": node_cfg.hardware_preset,
-        **({"ppm_override": node_cfg.ppm_override} if node_cfg.ppm_override is not None else {}),
-    }
+    try:
+        payload = build_generate_payload(
+            config.type, config.name, config.config,
+            icecast={
+                "icecast_host": settings.icecast_host,
+                "icecast_port": settings.icecast_port,
+                "icecast_mountpoint": settings.icecast_mount,
+                "icecast_password": settings.icecast_source_password,
+            },
+            hardware_preset=node_cfg.hardware_preset,
+            ppm_override=node_cfg.ppm_override,
+        )
+    except ValueError as e:
+        logger.error(f"System {config.name!r} has an unusable channel list: {e}")
+        return False
     return await op25_client.generate_config(payload)
 
 
@@ -306,8 +291,9 @@ async def lifespan(app: FastAPI):
     mqtt_manager.on_command = on_command
     mqtt_manager.on_config_push = on_config_push
     mqtt_manager.on_api_key = on_api_key
+    mqtt_manager.on_discord_desired = radio_bot.apply_desired
 
-    # Start services (radio_bot starts on-demand when a discord_join command arrives)
+    # Start services (radio_bot starts when C2's retained discord/desired arrives)
     await mqtt_manager.connect()
     await metadata_watcher.start()
     await call_recorder.start()  # persistent PulseAudio ring buffer
@@ -326,9 +312,12 @@ async def lifespan(app: FastAPI):
         if (node_cfg.override_system_id and node_cfg.override_config)
         else node_cfg.system_config
     )
-    if node_cfg.configured and active_config:
+    # Not gated on `configured` (set only by a C2 assignment): a standalone or
+    # not-yet-approved node listening to a system chosen on the local
+    # dashboard (an override) must come back on it after a reboot too.
+    if active_config:
         from app.internal.op25_client import op25_client
-        logger.info("Node is configured — waiting for OP25 API then generating config.")
+        logger.info("Node has a system to listen to — waiting for OP25 API then generating config.")
         for attempt in range(10):
             if await _generate_op25_config(active_config):
                 await op25_client.start()

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

@@ -1,12 +1,12 @@
 from fastapi import APIRouter, Depends, HTTPException, Body
 from pydantic import BaseModel
-from typing import Dict, List, Optional
+from typing import Dict, List, Literal, Optional
 import asyncio
 import httpx
 from app.config import settings
 from app.models import SystemConfig, SECONDARY_SDR_MODES
 from app.internal.op25_client import op25_client
-from app.internal.config_manager import load_node_config, save_node_config, apply_system_config
+from app.internal.config_manager import load_node_config, save_node_config
 from app.internal.call_recorder import call_recorder
 from app.internal.discord_radio import radio_bot
 from app.internal.metadata_watcher import metadata_watcher
@@ -36,8 +36,13 @@ async def get_status():
         if (node_cfg.override_system_id and node_cfg.override_config)
         else node_cfg.system_config
     )
+    system_type = None
+    system_kind = None
     if active_config:
+        from app.internal.radio_config import system_kind as _kind
         system_name = active_config.name
+        system_type = active_config.type
+        system_kind = _kind(active_config.type, active_config.config)
         if active_tgid:
             # Reverse engineer the TGID to talkgroup name based on active system object
             tgs = active_config.config.get("talkgroups", [])
@@ -58,6 +63,13 @@ async def get_status():
         "configured": node_cfg.configured,
         "assigned_system_id": node_cfg.assigned_system_id,
         "system_name": system_name,
+        "system_type": system_type,
+        # Human label that tells P25 trunked from P25 conventional (node-26#24).
+        "system_kind": system_kind,
+        # True when what OP25 is tuned to was chosen on this dashboard rather
+        # than assigned by C2 (a local override, revertible to C2's system).
+        "local_choice": bool(node_cfg.override_system_id and node_cfg.override_config),
+        "has_c2_system": node_cfg.system_config is not None,
         "is_recording": call_recorder.is_recording,
         # Health of the PulseAudio capture that feeds every recording — the single
         # most useful signal when recordings come back empty. Segment boundaries
@@ -77,15 +89,27 @@ async def get_status():
         "icecast_url": (
             f"http://{settings.icecast_host}:{settings.icecast_port}{settings.icecast_mount}"
         ),
+        # The browser builds its own URL from these + window.location.hostname;
+        # icecast_host is "localhost" from the node's point of view.
+        "icecast_port": settings.icecast_port,
+        "icecast_mount": settings.icecast_mount,
         "op25": op25_status,
     }
 
 
+@router.get("/health")
+async def get_health():
+    """Plain-language setup/status checklist for the dashboard (see internal/health.py)."""
+    from app.internal import health
+    facts = await health.gather_facts()
+    return {"checks": health.build_checks(facts), "last_heard": facts["console"].get("last_heard")}
+
+
 @router.post("/op25/start")
 async def start_op25():
-    ok = await op25_client.start()
-    if not ok:
-        raise HTTPException(500, "Failed to start OP25")
+    err = await op25_client.start_with_reason()
+    if err:
+        raise HTTPException(502, err)
     return {"ok": True}
 
 
@@ -93,30 +117,178 @@ async def start_op25():
 async def stop_op25():
     ok = await op25_client.stop()
     if not ok:
-        raise HTTPException(500, "Failed to stop OP25")
+        raise HTTPException(502, "The op25 container did not answer the stop request.")
     return {"ok": True}
 
 
+@router.get("/op25/log")
+async def op25_log(lines: int = 80):
+    out = await op25_client.log(lines)
+    if out is None:
+        raise HTTPException(502, "The op25 container isn't answering.")
+    return {"lines": out}
+
+
 @router.get("/config")
 async def get_config():
     return load_node_config()
 
 
+async def _restart_op25_with(config: SystemConfig) -> None:
+    """Generate OP25's config for `config` and restart OP25 on it."""
+    from app.main import _generate_op25_config
+    if not await _generate_op25_config(config):
+        raise HTTPException(
+            502, f"The op25 container did not accept the config for {config.name} — is it running?",
+        )
+    await op25_client.stop()
+    await asyncio.sleep(2)
+    err = await op25_client.start_with_reason()
+    if err:
+        raise HTTPException(502, f"Config saved, but OP25 did not start: {err}")
+
+
+async def _apply_override(config: SystemConfig) -> None:
+    """Make `config` what this node listens to, chosen locally.
+
+    Stored as an override so a C2-assigned system (if any) survives and can
+    be reverted to, and so C2 sees the node is overridden. A later C2 config
+    push replaces it.
+    """
+    node_cfg = load_node_config()
+    node_cfg.override_system_id = config.system_id
+    node_cfg.override_config = config
+    save_node_config(node_cfg)
+    await _restart_op25_with(config)
+    if mqtt_manager.is_connected:
+        await mqtt_manager.publish_checkin()
+
+
 @router.post("/config/system")
 async def set_system_config(config: SystemConfig):
     """
-    Apply a system config locally — called by the web UI or pushed by C2.
-    Writes the OP25 config and persists the node config.
+    Assign a system as if C2 had (persists as the node's own system, not an
+    override) and restart OP25 on it.
+
+    Used to write the raw `config` blob straight to active.cfg.json, which is
+    not an OP25 config at all — OP25 could not start from it.
     """
+    _validate_type(config.type)
     node_cfg = load_node_config()
     node_cfg.assigned_system_id = config.system_id
     node_cfg.system_config = config
     node_cfg.configured = True
     save_node_config(node_cfg)
-    apply_system_config(config)
+    await _restart_op25_with(config)
     return {"ok": True}
 
 
+def _validate_type(system_type: str) -> None:
+    if str(system_type).upper() == "DMR":
+        raise HTTPException(400, "DMR isn't supported by this node's decoder yet — use P25 or analog (NBFM).")
+    if str(system_type).upper() not in ("P25", "NBFM"):
+        raise HTTPException(400, f"Unknown system type {system_type!r} — use P25 or NBFM.")
+
+
+class Talkgroup(BaseModel):
+    id: int
+    name: str = ""
+
+
+class ListenBody(BaseModel):
+    """What to listen to, chosen on the local dashboard."""
+    mode: Literal["system", "p25", "p25conv", "analog"]
+    system_id: Optional[str] = None       # mode=system: a cached C2 system
+    name: Optional[str] = None
+    # MHz. p25 = control channels; p25conv = 1-2 voice channels (node-26#24);
+    # analog = exactly one.
+    frequencies: List[float] = []
+    talkgroups: List[Talkgroup] = []       # p25 only; empty = every talkgroup
+    nac: Optional[str] = None              # p25conv only, hex; empty = any NAC
+
+
+# RTL-SDR (R820T/R828D) tuning range.
+_MIN_MHZ, _MAX_MHZ = 24.0, 1766.0
+
+
+@router.post("/listen")
+async def listen(body: ListenBody):
+    """Point OP25 at a system or frequency and start it — no C2 needed."""
+    if body.mode == "system":
+        from app.internal.system_cacher import get_cached_system
+        cached = get_cached_system(body.system_id or "")
+        if not cached:
+            raise HTTPException(404, "That system isn't in this node's saved list.")
+        config = SystemConfig(**cached)
+        _validate_type(config.type)
+    else:
+        freqs = body.frequencies
+        if not freqs:
+            raise HTTPException(400, "Enter at least one frequency in MHz.")
+        bad = [f for f in freqs if not (_MIN_MHZ <= f <= _MAX_MHZ)]
+        if bad:
+            raise HTTPException(
+                400, f"{bad[0]:g} isn't a frequency an RTL-SDR can tune — enter MHz between "
+                f"{_MIN_MHZ:g} and {_MAX_MHZ:g} (e.g. 851.0125).",
+            )
+        if body.mode == "analog" and len(freqs) != 1:
+            raise HTTPException(400, "Analog listens to exactly one frequency.")
+        first = f"{freqs[0]:.5f}".rstrip("0").rstrip(".")
+        if body.mode == "p25conv":
+            config = _conventional_listen_config(body, freqs, first)
+        else:
+            is_p25 = body.mode == "p25"
+            label = (body.name or "").strip() or (f"P25 {first} MHz" if is_p25 else f"Analog {first} MHz")
+            config = SystemConfig(
+                system_id=f"local-{body.mode}",
+                name=label,
+                type="P25" if is_p25 else "NBFM",
+                config={
+                    "control_channels": freqs,
+                    "talkgroups": [tg.model_dump() for tg in body.talkgroups] if is_p25 else [],
+                },
+            )
+    await _apply_override(config)
+    return {"ok": True, "name": config.name}
+
+
+# One SDR's usable passband, less a channel's width — op25-container makes the
+# exact check; this one just turns the common mistake into a readable error.
+_CONVENTIONAL_MAX_SPAN_MHZ = 1.5
+
+
+def _conventional_listen_config(body: ListenBody, freqs: List[float], first: str) -> SystemConfig:
+    from app.internal.radio_config import MAX_CONVENTIONAL_CHANNELS, parse_nac
+    if len(freqs) > MAX_CONVENTIONAL_CHANNELS:
+        raise HTTPException(400, f"P25 conventional listens to at most {MAX_CONVENTIONAL_CHANNELS} channels.")
+    if len(set(freqs)) != len(freqs):
+        raise HTTPException(400, "The same frequency is entered twice.")
+    if max(freqs) - min(freqs) > _CONVENTIONAL_MAX_SPAN_MHZ:
+        raise HTTPException(
+            400, f"Those channels are {max(freqs) - min(freqs):.3f} MHz apart — one SDR covers about "
+            f"{_CONVENTIONAL_MAX_SPAN_MHZ:g} MHz. Pick channels closer together.",
+        )
+    try:
+        nac = parse_nac(body.nac)
+    except ValueError as e:
+        raise HTTPException(400, str(e))
+    label = (body.name or "").strip() or f"P25 conventional {first} MHz"
+    return SystemConfig(
+        system_id="local-p25conv",
+        name=label,
+        type="P25",
+        config={
+            "mode": "conventional",
+            "channels": [
+                {"freq": f, "name": f"{label} {f:.4f}" if len(freqs) > 1 else label,
+                 **({"nac": f"{nac:03X}"} if nac is not None else {})}
+                for f in freqs
+            ],
+            "talkgroups": [],
+        },
+    )
+
+
 @router.get("/systems")
 async def get_systems():
     from app.internal.system_cacher import load_cached_systems, fetch_and_cache_systems
@@ -129,9 +301,6 @@ async def set_override(
     system_id: Optional[str] = Body(None),
     system_config: Optional[dict] = Body(None)
 ):
-    node_cfg = load_node_config()
-    config = None
-
     if system_id:
         from app.internal.system_cacher import get_cached_system
         cached = get_cached_system(system_id)
@@ -145,19 +314,8 @@ async def set_override(
     else:
         raise HTTPException(400, "Must specify system_id or system_config.")
 
-    node_cfg.override_system_id = config.system_id
-    node_cfg.override_config = config
-    save_node_config(node_cfg)
-
-    from app.main import _generate_op25_config
-    if not await _generate_op25_config(config):
-        raise HTTPException(500, f"Failed to generate OP25 config for override: {config.name}")
-
-    await op25_client.stop()
-    await asyncio.sleep(2)
-    await op25_client.start()
-
-    await mqtt_manager._publish_checkin()
+    _validate_type(config.type)
+    await _apply_override(config)
     return {"ok": True}
 
 
@@ -172,15 +330,14 @@ async def revert_config():
     save_node_config(node_cfg)
 
     if node_cfg.system_config:
-        from app.main import _generate_op25_config
-        if not await _generate_op25_config(node_cfg.system_config):
-            raise HTTPException(500, "Failed to regenerate original OP25 config.")
-
+        await _restart_op25_with(node_cfg.system_config)
+    else:
+        # Nothing from C2 to go back to: stop rather than keep decoding a
+        # system the dashboard now says is cleared.
         await op25_client.stop()
-        await asyncio.sleep(2)
-        await op25_client.start()
 
-    await mqtt_manager._publish_checkin()
+    if mqtt_manager.is_connected:
+        await mqtt_manager.publish_checkin()
     return {"ok": True}
 
 
@@ -243,15 +400,25 @@ async def set_sdr(body: SdrSettingsBody):
     return {"ok": True, "running": running}
 
 
+@router.get("/discord")
+async def discord_status():
+    """What the node's radio bots are doing (desired state from C2, actual here)."""
+    return radio_bot.snapshot()
+
+
 @router.post("/discord/join")
-async def discord_join(guild_id: int, channel_id: int):
-    ok = await radio_bot.join(guild_id, channel_id)
-    if not ok:
-        raise HTTPException(500, "Failed to join voice channel")
-    return {"ok": True}
+async def discord_join():
+    # node-26#25 item 4: this used to call join() without a token and 500 every
+    # time. A node cannot lease a radio bot by itself — C2 owns the pool.
+    raise HTTPException(
+        409,
+        "Discord radio channels are started from Discord (/join) or the DRB web app; "
+        "the node cannot pick a radio bot on its own.",
+    )
 
 
 @router.post("/discord/leave")
 async def discord_leave():
-    await radio_bot.leave()
+    """Local stop: log every radio bot out until DRB sends a new channel list."""
+    await radio_bot.pause_locally()
     return {"ok": True}

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

@@ -124,7 +124,7 @@
     }
     .softkey:last-child { border-right: none; }
     
-    /* Overlay UI (Volume, Hold) */
+    /* Overlay UI (Volume, status) */
     .overlay {
       position: absolute;
       top: 50%; left: 50%;
@@ -153,7 +153,6 @@
     <ul>
       <li><span>Up Arrow</span> Volume Up</li>
       <li><span>Down Arrow</span> Volume Down</li>
-      <li><span>Enter / H</span> Toggle Hold (UI mock)</li>
       <li><span>P</span> Play/Pause Stream</li>
       <li><span>S</span> Select System</li>
       <li><span>R</span> Revert Override</li>
@@ -171,15 +170,14 @@
       <h3 style="color: var(--dept-color); margin-bottom: 6px; font-size: 13px;">Or Enter Freq Manually:</h3>
       <div style="display: grid; grid-template-columns: 1fr 1fr; gap: 5px; font-size: 11px; margin-bottom: 8px;">
         <div>
-          <label>Control Freq (MHz/Hz):</label>
+          <label>Freq (MHz) — P25 control channel or analog:</label>
           <input type="text" id="manual-freq" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;">
         </div>
         <div>
           <label>System Type:</label>
           <select id="manual-type" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;">
             <option value="P25">P25</option>
-            <option value="DMR">DMR</option>
-            <option value="NBFM">NBFM</option>
+            <option value="NBFM">Analog FM</option>
           </select>
         </div>
       </div>
@@ -208,11 +206,11 @@
 
   <div class="softkeys">
     <div class="softkey">VOL (+/-)</div>
-    <div class="softkey">HOLD (H)</div>
+    <div class="softkey">SYSTEM (S)</div>
     <div class="softkey">PLAY (P)</div>
   </div>
 
-  <div class="overlay" id="overlay">HOLD</div>
+  <div class="overlay" id="overlay"></div>
 
   <!-- Audio element for local playback -->
   <audio id="audio-player" autoplay></audio>
@@ -231,7 +229,6 @@
     const elOverlay = document.getElementById('overlay');
 
     let currentIcecastUrl = null;
-    let isHolding = false;
     let volume = 1.0;
 
     // Clock updater
@@ -309,7 +306,7 @@
       const container = document.getElementById('system-list');
       container.innerHTML = '';
       if (systems.length === 0) {
-        container.innerHTML = '<div style="padding: 10px; color: #888;">No cached systems found.</div>';
+        container.innerHTML = '<div style="padding: 10px; color: #888;">No server systems saved yet (they arrive once the node is approved). Enter a frequency below.</div>';
         return;
       }
       systems.forEach((sys, index) => {
@@ -373,25 +370,20 @@
       }
       closeSystemModal();
       showOverlay("APPLYING MANUAL", 3000);
-      const manualConfig = {
-        system_id: "manual-override",
-        name: `Freq ${freqVal} (${typeVal})`,
-        type: typeVal,
-        config: {
-          control_channels: [freqVal],
-          talkgroups: []
-        }
-      };
+      // Same validated path as the dashboard's "Listen locally" (MHz range check).
+      const mode = typeVal === 'NBFM' ? 'analog' : 'p25';
+      const freqs = freqVal.split(/[\s,;]+/).filter(Boolean).map(Number);
       try {
-        const res = await fetch('/api/config/override', {
+        const res = await fetch('/api/listen', {
           method: 'POST',
           headers: { 'Content-Type': 'application/json' },
-          body: JSON.stringify({ system_config: manualConfig })
+          body: JSON.stringify({ mode, frequencies: freqs })
         });
         if (res.ok) {
-          showOverlay('MANUAL OVERRIDE OK', 1500);
+          showOverlay('MANUAL OK', 1500);
         } else {
-          showOverlay('MANUAL OVERRIDE FAIL', 2000);
+          const d = await res.json().catch(() => ({}));
+          showOverlay((d.detail || 'MANUAL FAILED').slice(0, 60), 4000);
         }
       } catch (err) {
         showOverlay('ERROR APPLYING', 2000);
@@ -444,9 +436,6 @@
         audioPlayer.volume = volume;
         elVol.textContent = `VOL: ${Math.round(volume * 100)}%`;
         showOverlay(`VOL: ${Math.round(volume * 100)}%`);
-      } else if (e.key === 'Enter' || e.key === 'h' || e.key === 'H') {
-        isHolding = !isHolding;
-        showOverlay(isHolding ? 'HOLD ON' : 'HOLD OFF');
       } else if (e.key === 'p' || e.key === 'P') {
           if (audioPlayer.paused) {
               audioPlayer.play();

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

@@ -0,0 +1,402 @@
+"""
+Conventional channels: P25 conventional (node-26#24) and analog (node-26#20).
+
+Covers the generate-config payload (radio_config), the console normalizer
+(conventional.py) and how the watcher attributes conventional audio. Channel
+rows are shaped like tk_drb_conv's / tk_p25's conventional `channel_update`
+entries; grants like tk_p25.log_call entries.
+"""
+import pytest
+from unittest.mock import AsyncMock
+
+from app.config import settings
+from app.internal import radio_config as rc
+from app.internal.conventional import ConventionalChannels, channel_tgid, NON_DISTINCT_TGIDS
+from app.internal.metadata_watcher import MetadataWatcher
+from tests.test_metadata_watcher import FakeAudio, FakeClock, tick, update
+
+PD1, PD3 = 460_400_000, 460_425_000
+
+
+def conv_row(msgqid=0, freq=PD1, name="PD1 Dispatch", tgid=None, srcaddr=0, voice=False,
+             encrypted=0, **extra):
+    row = {
+        "freq": freq, "name": name, "tgid": tgid, "tag": name, "srcaddr": srcaddr,
+        "encrypted": encrypted, "conventional": True, "msgqid": msgqid, "voice": voice,
+    }
+    row.update(extra)
+    return row
+
+
+def conv_grant(rcvr, tgid, at, freq=PD1, rid=4242):
+    return {"time": at, "sysid": 0, "rcvr": rcvr, "rcvrtag": "", "freq": freq, "slot": 0,
+            "prio": 0, "tgid": tgid, "tgtag": "", "rid": rid, "rtag": ""}
+
+
+# ---------------------------------------------------------------------------
+# radio_config: system document -> op25 generate-config payload
+# ---------------------------------------------------------------------------
+
+ICE = {"icecast_host": "localhost", "icecast_port": 8000, "icecast_mountpoint": "/n", "icecast_password": "p"}
+
+WHITE_PLAINS = {
+    "mode": "conventional",
+    "channels": [
+        {"freq": 460.4, "name": "PD1 Dispatch"},
+        {"freq": 460.425, "name": "PD3 Tac", "nac": "293"},
+    ],
+    "talkgroups": [],
+}
+
+
+def test_conventional_payload():
+    p = rc.build_generate_payload("P25", "White Plains", WHITE_PLAINS, ICE, "rtl-sdr-v3", None)
+    assert p["type"] == "P25" and p["p25_mode"] == "conventional"
+    assert p["channels"] == []    # nothing that could become a control channel
+    assert p["conv_channels"] == [
+        {"frequency": PD1, "name": "PD1 Dispatch", "nac": None},
+        {"frequency": PD3, "name": "PD3 Tac", "nac": 0x293},
+    ]
+    assert "ppm_override" not in p
+
+
+def test_trunked_payload_unchanged_when_mode_missing():
+    cfg = {"control_channels": [851.0125, 851.5125], "talkgroups": [{"id": 9048, "name": "Ossining PD"}]}
+    p = rc.build_generate_payload("P25", "County", cfg, ICE, "other", 1.5)
+    assert "p25_mode" not in p
+    assert p["channels"] == [851_012_500, 851_512_500]
+    assert p["tags"] == [{"talkgroup": "Ossining PD", "tagDec": 9048}]
+    assert p["whitelist"] == [9048]
+    assert p["ppm_override"] == 1.5
+
+
+@pytest.mark.parametrize("cfg", [
+    {"channels": [{"freq": 154.43, "name": "Fire"}]},
+    {"control_channels": [154.43]},                     # saved before #24
+])
+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
+
+
+def test_system_mode():
+    assert rc.system_mode("P25", {}) == rc.TRUNKED
+    assert rc.system_mode("P25", {"mode": "Conventional"}) == rc.CONVENTIONAL
+    assert rc.system_mode("NBFM", {"mode": "conventional"}) == rc.ANALOG
+    assert rc.system_mode("DMR", {}) == rc.DMR
+
+
+@pytest.mark.parametrize("raw, want", [("293", 0x293), ("0x293", 0x293), (0x293, 0x293), ("", None), (None, None)])
+def test_parse_nac(raw, want):
+    assert rc.parse_nac(raw) == want
+
+
+@pytest.mark.parametrize("raw", ["xyz", "1000", 0, True])
+def test_parse_nac_rejects(raw):
+    with pytest.raises(ValueError):
+        rc.parse_nac(raw)
+
+
+def test_conventional_channel_without_freq_rejected():
+    with pytest.raises(ValueError):
+        rc.conventional_channels({"channels": [{"name": "no freq"}]})
+
+
+# ---------------------------------------------------------------------------
+# Normalizer
+# ---------------------------------------------------------------------------
+
+def test_synthetic_tgid_is_frequency_in_100hz_and_clear_of_p25_space():
+    assert channel_tgid(PD1) == 4_604_000
+    assert channel_tgid(460.4) == 4_604_000
+    assert channel_tgid(460_406_250) != channel_tgid(460_412_500)   # 6.25 kHz steps stay distinct
+    assert channel_tgid(24_000_000) > 0xFFFF                        # lowest RTL-SDR frequency
+
+
+def test_non_distinct_link_control_tgid_becomes_the_channel():
+    conv = ConventionalChannels()
+    up = update(channels=[conv_row(0, PD1, tgid=1, srcaddr=7, voice=True),
+                          conv_row(1, PD3, "PD3 Tac", tgid=1, srcaddr=8, voice=True)])
+    conv.normalize(up)
+    assert [r["tgid"] for r in up.channels] == [4_604_000, 4_604_250]
+    assert [r["lc_tgid"] for r in up.channels] == [1, 1]
+    assert [r["tag"] for r in up.channels] == ["PD1 Dispatch", "PD3 Tac"]
+    assert all(r["drb_active"] for r in up.channels)
+    assert 1 in NON_DISTINCT_TGIDS and 0xFFFF in NON_DISTINCT_TGIDS
+
+
+def test_distinct_link_control_tgid_is_kept():
+    conv = ConventionalChannels()
+    up = update(channels=[conv_row(0, tgid=101, srcaddr=7)],
+                call_log=[conv_grant(0, 101, 1.0)])
+    conv.normalize(up)
+    assert up.channels[0]["tgid"] == 101
+    assert up.call_log[0]["tgid"] == 101
+    assert up.call_log[0]["tgtag"] == "PD1 Dispatch"
+
+
+def test_grants_are_mapped_per_receiver():
+    conv = ConventionalChannels()
+    up = update(channels=[conv_row(0, PD1), conv_row(1, PD3, "PD3 Tac")],
+                call_log=[conv_grant(1, 1, 1.0, freq=PD3)])
+    conv.normalize(up)
+    assert up.call_log[0]["tgid"] == 4_604_250
+    assert up.call_log[0]["tgtag"] == "PD3 Tac"
+
+
+def test_voice_without_link_control_is_still_a_talker():
+    conv = ConventionalChannels()
+    up = update(channels=[conv_row(0, voice=True)])
+    conv.normalize(up)
+    assert up.channels[0]["tgid"] == 4_604_000 and up.channels[0]["drb_active"]
+
+
+def test_idle_and_encrypted_channels_are_not_talkers():
+    conv = ConventionalChannels()
+    up = update(channels=[conv_row(0), conv_row(1, PD3, "PD3 Tac", tgid=1, voice=True, encrypted=1)])
+    conv.normalize(up)
+    assert [r["tgid"] for r in up.channels] == [None, None]
+
+
+def test_trunked_rows_untouched():
+    conv = ConventionalChannels()
+    row = {"freq": 851_000_000, "tgid": 9048, "srcaddr": 5, "tag": "PD", "msgqid": 0}
+    up = update(channels=[dict(row)], call_log=[conv_grant(0, 9048, 1.0)])
+    conv.normalize(up)
+    assert up.channels[0] == row
+    assert up.call_log[0]["tgid"] == 9048
+    assert conv.sole_channel() is None
+
+
+def test_sole_channel_only_with_exactly_one_conventional_channel():
+    conv = ConventionalChannels()
+    conv.normalize(update(channels=[conv_row(0, 154_430_000, "Fire")]))
+    assert conv.sole_channel() == {"tgid": 1_544_300, "name": "Fire", "freq": 154_430_000}
+    conv.normalize(update(channels=[conv_row(0), conv_row(1, PD3)]))
+    assert conv.sole_channel() is None
+    # A poll without channel_update does not forget the channel list.
+    conv.normalize(update())
+    assert conv.sole_channel() is None
+
+
+def test_encryption_report():
+    conv = ConventionalChannels()
+    conv.normalize(update(channels=[
+        conv_row(0, enc_calls=3, clear_calls=0, last_enc_time=50.0, last_enc_algid=0x84,
+                 last_enc_keyid=0x1234, nac=0x293),
+        conv_row(1, PD3, "PD3 Tac", enc_calls=1, clear_calls=5, last_enc_time=10.0, last_clear_time=20.0),
+    ]))
+    pd1, pd3 = conv.encryption_report()
+    assert pd1["encrypted"] is True and pd1["algid"] == 0x84 and pd1["keyid"] == 0x1234 and pd1["nac"] == 0x293
+    assert pd3["encrypted"] is False   # its most recent call was clear
+
+
+# ---------------------------------------------------------------------------
+# Watcher attribution
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def clock():
+    return FakeClock()
+
+
+@pytest.fixture
+def audio(clock):
+    return FakeAudio(clock)
+
+
+@pytest.fixture
+def hearing(clock, audio):
+    w = MetadataWatcher()
+    w._clock = clock
+    w.on_call_start = AsyncMock()
+    w.on_call_end = AsyncMock()
+    w.audio_activity = audio
+    return w
+
+
+async def _finish(hearing, clock, channels):
+    clock.advance(settings.call_silence_timeout + 0.5)
+    await tick(hearing, update(channels=channels))
+    return hearing.on_call_end.call_args[0][0]
+
+
+@pytest.mark.asyncio
+async def test_conventional_call_attributed_to_its_channel(hearing, clock, audio):
+    """White Plains: both channels carry the default TGID 1; the keyed one wins."""
+    audio.speak()
+    await tick(hearing, update(
+        channels=[conv_row(0, tgid=1, srcaddr=4242, voice=True), conv_row(1, PD3, "PD3 Tac")],
+        call_log=[conv_grant(0, 1, clock.now)],
+    ))
+    assert hearing.is_active and hearing.current_tgid == 4_604_000
+
+    ended = await _finish(hearing, clock, [conv_row(0), conv_row(1, PD3, "PD3 Tac")])
+    assert ended["attributed"] is True
+    assert (ended["tgid"], ended["tgid_name"], ended["srcaddr"]) == (4_604_000, "PD1 Dispatch", 4242)
+    assert hearing.unattributed_segments == 0
+
+
+@pytest.mark.asyncio
+async def test_other_channel_keying_splits(hearing, clock, audio):
+    audio.speak()
+    await tick(hearing, update(
+        channels=[conv_row(0, tgid=1, voice=True), conv_row(1, PD3, "PD3 Tac")],
+        call_log=[conv_grant(0, 1, clock.now)],
+    ))
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(
+        channels=[conv_row(0), conv_row(1, PD3, "PD3 Tac", tgid=1, voice=True)],
+        call_log=[conv_grant(1, 1, clock.now, freq=PD3)],
+    ))
+    first = hearing.on_call_end.call_args[0][0]
+    assert first["tgid"] == 4_604_000 and first["end_reason"] == "tgid_change"
+    assert hearing.current_tgid == 4_604_250
+
+
+@pytest.mark.asyncio
+async def test_single_analog_channel_audio_is_kept_not_orphaned(hearing, clock, audio):
+    """node-26#20: analog never reports a talker; the only channel is the answer."""
+    audio.speak()
+    rows = [conv_row(0, 154_430_000, "County Fire")]
+    await tick(hearing, update(channels=rows))
+    # Not labelled at open — only at close.
+    assert hearing.on_call_start.call_args[0][0]["tgid"] is None
+
+    ended = await _finish(hearing, clock, rows)
+    assert ended["attributed"] is True
+    assert (ended["tgid"], ended["tgid_name"]) == (1_544_300, "County Fire")
+    assert hearing.unattributed_segments == 0
+
+
+@pytest.mark.asyncio
+async def test_two_channels_with_no_evidence_is_still_orphan(hearing, clock, audio):
+    audio.speak()
+    rows = [conv_row(0), conv_row(1, PD3, "PD3 Tac")]
+    await tick(hearing, update(channels=rows))
+    ended = await _finish(hearing, clock, rows)
+    assert ended["attributed"] is False
+    assert hearing.unattributed_segments == 1
+
+
+@pytest.mark.asyncio
+async def test_late_real_talkgroup_on_single_channel_does_not_split(hearing, clock, audio):
+    """Audio first, link-control grant a poll later: adopt it, don't split."""
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0)]))
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=9, voice=True)],
+                               call_log=[conv_grant(0, 101, clock.now - 0.3)]))
+    assert hearing.current_tgid == 101
+    hearing.on_call_end.assert_not_called()
+    ended = await _finish(hearing, clock, [conv_row(0)])
+    assert ended["tgid"] == 101 and ended["end_reason"] == "audio_silence"
+
+
+@pytest.mark.asyncio
+async def test_encrypted_channel_does_not_label_clear_audio(hearing, clock, audio):
+    """PD3 is keyed encrypted (muted) while PD1 talks in the clear."""
+    audio.speak()
+    await tick(hearing, update(channels=[
+        conv_row(0, tgid=1, voice=True),
+        conv_row(1, PD3, "PD3 Tac", tgid=1, voice=True, encrypted=1, algid=0x84),
+    ]))
+    ended = await _finish(hearing, clock, [conv_row(0), conv_row(1, PD3, "PD3 Tac")])
+    assert ended["tgid"] == 4_604_000
+
+
+@pytest.mark.asyncio
+async def test_conventional_channels_property(hearing, clock):
+    await tick(hearing, update(channels=[conv_row(0, encrypted=1, voice=True, algid=0x84, keyid=7)]))
+    (only,) = hearing.conventional_channels
+    assert only["encrypted"] and only["encrypted_now"] and only["algid"] == 0x84
+
+
+# drb-audio-review findings on #24: a header frame (voice, no link control yet)
+# must not label the call with the synthetic id and then split against the
+# real talkgroup a poll later; nor may a reply's header split it back.
+
+@pytest.mark.asyncio
+async def test_header_before_link_control_does_not_split(hearing, clock, audio):
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, voice=True)]))       # HDU only
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=9, voice=True)],
+                               call_log=[conv_grant(0, 101, clock.now - 0.2)]))
+    hearing.on_call_end.assert_not_called()
+    assert hearing.current_tgid == 101
+    ended = await _finish(hearing, clock, [conv_row(0)])
+    assert ended["tgid"] == 101 and ended["end_reason"] == "audio_silence"
+    assert hearing.on_call_end.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_reply_header_inside_conversation_does_not_ping_pong(hearing, clock, audio):
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=9, voice=True)],
+                               call_log=[conv_grant(0, 101, clock.now)]))
+    clock.advance(0.5)
+    await tick(hearing, update(channels=[conv_row(0)]))                   # TDU
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, voice=True)]))       # reply's HDU
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=12, voice=True)],
+                               call_log=[conv_grant(0, 101, clock.now)]))
+    hearing.on_call_end.assert_not_called()
+    ended = await _finish(hearing, clock, [conv_row(0)])
+    assert ended["tgid"] == 101 and hearing.on_call_end.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_never_decoded_link_control_then_real_tg_relabels(hearing, clock, audio):
+    """First-ever call on a channel: synthetic label, then the real TG adopts it."""
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, voice=True)]))
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, voice=True)]))
+    assert hearing.current_tgid == 4_604_000
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, tgid=101, voice=True)],
+                               call_log=[conv_grant(0, 101, clock.now)]))
+    hearing.on_call_end.assert_not_called()
+    assert hearing.current_tgid == 101
+
+
+def test_voice_without_link_control_uses_channels_last_talkgroup():
+    conv = ConventionalChannels()
+    conv.normalize(update(channels=[conv_row(0, tgid=101, voice=True)]))
+    up = update(channels=[conv_row(0, voice=True)])
+    conv.normalize(up)
+    assert up.channels[0]["tgid"] == 101
+
+
+def test_same_channel_relabel_is_conventional_only():
+    conv = ConventionalChannels()
+    assert conv.same_channel_relabel(4_604_000, PD1, 101, PD1) is None    # no conventional rows known
+    conv.normalize(update(channels=[conv_row(0), conv_row(1, PD3)]))
+    assert conv.same_channel_relabel(4_604_000, PD1, 101, PD1) == "adopt"
+    assert conv.same_channel_relabel(101, PD1, 4_604_000, PD1) == "keep"
+    assert conv.same_channel_relabel(4_604_000, PD1, 4_604_250, PD3) is None   # other channel: split
+
+
+def test_remembered_talkgroup_expires():
+    conv = ConventionalChannels()
+    t = [1000.0]
+    conv._clock = lambda: t[0]
+    conv.normalize(update(channels=[conv_row(0, tgid=101, voice=True)]))
+    t[0] += 10
+    up = update(channels=[conv_row(0, voice=True)])
+    conv.normalize(up)
+    assert up.channels[0]["tgid"] == 101
+    t[0] += 3600
+    up = update(channels=[conv_row(0, voice=True)])
+    conv.normalize(up)
+    assert up.channels[0]["tgid"] == 4_604_000

drb-edge-node/tests/test_discord_radio.py

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

drb-edge-node/tests/test_host_updater.py

@@ -24,21 +24,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

@@ -0,0 +1,305 @@
+"""
+Local dashboard: the health checklist (/api/health) and choosing what to
+listen to without C2 (/api/listen, /api/config/system, /api/config/revert).
+
+Same standalone-app pattern as test_auth_endpoints.py — the real routers,
+none of app.main's lifespan. OP25/MQTT are mocked at the client boundary.
+"""
+import base64
+import time
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.config import settings
+from app.internal import config_manager, credentials, health
+from app.internal.op25_client import parse_terminal_messages
+from app.routers import api, ui
+
+
+@pytest.fixture(autouse=True)
+def isolated(tmp_path, monkeypatch):
+    monkeypatch.setattr(credentials, "_CREDS_FILE", tmp_path / "credentials.json")
+    monkeypatch.setattr(credentials, "_api_key", None)
+    monkeypatch.setattr(credentials, "_auth_salt", None)
+    monkeypatch.setattr(credentials, "_session_secret", None)
+    monkeypatch.setattr(config_manager, "_CONFIG_FILE", tmp_path / "node_config.json")
+    monkeypatch.setattr(settings, "dashboard_username", "tester")
+    monkeypatch.setattr(settings, "dashboard_password", "s3cret-pass")
+    yield
+
+
+@pytest.fixture
+def client():
+    app = FastAPI()
+    app.include_router(api.router)
+    app.include_router(ui.router)
+    auth = base64.b64encode(b"tester:s3cret-pass").decode()
+    with TestClient(app, headers={"Authorization": f"Basic {auth}"}) as c:
+        yield c
+
+
+@pytest.fixture
+def op25():
+    """Mock the OP25 side of an apply: generate, stop, start, checkin."""
+    with patch("app.main._generate_op25_config", AsyncMock(return_value=True)) as gen, \
+         patch("app.routers.api.op25_client.stop", AsyncMock(return_value=True)) as stop, \
+         patch("app.routers.api.op25_client.start_with_reason", AsyncMock(return_value=None)) as start, \
+         patch("app.routers.api.asyncio.sleep", AsyncMock()), \
+         patch("app.routers.api.mqtt_manager.publish_checkin", AsyncMock()) as checkin:
+        yield {"gen": gen, "stop": stop, "start": start, "checkin": checkin}
+
+
+# ---------------------------------------------------------------------------
+# Health checklist — pure function of facts
+# ---------------------------------------------------------------------------
+
+def _facts(**over):
+    now = 1_000_000.0
+    f = {
+        "now": now, "node_id": "toughny", "default_password": False,
+        "c2_url": "https://api.example", "has_api_key": True, "mqtt_connected": True,
+        "mqtt_broker": "mqtt.example", "mqtt_port": 8883, "sdr_count": 1,
+        "source": {"name": "County", "type": "P25", "system_id": "s1", "local": False},
+        "op25_status": "running",
+        "console": {"last_poll_epoch": now - 1, "channels": [], "trunk": [{"last_tsbk": now - 1, "rxchan": 851012500}],
+                    "last_heard": None},
+        "audio_capturing": True, "last_voice_epoch": now - 30, "icecast_live": True, "icecast_port": 8000,
+    }
+    f.update(over)
+    return f
+
+
+def _by_id(checks):
+    return {c["id"]: c for c in checks}
+
+
+def test_healthy_node_is_all_ok():
+    checks = health.build_checks(_facts())
+    assert all(c["state"] == "ok" for c in checks), [c for c in checks if c["state"] != "ok"]
+    assert "851.0125 MHz" in _by_id(checks)["decode"]["summary"]
+
+
+def test_unapproved_node_explains_wait_and_mqtt_refusal():
+    """Toughny's state: C2 set, no API key, broker refuses, nothing chosen."""
+    c = _by_id(health.build_checks(_facts(has_api_key=False, mqtt_connected=False, source=None,
+                                          op25_status="stopped")))
+    assert c["c2"]["state"] == "warn" and "approve" in c["c2"]["action"]
+    assert "toughny" in c["c2"]["action"]
+    assert c["mqtt"]["state"] == "warn"
+    assert c["source"]["state"] == "warn" and "Listen locally" in c["source"]["action"]
+    assert c["op25"]["state"] == "off"
+    assert c["decode"]["state"] == "off"
+
+
+def test_key_on_disk_but_mqtt_down_says_restart():
+    c = _by_id(health.build_checks(_facts(mqtt_connected=False)))
+    assert c["mqtt"]["state"] == "error"
+    assert "restart the edge-node container" in c["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"
+    assert c["mqtt"]["state"] == "off"
+
+
+def test_no_sdr_and_crashed_op25():
+    c = _by_id(health.build_checks(_facts(sdr_count=0, op25_status="crashed")))
+    assert c["sdr"]["state"] == "error"
+    assert c["op25"]["state"] == "error" and "log" in c["op25"]["action"]
+
+
+def test_running_but_no_control_channel():
+    now = 1_000_000.0
+    c = _by_id(health.build_checks(_facts(console={"last_poll_epoch": now, "channels": [],
+                                                   "trunk": [{"last_tsbk": 0}], "last_heard": None})))
+    assert c["decode"]["state"] == "error"
+
+
+def test_analog_needs_no_control_channel():
+    now = 1_000_000.0
+    c = _by_id(health.build_checks(_facts(
+        source={"name": "Fire", "type": "NBFM", "system_id": "local-analog", "local": True},
+        console={"last_poll_epoch": now, "channels": [{"freq": 154430000}], "trunk": [], "last_heard": None},
+    )))
+    assert c["decode"]["state"] == "ok" and "154.4300 MHz" in c["decode"]["summary"]
+    # node-26#20: analog calls are recorded now (single-channel attribution).
+    assert c["decode"]["action"] == ""
+    assert "this dashboard" in c["source"]["summary"]
+
+
+_CONV_SRC = {"name": "White Plains", "type": "P25", "mode": "conventional", "system_id": "wp", "local": False}
+_PD1 = {"name": "PD1 Dispatch", "freq": 460400000, "nac": None, "encrypted": False, "algid": None, "keyid": None}
+_PD3 = {"name": "PD3 Tac", "freq": 460425000, "nac": None, "encrypted": False, "algid": None, "keyid": None}
+
+
+def _conv_facts(chans):
+    now = 1_000_000.0
+    return _facts(source=_CONV_SRC, conventional=chans,
+                  console={"last_poll_epoch": now, "channels": [], "trunk": [], "last_heard": None})
+
+
+def test_conventional_quiet_channels_are_ok_not_a_control_channel_error():
+    c = _by_id(health.build_checks(_conv_facts([_PD1, _PD3])))
+    assert c["decode"]["state"] == "ok"
+    assert "PD1 Dispatch (460.4000 MHz)" in c["decode"]["summary"]
+    assert "control channel" not in c["decode"]["summary"]
+    assert "P25 conventional" in c["source"]["summary"]
+
+
+def test_conventional_decoding_shows_nac():
+    c = _by_id(health.build_checks(_conv_facts([dict(_PD1, nac=0x293), _PD3])))
+    assert c["decode"]["state"] == "ok" and "NAC 0x293" in c["decode"]["summary"]
+
+
+def test_conventional_all_encrypted_is_an_error_naming_the_algorithm():
+    enc = [dict(_PD1, encrypted=True, algid=0x84, keyid=0x1234), dict(_PD3, encrypted=True, algid=0x84)]
+    c = _by_id(health.build_checks(_conv_facts(enc)))
+    assert c["decode"]["state"] == "error"
+    assert "AES-256" in c["decode"]["summary"] and "0x1234" in c["decode"]["summary"]
+    assert "different system" in c["decode"]["action"]
+
+
+def test_conventional_partly_encrypted_warns():
+    c = _by_id(health.build_checks(_conv_facts([dict(_PD1, encrypted=True, algid=0x81), _PD3])))
+    assert c["decode"]["state"] == "warn" and "PD1 Dispatch" in c["decode"]["summary"]
+
+
+def test_op25_unreachable_and_icecast_down():
+    c = _by_id(health.build_checks(_facts(op25_status=None, icecast_live=None)))
+    assert c["op25"]["state"] == "error"
+    assert c["stream"]["state"] == "error"
+
+
+def test_default_password_warns():
+    assert _by_id(health.build_checks(_facts(default_password=True)))["password"]["state"] == "warn"
+
+
+def test_trunk_update_is_parsed():
+    update = parse_terminal_messages([
+        {"json_type": "trunk_update", "0": {"last_tsbk": 5.0, "rxchan": 851012500}, "nac": 0},
+        {"json_type": "channel_update", "channels": ["0"], "0": {"freq": 1, "tgid": None}},
+    ])
+    assert update.trunk == [{"last_tsbk": 5.0, "rxchan": 851012500}]
+    assert len(update.channels) == 1
+
+
+def test_health_route_returns_checks(client):
+    facts = _facts(now=time.time())
+    with patch("app.internal.health.gather_facts", AsyncMock(return_value=facts)):
+        r = client.get("/api/health")
+    assert r.status_code == 200
+    assert [c["id"] for c in r.json()["checks"]][:3] == ["password", "c2", "mqtt"]
+
+
+# ---------------------------------------------------------------------------
+# Choosing what to listen to
+# ---------------------------------------------------------------------------
+
+def test_listen_p25_saves_override_and_restarts_op25(client, op25):
+    r = client.post("/api/listen", json={
+        "mode": "p25", "frequencies": [851.0125, 851.5125], "talkgroups": [{"id": 1001, "name": "Fire"}],
+    })
+    assert r.status_code == 200, r.text
+    cfg = config_manager.load_node_config()
+    assert cfg.override_config.type == "P25"
+    assert cfg.override_config.config["control_channels"] == [851.0125, 851.5125]
+    assert cfg.override_config.config["talkgroups"] == [{"id": 1001, "name": "Fire"}]
+    op25["gen"].assert_awaited_once()
+    op25["start"].assert_awaited_once()
+    op25["checkin"].assert_not_awaited()  # MQTT not connected in tests
+
+
+def test_listen_analog_single_frequency(client, op25):
+    r = client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43], "name": "Fire"})
+    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])
+
+
+def test_listen_p25_conventional(client, op25):
+    r = client.post("/api/listen", json={
+        "mode": "p25conv", "frequencies": [460.4, 460.425], "name": "White Plains", "nac": "0x293",
+    })
+    assert r.status_code == 200, r.text
+    cfg = config_manager.load_node_config().override_config
+    assert cfg.type == "P25" and cfg.config["mode"] == "conventional"
+    assert "control_channels" not in cfg.config
+    assert [c["freq"] for c in cfg.config["channels"]] == [460.4, 460.425]
+    assert {c["nac"] for c in cfg.config["channels"]} == {"293"}
+    op25["gen"].assert_awaited_once()
+    status = client.get("/api/status").json()
+    assert status["system_kind"] == "P25 conventional"
+
+
+@pytest.mark.parametrize("body,needle", [
+    ({"mode": "p25conv", "frequencies": [460.4, 460.425, 460.45]}, "at most 2"),
+    ({"mode": "p25conv", "frequencies": [460.4, 462.6]}, "apart"),
+    ({"mode": "p25conv", "frequencies": [460.4, 460.4]}, "twice"),
+    ({"mode": "p25conv", "frequencies": [460.4], "nac": "zz"}, "hex"),
+    ({"mode": "p25", "frequencies": []}, "at least one"),
+    ({"mode": "p25", "frequencies": [851012500]}, "MHz"),
+    ({"mode": "analog", "frequencies": [154.43, 155.1]}, "exactly one"),
+    ({"mode": "system", "system_id": "nope"}, "saved list"),
+])
+def test_listen_rejects_bad_input_with_a_readable_reason(client, op25, body, needle):
+    r = client.post("/api/listen", json=body)
+    assert r.status_code in (400, 404)
+    assert needle in r.json()["detail"]
+    op25["gen"].assert_not_awaited()
+
+
+def test_listen_reports_op25_start_failure(client, op25):
+    op25["start"].return_value = "No OP25 config yet — choose a system or frequency first."
+    r = client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
+    assert r.status_code == 502
+    assert "did not start" in r.json()["detail"]
+
+
+def test_dmr_is_refused_up_front(client, op25):
+    r = client.post("/api/config/override", json={"system_config": {
+        "system_id": "x", "name": "x", "type": "DMR", "config": {"control_channels": [450.1]},
+    }})
+    assert r.status_code == 400 and "DMR" in r.json()["detail"]
+    op25["gen"].assert_not_awaited()
+
+
+def test_config_system_goes_through_op25_generation(client, op25, tmp_path, monkeypatch):
+    """It used to dump the raw blob into active.cfg.json — not an OP25 config."""
+    monkeypatch.setattr(settings, "config_path", str(tmp_path))
+    r = client.post("/api/config/system", json={
+        "system_id": "s1", "name": "County", "type": "P25", "config": {"control_channels": [851.0125]},
+    })
+    assert r.status_code == 200, r.text
+    op25["gen"].assert_awaited_once()
+    assert not (tmp_path / "active.cfg.json").exists()
+    cfg = config_manager.load_node_config()
+    assert cfg.configured and cfg.system_config.name == "County"
+
+
+def test_revert_without_c2_system_stops_op25(client, op25):
+    client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
+    op25["stop"].reset_mock()
+    r = client.post("/api/config/revert")
+    assert r.status_code == 200
+    assert config_manager.load_node_config().override_config is None
+    op25["stop"].assert_awaited_once()
+
+
+def test_status_exposes_stream_parts_and_source(client, op25):
+    client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
+    with patch("app.routers.api.op25_client.status", AsyncMock(return_value={"status": "running"})):
+        d = client.get("/api/status").json()
+    assert d["icecast_port"] == settings.icecast_port and d["icecast_mount"] == settings.icecast_mount
+    assert d["local_choice"] is True and d["has_c2_system"] is False
+    assert d["system_type"] == "NBFM"
+
+
+def test_dashboard_page_has_no_dead_api_instruction(client):
+    client.post("/login", data={"username": "tester", "password": "s3cret-pass"}, follow_redirects=False)
+    html = client.get("/").text
+    assert "POST /api/config/system" not in html
+    assert "/api/listen" in html and "/api/health" in html

drb-edge-node/tests/test_mqtt_manager.py

@@ -140,3 +140,16 @@ 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}

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
-#       the running Python is the CLONED REF's code against the pulled image's
-#       dependencies. `v1` == the commit CI built the current :latest/:stable
-#       from, so today they match — but the moment `v1` and the image tags
-#       diverge this silently mixes them. Fix: drop those two mounts from the
-#       prod compose path, or always retag images from the same ref as `v1`.
+#       ./op25-container/app OVER the image's /app, so the running Python is
+#       the CLONED REF's code against the pulled image's dependencies. Those
+#       used to be :latest/:stable, built from main — silently a different
+#       commit from `v1`. Since node-26#19 every image is pinned to
+#       src-<git tree hash of its own directory at the checked-out ref>
+#       (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" ;;
-  *) warn "arch is $ARCH — CI only builds linux/arm64 images. Prebuilt pull will fail; use --build." ;;
+  arm64|aarch64|amd64|x86_64) ok "arch $ARCH" ;;
+  *) 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." ;;
-      404) die "C2 no longer knows node '$NODE_ID' (404) — it was deleted. Re-run with --token DRB-… to enroll again." ;;
+      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 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"
-  [ -n "$ENROLLMENT_TOKEN" ] || die "no enrollment token — mint one at Settings -> Nodes, then re-run with --token"
+  need_token || die "no enrollment token — $NO_TOKEN_HELP"
 
   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. Re-enrolling with --token."
+      warn "saved pickup secret rejected (401) — likely rotated by a re-enroll elsewhere."
+      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
-  --token DRB-… to re-enroll (only works while the node is still pending), or
-  have an admin Reissue key for an approved node and write $CREDS by hand."
+        die "saved pickup secret is stale (401) and no enrollment token was given.
+  Re-enrolling only works while the node is still pending: $NO_TOKEN_HELP.
+  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. Re-enrolling with --token."
+      warn "C2 does not know node '$NODE_ID' (404) — deleted server-side or never fully enrolled."
+      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.
-  Re-run with --token DRB-… to enroll it again."
+        die "C2 does not know node '$NODE_ID' (404) and no enrollment token was given:
+  $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

@@ -31,6 +31,10 @@ RUN ./install.sh -f
 COPY requirements.txt /tmp/requirements.txt
 RUN pip3 install --no-cache-dir -r /tmp/requirements.txt
 
+# DRB's trunking module for P25 conventional channels (node-26#24): reports
+# encryption + NAC and applies the NAC filter, which stock tk_p25 does not.
+COPY op25_addons/tk_drb_conv.py /op25/op25/gr-op25_repeater/apps/tk_drb_conv.py
+
 # Create the run_multi-rx_service.sh script
 COPY run_multi-rx_service.sh /op25/op25/gr-op25_repeater/apps/run_multi-rx_service.sh
 RUN chmod +x /op25/op25/gr-op25_repeater/apps/run_multi-rx_service.sh

op25-container/app/internal/conventional_config.py

@@ -0,0 +1,118 @@
+"""
+P25 conventional OP25 config (node-26#24).
+
+A conventional P25 channel is a fixed voice frequency with no control channel:
+radios key up on it directly, C4FM (fsk4), and identify themselves in the link
+control of each LDU1 (talkgroup + unit id) and the ESS of each LDU2
+(encryption algorithm + key id). Building a `tk_p25.py` trunking config for one
+— what "P25" always meant before — makes OP25 hunt forever for a control
+channel that does not exist, so nothing is ever decoded.
+
+Layout this produces, per boatbod OP25 gr310 `p25_conventional_example.json`:
+
+  * one non-tunable device parked so every channel sits inside its passband;
+  * one channel per frequency, `trunking_sysname: ""` (so the trunking module
+    treats it as conventional), `demod_type: fsk4`;
+  * trunking module `tk_drb_conv.py` with no chans — our thin subclass of
+    tk_p25.rx_ctl that reports encryption and the NAC for conventional
+    channels and applies the per-channel NAC filter (tk_p25 itself does
+    neither for FDMA conventional);
+  * channel N's audio to UDP 23456+N; with two channels audio.py runs in
+    two-channel mode (left = first channel, right = second).
+
+Pure: no file I/O, so it is unit-testable outside the container.
+"""
+from typing import Any, Dict, List, Tuple
+
+from models import ChannelConfig, ConventionalChannel, DeviceConfig, TerminalConfig
+
+# Two is what one audio.py instance can carry (its A and B sockets). A third
+# channel would need a second audio pipeline into PulseAudio.
+MAX_CONVENTIONAL_CHANNELS = 2
+
+AUDIO_UDP_BASE_PORT = 23456
+
+DEVICE_RATE = 1_920_000
+DEVICE_USABLE_BW_PCT = 0.85
+# Half the occupied bandwidth of a 12.5 kHz P25 channel.
+CHANNEL_HALF_BW_HZ = 6_250
+# Keep the RTL-SDR's DC spike off the channels. A single channel (or two very
+# close together) is offset from the centre; channels further apart than
+# 2x this already straddle the spike, so the centre goes between them.
+DC_SPIKE_OFFSET_HZ = 250_000
+DC_SPIKE_CLEARANCE_HZ = 50_000
+
+CONVENTIONAL_TRUNKING_MODULE = "tk_drb_conv.py"
+
+
+def device_center(freqs: List[int]) -> int:
+    lo, hi = min(freqs), max(freqs)
+    mid = (lo + hi) // 2
+    if (hi - lo) // 2 >= DC_SPIKE_CLEARANCE_HZ:
+        return mid
+    return mid + DC_SPIKE_OFFSET_HZ
+
+
+def validate_channels(channels: List[ConventionalChannel]) -> None:
+    """Raise ValueError with an operator-readable reason."""
+    if not channels:
+        raise ValueError("P25 conventional needs at least one channel frequency.")
+    if len(channels) > MAX_CONVENTIONAL_CHANNELS:
+        raise ValueError(
+            f"P25 conventional supports up to {MAX_CONVENTIONAL_CHANNELS} channels on one SDR "
+            f"(got {len(channels)})."
+        )
+    freqs = [c.frequency for c in channels]
+    if len(set(freqs)) != len(freqs):
+        raise ValueError("The same conventional frequency is listed twice.")
+    for c in channels:
+        if c.nac is not None and not (0 < c.nac <= 0xFFF):
+            raise ValueError(f"NAC {c.nac:#x} on {c.frequency / 1e6:.5f} MHz is outside 0x001-0xFFF.")
+    center = device_center(freqs)
+    half = DEVICE_RATE * DEVICE_USABLE_BW_PCT / 2
+    for f in freqs:
+        if abs(f - center) + CHANNEL_HALF_BW_HZ > half:
+            span = (max(freqs) - min(freqs)) / 1e6
+            raise ValueError(
+                f"These channels are {span:.3f} MHz apart — too far for one SDR "
+                f"(max ~{(2 * (half - CHANNEL_HALF_BW_HZ)) / 1e6:.2f} MHz)."
+            )
+
+
+def build_conventional_config(
+    system_name: str,
+    channels: List[ConventionalChannel],
+    gains: str,
+    ppm: float,
+) -> Tuple[Dict[str, Any], bool]:
+    """Return (OP25 config dict, two_channel_audio)."""
+    validate_channels(channels)
+    freqs = [c.frequency for c in channels]
+    center = device_center(freqs)
+
+    chans = []
+    for i, c in enumerate(channels):
+        chans.append(ChannelConfig(
+            name=c.name or f"{system_name} {c.frequency / 1e6:.4f}",
+            trunking_sysname="",
+            enable_analog="off",
+            demod_type="fsk4",
+            filter_type="rc",
+            frequency=c.frequency,
+            meta_stream_name="",
+            destination=f"udp://127.0.0.1:{AUDIO_UDP_BASE_PORT + i}",
+            nac=c.nac,
+            crypt_behavior=2,
+        ))
+
+    device = DeviceConfig(
+        gains=gains, ppm=ppm, tunable=False, frequency=center,
+        rate=DEVICE_RATE, usable_bw_pct=DEVICE_USABLE_BW_PCT,
+    )
+    config = {
+        "channels": [ch.model_dump() for ch in chans],
+        "devices": [device.model_dump()],
+        "trunking": {"module": CONVENTIONAL_TRUNKING_MODULE, "chans": []},
+        "terminal": TerminalConfig().model_dump(),
+    }
+    return config, len(chans) > 1

op25-container/app/internal/liquidsoap_config_utils.py

@@ -2,7 +2,26 @@ import os
 from internal.op25_liq_template import liquidsoap_config_template
 from models import IcecastConfig
 
-def generate_liquid_script(config: IcecastConfig):
+# P25 conventional with two channels (node-26#24): OP25 sends channel A's
+# audio to UDP 23456 and channel B's to 23457, and audio.py -2 puts them on the
+# left and right of the stereo stream. Every output below is mean(input) — mono
+# — so a lone talker on one side would come out 6 dB down; doubling first keeps
+# the common case (one channel keyed) at the same level as single-channel mode.
+# Liquidsoap mixes in float, so the doubling cannot clip before the output.
+_TWO_CHANNEL_AUDIO_ARGS = " -2"
+_TWO_CHANNEL_MIX = "input = amplify(2.0, input)  # two conventional channels, L=A R=B"
+
+
+def render_liquid_script(config: IcecastConfig, two_channel: bool = False) -> str:
+    content = liquidsoap_config_template
+    for key, value in config.model_dump().items():
+        content = content.replace(f"${{{key}}}", str(value))
+    content = content.replace("${audio_py_args}", _TWO_CHANNEL_AUDIO_ARGS if two_channel else "")
+    content = content.replace("${two_channel_mix}", _TWO_CHANNEL_MIX if two_channel else "")
+    return content
+
+
+def generate_liquid_script(config: IcecastConfig, two_channel: bool = False):
     """
     Generates the "*.liq" file that's run by OP25 on startup. 
 
@@ -13,13 +32,7 @@ def generate_liquid_script(config: IcecastConfig):
                           Keys should match the variable names in the template (e.g., 'icecast_host').
     """
     try:
-        content = liquidsoap_config_template
-        # Replace variables
-        for key, value in config.model_dump().items():
-            placeholder = f"${{{key}}}"
-            # Ensure the value is converted to string for replacement
-            content = content.replace(placeholder, str(value))
-            print(f"  - Replaced placeholder {placeholder}")
+        content = render_liquid_script(config, two_channel)
 
         # Write the processed content to the output path
         output_path = "/configs/op25.liq"

op25-container/app/internal/op25_liq_template.py

@@ -17,8 +17,9 @@ ICE_DESCRIPTION = "${icecast_description}"
 ICE_GENRE = "${icecast_genre}"
 # ==========================================================
 
-input = mksafe(input.external(buffer=0.25, channels=2, samplerate=8000, restart_on_error=false, "./audio.py -x 2.5 -s"))
+input = mksafe(input.external(buffer=0.25, channels=2, samplerate=8000, restart_on_error=false, "./audio.py -x 2.5 -s${audio_py_args}"))
 # Consider increasing the buffer value on slow systems such as RPi3. e.g. buffer=0.25
+${two_channel_mix}
 
 # Compression
 input = compress(input, attack = 2.0, gain = 0.0, knee = 13.0, ratio = 2.0, release = 12.3, threshold = -18.0)

op25-container/app/models.py

@@ -49,15 +49,35 @@ class IcecastConfig(BaseModel):
     icecast_genre: Optional[str] = "Public Safety"
 
 
+class P25Mode(str, Enum):
+    TRUNKED = "trunked"
+    CONVENTIONAL = "conventional"
+
+
+class ConventionalChannel(BaseModel):
+    """One fixed P25 conventional voice channel (node-26#24)."""
+    frequency: int                 # Hz
+    name: str = ""
+    nac: Optional[int] = None      # 0x001-0xFFF; None = accept any NAC
+
+
 class ConfigGenerator(BaseModel):
     type: DecodeMode
     systemName: str
-    channels: List[Union[int, str]]
-    tags: Optional[List[TalkgroupTag]]
-    whitelist: Optional[List[int]]
-    icecastConfig: Optional[IcecastConfig]
+    # P25 trunked: control channel list (Hz). NBFM: channels[0] is the one
+    # analog frequency to listen to (Hz). P25 conventional: unused — see
+    # conv_channels.
+    channels: List[Union[int, str]] = []
+    # P25 only. Missing = trunked, so every existing caller is unchanged.
+    p25_mode: P25Mode = P25Mode.TRUNKED
+    conv_channels: List[ConventionalChannel] = []
+    tags: Optional[List[TalkgroupTag]] = None
+    whitelist: Optional[List[int]] = None
+    icecastConfig: Optional[IcecastConfig] = None
     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).
+    nbfm_squelch: Optional[float] = None
 
 class DemodType(str, Enum):
     CQPSK = "cqpsk"
@@ -69,16 +89,24 @@ class FilterType(str, Enum):
 
 class ChannelConfig(BaseModel):
     name: str
-    trunking_sysname: Optional[str]
+    trunking_sysname: Optional[str] = None
     enable_analog: str
     meta_stream_name: str
     demod_type: DemodType
     filter_type: FilterType
     device: Optional[str] = "sdr"
     cqpsk_tracking: Optional[bool] = None
-    frequency: Optional[float] = None
-    nbfmSquelch: Optional[float] = None
+    frequency: Optional[int] = None
+    # Key name op25_nbfm.py actually reads (from_dict(config, 'nbfm_squelch_threshold', -60)).
+    nbfm_squelch_threshold: 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
+    # through, so an extra key is harmless.
+    nac: Optional[int] = None
+    # Set on the channel itself for conventional (multi_rx forwards it to the
+    # frame assembler). 2 = drop encrypted voice instead of decoding noise.
+    crypt_behavior: Optional[int] = None
     tracking_threshold: Optional[int] = 120
     tracking_feedback: Optional[float] = 0.75
     excess_bw: Optional[float] = 0.2
@@ -93,6 +121,9 @@ class DeviceConfig(BaseModel):
     gains: Optional[str] = "lna:39"
     gain_mode: Optional[bool] = False
     name: Optional[str] = "sdr"
+    # Centre frequency (Hz). Unset for trunked P25, where tk_p25 tunes the
+    # device; required for a fixed analog channel (tunable=False).
+    frequency: Optional[int] = None
     offset: Optional[int] = 0
     ppm: Optional[float] = 0.0
     rate: Optional[int] = 1920000

op25-container/app/routers/op25_controller.py

@@ -3,8 +3,13 @@ import subprocess
 import os
 import re
 import signal
+import sys
 import json
-from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS
+import threading
+from collections import deque
+from typing import Optional
+from models import ConfigGenerator, DecodeMode, P25Mode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS
+from internal.conventional_config import build_conventional_config, CONVENTIONAL_TRUNKING_MODULE
 from internal.logger import create_logger
 from internal.op25_config_utls import save_talkgroup_tags, save_whitelist, del_none_in_dict, get_current_system_from_config
 from internal.liquidsoap_config_utils import generate_liquid_script
@@ -14,6 +19,16 @@ LOGGER = create_logger(__name__)
 OP25_PATH = "/op25/op25/gr-op25_repeater/apps/"
 OP25_SCRIPT = "run_multi-rx_service.sh"
 _PGID_FILE = "/tmp/op25.pgid"
+_ACTIVE_CFG = "/configs/active.cfg.json"
+
+# Analog: park the dongle's centre this far from the channel so the channel
+# does not sit on the RTL-SDR's DC spike. Well inside the 1.92 MS/s window.
+ANALOG_CENTER_OFFSET_HZ = 250_000
+
+# Last lines multi_rx/liquidsoap printed — the only place the reason OP25
+# died (SDR busy, bad config, no lock) is visible without SSH + docker logs.
+_log: deque = deque(maxlen=300)
+_proc: Optional[subprocess.Popen] = None
 
 
 def _save_pgid(pgid: int) -> None:
@@ -28,15 +43,60 @@ def _read_pgid():
         return None
 
 
+def _multi_rx_alive(pgid: int) -> bool:
+    """True only if a live multi_rx.py process is in OP25's process group.
+
+    node-26#12: killpg(pgid, 0) succeeds while the wrapper shell or
+    liquidsoap outlives a dead multi_rx (and even for a zombie leader), so
+    the old check reported a dead scanner as "running".
+    """
+    try:
+        pids = [d for d in os.listdir("/proc") if d.isdigit()]
+    except OSError:
+        return False
+    for pid in pids:
+        try:
+            with open(f"/proc/{pid}/stat") as f:
+                fields = f.read().rsplit(")", 1)[1].split()
+            state, pgrp = fields[0], int(fields[2])
+            if pgrp != pgid or state == "Z":
+                continue
+            with open(f"/proc/{pid}/cmdline", "rb") as f:
+                if b"multi_rx.py" in f.read():
+                    return True
+        except (OSError, IndexError, ValueError):
+            continue
+    return False
+
+
+def _reap() -> None:
+    if _proc is not None:
+        _proc.poll()
+
+
 def _is_running() -> bool:
     pgid = _read_pgid()
     if pgid is None:
         return False
+    _reap()
+    return _multi_rx_alive(pgid)
+
+
+def _kill_group(pgid: int) -> None:
     try:
-        os.killpg(pgid, 0)
-        return True
+        os.killpg(pgid, signal.SIGTERM)
     except OSError:
-        return False
+        pass
+
+
+def _pump(proc: subprocess.Popen) -> None:
+    """Mirror OP25's output to the container log and keep the tail for /log."""
+    for line in proc.stdout:
+        sys.stdout.write(line)
+        sys.stdout.flush()
+        _log.append(line.rstrip("\n"))
+    rc = proc.wait()
+    _log.append(f"[OP25 exited with code {rc}]")
 
 
 def create_op25_router():
@@ -44,11 +104,28 @@ def create_op25_router():
 
     @router.post("/start")
     async def start_op25():
+        global _proc
         if _is_running():
             return {"status": "OP25 already running"}
+        if not os.path.exists(_ACTIVE_CFG):
+            raise HTTPException(
+                status_code=409,
+                detail="No OP25 config yet — choose a system or frequency first.",
+            )
+        stale = _read_pgid()
+        if stale is not None:
+            # multi_rx died but liquidsoap/the wrapper may still hold the
+            # Icecast mount — clear the old group before starting a new one.
+            _kill_group(stale)
         try:
-            proc = subprocess.Popen(os.path.join(OP25_PATH, OP25_SCRIPT), shell=True, preexec_fn=os.setsid, cwd=OP25_PATH)
+            _log.clear()
+            proc = subprocess.Popen(
+                os.path.join(OP25_PATH, OP25_SCRIPT), shell=True, preexec_fn=os.setsid, cwd=OP25_PATH,
+                stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
+            )
+            _proc = proc
             _save_pgid(proc.pid)
+            threading.Thread(target=_pump, args=(proc,), daemon=True).start()
             LOGGER.debug(proc)
             return {"status": "OP25 started"}
         except Exception as e:
@@ -57,10 +134,10 @@ def create_op25_router():
     @router.post("/stop")
     async def stop_op25():
         pgid = _read_pgid()
-        if pgid is None or not _is_running():
+        if pgid is None:
             return {"status": "OP25 is not running"}
         try:
-            os.killpg(pgid, signal.SIGTERM)
+            _kill_group(pgid)
             os.remove(_PGID_FILE)
             return {"status": "OP25 stopped"}
         except Exception as e:
@@ -68,7 +145,18 @@ def create_op25_router():
 
     @router.get("/status")
     async def get_status():
-        return {"status": "running" if _is_running() else "stopped"}
+        if _is_running():
+            return {"status": "running"}
+        # Started (pgid file present, never stopped via the API) but multi_rx
+        # is gone: it crashed. The tail of /log says why.
+        if _read_pgid() is not None:
+            return {"status": "crashed", "configured": os.path.exists(_ACTIVE_CFG)}
+        return {"status": "stopped", "configured": os.path.exists(_ACTIVE_CFG)}
+
+    @router.get("/log")
+    async def get_log(lines: int = 80):
+        lines = max(1, min(lines, _log.maxlen))
+        return {"lines": list(_log)[-lines:]}
 
     @router.get("/devices")
     async def list_sdr_devices():
@@ -91,7 +179,20 @@ def create_op25_router():
     @router.post("/generate-config")
     async def generate_config(generator: ConfigGenerator):
         try:
-            if generator.type == DecodeMode.P25:
+            if generator.type == DecodeMode.P25 and generator.p25_mode == P25Mode.CONVENTIONAL:
+                # node-26#24: fixed voice channels, no control channel to hunt.
+                preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
+                ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
+                try:
+                    config_dict, two_channel = build_conventional_config(
+                        generator.systemName, generator.conv_channels, preset["gains"], ppm,
+                    )
+                except ValueError as e:
+                    raise HTTPException(status_code=400, detail=str(e))
+                if generator.icecastConfig:
+                    generate_liquid_script(generator.icecastConfig, two_channel=two_channel)
+
+            elif generator.type == DecodeMode.P25:
                 preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
                 ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
                 channels = [ChannelConfig(
@@ -104,7 +205,7 @@ def create_op25_router():
                     meta_stream_name="stream_0"
                 )]
                 devices = [DeviceConfig(gains=preset["gains"], ppm=ppm)]
-                save_talkgroup_tags(generator.tags)
+                save_talkgroup_tags(generator.tags or [])
                 save_whitelist(generator.whitelist)
                 has_talkgroups = bool(generator.whitelist)
                 trunking = TrunkingConfig(
@@ -142,30 +243,53 @@ def create_op25_router():
                     "terminal": terminal.dict()
                 }
 
+
             elif generator.type == DecodeMode.ANALOG:
-                generator = generator.config
+                if not generator.channels:
+                    raise HTTPException(status_code=400, detail="Analog needs a frequency.")
+                preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
+                ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
+                freq = int(float(generator.channels[0]))
+                # One fixed conventional channel: a trunking module with no
+                # chans, so the channel is reported as `conventional` on the
+                # HTTP terminal.
                 channels = [ChannelConfig(
-                    channelName=generator.systemName,
-                    enableAnalog="on",
-                    demodType="fsk4",
-                    frequency=generator.frequency,
-                    filterType="widepulse",
-                    nbfmSquelch=generator.nbfmSquelch
+                    name=generator.systemName,
+                    trunking_sysname="",
+                    enable_analog="on",
+                    demod_type="fsk4",
+                    filter_type="rc",
+                    frequency=freq,
+                    meta_stream_name="",
+                    nbfm_squelch_threshold=generator.nbfm_squelch,
                 )]
-                devices = [DeviceConfig(gain="LNA:32")]
+                devices = [DeviceConfig(
+                    gains=preset["gains"], ppm=ppm, tunable=False,
+                    frequency=freq + ANALOG_CENTER_OFFSET_HZ,
+                )]
+                if generator.icecastConfig:
+                    generate_liquid_script(generator.icecastConfig)
 
                 config_dict = {
                     "channels": [channel.dict() for channel in channels],
-                    "devices": [device.dict() for device in devices]
+                    "devices": [device.dict() for device in devices],
+                    # Same module as P25 conventional (node-26#24): it reports
+                    # the channel as `conventional` whatever tk_p25 version the
+                    # image was built with, which is what the edge node's
+                    # single-channel attribution keys on (node-26#20).
+                    "trunking": {"module": CONVENTIONAL_TRUNKING_MODULE, "chans": []},
+                    "terminal": TerminalConfig().dict(),
                 }
 
             else:
-                raise HTTPException(status_code=400, detail="Invalid configuration type. Must be 'p25' or 'nbfm'.")
+                raise HTTPException(status_code=400, detail="Unsupported system type. Must be 'P25' or 'NBFM' (DMR is not implemented).")
 
-            with open('/configs/active.cfg.json', 'w') as f:
+            with open(_ACTIVE_CFG, 'w') as f:
                 json.dump(del_none_in_dict(config_dict), f, indent=2)
 
-            return {"message": "Config exported to '/configs/active.cfg.json'"}
+            return {"message": f"Config exported to '{_ACTIVE_CFG}'"}
+        except HTTPException:
+            raise
         except Exception as e:
             raise HTTPException(status_code=500, detail=str(e))
 

op25-container/op25_addons/tk_drb_conv.py

@@ -0,0 +1,263 @@
+"""
+tk_drb_conv — DRB's trunking module for P25 CONVENTIONAL channels (node-26#24).
+
+Copied into OP25's apps directory by the Dockerfile and named as the trunking
+module by op25-container's conventional config, so multi_rx imports it in place
+of tk_p25. It is tk_p25.rx_ctl with the conventional-channel handling replaced;
+trunked receivers (none in a conventional config) still go to the base class.
+
+Why not stock tk_p25: for an FDMA conventional channel it reports the talkgroup
+and unit id from LDU1 link control, but
+
+  * never reports encryption — the encrypted/algid/keyid JSON OP25 sends after
+    every LDU2 carries no grpaddr, and tk_p25 only records `encrypted` when
+    grpaddr != 0 — so an encrypted channel looks exactly like a quiet one;
+  * never reports the NAC it is hearing, and has no per-channel NAC filter;
+  * has no "voice frames are arriving" signal of its own, so a transmission
+    whose link control is unreadable is invisible to the console.
+
+All three matter to the edge node: it attributes audio to a channel from what
+this module reports, and must tell the owner when a channel is encrypted rather
+than silently recording nothing.
+
+Each conventional channel row in `channel_update` keeps every key tk_p25 sends
+(freq, tgid, srcaddr, tag, encrypted, conventional, ...) and adds:
+
+    voice            True while voice frames are arriving (HDU/LDU1/LDU2 seen,
+                     no TDU or sync timeout since)
+    nac              last NAC heard on the channel (int) or None
+    nac_filter       the configured NAC filter (int) or None
+    algid, keyid     ALGID/key id of the current call (None when clear/idle)
+    enc_calls, clear_calls     calls heard since start, by encryption
+    last_enc_algid, last_enc_keyid, last_enc_time, last_clear_time
+
+Message types handled (lower 16 bits of msg.type(), P25 protocol only):
+    0          HDU                 -> voice start
+    3, 15      TDU / TDULC         -> transmission end
+    19         LDU1 link control   -> tgid, srcaddr, service options (E bit)
+    -3         LDU2 ESS JSON       -> encrypted / algid / keyid
+    -1         sync timeout        -> transmission end
+"""
+import ctypes
+import json
+import sys
+import time
+
+import tk_p25
+from helper_funcs import from_dict, get_ordinals
+from log_ts import log_ts
+
+ALGID_CLEAR = 0x80
+SVCOPT_ENCRYPTED = 0x40
+LCW_PROTECTED = 0x80
+LCO_GROUP_VOICE = 0x00
+DUID_HDU, DUID_TDU, DUID_TDULC, DUID_LCW = 0, 3, 15, 19
+M_TIMEOUT, M_JSON_DATA = -1, -3
+
+
+def _parse_nac(value):
+    if value is None or value == "":
+        return None
+    try:
+        nac = int(value, 16) if isinstance(value, str) else int(value)
+    except (TypeError, ValueError):
+        return None
+    return nac if 0 < nac <= 0xFFF else None
+
+
+class rx_ctl(tk_p25.rx_ctl):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.drb_conv = {}   # msgq_id -> state dict
+
+    def add_receiver(self, msgq_id, config, meta_q=None, freq=0):
+        super().add_receiver(msgq_id, config, meta_q=meta_q, freq=freq)
+        rcvr = self.receivers.get(msgq_id)
+        if rcvr is not None and rcvr.get('rx_rcvr') is not None:
+            return   # trunked receiver: base class owns it
+        self.drb_conv[msgq_id] = {
+            'name': from_dict(config, 'name', str(msgq_id)),
+            'freq': freq,
+            'nac_filter': _parse_nac(config.get('nac')),
+            'nac': None,
+            'voice': False,
+            'tgid': None,
+            'srcaddr': 0,
+            'svc_encrypted': False,
+            'ess_encrypted': False,
+            'algid': None,
+            'keyid': None,
+            'call_start': None,
+            'call_encrypted': False,
+            'logged_tgid': None,
+            'prev_type': None,
+            'enc_calls': 0,
+            'clear_calls': 0,
+            'last_enc_algid': None,
+            'last_enc_keyid': None,
+            'last_enc_time': None,
+            'last_clear_time': None,
+            'last_voice_time': None,
+        }
+
+    def post_init(self):
+        super().post_init()
+        for msgq_id, cs in self.drb_conv.items():
+            if cs['nac_filter'] is not None and self.fa_ctrl is not None:
+                self.fa_ctrl({'tuner': msgq_id, 'cmd': 'set_nac', 'nac': cs['nac_filter']})
+                sys.stderr.write("%s [%d] conventional NAC filter 0x%03x\n" % (log_ts.get(), msgq_id, cs['nac_filter']))
+
+    # ------------------------------------------------------------------
+
+    def _encrypted(self, cs):
+        return cs['svc_encrypted'] or cs['ess_encrypted']
+
+    def _voice_start(self, cs, now):
+        if not cs['voice']:
+            cs['voice'] = True
+            cs['call_start'] = now
+            cs['call_encrypted'] = False
+            cs['logged_tgid'] = None
+        cs['last_voice_time'] = now
+
+    def _voice_end(self, cs, now):
+        if cs['voice']:
+            if cs['call_encrypted']:
+                cs['enc_calls'] += 1
+                cs['last_enc_time'] = now
+            else:
+                cs['clear_calls'] += 1
+                cs['last_clear_time'] = now
+        cs['voice'] = False
+        cs['tgid'] = None
+        cs['srcaddr'] = 0
+        cs['svc_encrypted'] = False
+        cs['ess_encrypted'] = False
+        cs['algid'] = None
+        cs['keyid'] = None
+        cs['call_start'] = None
+        cs['logged_tgid'] = None
+
+    def _note_encryption(self, cs, now):
+        if self._encrypted(cs):
+            cs['call_encrypted'] = True
+            if cs['algid'] is not None:
+                cs['last_enc_algid'] = cs['algid']
+                cs['last_enc_keyid'] = cs['keyid']
+            cs['last_enc_time'] = now
+
+    def _maybe_log_call(self, msgq_id, cs):
+        # One call_log entry per (call, talkgroup), and none for an encrypted
+        # call: it will produce no audio, so a "grant" would only mislead the
+        # edge node's attribution.
+        if cs['tgid'] is None or self._encrypted(cs) or cs['logged_tgid'] == cs['tgid']:
+            return
+        cs['logged_tgid'] = cs['tgid']
+        try:
+            self.log_call(0, msgq_id, cs['freq'], 0, 0, cs['tgid'], cs['name'], cs['srcaddr'], '')
+        except Exception as e:   # older/newer tk_p25 log_call signature
+            sys.stderr.write("%s [%d] conventional call_log failed: %s\n" % (log_ts.get(), msgq_id, e))
+
+    def _process_conv(self, msgq_id, cs, m_type, msg, now):
+        if m_type == M_TIMEOUT or m_type in (DUID_TDU, DUID_TDULC):
+            self._voice_end(cs, now)
+        elif m_type == DUID_HDU:
+            s = msg.to_string()
+            cs['nac'] = get_ordinals(s[0:2]) if len(s) >= 2 else cs['nac']
+            self._voice_start(cs, now)
+        elif m_type == DUID_LCW:
+            s = msg.to_string()
+            if len(s) < 11:
+                return
+            cs['nac'] = get_ordinals(s[0:2])
+            if cs['prev_type'] in (DUID_TDU, DUID_TDULC):
+                return   # TDULC's link control follows its TDU: the call is over
+            lcw = s[2:]
+            self._voice_start(cs, now)
+            pb_sf_lco = get_ordinals(lcw[0:1])
+            if pb_sf_lco & LCW_PROTECTED:
+                # Link control itself is encrypted — the call certainly is.
+                cs['svc_encrypted'] = True
+            elif (pb_sf_lco & 0x3f) == LCO_GROUP_VOICE:
+                svcopts = get_ordinals(lcw[2:3])
+                ga = get_ordinals(lcw[4:6])
+                sa = get_ordinals(lcw[6:9])
+                cs['svc_encrypted'] = bool(svcopts & SVCOPT_ENCRYPTED)
+                if ga:
+                    cs['tgid'] = ga
+                if sa:
+                    cs['srcaddr'] = sa
+            self._note_encryption(cs, now)
+            self._maybe_log_call(msgq_id, cs)
+        elif m_type == M_JSON_DATA:
+            try:
+                js = json.loads(msg.to_string())
+            except (ValueError, TypeError):
+                return
+            if 'encrypted' in js:
+                self._voice_start(cs, now)
+                algid = from_dict(js, 'algid', None)
+                cs['ess_encrypted'] = bool(js.get('encrypted')) and algid != ALGID_CLEAR
+                cs['algid'] = algid if cs['ess_encrypted'] else None
+                cs['keyid'] = from_dict(js, 'keyid', None) if cs['ess_encrypted'] else None
+                self._note_encryption(cs, now)
+            grpaddr = from_dict(js, 'grpaddr', 0)
+            if grpaddr:
+                cs['tgid'] = grpaddr
+                cs['srcaddr'] = from_dict(js, 'srcaddr', 0) or cs['srcaddr']
+                self._maybe_log_call(msgq_id, cs)
+
+    def process_qmsg(self, msg):
+        m_proto = ctypes.c_int16(msg.type() >> 16).value
+        m_rxid = int(msg.arg1()) >> 1
+        if m_proto != 0 or m_rxid not in self.drb_conv:
+            return super().process_qmsg(msg)
+        m_type = ctypes.c_int16(msg.type() & 0xffff).value
+        cs = self.drb_conv[m_rxid]
+        try:
+            self._process_conv(m_rxid, cs, m_type, msg, time.time())
+        except Exception as e:   # never take multi_rx down over a bad frame
+            sys.stderr.write("%s [%d] conventional message type %d failed: %s\n" % (log_ts.get(), m_rxid, m_type, e))
+        cs['prev_type'] = m_type
+
+    def get_chan_status(self):
+        d = json.loads(super().get_chan_status())
+        ids = list(d.get('channels') or [])
+        for msgq_id, cs in self.drb_conv.items():
+            key = str(msgq_id)
+            row = d.get(key) or {}
+            row.update({
+                'freq': cs['freq'],
+                'name': cs['name'],
+                'system': cs['name'],
+                'tdma': 0,
+                'tgid': cs['tgid'],
+                'tag': cs['name'],
+                'srcaddr': cs['srcaddr'],
+                'srctag': '',
+                'svcopts': 0,
+                'encrypted': 1 if self._encrypted(cs) else 0,
+                'emergency': 0,
+                'hold_tgid': 0,
+                'mode': None,
+                'stream': '',
+                'msgqid': msgq_id,
+                'conventional': True,
+                'voice': cs['voice'],
+                'nac': cs['nac'],
+                'nac_filter': cs['nac_filter'],
+                'algid': cs['algid'],
+                'keyid': cs['keyid'],
+                'enc_calls': cs['enc_calls'],
+                'clear_calls': cs['clear_calls'],
+                'last_enc_algid': cs['last_enc_algid'],
+                'last_enc_keyid': cs['last_enc_keyid'],
+                'last_enc_time': cs['last_enc_time'],
+                'last_clear_time': cs['last_clear_time'],
+                'last_voice_time': cs['last_voice_time'],
+            })
+            d[key] = row
+            if key not in ids:
+                ids.append(key)
+        d['channels'] = ids
+        return json.dumps(d)

op25-container/run_multi-rx_service.sh

@@ -20,3 +20,12 @@ LIQ_PID=$! # Store the PID of the op25.liq process
 # The trap command ensures that SIGTERM is passed to the background jobs.
 trap "kill $MULTI_RX_PID $LIQ_PID" SIGTERM SIGINT
 wait $MULTI_RX_PID
+RC=$?
+
+# multi_rx is gone (crashed, or SDR could not be opened). Take liquidsoap down
+# with it so nothing in this process group outlives the decoder — a leftover
+# liquidsoap kept /op25/status reporting "running" (node-26#12) and held the
+# Icecast mount against the next start.
+echo "multi_rx.py exited (code $RC) — stopping liquidsoap."
+kill $LIQ_PID 2>/dev/null
+exit $RC

op25-container/tests/test_conventional.py

@@ -0,0 +1,305 @@
+"""
+P25 conventional (node-26#24): config generation, the liquidsoap two-channel
+variant, and the tk_drb_conv state machine — all without OP25/GNU Radio.
+
+tk_drb_conv imports tk_p25/helper_funcs/log_ts from OP25's apps dir; those are
+stubbed here with the minimum of tk_p25.rx_ctl it relies on. The message
+shapes fed in are the ones boatbod gr310 p25p1_fdma.cc sends: LCW = 2-byte NAC
++ 9 LC bytes (type 19), LDU2 ESS = JSON (type -3), TDU = 2-byte NAC (type 3/15),
+sync timeout = type -1.
+"""
+import json
+import os
+import sys
+import types
+
+import pytest
+
+HERE = os.path.dirname(__file__)
+sys.path.insert(0, os.path.join(HERE, "..", "app"))
+sys.path.insert(0, os.path.join(HERE, "..", "op25_addons"))
+
+from fastapi.testclient import TestClient  # noqa: E402
+from fastapi import FastAPI  # noqa: E402
+
+from models import ConventionalChannel  # noqa: E402
+from internal import conventional_config as cc  # noqa: E402
+from internal.liquidsoap_config_utils import render_liquid_script  # noqa: E402
+from models import IcecastConfig  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# Config generation
+# ---------------------------------------------------------------------------
+
+WP = [
+    ConventionalChannel(frequency=460_400_000, name="PD1 Dispatch"),
+    ConventionalChannel(frequency=460_425_000, name="PD3 Tac", nac=0x293),
+]
+
+
+def test_white_plains_two_channels_one_sdr():
+    config, two = cc.build_conventional_config("White Plains", WP, "lna:34", 0.0)
+    assert two is True
+    assert "trunking" in config and config["trunking"]["module"] == "tk_drb_conv.py"
+    assert config["trunking"]["chans"] == []          # nothing for tk_p25 to hunt
+    a, b = config["channels"]
+    for ch in (a, b):
+        assert ch["demod_type"] == "fsk4"
+        assert ch["trunking_sysname"] == ""
+        assert ch["enable_analog"] == "off"
+        assert ch["crypt_behavior"] == 2
+        assert "control_channel_list" not in ch
+    assert (a["frequency"], b["frequency"]) == (460_400_000, 460_425_000)
+    assert a["destination"].endswith(":23456") and b["destination"].endswith(":23457")
+    assert a["name"] == "PD1 Dispatch" and b["nac"] == 0x293
+    dev = config["devices"][0]
+    assert dev["tunable"] is False
+    # 25 kHz apart: centre is offset off both channels (DC spike), and both
+    # still sit inside the usable passband.
+    half = dev["rate"] * dev["usable_bw_pct"] / 2
+    for f in (460_400_000, 460_425_000):
+        assert 100_000 < abs(f - dev["frequency"]) < half
+
+
+def test_single_channel_is_not_two_channel_audio():
+    config, two = cc.build_conventional_config("X", WP[:1], "lna:34", 0.0)
+    assert two is False
+    assert len(config["channels"]) == 1
+
+
+def test_wide_pair_centres_between_channels():
+    chans = [ConventionalChannel(frequency=453_000_000), ConventionalChannel(frequency=454_000_000)]
+    config, _ = cc.build_conventional_config("X", chans, "lna:34", 0.0)
+    assert config["devices"][0]["frequency"] == 453_500_000
+
+
+@pytest.mark.parametrize("chans, needle", [
+    ([], "at least one"),
+    ([ConventionalChannel(frequency=f) for f in (460_000_000, 460_025_000, 460_050_000)], "up to 2"),
+    ([ConventionalChannel(frequency=460_000_000), ConventionalChannel(frequency=462_000_000)], "too far"),
+    ([ConventionalChannel(frequency=460_000_000), ConventionalChannel(frequency=460_000_000)], "twice"),
+    ([ConventionalChannel(frequency=460_000_000, nac=0x1000)], "NAC"),
+])
+def test_rejects_bad_channel_sets(chans, needle):
+    with pytest.raises(ValueError, match=needle):
+        cc.build_conventional_config("X", chans, "lna:34", 0.0)
+
+
+ICE = IcecastConfig(icecast_host="localhost", icecast_port=8000, icecast_mountpoint="/n", icecast_password="p")
+
+
+def test_liquidsoap_single_channel_unchanged_command():
+    liq = render_liquid_script(ICE, two_channel=False)
+    assert '"./audio.py -x 2.5 -s"' in liq
+    assert "amplify" not in liq and "${" not in liq
+
+
+def test_liquidsoap_two_channel():
+    liq = render_liquid_script(ICE, two_channel=True)
+    assert '"./audio.py -x 2.5 -s -2"' in liq
+    assert "input = amplify(2.0, input)" in liq
+    assert "${" not in liq
+
+
+def test_generate_config_route(tmp_path, monkeypatch):
+    from routers import op25_controller
+    monkeypatch.setattr(op25_controller, "_ACTIVE_CFG", str(tmp_path / "active.cfg.json"))
+    written = {}
+    monkeypatch.setattr(op25_controller, "generate_liquid_script",
+                        lambda ice, two_channel=False: written.setdefault("two", two_channel))
+    monkeypatch.setattr(op25_controller, "save_talkgroup_tags", lambda tags: None)
+    monkeypatch.setattr(op25_controller, "save_whitelist", lambda wl: None)
+    app = FastAPI()
+    app.include_router(op25_controller.create_op25_router(), prefix="/op25")
+    client = TestClient(app)
+    ice = ICE.model_dump()
+    r = client.post("/op25/generate-config", json={
+        "type": "P25", "systemName": "White Plains", "p25_mode": "conventional",
+        "conv_channels": [{"frequency": 460400000, "name": "PD1"}, {"frequency": 460425000, "name": "PD3"}],
+        "icecastConfig": ice,
+    })
+    assert r.status_code == 200, r.text
+    cfg = json.loads((tmp_path / "active.cfg.json").read_text())
+    assert cfg["trunking"]["module"] == "tk_drb_conv.py"
+    assert [c["frequency"] for c in cfg["channels"]] == [460400000, 460425000]
+    assert written["two"] is True
+
+    # Trunked P25 without p25_mode is exactly what it was.
+    r = client.post("/op25/generate-config", json={
+        "type": "P25", "systemName": "County", "channels": [851012500], "icecastConfig": ice,
+    })
+    assert r.status_code == 200, r.text
+    cfg = json.loads((tmp_path / "active.cfg.json").read_text())
+    assert cfg["trunking"]["module"] == "tk_p25.py"
+    assert cfg["trunking"]["chans"][0]["control_channel_list"] == "851012500"
+    assert cfg["channels"][0]["demod_type"] == "cqpsk"
+
+    r = client.post("/op25/generate-config", json={
+        "type": "P25", "systemName": "X", "p25_mode": "conventional", "conv_channels": [],
+    })
+    assert r.status_code == 400
+
+    # Analog uses the same module, so the edge node sees a `conventional` row
+    # whatever tk_p25 the image was built with (node-26#20).
+    r = client.post("/op25/generate-config", json={
+        "type": "NBFM", "systemName": "Fire", "channels": [154430000], "icecastConfig": ice,
+    })
+    assert r.status_code == 200, r.text
+    cfg = json.loads((tmp_path / "active.cfg.json").read_text())
+    assert cfg["trunking"]["module"] == "tk_drb_conv.py"
+    assert cfg["channels"][0]["frequency"] == 154430000
+
+
+# ---------------------------------------------------------------------------
+# tk_drb_conv state machine
+# ---------------------------------------------------------------------------
+
+class _BaseRxCtl:
+    """The slice of tk_p25.rx_ctl that tk_drb_conv relies on."""
+
+    def __init__(self, debug=0, frequency_set=None, nbfm_ctrl=None, fa_ctrl=None, chans={}):
+        self.fa_ctrl = fa_ctrl
+        self.receivers = {}
+        self.call_log = []
+
+    def add_receiver(self, msgq_id, config, meta_q=None, freq=0):
+        self.receivers[msgq_id] = {'msgq_id': msgq_id, 'config': config, 'rx_rcvr': None,
+                                   'conv_state': {'tgid': None}}
+
+    def post_init(self):
+        pass
+
+    def process_qmsg(self, msg):
+        self.base_saw = msg
+
+    def log_call(self, sysid, rcvr, freq, slot, prio, tgid, tgtag, rid, rtag):
+        self.call_log.append({"rcvr": rcvr, "freq": freq, "tgid": tgid, "tgtag": tgtag, "rid": rid})
+
+    def get_chan_status(self):
+        return json.dumps({'json_type': 'channel_update', 'channels': []})
+
+
+def _stub_op25_modules():
+    tk = types.ModuleType("tk_p25")
+    tk.rx_ctl = _BaseRxCtl
+    hf = types.ModuleType("helper_funcs")
+
+    def from_dict(d, key, def_val):
+        return d[key] if key in d and d[key] != "" else def_val
+
+    def get_ordinals(s):
+        t = 0
+        for c in s:
+            t = (t << 8) + c
+        return t
+    hf.from_dict = from_dict
+    hf.get_ordinals = get_ordinals
+    lt = types.ModuleType("log_ts")
+    lt.log_ts = types.SimpleNamespace(get=lambda *a: "ts")
+    sys.modules.update({"tk_p25": tk, "helper_funcs": hf, "log_ts": lt})
+
+
+class _Msg:
+    def __init__(self, m_type, rxid, payload=b""):
+        self._type = m_type & 0xffff   # proto 0 in the upper 16 bits
+        self._arg1 = rxid << 1
+        self._payload = payload
+
+    def type(self):
+        return self._type
+
+    def arg1(self):
+        return self._arg1
+
+    def to_string(self):
+        return self._payload
+
+
+def _lcw(nac, tgid, src, svcopts=0, lco=0x00):
+    lc = bytes([lco, 0x00, svcopts, 0x00]) + tgid.to_bytes(2, "big") + src.to_bytes(3, "big")
+    return nac.to_bytes(2, "big") + lc
+
+
+def _ess(encrypted, algid, keyid):
+    return json.dumps({"encrypted": encrypted, "algid": algid, "keyid": keyid}).encode()
+
+
+@pytest.fixture
+def tk():
+    _stub_op25_modules()
+    sys.modules.pop("tk_drb_conv", None)
+    import tk_drb_conv
+    sent = []
+    rx = tk_drb_conv.rx_ctl(fa_ctrl=sent.append, chans=[])
+    rx.add_receiver(0, {"name": "PD1 Dispatch", "nac": 0x293}, freq=460_400_000)
+    rx.add_receiver(1, {"name": "PD3 Tac"}, freq=460_425_000)
+    rx.post_init()
+    rx.sent = sent
+    return rx
+
+
+def _rows(rx):
+    d = json.loads(rx.get_chan_status())
+    return {cid: d[cid] for cid in d["channels"]}
+
+
+def test_nac_filter_applied_only_where_configured(tk):
+    assert tk.sent == [{'tuner': 0, 'cmd': 'set_nac', 'nac': 0x293}]
+
+
+def test_clear_call_reports_tgid_unit_nac_and_logs_once(tk):
+    tk.process_qmsg(_Msg(0, 0, b"\x02\x93"))                       # HDU
+    tk.process_qmsg(_Msg(19, 0, _lcw(0x293, 101, 4242)))            # LDU1
+    tk.process_qmsg(_Msg(-3, 0, _ess(0, 0x80, 0)))                  # LDU2 clear
+    tk.process_qmsg(_Msg(19, 0, _lcw(0x293, 101, 4242)))
+    row = _rows(tk)["0"]
+    assert row["conventional"] and row["voice"]
+    assert (row["tgid"], row["srcaddr"], row["nac"], row["encrypted"]) == (101, 4242, 0x293, 0)
+    assert row["algid"] is None
+    assert [e["tgid"] for e in tk.call_log] == [101]
+    assert tk.call_log[0]["tgtag"] == "PD1 Dispatch"
+    # The other channel is idle.
+    assert _rows(tk)["1"]["voice"] is False
+
+    tk.process_qmsg(_Msg(3, 0, b"\x02\x93"))                         # TDU
+    row = _rows(tk)["0"]
+    assert row["voice"] is False and row["tgid"] is None
+    assert (row["clear_calls"], row["enc_calls"]) == (1, 0)
+
+
+def test_tdulc_link_control_does_not_reopen_the_call(tk):
+    tk.process_qmsg(_Msg(19, 1, _lcw(0x111, 7, 9)))
+    tk.process_qmsg(_Msg(15, 1, b"\x01\x11"))                        # TDULC
+    tk.process_qmsg(_Msg(19, 1, _lcw(0x111, 7, 9)))                 # its LC
+    assert _rows(tk)["1"]["voice"] is False
+
+
+def test_encrypted_call_is_reported_and_never_logged(tk):
+    tk.process_qmsg(_Msg(19, 1, _lcw(0x111, 7, 9, svcopts=0x40)))  # E bit
+    tk.process_qmsg(_Msg(-3, 1, _ess(1, 0x84, 0x1234)))
+    row = _rows(tk)["1"]
+    assert row["encrypted"] == 1 and row["algid"] == 0x84 and row["keyid"] == 0x1234
+    assert tk.call_log == []
+    tk.process_qmsg(_Msg(-1, 1))                                    # sync timeout
+    row = _rows(tk)["1"]
+    assert row["encrypted"] == 0 and row["voice"] is False
+    assert (row["enc_calls"], row["last_enc_algid"], row["last_enc_keyid"]) == (1, 0x84, 0x1234)
+
+
+def test_encryption_seen_only_in_ess_still_counts(tk):
+    tk.process_qmsg(_Msg(-3, 0, _ess(1, 0x81, 0x0001)))
+    assert _rows(tk)["0"]["encrypted"] == 1
+    tk.process_qmsg(_Msg(3, 0, b"\x02\x93"))
+    assert _rows(tk)["0"]["enc_calls"] == 1
+
+
+def test_protected_link_control_marks_encrypted(tk):
+    tk.process_qmsg(_Msg(19, 0, _lcw(0x293, 0, 0, lco=0x80)))
+    assert _rows(tk)["0"]["encrypted"] == 1
+
+
+def test_trunked_receivers_and_other_protocols_go_to_base(tk):
+    msg = _Msg(7, 5)
+    tk.process_qmsg(msg)
+    assert tk.base_saw is msg

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
-  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; }
+  if [ "$pin" = 1 ]; then
+    drb_pin_image_tags "$INSTALL_DIR/.env" "$INSTALL_DIR" "$new" || { err "could not write image pins to .env"; return 1; }
+  fi
 
   if [ "$BUILD" = 1 ]; then
-    docker compose build || return 1
-  else
-    docker compose pull || return 1
+    docker compose build || return 1   # tags the local builds with the same pins
   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
+}