discord_radio: don't count the bot's own voice state toward a full channel; timeout text points at the handshake

Permissions are pre-checked now, so a connect timeout is the voice handshake
(pre-2.7 discord.py: DAVE close 4017 retried to the 30s timeout) or UDP egress,
not permissions.

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/publish-public.yml

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

.gitea/workflows/release-images.yml

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

.gitignore

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

README.md

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

docker-compose.yml

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

drb-edge-node/app/config.py

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

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

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

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

@@ -11,6 +11,7 @@ from app.internal import credentials, key_pickup
 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 +47,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 +57,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 +137,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 +166,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 +294,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

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":
+    if action in ("discord_join", "discord_leave"):
-        token = payload.get("token")
+        # Pre-desired-state C2. Discord streams now arrive on
-        if not token:
+        # nodes/{id}/discord/desired (node-26#25); a raw-token join is refused
-            logger.error("discord_join command missing token — ignoring.")
+        # so a bot token is never taken from an MQTT command again.
-            return
+        logger.warning(f"Ignoring legacy {action} command — Discord radio is driven by desired state now.")
-        await radio_bot.join(
-            guild_id=int(payload["guild_id"]),
-            channel_id=int(payload["channel_id"]),
-            token=token,
-            call_active=metadata_watcher.is_active,
-            system_name=payload.get("system_name"),
-        )
-    elif action == "discord_leave":
-        await radio_bot.leave()
     elif action in ("set_sdr_config", "set_secondary_priority"):
         from app.internal.sdr_settings import set_sdr_settings
         try:
@@ -300,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

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

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

drb-edge-node/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_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,7 +147,7 @@ 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
 
 
@@ -153,7 +174,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 +198,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()

install.sh

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

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
@@ -146,6 +148,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,19 +196,28 @@ do_update() {
     bash scripts/install-updater.sh --refresh || log "unit refresh failed (continuing with the old units)"
   fi
 
-  local env_registry
+  if [ "$pin" = 1 ]; then
-  env_registry="$(grep -E '^IMAGE_REGISTRY=' .env | head -1 | cut -d= -f2- | tr -d '"' || true)"
+    drb_pin_image_tags "$INSTALL_DIR/.env" "$INSTALL_DIR" "$new" || { err "could not write image pins to .env"; return 1; }
-  [ "${env_registry:-git.vpn.cusano.net}" = "$REGISTRY" ] \
+  fi
-    || { err ".env IMAGE_REGISTRY '$env_registry' != configured '$REGISTRY' — refusing"; return 1; }
 
   if [ "$BUILD" = 1 ]; then
-    docker compose build || return 1
+    docker compose build || return 1   # tags the local builds with the same pins
-  else
-    docker compose pull || return 1
   fi
   # --force-recreate: app code is bind-mounted from the checkout (install.sh
   # D3), so a code-only change also needs every container restarted.
   docker compose up -d --no-build --force-recreate --remove-orphans || return 1
+
+  # Drop pinned images this node no longer runs. `image prune` only removes
+  # dangling images, and every previous src- tag is still tagged. `image rm`
+  # without -f refuses anything a container still uses.
+  if [ "$pin" = 1 ]; then
+    local keep ref
+    keep="$(drb_image_tags "$INSTALL_DIR" "$new" | cut -d= -f2)"
+    docker images --format '{{.Repository}}:{{.Tag}}' | grep -E ':src-[0-9a-f]{40}$' \
+      | while IFS= read -r ref; do
+          grep -qxF "${ref##*:}" <<< "$keep" || docker image rm "$ref" >/dev/null 2>&1 || true
+        done || true
+  fi
   docker image prune -f >/dev/null 2>&1 || true
 }
 

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
+}