P25 conventional decode mode + conventional-channel attribution (node-26#24, #20)

op25-container
- generate-config takes p25_mode "conventional" + conv_channels [{frequency,
  name, nac}]. It builds fixed fsk4 channels (up to 2 on one non-tunable SDR,
  DC spike kept off the channels) with no control channel to hunt. Two
  channels send audio to UDP 23456/23457, and audio.py -2 + amplify(2) keeps
  levels.
- tk_drb_conv.py (a tk_p25.rx_ctl subclass, copied into OP25 apps) reports
  voice / NAC / encryption (LC service options + LDU2 ESS algid/keyid) for
  conventional channels and applies the per-channel NAC filter, which stock
  tk_p25 does not. Analog uses it too.
- Trunked P25 config is unchanged when p25_mode is absent.

drb-edge-node
- radio_config: system kind from type + config.mode (missing = trunked).
- conventional.py normalizes the console: a distinctive LC TGID, else the
  channel's last TGID (30 s), else a synthetic per-channel id (freq/100 Hz).
  Encrypted rows are never talkers. With exactly one conventional channel,
  orphan audio is attributed to it at close, which fixes analog recording
  (#20). The synthetic id and the real TGID on one channel never split.
- Health/dashboard: P25 conventional listen tab (/api/listen p25conv), and
  conventional decode + encryption checks. The checkin carries radio_channels.
- Tests: edge 296 pass; op25-container 18 (new CI job). Reviewed by
  drb-audio-review (findings fixed; simultaneous two-channel mixdown filed
  as #26) and drb-qa-review.

Not verified on SDR hardware.

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

.gitea/workflows/ci.yml

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

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

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

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

@@ -3,8 +3,12 @@ Manages the persisted node API key, plus the local-auth signing material used
 by app/internal/auth.py.
 
 The API key is provisioned by the C2 server after an admin approves the node.
-It arrives via MQTT and is saved to /configs/credentials.json so it survives
+install.sh fetches it over HTTPS (GET /nodes/{id}/credentials + the pickup
-container restarts.
+secret it saved to /configs/pickup_secret) and writes /configs/credentials.json;
+the running app can also fetch it itself (app/internal/key_pickup.py) and
+re-read the file at any time (reload_api_key), so a key written after startup
+is picked up without a container restart (node-26#14). The legacy MQTT
+delivery (nodes/{id}/api_key) still saves here too.
 
 The scrypt salt and session-signing secret are generated locally on first boot
 (never provisioned externally) and persisted the same way, so dashboard
@@ -12,12 +16,14 @@ sessions survive a container restart instead of forcing every operator to
 re-login whenever the node restarts.
 """
 import json
+import os
 import secrets
 from pathlib import Path
 from app.config import settings
 from app.internal.logger import logger
 
 _CREDS_FILE = Path(settings.config_path) / "credentials.json"
+_PICKUP_FILE = Path(settings.config_path) / "pickup_secret"
 _api_key: str | None = None
 _auth_salt: bytes | None = None
 _session_secret: bytes | None = None
@@ -74,6 +80,37 @@ def get_api_key() -> str | None:
     return _api_key
 
 
+def reload_api_key() -> str | None:
+    """Re-read the api_key from disk (install.sh, or an operator, may have
+    written it after startup). Returns the key now in effect. Only the api_key
+    is taken from the file — the auth material already in memory stays."""
+    global _api_key
+    try:
+        data = json.loads(_CREDS_FILE.read_text())
+    except FileNotFoundError:
+        return _api_key
+    except Exception as e:
+        logger.warning(f"Could not re-read credentials file: {e}")
+        return _api_key
+    key = data.get("api_key")
+    if key and key != _api_key:
+        _api_key = key
+        logger.info("Node API key re-read from disk (changed since it was last loaded).")
+    return _api_key
+
+
+def get_pickup_secret() -> str | None:
+    """The one-time pickup secret install.sh saved at enrollment, if any."""
+    try:
+        secret = _PICKUP_FILE.read_text().strip()
+    except FileNotFoundError:
+        return None
+    except Exception as e:
+        logger.warning(f"Could not read pickup secret: {e}")
+        return None
+    return secret or None
+
+
 def save_api_key(key: str) -> None:
     global _api_key
     _api_key = key
@@ -88,4 +125,10 @@ def _write() -> None:
         data["auth_salt"] = _auth_salt.hex()
     if _session_secret is not None:
         data["session_secret"] = _session_secret.hex()
-    _CREDS_FILE.write_text(json.dumps(data))
+    # Atomic replace with 0600: install.sh on the host writes this same file,
+    # and a torn half-written JSON would lose the key AND the auth material.
+    tmp = _CREDS_FILE.with_name(_CREDS_FILE.name + ".tmp")
+    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+    with os.fdopen(fd, "w") as f:
+        f.write(json.dumps(data))
+    os.replace(tmp, _CREDS_FILE)

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

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

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

@@ -0,0 +1,73 @@
+"""
+HTTPS api_key pickup — the node's own half of C2's enrollment flow
+(Server/drb-c2-core/app/routers/enrollment.py, GET /nodes/{id}/credentials).
+
+Why this exists (node-26#14): key delivery used to be MQTT-only
+(nodes/{id}/api_key), and a dynsec broker refuses a node that has no key — or
+a stale one — so a locked-out node could never receive the key that would let
+it in. install.sh already did this pickup once at install time; doing it from
+the running app too means a node approved after install, or whose key was
+reissued, recovers on its own with no SSH and no container restart.
+
+Auth is the pickup secret install.sh saved to /configs/pickup_secret at
+enrollment (the same file, bind-mounted). Only ever sent to settings.c2_url,
+over whatever scheme the operator configured (https in production).
+"""
+from typing import Optional
+
+import httpx
+
+from app.config import settings
+from app.internal import credentials
+from app.internal.logger import logger
+
+_TIMEOUT = 15
+
+
+async def fetch_api_key() -> Optional[str]:
+    """Ask C2 for this node's api_key. Returns it, or None when there isn't one
+    to be had yet (not approved, no pickup secret, C2 unreachable, rejected).
+    Never raises — the caller retries on its own backoff."""
+    if not settings.c2_url:
+        logger.warning("Key pickup skipped: C2_URL is not configured.")
+        return None
+    secret = credentials.get_pickup_secret()
+    if not secret:
+        logger.warning(
+            f"Key pickup skipped: no {settings.config_path}/pickup_secret. This node was not enrolled by "
+            "install.sh on this machine — re-run install.sh, or have an admin Reissue key and write it in."
+        )
+        return None
+
+    url = f"{settings.c2_url.rstrip('/')}/nodes/{settings.node_id}/credentials"
+    try:
+        async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
+            r = await client.get(url, headers={"X-Pickup-Secret": secret})
+    except Exception as e:
+        logger.warning(f"Key pickup: could not reach C2 ({e}).")
+        return None
+
+    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."
+        )
+        return None
+    if r.status_code == 404:
+        logger.error("Key pickup: C2 does not know this node (404) — it was deleted server-side. Re-run install.sh.")
+        return None
+    if r.status_code != 200:
+        logger.warning(f"Key pickup: unexpected HTTP {r.status_code} from C2.")
+        return None
+
+    try:
+        body = r.json()
+    except ValueError:
+        logger.warning("Key pickup: C2 returned a non-JSON body.")
+        return None
+    key = body.get("api_key") if isinstance(body, dict) else None
+    if not key or not isinstance(key, str):
+        status = body.get("approval_status") if isinstance(body, dict) else None
+        logger.info(f"Key pickup: no api_key yet (approval_status={status!r}) — waiting for an admin to approve.")
+        return None
+    return key

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

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

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

@@ -6,12 +6,30 @@ from typing import Optional, Callable, Awaitable, Dict, Any
 import paho.mqtt.client as mqtt
 from app.config import settings
 from app.internal.logger import logger
-from app.internal import credentials
+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]]
 
+# 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
+# (134 Bad user name or password, 135 Not authorized).
+_AUTH_FAILURE_CODES = {4, 5, 134, 135}
+
+# Credential-recovery backoff (node-26#14). The pickup endpoint is cheap, but a
+# node that is simply not approved yet must not hammer C2 for days.
+RECOVERY_MIN_DELAY = 30
+RECOVERY_MAX_DELAY = 900
+
+
+def _is_auth_failure(reason_code) -> bool:
+    value = getattr(reason_code, "value", reason_code)
+    if isinstance(value, int) and value in _AUTH_FAILURE_CODES:
+        return True
+    text = str(reason_code).lower()
+    return "not authori" in text or "bad user name or password" in text
+
 
 class MQTTManager:
     def __init__(self):
@@ -19,6 +37,11 @@ class MQTTManager:
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._connected = False
         self._connect_task: Optional[asyncio.Task] = None
+        # The api_key the CURRENT paho client was built with — not necessarily
+        # credentials.get_api_key(), which can move ahead of it (MQTT key push,
+        # install.sh writing the file). Recovery compares against this.
+        self._client_key: Optional[str] = None
+        self._recovery_task: Optional[asyncio.Task] = None
 
         self.on_command: Optional[CommandCallback] = None
         self.on_config_push: Optional[ConfigCallback] = None
@@ -53,6 +76,7 @@ class MQTTManager:
         )
 
         api_key = credentials.get_api_key()
+        self._client_key = api_key
         if api_key:
             # Post-cutover auth: broker's dynsec plugin authenticates this
             # exact (username, password) pair as this node's own client — see
@@ -70,18 +94,14 @@ class MQTTManager:
             client.username_pw_set(settings.mqtt_user, settings.mqtt_pass)
         else:
             # No api_key on disk and no legacy shared login configured. A
-            # dynsec broker (allow_anonymous false) refuses this outright —
+            # dynsec broker (allow_anonymous false) refuses this outright.
-            # expected, not a bug to route around here: this node hasn't been
+            # connect() starts _recover_credentials() in this case, which
-            # enrolled/approved yet, and the enrollment flow that would fix
+            # fetches the key over HTTPS (key_pickup.py) and rebuilds this
-            # that client-side is a later, separate pass (out of scope here;
+            # client with it — node-26#14. paho's reconnect_delay_set() below
-            # see MQTT-PUBLIC-AUTH-PLAN.md). paho's reconnect_delay_set()
+            # bounds the refusal loop meanwhile (2..60s backoff).
-            # below bounds the retry rate (2..60s exponential backoff), so
-            # this degrades to a slow, clearly-logged refusal loop via
-            # _on_connect's "MQTT connect refused" line — not a hot spin.
             logger.warning(
                 "No API key on disk and no legacy MQTT_USER configured — "
-                "connecting without credentials; the broker is expected to "
+                "connecting without credentials while the key is fetched from C2."
-                "refuse this until the node is enrolled/approved."
             )
 
         if settings.mqtt_tls:
@@ -120,6 +140,10 @@ class MQTTManager:
             asyncio.run_coroutine_threadsafe(self._flush_offline_buffer(), self._loop)
         else:
             logger.error(f"MQTT connect refused: {reason_code}")
+            if _is_auth_failure(reason_code) and self._loop is not None:
+                # paho thread -> event loop. A refused login never fixes itself
+                # by retrying the same password, so go get the right one.
+                self._loop.call_soon_threadsafe(self._start_recovery, f"broker refused login ({reason_code})")
 
     def _on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties):
         self._connected = False
@@ -144,6 +168,72 @@ class MQTTManager:
         self._loop = asyncio.get_event_loop()
         self._client = self._build_client()
         self._connect_task = asyncio.create_task(self._connect_with_retry())
+        if not self._client_key and not settings.mqtt_user:
+            self._start_recovery("no API key at startup")
+
+    # ------------------------------------------------------------------
+    # Credential self-heal (node-26#14)
+    #
+    # A node with no key, or a stale one, used to sit in an MQTT refusal loop
+    # forever: key delivery was MQTT-only, and credentials were read once at
+    # startup. Now: re-read credentials.json (install.sh may have written it),
+    # else fetch from C2 over HTTPS with the pickup secret; if that yields a
+    # key different from the one the client is using, rebuild the client with
+    # it in place. No container restart.
+    # ------------------------------------------------------------------
+    def _start_recovery(self, why: str) -> None:
+        if self._recovery_task and not self._recovery_task.done():
+            return
+        logger.warning(f"MQTT credential recovery started: {why}")
+        self._recovery_task = asyncio.create_task(self._recover_credentials())
+
+    async def _recover_credentials(self) -> None:
+        delay = RECOVERY_MIN_DELAY
+        while True:
+            tried = self._client_key
+            key = credentials.reload_api_key()
+            if not key or key == tried:
+                fetched = await key_pickup.fetch_api_key()
+                if fetched and fetched != tried:
+                    credentials.save_api_key(fetched)
+                    key = fetched
+            if key and key != tried:
+                logger.info("MQTT credential recovery: new API key in hand — reconnecting with it.")
+                await self._rebuild_client()
+                return
+            if self._connected:
+                return
+            if key and key == tried:
+                logger.error(
+                    "MQTT credential recovery: C2 returns the same key the broker refuses. The broker's login for "
+                    "this node is out of sync with C2 (dynsec) — an admin re-approve re-syncs it without rotating."
+                )
+            logger.info(f"MQTT credential recovery: retrying in {delay}s.")
+            await asyncio.sleep(delay)
+            delay = min(delay * 2, RECOVERY_MAX_DELAY)
+
+    async def _rebuild_client(self) -> None:
+        """Swap the paho client for one built from the current credentials."""
+        old = self._client
+        if self._connect_task and not self._connect_task.done():
+            self._connect_task.cancel()
+        if old is not None:
+            # loop_stop() joins paho's thread, which can be mid-backoff — off the event loop.
+            await asyncio.to_thread(old.loop_stop)
+            try:
+                old.disconnect()
+            except Exception:
+                pass
+        self._connected = False
+        self._client = self._build_client()
+        self._connect_task = asyncio.create_task(self._connect_with_retry())
+
+    def update_api_key(self, key: str) -> None:
+        """A key arrived while connected (legacy MQTT push / reissue): use it on
+        the next reconnect without dropping the current, working session."""
+        if self._client is not None and key != self._client_key:
+            self._client.username_pw_set(settings.node_id, key)
+            self._client_key = key
 
     async def _connect_with_retry(self):
         """Attempt the initial TCP connect, retrying with backoff until it succeeds."""
@@ -161,6 +251,8 @@ class MQTTManager:
                 delay = min(delay * 2, 60)
 
     async def disconnect(self):
+        if self._recovery_task:
+            self._recovery_task.cancel()
         if self._connect_task:
             self._connect_task.cancel()
         if self._client:
@@ -235,7 +327,18 @@ class MQTTManager:
             "secondary_sdr_priority": config.secondary_sdr_priority,
         }
         payload["sdr_pins"] = config.sdr_pins
+        # Conventional channels (node-26#24): name/freq/NAC and whether the
+        # agency encrypts them — the one decode failure nobody can fix on the
+        # node, so C2 has to be told. Empty list on a trunked system.
+        from app.internal.metadata_watcher import metadata_watcher
+        payload["radio_channels"] = metadata_watcher.conventional_channels
         secondary = await secondary_sdr_client.status()
+        # Unreachable secondary-sdr: say so with explicit nulls. C2 only
+        # overwrites keys present in the checkin, so omitting them would leave
+        # a dead decoder showing "Running" forever.
+        payload["secondary_sdr_running"] = None
+        payload["sdr_devices"] = None
+        payload["op25_sdr_serial"] = None
         if secondary is not None:
             payload["secondary_sdr_running"] = [r["mode"] for r in secondary.get("running", [])]
             if secondary.get("devices") is not None:

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

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

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

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

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

@@ -0,0 +1,41 @@
+"""
+Hand an MQTT `node_update` command to the host-level updater (node-26#15).
+
+The edge-node container has no business updating its own host: it writes a
+sentinel file into /configs (bind-mounted from <install dir>/configs), and the
+host's systemd drb-update.path unit — installed by install.sh — triggers
+drb-update.service, which runs Client/scripts/drb-update.sh. The host treats
+the file as a bare "please update" flag and never reads its content as input,
+so nothing C2 sends can pick the ref, image or command that runs.
+"""
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+
+from app.config import settings
+from app.internal.logger import logger
+
+SENTINEL_NAME = "update-requested"
+
+
+def sentinel_path() -> Path:
+    return Path(settings.config_path) / SENTINEL_NAME
+
+
+def request_host_update() -> bool:
+    """Drop the sentinel. Returns False if it could not be written."""
+    path = sentinel_path()
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = path.with_name(SENTINEL_NAME + ".tmp")
+        tmp.write_text(json.dumps({"requested_at": datetime.now(timezone.utc).isoformat()}))
+        os.replace(tmp, path)
+    except OSError as e:
+        logger.error(f"Node update requested but the sentinel could not be written ({e}).")
+        return False
+    logger.info(
+        f"Node update requested — handed to the host updater via {path}. If nothing happens within a minute, "
+        "this node predates the updater: re-run install.sh on it once."
+    )
+    return True

drb-edge-node/app/main.py

@@ -170,17 +170,15 @@ async def on_command(payload: dict):
         await asyncio.sleep(2)
         await op25_client.start()
     elif action == "node_update":
-        # TODO: Full OTA update — register a host-level systemd service (e.g. drb-update.service)
+        # node-26#15: the container cannot (and must not) update itself. It
-        # that stops all DRB containers, runs `docker compose pull`, then `docker compose up -d`.
+        # drops a sentinel in /configs; the host's drb-update.path unit
-        # The C2 server triggers it by sending this MQTT command; the host service watches for the
+        # (installed by install.sh) sees it and runs drb-update.service, which
-        # restart signal (e.g. via a Unix socket, a sentinel file, or a lightweight webhook).
+        # pulls the node's own pinned repo ref + images and recreates the
-        # Not implemented yet — for now, just restart the container so any pre-pulled image
+        # stack. The sentinel's content is ignored by the host — C2 cannot
-        # is picked up (requires a prior `docker compose pull` on the host).
+        # choose what gets installed. Offline/locked-out nodes get the same
-        logger.info("Node update requested — restarting container to pick up latest image.")
+        # request over HTTPS via the updater's timer instead.
-        await mqtt_manager.publish_status("offline")
+        from app.internal.update_request import request_host_update
-        await asyncio.sleep(1)
+        request_host_update()
-        import os
-        os._exit(0)  # Docker restart=unless-stopped will bring the container back up
     else:
         logger.warning(f"Unknown command: {action}")
 
@@ -189,39 +187,34 @@ async def on_api_key(payload: dict):
     key = payload.get("api_key")
     if key:
         credentials.save_api_key(key)
+        mqtt_manager.update_api_key(key)
         logger.info("Node API key received and saved.")
 
 
-def _to_hz(freq) -> int:
-    """Convert a frequency to Hz. Accepts MHz floats (< 1e6) or Hz ints."""
-    f = float(freq)
-    return int(f * 1_000_000) if f < 1_000_000 else int(f)
-
-
 async def _generate_op25_config(config: SystemConfig) -> bool:
-    """Translate a SystemConfig (Firestore format) into OP25 active.cfg.json + op25.liq."""
+    """Translate a SystemConfig (Firestore format) into OP25 active.cfg.json + op25.liq.
+
+    Trunked / conventional / analog are told apart in radio_config (node-26#24);
+    a system saved before that has no `mode` and is trunked, as it always was.
+    """
     from app.internal.op25_client import op25_client
+    from app.internal.radio_config import build_generate_payload
     node_cfg = load_node_config()
-    raw = config.config
+    try:
-    payload = {
+        payload = build_generate_payload(
-        "type": config.type,
+            config.type, config.name, config.config,
-        "systemName": config.name,
+            icecast={
-        "channels": [_to_hz(ch) for ch in raw.get("control_channels", [])],
+                "icecast_host": settings.icecast_host,
-        "tags": [
+                "icecast_port": settings.icecast_port,
-            {"talkgroup": str(tg.get("name", "")), "tagDec": int(tg["id"])}
+                "icecast_mountpoint": settings.icecast_mount,
-            for tg in raw.get("talkgroups", [])
+                "icecast_password": settings.icecast_source_password,
-            if tg.get("id") is not None
+            },
-        ],
+            hardware_preset=node_cfg.hardware_preset,
-        "whitelist": [int(tg["id"]) for tg in raw.get("talkgroups", []) if tg.get("id") is not None],
+            ppm_override=node_cfg.ppm_override,
-        "icecastConfig": {
+        )
-            "icecast_host": settings.icecast_host,
+    except ValueError as e:
-            "icecast_port": settings.icecast_port,
+        logger.error(f"System {config.name!r} has an unusable channel list: {e}")
-            "icecast_mountpoint": settings.icecast_mount,
+        return False
-            "icecast_password": settings.icecast_source_password,
-        },
-        "hardware_preset": node_cfg.hardware_preset,
-        **({"ppm_override": node_cfg.ppm_override} if node_cfg.ppm_override is not None else {}),
-    }
     return await op25_client.generate_config(payload)
 
 
@@ -327,9 +320,12 @@ async def lifespan(app: FastAPI):
         if (node_cfg.override_system_id and node_cfg.override_config)
         else node_cfg.system_config
     )
-    if node_cfg.configured and active_config:
+    # Not gated on `configured` (set only by a C2 assignment): a standalone or
+    # not-yet-approved node listening to a system chosen on the local
+    # dashboard (an override) must come back on it after a reboot too.
+    if active_config:
         from app.internal.op25_client import op25_client
-        logger.info("Node is configured — waiting for OP25 API then generating config.")
+        logger.info("Node has a system to listen to — waiting for OP25 API then generating config.")
         for attempt in range(10):
             if await _generate_op25_config(active_config):
                 await op25_client.start()

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

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

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

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

drb-edge-node/tests/test_conventional.py

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

drb-edge-node/tests/test_credential_recovery.py

@@ -0,0 +1,205 @@
+"""
+node-26#14 — a node with no key, or a stale key, recovers on its own:
+re-read credentials.json, else fetch over HTTPS with the pickup secret, then
+rebuild the MQTT client in place. node-26#15 — node_update hands off to the
+host updater via a sentinel file.
+"""
+import asyncio
+import json
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+
+from app.config import settings
+from app.internal import credentials, key_pickup, mqtt_manager as mm_mod, update_request
+from app.internal.mqtt_manager import MQTTManager, _is_auth_failure
+
+
+@pytest.fixture(autouse=True)
+def isolated(tmp_path, monkeypatch):
+    monkeypatch.setattr(credentials, "_CREDS_FILE", tmp_path / "credentials.json")
+    monkeypatch.setattr(credentials, "_PICKUP_FILE", tmp_path / "pickup_secret")
+    monkeypatch.setattr(credentials, "_api_key", None)
+    monkeypatch.setattr(settings, "config_path", str(tmp_path))
+    monkeypatch.setattr(settings, "c2_url", "https://c2.example")
+    monkeypatch.setattr(settings, "mqtt_user", None)
+    monkeypatch.setattr(mm_mod, "RECOVERY_MIN_DELAY", 0)
+    yield tmp_path
+
+
+def _manager(client_key=None):
+    m = MQTTManager()
+    m._client_key = client_key
+    m._rebuild_client = AsyncMock()
+    return m
+
+
+# --- credentials.py -----------------------------------------------------------
+
+def test_reload_picks_up_a_key_written_after_startup(isolated):
+    credentials.load()
+    assert credentials.get_api_key() is None
+    data = json.loads((isolated / "credentials.json").read_text())
+    data["api_key"] = "written-by-install-sh"
+    (isolated / "credentials.json").write_text(json.dumps(data))
+    assert credentials.reload_api_key() == "written-by-install-sh"
+
+
+def test_write_is_atomic_and_private(isolated):
+    credentials.save_api_key("k1")
+    f = isolated / "credentials.json"
+    assert json.loads(f.read_text())["api_key"] == "k1"
+    assert (f.stat().st_mode & 0o777) == 0o600
+    assert not (isolated / "credentials.json.tmp").exists()
+
+
+def test_pickup_secret_read_and_absent(isolated):
+    assert credentials.get_pickup_secret() is None
+    (isolated / "pickup_secret").write_text("abc123\n")
+    assert credentials.get_pickup_secret() == "abc123"
+
+
+# --- auth-failure classification -------------------------------------------------
+
+@pytest.mark.parametrize("code", [4, 5, 134, 135])
+def test_auth_failure_codes(code):
+    assert _is_auth_failure(code)
+
+
+def test_non_auth_failures_do_not_trigger_recovery():
+    assert not _is_auth_failure(3)  # server unavailable
+    assert not _is_auth_failure(136)
+
+
+def test_paho_reason_code_objects_classified():
+    from paho.mqtt.reasoncodes import ReasonCode
+    from paho.mqtt.packettypes import PacketTypes
+    assert _is_auth_failure(ReasonCode(PacketTypes.CONNACK, "Not authorized"))
+    assert _is_auth_failure(ReasonCode(PacketTypes.CONNACK, "Bad user name or password"))
+    assert not _is_auth_failure(ReasonCode(PacketTypes.CONNACK, "Server unavailable"))
+
+
+# --- recovery loop ----------------------------------------------------------------
+
+async def test_recovery_uses_key_already_on_disk_without_calling_c2(isolated, monkeypatch):
+    credentials.save_api_key("disk-key")
+    monkeypatch.setattr(credentials, "_api_key", None)  # process started before the write
+    fetch = AsyncMock()
+    monkeypatch.setattr(key_pickup, "fetch_api_key", fetch)
+    m = _manager(client_key=None)
+    await m._recover_credentials()
+    m._rebuild_client.assert_awaited_once()
+    fetch.assert_not_awaited()
+
+
+async def test_recovery_fetches_over_https_and_persists(isolated, monkeypatch):
+    monkeypatch.setattr(key_pickup, "fetch_api_key", AsyncMock(return_value="fresh-key"))
+    m = _manager(client_key="stale-key")
+    monkeypatch.setattr(credentials, "_api_key", "stale-key")
+    await m._recover_credentials()
+    m._rebuild_client.assert_awaited_once()
+    assert credentials.get_api_key() == "fresh-key"
+    assert json.loads((isolated / "credentials.json").read_text())["api_key"] == "fresh-key"
+
+
+async def test_recovery_backs_off_until_approved(isolated, monkeypatch):
+    fetch = AsyncMock(side_effect=[None, None, "approved-key"])
+    monkeypatch.setattr(key_pickup, "fetch_api_key", fetch)
+    m = _manager()
+    await asyncio.wait_for(m._recover_credentials(), timeout=5)
+    assert fetch.await_count == 3
+    m._rebuild_client.assert_awaited_once()
+
+
+async def test_recovery_stops_if_connection_recovers_meanwhile(isolated, monkeypatch):
+    monkeypatch.setattr(key_pickup, "fetch_api_key", AsyncMock(return_value=None))
+    m = _manager()
+    m._connected = True
+    await asyncio.wait_for(m._recover_credentials(), timeout=5)
+    m._rebuild_client.assert_not_awaited()
+
+
+async def test_start_recovery_is_single_flight(isolated, monkeypatch):
+    gate = asyncio.Event()
+
+    async def slow_fetch():
+        await gate.wait()
+        return "k"
+    monkeypatch.setattr(key_pickup, "fetch_api_key", slow_fetch)
+    m = _manager()
+    m._start_recovery("a")
+    first = m._recovery_task
+    m._start_recovery("b")
+    assert m._recovery_task is first
+    gate.set()
+    await first
+
+
+async def test_connect_without_key_starts_recovery(isolated, monkeypatch):
+    m = MQTTManager()
+    started = []
+    monkeypatch.setattr(m, "_connect_with_retry", AsyncMock())
+    monkeypatch.setattr(m, "_start_recovery", lambda why: started.append(why))
+    await m.connect()
+    assert started == ["no API key at startup"]
+
+
+def test_update_api_key_changes_creds_for_next_reconnect(isolated):
+    m = MQTTManager()
+    m._client = m._build_client()
+    m.update_api_key("rotated")
+    assert m._client._password == b"rotated"
+    assert m._client_key == "rotated"
+
+
+# --- key_pickup.fetch_api_key -----------------------------------------------------
+
+def _mock_http(monkeypatch, handler):
+    real = httpx.AsyncClient
+
+    def factory(*a, **kw):
+        kw["transport"] = httpx.MockTransport(handler)
+        return real(*a, **kw)
+    monkeypatch.setattr(key_pickup.httpx, "AsyncClient", factory)
+
+
+async def test_fetch_sends_pickup_secret_and_returns_key(isolated, monkeypatch):
+    (isolated / "pickup_secret").write_text("the-secret")
+    seen = {}
+
+    def handler(request):
+        seen["url"] = str(request.url)
+        seen["secret"] = request.headers.get("X-Pickup-Secret")
+        return httpx.Response(200, json={"approval_status": "approved", "api_key": "K"})
+    _mock_http(monkeypatch, handler)
+    assert await key_pickup.fetch_api_key() == "K"
+    assert seen == {"url": f"https://c2.example/nodes/{settings.node_id}/credentials", "secret": "the-secret"}
+
+
+@pytest.mark.parametrize("resp", [
+    httpx.Response(200, json={"approval_status": "pending", "api_key": None}),
+    httpx.Response(401, json={"detail": "Invalid pickup secret"}),
+    httpx.Response(404),
+    httpx.Response(500),
+    httpx.Response(200, text="not json"),
+])
+async def test_fetch_returns_none_when_no_key(isolated, monkeypatch, resp):
+    (isolated / "pickup_secret").write_text("s")
+    _mock_http(monkeypatch, lambda request: resp)
+    assert await key_pickup.fetch_api_key() is None
+
+
+async def test_fetch_without_pickup_secret_does_not_call_c2(isolated, monkeypatch):
+    def handler(request):
+        raise AssertionError("must not call C2 without a pickup secret")
+    _mock_http(monkeypatch, handler)
+    assert await key_pickup.fetch_api_key() is None
+
+
+# --- node_update -> host sentinel ---------------------------------------------------
+
+def test_node_update_writes_sentinel(isolated):
+    assert update_request.request_host_update() is True
+    data = json.loads((isolated / "update-requested").read_text())
+    assert "requested_at" in data

drb-edge-node/tests/test_host_updater.py

@@ -0,0 +1,179 @@
+"""
+node-26#15 — Client/scripts/drb-update.sh, the host-level updater.
+
+Runs the real script with stub git/docker/curl on PATH, so what is checked is
+the script's decision logic and its security guards: it only updates when
+asked (MQTT sentinel or a new C2 request), never takes anything but "update
+now" from C2, refuses a mismatched git origin or registry, and keeps node
+credentials off curl's argv.
+"""
+import json
+import os
+import shutil
+import stat
+import subprocess
+from pathlib import Path
+
+import pytest
+
+SCRIPT = Path(__file__).resolve().parents[2] / "scripts" / "drb-update.sh"
+
+pytestmark = pytest.mark.skipif(
+    not (SCRIPT.exists() and shutil.which("bash") and shutil.which("jq") and shutil.which("flock")),
+    reason="needs the repo's scripts/ (absent inside the image) plus bash, jq and flock",
+)
+
+REPO = "https://git.example.net/logan/node-26.git"
+OLD, NEW = "a" * 40, "b" * 40
+
+STUB_GIT = r'''#!/usr/bin/env bash
+echo "git $*" >> "$STUB/calls"
+while [ "$1" = -c ]; do shift 2; done
+case "$1" in
+  rev-parse) if [ "$2" = HEAD ]; then cat "$STUB/head"; else echo "$STUB_NEW"; fi ;;
+  remote) echo "$STUB_ORIGIN" ;;
+  fetch) exit 0 ;;
+  show-ref) [ "$4" = "refs/tags/v1" ] ;;
+  checkout) echo "${@: -1}" > "$STUB/head" ;;
+  *) exit 0 ;;
+esac
+'''
+STUB_DOCKER = '#!/usr/bin/env bash\necho "docker $*" >> "$STUB/calls"\n'
+STUB_CURL = r'''#!/usr/bin/env bash
+echo "curl $*" >> "$STUB/calls"
+for a in "$@"; do case "$a" in @*) cat "${a#@}" >> "$STUB/headers" ;; esac; done
+cat "$STUB/response"
+'''
+
+
+def _exe(path: Path, body: str):
+    path.write_text(body)
+    path.chmod(path.stat().st_mode | stat.S_IEXEC)
+
+
+@pytest.fixture
+def env(tmp_path):
+    stub = tmp_path / "stub"
+    bindir = stub / "bin"
+    bindir.mkdir(parents=True)
+    _exe(bindir / "git", STUB_GIT)
+    _exe(bindir / "docker", STUB_DOCKER)
+    _exe(bindir / "curl", STUB_CURL)
+    (stub / "head").write_text(OLD + "\n")
+    (stub / "calls").write_text("")
+    (stub / "response").write_text(json.dumps({"update_requested_at": None}))
+
+    inst = tmp_path / "node"
+    (inst / ".git").mkdir(parents=True)
+    (inst / "configs").mkdir()
+    (inst / ".env").write_text(
+        "NODE_ID=node-7\nC2_URL=https://api.example.net\nIMAGE_REGISTRY=git.example.net\n"
+    )
+    (inst / "configs" / "credentials.json").write_text(json.dumps({"api_key": "k" * 64}))
+    (inst / "configs" / "pickup_secret").write_text("p" * 48)
+
+    conf = tmp_path / "updater.conf"
+    conf.write_text(f"INSTALL_DIR={inst}\nREPO_URL={REPO}\nREF=v1\nREGISTRY=git.example.net\nBUILD=0\n")
+
+    e = dict(os.environ)
+    e.update({
+        "PATH": f"{bindir}:{os.environ['PATH']}",
+        "STUB": str(stub), "STUB_NEW": NEW, "STUB_ORIGIN": REPO,
+        "DRB_UPDATER_CONF": str(conf),
+        "DRB_UPDATER_STATE": str(tmp_path / "state"),
+        "DRB_UPDATER_LOCK": str(tmp_path / "lock"),
+    })
+    return {"env": e, "stub": stub, "inst": inst, "conf": conf, "state": tmp_path / "state"}
+
+
+def _run(env):
+    return subprocess.run(["bash", str(SCRIPT)], env=env["env"], capture_output=True, text=True, timeout=30)
+
+
+def _calls(env):
+    return (env["stub"] / "calls").read_text()
+
+
+def test_no_request_means_checkin_only(env):
+    r = _run(env)
+    assert r.returncode == 0, r.stderr
+    calls = _calls(env)
+    assert "update-check" in calls
+    assert "docker" not in calls
+    assert "fetch" not in calls
+
+
+def test_c2_request_updates_once_and_reports(env):
+    (env["stub"] / "response").write_text(json.dumps({"update_requested_at": "2026-10-04T12:00:00+00:00"}))
+    r = _run(env)
+    assert r.returncode == 0, r.stderr + r.stdout
+    calls = _calls(env)
+    assert "docker compose pull" in 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
+    assert '"result":"ok"' in calls
+    # Same request again: nothing to do.
+    (env["stub"] / "calls").write_text("")
+    assert _run(env).returncode == 0
+    assert "docker" not in _calls(env)
+
+
+def test_mqtt_sentinel_triggers_update_and_is_consumed(env):
+    sentinel = env["inst"] / "configs" / "update-requested"
+    sentinel.write_text('{"ref": "evil", "cmd": "rm -rf /"}')
+    r = _run(env)
+    assert r.returncode == 0, r.stderr
+    assert not sentinel.exists()
+    calls = _calls(env)
+    assert "docker compose pull" in calls
+    assert "evil" not in calls and "rm -rf" not in calls
+
+
+def test_c2_response_is_never_used_as_a_ref_or_command(env):
+    (env["stub"] / "response").write_text(json.dumps({"update_requested_at": "$(touch /tmp/pwned); main"}))
+    r = _run(env)
+    assert r.returncode == 0
+    assert "docker" not in _calls(env)  # malformed request is ignored, not executed
+
+
+def test_refuses_when_git_origin_differs(env):
+    env["env"]["STUB_ORIGIN"] = "https://attacker.example/node-26.git"
+    (env["inst"] / "configs" / "update-requested").write_text("")
+    r = _run(env)
+    assert r.returncode != 0
+    assert "refusing" in r.stderr
+    assert "docker" not in _calls(env)
+
+
+def test_refuses_when_env_registry_differs(env):
+    (env["inst"] / ".env").write_text(
+        "NODE_ID=node-7\nC2_URL=https://api.example.net\nIMAGE_REGISTRY=evil.example\n"
+    )
+    (env["inst"] / "configs" / "update-requested").write_text("")
+    r = _run(env)
+    assert r.returncode != 0
+    assert "docker compose pull" not in _calls(env)
+
+
+def test_non_https_repo_in_conf_is_rejected(env):
+    env["conf"].write_text(env["conf"].read_text().replace("https://", "http://"))
+    r = _run(env)
+    assert r.returncode != 0
+    assert "git" not in _calls(env)
+
+
+def test_credentials_go_in_a_header_file_never_argv(env):
+    _run(env)
+    calls = _calls(env)
+    assert "k" * 64 not in calls and "p" * 48 not in calls
+    headers = (env["stub"] / "headers").read_text()
+    assert f"Authorization: Bearer {'k' * 64}" in headers
+    assert f"X-Pickup-Secret: {'p' * 48}" in headers
+
+
+def test_plain_http_c2_skips_checkin(env):
+    (env["inst"] / ".env").write_text("NODE_ID=node-7\nC2_URL=http://api.example.net\nIMAGE_REGISTRY=git.example.net\n")
+    r = _run(env)
+    assert r.returncode == 0
+    assert "curl" not in _calls(env)

drb-edge-node/tests/test_local_dashboard.py

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

install.sh

@@ -17,9 +17,13 @@
 #   2. Install Docker + compose plugin + git + curl + jq
 #   3. Clone logan/node-26 at a PINNED ref into $INSTALL_DIR
 #   4. Write .env (non-interactive from flags/env, interactive fallback)
-#   5. Enroll with C2 (POST /nodes/enroll) and poll for the api_key
+#   5. Enroll with C2 (POST /nodes/enroll), then BLOCK until an admin approves
-#   6. docker compose pull && up -d   (prebuilt; --build opts into the ~1h build)
+#      and the api_key is written (node-26#14). Ctrl-C safe: re-run resumes
-#   7. Print the approval step
+#      from configs/pickup_secret. A re-run also refreshes a stale key.
+#   6. docker compose pull && up -d, ALWAYS recreating edge-node so it loads
+#      the key on disk  (prebuilt; --build opts into the ~1h build)
+#   7. Install the host updater (systemd timer + path unit, node-26#15)
+#   8. Print a summary
 #
 # It does NOT set up WireGuard. WireGuard-per-node was evaluated and rejected
 # (Server/MQTT-PUBLIC-AUTH-PLAN.md, Server/infra/main.tf:67-73). A field node
@@ -85,7 +89,14 @@ REGISTRY_PASS="${DRB_REGISTRY_PASS:-}"
 DO_BUILD=0        # 0 = pull prebuilt images (default), 1 = build on the Pi (~1h for op25)
 DO_START=1
 ASSUME_YES=0
-ENROLL_WAIT="${DRB_ENROLL_WAIT:-0}"   # seconds to block waiting for admin approval; 0 = don't block
+# How long to block waiting for admin approval. Default: forever (node-26#14 —
+# the old default of 0 started a keyless stack that never came online).
+# N>0 = give up after N seconds and start anyway; 0 = don't wait at all. Either
+# way the edge-node app keeps fetching the key itself and comes online on its
+# own once approved.
+ENROLL_WAIT="${DRB_ENROLL_WAIT:-forever}"
+APP_URL="${DRB_APP_URL:-}"
+INSTALL_UPDATER=1
 
 C='\033[0;36m'; G='\033[0;32m'; Y='\033[1;33m'; R='\033[0;31m'; N='\033[0m'
 say()  { printf "${C}==>${N} %s\n" "$*"; }
@@ -111,7 +122,12 @@ Usage: install.sh [options]
   --dir PATH           install location (default /opt/drb/node-26)
   --build              build images locally instead of pulling (~1h for op25)
   --no-start           configure and enroll, but do not start containers
-  --wait-approval SEC  block up to SEC seconds polling for admin approval
+  --wait-approval SEC  stop waiting for admin approval after SEC seconds
+                       (default: wait forever; 0 = don't wait). The node
+                       still fetches its key by itself once approved.
+  --app-url URL        admin UI base for the approval message
+                       (default: C2 URL without its "api." prefix)
+  --no-updater         don't install the host updater (no remote updates)
   -y, --yes            never prompt; fail instead of asking
 
 Every option also has an env var: DRB_NODE_ID, DRB_C2_URL, DRB_ENROLLMENT_TOKEN,
@@ -138,12 +154,17 @@ while [ $# -gt 0 ]; do
     --build)          DO_BUILD=1; shift ;;
     --no-start)       DO_START=0; shift ;;
     --wait-approval)  ENROLL_WAIT="$2"; shift 2 ;;
+    --app-url)        APP_URL="$2"; shift 2 ;;
+    --no-updater)     INSTALL_UPDATER=0; shift ;;
     -y|--yes)         ASSUME_YES=1; shift ;;
     -h|--help)        usage; exit 0 ;;
     *)                die "unknown option: $1 (try --help)" ;;
   esac
 done
 
+[ "$ENROLL_WAIT" = forever ] || [[ "$ENROLL_WAIT" =~ ^[0-9]+$ ]] \
+  || die "--wait-approval takes a number of seconds (or omit it to wait until approved)"
+
 # Prompts must come from the terminal, not from the `curl |` pipe on stdin.
 ask() {  # ask VAR "prompt" "default"
   local __v="$1" __p="$2" __d="${3:-}" __r=""
@@ -226,7 +247,9 @@ mkdir -p "$(dirname "$INSTALL_DIR")"
 if [ -d "$INSTALL_DIR/.git" ]; then
   ok "existing install at $INSTALL_DIR — updating in place (.env is preserved)"
   git -C "$INSTALL_DIR" remote set-url origin "$REPO_URL"
-  git -C "$INSTALL_DIR" fetch --tags --prune origin
+  # --force: a moved release tag (v1 re-cut) must replace the local copy,
+  # otherwise fetch refuses with "would clobber existing tag" and set -e dies.
+  git -C "$INSTALL_DIR" fetch --force --tags --prune origin
 else
   git clone --no-checkout "$REPO_URL" "$INSTALL_DIR"
 fi
@@ -345,39 +368,68 @@ PICKUP_FILE="$INSTALL_DIR/configs/pickup_secret"
 # GET /nodes/{id}/credentials. Sets CRED_HTTP + CRED_BODY (no -f: we need the
 # body and status on a 4xx). One implementation so first-run and re-run agree.
 creds_pickup() {  # creds_pickup PICKUP_SECRET
-  local _tmp; _tmp="$(mktemp)"
+  local _tmp _hdr; _tmp="$(mktemp)"; _hdr="$(mktemp)"; chmod 600 "$_hdr"
+  # Secret via a 0600 header file, never curl's argv (visible in `ps`).
+  printf 'X-Pickup-Secret: %s\n' "$1" > "$_hdr"
   CRED_HTTP="$(curl -sS -o "$_tmp" -w '%{http_code}' \
-      "$C2_URL/nodes/$NODE_ID/credentials" -H "X-Pickup-Secret: $1" 2>/dev/null || echo 000)"
+      "$C2_URL/nodes/$NODE_ID/credentials" -H "@$_hdr" 2>/dev/null || echo 000)"
   CRED_BODY="$(cat "$_tmp" 2>/dev/null || true)"
-  rm -f "$_tmp"
+  rm -f "$_tmp" "$_hdr"
 }
 cred_field() { printf '%s' "${CRED_BODY:-}" | jq -r "$1 // empty" 2>/dev/null || true; }
 
+KEY_WRITTEN=0
+# Merge into the existing file: the edge-node app keeps its dashboard auth
+# salt + session secret in the same JSON, and clobbering them logs everyone
+# out. Atomic (tmp + mv) so a Ctrl-C can never leave half a file.
 write_creds() {  # write_creds API_KEY
-  umask 077; jq -n --arg k "$1" '{api_key:$k}' > "$CREDS"; umask 022
+  local _tmp="$CREDS.tmp"
-  ok "api_key received and written to configs/credentials.json"
+  umask 077
+  if [ -s "$CREDS" ] && jq -e . "$CREDS" >/dev/null 2>&1; then
+    jq --arg k "$1" '.api_key = $k' "$CREDS" > "$_tmp"
+  else
+    jq -n --arg k "$1" '{api_key:$k}' > "$_tmp"
+  fi
+  umask 022
+  mv -f "$_tmp" "$CREDS"
+  KEY_WRITTEN=1
+  ok "api_key written to configs/credentials.json"
 }
 
+[ -n "$APP_URL" ] || { APP_URL="${C2_URL/:\/\/api./://}"; }
+APP_URL="${APP_URL%/}"
+
 say_pending() {  # say_pending APPROVAL_STATUS  — not an error: node is enrolled, key not minted yet
   warn "not approved yet — C2 reports approval_status=${1:-pending}, no api_key minted."
-  warn "an admin must, at  <app-url>/settings/nodes :  Approve '$NODE_ID'  (assigning a system is separate)."
+  warn "an admin must approve '$NODE_ID' at ${APP_URL}/settings/nodes. The node keeps fetching"
-  warn "then re-run this installer — it reuses configs/pickup_secret — or fetch it directly:"
+  warn "its key by itself and comes online once approved; re-running this installer also works."
-  warn "  curl -fsS $C2_URL/nodes/$NODE_ID/credentials -H \"X-Pickup-Secret: \$(cat $PICKUP_FILE)\" | jq -r .api_key"
 }
 
-# Poll creds_pickup for up to ENROLL_WAIT seconds while still pending. Result
+# Block until the node is approved and C2 hands back an api_key (node-26#14).
-# left in CRED_HTTP/CRED_BODY. --wait-approval is what would have avoided the
+# Result left in CRED_HTTP/CRED_BODY. Returns 0 with a key, 1 on timeout/0-wait,
-# original prod bug; the default (0) does not block, so the re-run path below
+# dies on a credential error (401/404) that waiting cannot fix.
-# must stand on its own.
 wait_for_key() {  # wait_for_key PICKUP_SECRET
-  [ "${ENROLL_WAIT:-0}" -gt 0 ] || return 0
+  [ "$ENROLL_WAIT" = 0 ] && return 1
-  local _end; _end=$(( $(date +%s) + ENROLL_WAIT ))
+  local _end=0 _shown=0 _delay=5
-  say "Waiting up to ${ENROLL_WAIT}s for an admin to approve '$NODE_ID'"
+  [ "$ENROLL_WAIT" = forever ] || _end=$(( $(date +%s) + ENROLL_WAIT ))
-  while [ "$(date +%s)" -lt "$_end" ]; do
+  trap 'printf "\n"; warn "interrupted — nothing is lost: re-run the same command to resume waiting (configs/pickup_secret is saved)."; exit 130' INT TERM
-    sleep 10
+  while :; do
+    case "$CRED_HTTP" in
+      200) [ -n "$(cred_field '.api_key')" ] && { trap - INT TERM; return 0; } ;;
+      401) die "C2 rejected the saved pickup secret (401) while waiting — re-run with --token DRB-… to re-enroll." ;;
+      404) die "C2 no longer knows node '$NODE_ID' (404) — it was deleted. Re-run with --token DRB-… to enroll again." ;;
+      *)   : ;;  # 000/5xx: transient, keep waiting
+    esac
+    if [ "$_shown" = 0 ]; then
+      printf "\n${Y}Please have an admin approve node '%s' at %s/settings/nodes to continue.${N}\n" \
+        "$NODE_ID" "$APP_URL"
+      printf "Waiting for approval%s… (Ctrl-C is safe — re-run the same command to resume)\n\n" \
+        "$([ "$_end" = 0 ] && echo "" || echo " for up to ${ENROLL_WAIT}s")"
+      _shown=1
+    fi
+    if [ "$_end" != 0 ] && [ "$(date +%s)" -ge "$_end" ]; then trap - INT TERM; return 1; fi
+    sleep "$_delay"; [ "$_delay" -lt 30 ] && _delay=$(( _delay + 5 ))
     creds_pickup "$1"
-    [ "$CRED_HTTP" = 200 ] || return 0
-    [ -z "$(cred_field '.api_key')" ] || return 0
   done
 }
 
@@ -390,12 +442,15 @@ do_fresh_enroll() {
   BODY="$(jq -nc --arg id "$NODE_ID" --arg n "${NODE_NAME:-$NODE_ID}" \
             --argjson lat "${NODE_LAT:-0}" --argjson lon "${NODE_LON:-0}" \
             '{node_id:$id,name:$n,lat:$lat,lon:$lon}')"
-  # Token goes in a header from a shell variable — never on a process command
+  # Token goes in a 0600 header file — never on a process command line (curl's
-  # line, never echoed.
+  # argv is visible in `ps`; CISO condition, server-26#61), never echoed.
+  local _ehdr; _ehdr="$(mktemp)"; chmod 600 "$_ehdr"
+  printf 'X-Enrollment-Token: %s\n' "$ENROLLMENT_TOKEN" > "$_ehdr"
   if ! RESP="$(curl -fsS -X POST "$C2_URL/nodes/enroll" \
             -H "Content-Type: application/json" \
-            -H "X-Enrollment-Token: $ENROLLMENT_TOKEN" \
+            -H "@$_ehdr" \
             --data "$BODY" 2>&1)"; then
+    rm -f "$_ehdr"
     case "$RESP" in
       *403*) die "enroll refused (403): node '$NODE_ID' is already approved on C2, and
   this machine has no configs/pickup_secret to collect its key with. A token
@@ -409,6 +464,7 @@ do_fresh_enroll() {
       *)     die "enroll failed: $RESP" ;;
     esac
   fi
+  rm -f "$_ehdr"
   PICKUP="$(printf '%s' "$RESP" | jq -r '.pickup_secret')"
   STATUS="$(printf '%s' "$RESP" | jq -r '.approval_status')"
   [ -n "$PICKUP" ] && [ "$PICKUP" != null ] || die "enroll returned no pickup_secret: $RESP"
@@ -418,21 +474,32 @@ do_fresh_enroll() {
 
   creds_pickup "$PICKUP"
   [ "$CRED_HTTP" = 200 ] || die "post-enroll credential fetch failed (HTTP $CRED_HTTP): ${CRED_BODY:-<no body>}"
-  KEY="$(cred_field '.api_key')"
+  if wait_for_key "$PICKUP"; then
-  if [ -z "$KEY" ]; then
+    write_creds "$(cred_field '.api_key')"
-    wait_for_key "$PICKUP" || true
-    KEY="$(cred_field '.api_key')"
-  fi
-  if [ -n "$KEY" ]; then
-    write_creds "$KEY"
   else
     say_pending "$(cred_field '.approval_status')"
   fi
 }
 
-if [ -s "$CREDS" ] && jq -e '.api_key // empty' "$CREDS" >/dev/null 2>&1; then
+DISK_KEY="$(jq -r '.api_key // empty' "$CREDS" 2>/dev/null || true)"
+if [ -n "$DISK_KEY" ] && [ -s "$PICKUP_FILE" ] && [ -n "${C2_URL:-}" ]; then
+  # Key on disk, but it may be STALE (an admin Reissue, or the old approve
+  # that rotated on every click). Ask C2 for the current one and replace it if
+  # different. Best-effort: on any error keep the key we have.
+  say "Enrollment — key on disk; checking it is still current"
+  creds_pickup "$(cat "$PICKUP_FILE")"
+  SERVER_KEY="$(cred_field '.api_key')"
+  if [ "$CRED_HTTP" = 200 ] && [ -n "$SERVER_KEY" ] && [ "$SERVER_KEY" != "$DISK_KEY" ]; then
+    warn "the key on disk is not the one C2 has on file (reissued?) — replacing it"
+    write_creds "$SERVER_KEY"
+  elif [ "$CRED_HTTP" = 200 ] && [ -n "$SERVER_KEY" ]; then
+    ok "api_key on disk matches C2"
+  else
+    warn "could not confirm the key with C2 (HTTP $CRED_HTTP) — keeping the key on disk"
+  fi
+elif [ -n "$DISK_KEY" ]; then
   say "Enrollment"
-  ok "api_key already on disk — skipping enrollment"
+  ok "api_key already on disk (no pickup secret to re-check it with) — skipping enrollment"
 elif [ -z "${C2_URL:-}" ]; then
   say "Enrollment"
   warn "no C2_URL — skipping enrollment"
@@ -446,16 +513,11 @@ elif [ -s "$PICKUP_FILE" ]; then
   creds_pickup "$PICKUP"
   case "$CRED_HTTP" in
     200)
-      API_KEY="$(cred_field '.api_key')"
+      if wait_for_key "$PICKUP"; then
-      if [ -z "$API_KEY" ]; then
+        write_creds "$(cred_field '.api_key')"
-        wait_for_key "$PICKUP" || true
-        API_KEY="$(cred_field '.api_key')"
-      fi
-      if [ -n "$API_KEY" ]; then
-        write_creds "$API_KEY"
       else
-        # Still pending. Enrolled and idempotent — next run collects the key.
+        # Only with --wait-approval N/0. Enrolled and idempotent; the app
-        # Clean exit, fall through to start the stack. NOT a failure.
+        # fetches the key itself once approved. NOT a failure.
         say_pending "$(cred_field '.approval_status')"
       fi
       ;;
@@ -510,12 +572,30 @@ if [ "$DO_START" = 1 ]; then
     fi
     docker compose up --no-build -d
   fi
-  ok "stack started"
+  # ALWAYS recreate edge-node (node-26#14): `up -d` leaves an already-running
+  # container alone, and a process started before the key was written would
+  # otherwise keep running keyless. Cheap, and makes every re-run a fix.
+  docker compose up -d --no-build --force-recreate --no-deps edge-node
+  ok "stack started (edge-node recreated so it loads configs/credentials.json)"
 else
   say "Skipping start (--no-start). Run: cd $INSTALL_DIR && make up-prebuilt"
 fi
 
-# ── 7. What the operator does next ──────────────────────────────────────────
+# ── 7. Host updater (node-26#15) ───────────────────────────────────────────
+UPDATER_STATE="not installed (--no-updater)"
+if [ "$INSTALL_UPDATER" = 1 ]; then
+  say "Installing the host updater"
+  UPD_ARGS=(--dir "$INSTALL_DIR" --repo-url "$REPO_URL" --ref "$REF" --registry "$REGISTRY")
+  [ "$DO_BUILD" = 1 ] && UPD_ARGS+=(--build)
+  if bash "$INSTALL_DIR/scripts/install-updater.sh" "${UPD_ARGS[@]}"; then
+    UPDATER_STATE="enabled — checks C2 every ~15 min; 'Update Node Software' in the admin UI"
+  else
+    warn "host updater NOT installed — this node cannot be updated remotely"
+    UPDATER_STATE="FAILED to install (see above)"
+  fi
+fi
+
+# ── 8. Summary ──────────────────────────────────────────────────────────────
 IP="$(hostname -I 2>/dev/null | awk '{print $1}')"
 cat <<EOF
 
@@ -525,20 +605,18 @@ $(printf "${G}Node '%s' installed at %s${N}" "$NODE_ID" "$INSTALL_DIR")
   images         $([ "$DO_BUILD" = 1 ] && echo "built locally" || echo "pulled from $REGISTRY")
   dashboard      http://${IP:-<node-ip>}/    (user: ${DASHBOARD_USER})
   logs           cd $INSTALL_DIR && docker compose logs -f edge-node
+  updater        ${UPDATER_STATE}
+  api key        $([ -n "$(jq -r '.api_key // empty' "$CREDS" 2>/dev/null)" ] && echo "present" || echo "NOT YET — node fetches it once approved")
 
-NEXT — an admin must approve this node before it can do anything:
+NEXT — assign this node a radio system at ${APP_URL}/nodes/${NODE_ID}
-
-  1. Open  <app-url>/settings/nodes
-  2. Approve  "$NODE_ID"
-  3. Assign it a radio system
 
 EOF
 if [ "${GENERATED_DASH:-0}" = 1 ]; then
   printf "${Y}Generated dashboard password (shown once): %s${N}\n\n" "$DASHBOARD_PASS"
 fi
-if [ ! -s "$CREDS" ]; then
+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. After an admin approves it, re-run the same\ninstall command — it reuses configs/pickup_secret and will collect the key\n(no --token needed for the re-run).${N}\n\n"
+    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"
   fi

op25-container/Dockerfile

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

op25-container/app/internal/conventional_config.py

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

op25-container/app/internal/liquidsoap_config_utils.py

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

op25-container/app/internal/op25_liq_template.py

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

op25-container/app/models.py

@@ -1,147 +1,178 @@
-from pydantic import BaseModel
+from pydantic import BaseModel
-from typing import List, Optional, Union
+from typing import List, Optional, Union
-from enum import Enum
+from enum import Enum
-from config import bind_host
+from config import bind_host
-
+
-# Preset device settings for common RTL-SDR hardware.
+# Preset device settings for common RTL-SDR hardware.
-# gains: OP25 gain string passed to the device block.
+# gains: OP25 gain string passed to the device block.
-# ppm: frequency offset correction (user should calibrate and override).
+# ppm: frequency offset correction (user should calibrate and override).
-# cqpsk_tracking: disable for TCXO dongles (RTL-SDR v3, NESDR) once PPM is dialled in;
+# cqpsk_tracking: disable for TCXO dongles (RTL-SDR v3, NESDR) once PPM is dialled in;
-#                 leave enabled for unknown hardware as a safety net.
+#                 leave enabled for unknown hardware as a safety net.
-HARDWARE_PRESETS: dict = {
+HARDWARE_PRESETS: dict = {
-    "rtl-sdr-v3": {
+    "rtl-sdr-v3": {
-        "gains": "lna:34",
+        "gains": "lna:34",
-        "ppm": 0.0,
+        "ppm": 0.0,
-        "cqpsk_tracking": False,
+        "cqpsk_tracking": False,
-    },
+    },
-    "nesdr-smart-v4": {
+    "nesdr-smart-v4": {
-        "gains": "lna:32",
+        "gains": "lna:32",
-        "ppm": 0.0,
+        "ppm": 0.0,
-        "cqpsk_tracking": False,
+        "cqpsk_tracking": False,
-    },
+    },
-    "other": {
+    "other": {
-        "gains": "lna:32",
+        "gains": "lna:32",
-        "ppm": 0.0,
+        "ppm": 0.0,
-        "cqpsk_tracking": True,
+        "cqpsk_tracking": True,
-    },
+    },
-}
+}
-
+
-class DecodeMode(str, Enum):
+class DecodeMode(str, Enum):
-    P25 = "P25"
+    P25 = "P25"
-    DMR = "DMR"
+    DMR = "DMR"
-    ANALOG = "NBFM"
+    ANALOG = "NBFM"
-
+
-class TalkgroupTag(BaseModel):
+class TalkgroupTag(BaseModel):
-    talkgroup: str
+    talkgroup: str
-    tagDec: int
+    tagDec: int
-
+
-# Defined before ConfigGenerator, which annotates a field with it. Under
+# Defined before ConfigGenerator, which annotates a field with it. Under
-# Python 3.14 (PEP 649) annotations are evaluated lazily, so the original
+# Python 3.14 (PEP 649) annotations are evaluated lazily, so the original
-# order happened to work in the container; on 3.13 or earlier it is a hard
+# order happened to work in the container; on 3.13 or earlier it is a hard
-# NameError at import. Keep the definition above its first use so this file
+# NameError at import. Keep the definition above its first use so this file
-# does not depend on the base image's Python version.
+# does not depend on the base image's Python version.
-class IcecastConfig(BaseModel):
+class IcecastConfig(BaseModel):
-    icecast_host: str
+    icecast_host: str
-    icecast_port: int
+    icecast_port: int
-    icecast_mountpoint: str
+    icecast_mountpoint: str
-    icecast_password: str
+    icecast_password: str
-    icecast_description: Optional[str] = "OP25"
+    icecast_description: Optional[str] = "OP25"
-    icecast_genre: Optional[str] = "Public Safety"
+    icecast_genre: Optional[str] = "Public Safety"
-
+
-
+
-class ConfigGenerator(BaseModel):
+class P25Mode(str, Enum):
-    type: DecodeMode
+    TRUNKED = "trunked"
-    systemName: str
+    CONVENTIONAL = "conventional"
-    channels: List[Union[int, str]]
+
-    tags: Optional[List[TalkgroupTag]]
+
-    whitelist: Optional[List[int]]
+class ConventionalChannel(BaseModel):
-    icecastConfig: Optional[IcecastConfig]
+    """One fixed P25 conventional voice channel (node-26#24)."""
-    hardware_preset: str = "rtl-sdr-v3"
+    frequency: int                 # Hz
-    ppm_override: Optional[float] = None  # when set, overrides the preset's default PPM
+    name: str = ""
-
+    nac: Optional[int] = None      # 0x001-0xFFF; None = accept any NAC
-class DemodType(str, Enum):
+
-    CQPSK = "cqpsk"
+
-    FSK4 = "fsk4"
+class ConfigGenerator(BaseModel):
-
+    type: DecodeMode
-class FilterType(str, Enum):
+    systemName: str
-    RC = "rc"
+    # P25 trunked: control channel list (Hz). NBFM: channels[0] is the one
-    WIDEPULSE = "widepulse"
+    # analog frequency to listen to (Hz). P25 conventional: unused — see
-
+    # conv_channels.
-class ChannelConfig(BaseModel):
+    channels: List[Union[int, str]] = []
-    name: str
+    # P25 only. Missing = trunked, so every existing caller is unchanged.
-    trunking_sysname: Optional[str]
+    p25_mode: P25Mode = P25Mode.TRUNKED
-    enable_analog: str
+    conv_channels: List[ConventionalChannel] = []
-    meta_stream_name: str
+    tags: Optional[List[TalkgroupTag]] = None
-    demod_type: DemodType
+    whitelist: Optional[List[int]] = None
-    filter_type: FilterType
+    icecastConfig: Optional[IcecastConfig] = None
-    device: Optional[str] = "sdr"
+    hardware_preset: str = "rtl-sdr-v3"
-    cqpsk_tracking: Optional[bool] = None
+    ppm_override: Optional[float] = None  # when set, overrides the preset's default PPM
-    frequency: Optional[float] = None
+    # NBFM only: power squelch threshold in dB (op25_nbfm default is -60).
-    nbfmSquelch: Optional[float] = None
+    nbfm_squelch: Optional[float] = None
-    destination: Optional[str] = "udp://127.0.0.1:23456"
+
-    tracking_threshold: Optional[int] = 120
+class DemodType(str, Enum):
-    tracking_feedback: Optional[float] = 0.75
+    CQPSK = "cqpsk"
-    excess_bw: Optional[float] = 0.2
+    FSK4 = "fsk4"
-    if_rate: Optional[int] = 24000
+
-    plot: Optional[str] = ""
+class FilterType(str, Enum):
-    symbol_rate: Optional[int] = 4800
+    RC = "rc"
-    blacklist: Optional[str] = ""
+    WIDEPULSE = "widepulse"
-    whitelist: Optional[str] = ""
+
-
+class ChannelConfig(BaseModel):
-class DeviceConfig(BaseModel):
+    name: str
-    args: Optional[str] = "rtl"
+    trunking_sysname: Optional[str] = None
-    gains: Optional[str] = "lna:39"
+    enable_analog: str
-    gain_mode: Optional[bool] = False
+    meta_stream_name: str
-    name: Optional[str] = "sdr"
+    demod_type: DemodType
-    offset: Optional[int] = 0
+    filter_type: FilterType
-    ppm: Optional[float] = 0.0
+    device: Optional[str] = "sdr"
-    rate: Optional[int] = 1920000
+    cqpsk_tracking: Optional[bool] = None
-    usable_bw_pct: Optional[float] = 0.85
+    frequency: Optional[int] = None
-    tunable: Optional[bool] = True
+    # Key name op25_nbfm.py actually reads (from_dict(config, 'nbfm_squelch_threshold', -60)).
-
+    nbfm_squelch_threshold: Optional[float] = None
-class TrunkingChannelConfig(BaseModel):
+    destination: Optional[str] = "udp://127.0.0.1:23456"
-    sysname: str
+    # P25 conventional only, read by tk_drb_conv.py (not by multi_rx): the NAC
-    control_channel_list: str
+    # to lock the frame assembler to. multi_rx passes the whole channel dict
-    tagsFile: Optional[str] = None
+    # through, so an extra key is harmless.
-    whitelist: Optional[str] = None
+    nac: Optional[int] = None
-    nac: Optional[str] = ""
+    # Set on the channel itself for conventional (multi_rx forwards it to the
-    wacn: Optional[str] = ""
+    # frame assembler). 2 = drop encrypted voice instead of decoding noise.
-    tdma_cc: Optional[bool] = False
+    crypt_behavior: Optional[int] = None
-    crypt_behavior: Optional[int] = 2
+    tracking_threshold: Optional[int] = 120
-
+    tracking_feedback: Optional[float] = 0.75
-class TrunkingConfig(BaseModel):
+    excess_bw: Optional[float] = 0.2
-    module: str
+    if_rate: Optional[int] = 24000
-    chans: List[TrunkingChannelConfig]
+    plot: Optional[str] = ""
-
+    symbol_rate: Optional[int] = 4800
-class MetadataStreamConfig(BaseModel):
+    blacklist: Optional[str] = ""
-    stream_name: str = "stream_0"
+    whitelist: Optional[str] = ""
-    meta_format_idle: str = "[idle]"
+
-    meta_format_tgid: str = "[%TGID%]"
+class DeviceConfig(BaseModel):
-    meta_format_tag: str = "[%TGID%] %TAG%"
+    args: Optional[str] = "rtl"
-    icecastServerAddress: str = "localhost"
+    gains: Optional[str] = "lna:39"
-    icecastMountpoint: str = "NODE_ID"
+    gain_mode: Optional[bool] = False
-    icecastMountExt: str = ".xspf"
+    name: Optional[str] = "sdr"
-    icecastPass: str = "PASSWORD"
+    # Centre frequency (Hz). Unset for trunked P25, where tk_p25 tunes the
-    delay: float = 0.0
+    # device; required for a fixed analog channel (tunable=False).
-
+    frequency: Optional[int] = None
-class MetadataConfig(BaseModel):
+    offset: Optional[int] = 0
-    module: str = "icemeta.py"
+    ppm: Optional[float] = 0.0
-    streams: List[MetadataStreamConfig]
+    rate: Optional[int] = 1920000
-
+    usable_bw_pct: Optional[float] = 0.85
-class TerminalConfig(BaseModel):
+    tunable: Optional[bool] = True
-    module: Optional[str] = "terminal.py"
+
-    # Bind address comes from OP25_DEBUG_EXPOSE (config.py) — 127.0.0.1 unless
+class TrunkingChannelConfig(BaseModel):
-    # that flag is set. See config.py for why.
+    sysname: str
-    terminal_type: Optional[str] = f"http:{bind_host()}:8081"
+    control_channel_list: str
-    terminal_timeout: Optional[float] = 5.0
+    tagsFile: Optional[str] = None
-    curses_plot_interval: Optional[float] = 0.2
+    whitelist: Optional[str] = None
-    http_plot_interval: Optional[float] = 1.0
+    nac: Optional[str] = ""
-    http_plot_directory: Optional[str] = "../www/images"
+    wacn: Optional[str] = ""
-    tuning_step_large: Optional[int] = 1200
+    tdma_cc: Optional[bool] = False
-    tuning_step_small: Optional[int] = 100
+    crypt_behavior: Optional[int] = 2
-
+
-
+class TrunkingConfig(BaseModel):
-
+    module: str
-### ======================================================
+    chans: List[TrunkingChannelConfig]
-# Icecast models
+
+class MetadataStreamConfig(BaseModel):
+    stream_name: str = "stream_0"
+    meta_format_idle: str = "[idle]"
+    meta_format_tgid: str = "[%TGID%]"
+    meta_format_tag: str = "[%TGID%] %TAG%"
+    icecastServerAddress: str = "localhost"
+    icecastMountpoint: str = "NODE_ID"
+    icecastMountExt: str = ".xspf"
+    icecastPass: str = "PASSWORD"
+    delay: float = 0.0
+
+class MetadataConfig(BaseModel):
+    module: str = "icemeta.py"
+    streams: List[MetadataStreamConfig]
+
+class TerminalConfig(BaseModel):
+    module: Optional[str] = "terminal.py"
+    # Bind address comes from OP25_DEBUG_EXPOSE (config.py) — 127.0.0.1 unless
+    # that flag is set. See config.py for why.
+    terminal_type: Optional[str] = f"http:{bind_host()}:8081"
+    terminal_timeout: Optional[float] = 5.0
+    curses_plot_interval: Optional[float] = 0.2
+    http_plot_interval: Optional[float] = 1.0
+    http_plot_directory: Optional[str] = "../www/images"
+    tuning_step_large: Optional[int] = 1200
+    tuning_step_small: Optional[int] = 100
+
+
+
+### ======================================================
+# Icecast models
 # (IcecastConfig itself is defined above ConfigGenerator, which references it.)

op25-container/app/routers/op25_controller.py

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

op25-container/op25_addons/tk_drb_conv.py

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

op25-container/run_multi-rx_service.sh

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

op25-container/tests/test_conventional.py

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

scripts/drb-update.sh

@@ -0,0 +1,172 @@
+#!/usr/bin/env bash
+#
+# DRB node host updater (node-26#15). Runs as root from systemd:
+#   drb-update.timer  -> every ~15 min: HTTPS check-in with C2; update if asked
+#   drb-update.path   -> fires when the edge-node container drops
+#                        <install>/configs/update-requested (MQTT node_update)
+# Both start drb-update.service, which runs this script. install-updater.sh
+# writes the units and /etc/drb/updater.conf; install.sh calls it.
+#
+# SECURITY MODEL — this is a remote-update path onto devices in the field:
+#   * What gets installed comes ONLY from /etc/drb/updater.conf (root-owned,
+#     written by install.sh): the repo URL, the ref, the image registry. C2's
+#     answer is a single timestamp meaning "update now"; it is never used as a
+#     ref, URL, path or command. The MQTT sentinel's content is never read.
+#   * git origin must equal the configured REPO_URL (https only) and the .env
+#     IMAGE_REGISTRY must equal the configured REGISTRY, or nothing runs.
+#   * 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).
+#
+# The whole body is inside main(), called on the last line: bash parses the
+# function completely before running it, so `git checkout` replacing this
+# very file mid-run cannot change what is executing.
+set -euo pipefail
+
+main() {
+  local CONF="${DRB_UPDATER_CONF:-/etc/drb/updater.conf}"
+  local STATE_DIR="${DRB_UPDATER_STATE:-/var/lib/drb-update}"
+  local LOCK="${DRB_UPDATER_LOCK:-/run/drb-update.lock}"
+
+  log()  { printf 'drb-update: %s\n' "$*"; }
+  err()  { printf 'drb-update: ERROR: %s\n' "$*" >&2; }
+  conf() { grep -E "^$1=" "$CONF" 2>/dev/null | head -1 | cut -d= -f2- || true; }
+
+  [ -r "$CONF" ] || { err "$CONF missing — re-run install.sh"; return 1; }
+  local INSTALL_DIR REPO_URL REF REGISTRY BUILD
+  INSTALL_DIR="$(conf INSTALL_DIR)"; REPO_URL="$(conf REPO_URL)"; REF="$(conf REF)"
+  REGISTRY="$(conf REGISTRY)"; BUILD="$(conf BUILD)"
+
+  [[ "$INSTALL_DIR" =~ ^/[A-Za-z0-9._/-]+$ ]] && [ -d "$INSTALL_DIR/.git" ] \
+    || { err "bad INSTALL_DIR '$INSTALL_DIR' in $CONF"; return 1; }
+  [[ "$REPO_URL" =~ ^https://[A-Za-z0-9._/:-]+$ ]] || { err "REPO_URL must be https: '$REPO_URL'"; return 1; }
+  [[ "$REF" =~ ^[A-Za-z0-9._][A-Za-z0-9._/-]*$ ]] || { err "bad REF '$REF'"; return 1; }
+  [[ "$REGISTRY" =~ ^[A-Za-z0-9.-]+(:[0-9]+)?$ ]] || { err "bad REGISTRY '$REGISTRY'"; return 1; }
+
+  mkdir -p "$STATE_DIR"; chmod 700 "$STATE_DIR"
+  exec 9>"$LOCK"
+  flock -n 9 || { log "another update is already running"; return 0; }
+
+  cd "$INSTALL_DIR"
+  envget() { grep -E "^$1=" .env 2>/dev/null | head -1 | cut -d= -f2- | tr -d '"' || true; }
+  local NODE_ID C2_URL
+  NODE_ID="$(envget NODE_ID)"; C2_URL="$(envget C2_URL)"; C2_URL="${C2_URL%/}"
+  [[ "$NODE_ID" =~ ^[A-Za-z0-9_-]+$ ]] || { err "no valid NODE_ID in $INSTALL_DIR/.env"; return 1; }
+
+  local reason=""
+  local SENTINEL="$INSTALL_DIR/configs/update-requested"
+  if [ -e "$SENTINEL" ] || [ -L "$SENTINEL" ]; then
+    rm -f "$SENTINEL"          # content deliberately ignored
+    reason="mqtt node_update"
+  fi
+
+  # ── HTTPS check-in ──────────────────────────────────────────────────────
+  local tmpd; tmpd="$(mktemp -d)"; chmod 700 "$tmpd"
+  # shellcheck disable=SC2064
+  trap "rm -rf '$tmpd'" RETURN
+
+  local sha; sha="$(git rev-parse HEAD 2>/dev/null || echo '')"
+  local requested=""
+
+  checkin() {  # checkin APPLIED_AT RESULT -> stdout: response body
+    [[ "$C2_URL" == https://* ]] || return 1
+    local key="" pickup=""
+    key="$(jq -r '.api_key // empty' configs/credentials.json 2>/dev/null || true)"
+    pickup="$(cat configs/pickup_secret 2>/dev/null || true)"
+    : > "$tmpd/h"; chmod 600 "$tmpd/h"
+    if [[ "$key" =~ ^[A-Za-z0-9]+$ ]]; then
+      printf 'Authorization: Bearer %s\n' "$key" >> "$tmpd/h"
+    fi
+    if [[ "$pickup" =~ ^[A-Za-z0-9]+$ ]]; then
+      printf 'X-Pickup-Secret: %s\n' "$pickup" >> "$tmpd/h"
+    fi
+    [ -s "$tmpd/h" ] || return 1
+    local body
+    body="$(jq -nc --arg sha "$sha" --arg a "$1" --arg r "$2" \
+      '{sha:$sha} + (if $a != "" then {applied_request_at:$a} else {} end)
+                  + (if $r != "" then {result:$r} else {} end)')"
+    curl -fsS --proto '=https' --max-time 20 -X POST \
+      "$C2_URL/nodes/$NODE_ID/update-check" \
+      -H "Content-Type: application/json" -H "@$tmpd/h" --data "$body"
+  }
+
+  local resp
+  if [[ "$C2_URL" != https://* ]]; then
+    log "C2_URL is not https — skipping the HTTPS check-in"
+  elif resp="$(checkin "" "" 2>"$tmpd/err")"; then
+    requested="$(printf '%s' "$resp" | jq -r '.update_requested_at // empty' 2>/dev/null || true)"
+    if [ -n "$requested" ] && ! [[ "$requested" =~ ^[0-9T:.+Z-]{10,40}$ ]]; then
+      log "ignoring malformed update_requested_at from C2"
+      requested=""
+    fi
+  else
+    log "HTTPS check-in failed: $(tr -d '\n' < "$tmpd/err" | head -c 200)"
+  fi
+
+  local last=""; last="$(cat "$STATE_DIR/last_request" 2>/dev/null || true)"
+  local new_request=""
+  if [ -n "$requested" ] && [ "$requested" != "$last" ]; then
+    new_request="$requested"
+    reason="${reason:-C2 request $requested}"
+  fi
+  [ -n "$reason" ] || { log "no update requested"; return 0; }
+
+  # ── Update ─────────────────────────────────────────────────────────────
+  log "updating ($reason) — repo $REPO_URL @ $REF"
+  local result=ok
+  if do_update; then
+    log "update complete: $(git rev-parse --short HEAD)"
+  else
+    result=failed
+    log "update FAILED — will retry on the next timer run"
+  fi
+  if [ -n "$new_request" ]; then
+    [ "$result" = ok ] && printf '%s' "$new_request" > "$STATE_DIR/last_request"
+    sha="$(git rev-parse HEAD 2>/dev/null || echo '')"
+    checkin "$new_request" "$result" >/dev/null 2>&1 || log "could not report the result to C2"
+  fi
+  [ "$result" = ok ]
+}
+
+do_update() {
+  local origin; origin="$(git remote get-url origin)"
+  [ "$origin" = "$REPO_URL" ] || { err "git origin '$origin' != configured '$REPO_URL' — refusing"; return 1; }
+
+  # --force: a moved release tag (v1) must replace the local copy.
+  git -c protocol.allow=never -c protocol.https.allow=always fetch --force --tags --prune origin || return 1
+
+  local target
+  if git show-ref --verify -q "refs/remotes/origin/$REF"; then
+    target="refs/remotes/origin/$REF"
+  elif git show-ref --verify -q "refs/tags/$REF"; then
+    target="refs/tags/$REF"
+  else
+    err "ref '$REF' not found on origin"; return 1
+  fi
+  local old new
+  old="$(git rev-parse HEAD)"; new="$(git rev-parse "$target^{commit}")"
+  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}"
+    # Units/conf may have changed with the new scripts.
+    bash scripts/install-updater.sh --refresh || log "unit refresh failed (continuing with the old units)"
+  fi
+
+  local env_registry
+  env_registry="$(grep -E '^IMAGE_REGISTRY=' .env | head -1 | cut -d= -f2- | tr -d '"' || true)"
+  [ "${env_registry:-git.vpn.cusano.net}" = "$REGISTRY" ] \
+    || { err ".env IMAGE_REGISTRY '$env_registry' != configured '$REGISTRY' — refusing"; return 1; }
+
+  if [ "$BUILD" = 1 ]; then
+    docker compose build || return 1
+  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
+  docker image prune -f >/dev/null 2>&1 || true
+}
+
+main "$@"

scripts/install-updater.sh

@@ -0,0 +1,107 @@
+#!/usr/bin/env bash
+#
+# Install (or refresh) the DRB host updater units (node-26#15). Run as root.
+#
+#   install-updater.sh --dir /opt/drb/node-26 --repo-url URL --ref v1 \
+#                      --registry git.vpn.cusano.net [--build]
+#   install-updater.sh --refresh      # re-render units from /etc/drb/updater.conf
+#
+# Writes /etc/drb/updater.conf (root:root 0644 — the ONLY source of what the
+# updater may install), drb-update.{service,timer,path}, and enables the timer
+# and the path watcher. install.sh calls this; drb-update.sh calls --refresh
+# after a checkout so unit changes ship with the repo.
+set -euo pipefail
+
+CONF=/etc/drb/updater.conf
+UNIT_DIR="${DRB_UNIT_DIR:-/etc/systemd/system}"
+conf() { grep -E "^$1=" "$CONF" 2>/dev/null | head -1 | cut -d= -f2- || true; }
+die()  { printf 'install-updater: %s\n' "$*" >&2; exit 1; }
+
+INSTALL_DIR="" REPO_URL="" REF="" REGISTRY="" BUILD=0 REFRESH=0
+while [ $# -gt 0 ]; do
+  case "$1" in
+    --dir)       INSTALL_DIR="$2"; shift 2 ;;
+    --repo-url)  REPO_URL="$2"; shift 2 ;;
+    --ref)       REF="$2"; shift 2 ;;
+    --registry)  REGISTRY="$2"; shift 2 ;;
+    --build)     BUILD=1; shift ;;
+    --refresh)   REFRESH=1; shift ;;
+    *)           die "unknown option: $1" ;;
+  esac
+done
+
+[ "$(id -u)" -eq 0 ] || die "must run as root"
+if [ "$REFRESH" = 1 ]; then
+  [ -r "$CONF" ] || die "$CONF missing — run install.sh"
+  INSTALL_DIR="$(conf INSTALL_DIR)"; REPO_URL="$(conf REPO_URL)"; REF="$(conf REF)"
+  REGISTRY="$(conf REGISTRY)"; BUILD="$(conf BUILD)"
+fi
+
+[[ "$INSTALL_DIR" =~ ^/[A-Za-z0-9._/-]+$ ]] || die "bad --dir '$INSTALL_DIR' (absolute, no spaces)"
+[[ "$REPO_URL" =~ ^https://[A-Za-z0-9._/:-]+$ ]] || die "--repo-url must be https: '$REPO_URL'"
+[[ "$REF" =~ ^[A-Za-z0-9._][A-Za-z0-9._/-]*$ ]] || die "bad --ref '$REF'"
+[[ "$REGISTRY" =~ ^[A-Za-z0-9.-]+(:[0-9]+)?$ ]] || die "bad --registry '$REGISTRY'"
+[ "$BUILD" = 1 ] || BUILD=0
+
+if ! command -v systemctl >/dev/null 2>&1 || [ ! -d /run/systemd/system ]; then
+  die "systemd not running — the host updater needs systemd (remote updates disabled on this node)"
+fi
+
+install -d -m 755 /etc/drb
+umask 022
+cat > "$CONF.tmp" <<EOF
+# DRB host updater — written by install-updater.sh. The updater installs ONLY
+# from these values; nothing C2 sends can change them. Edit = root only.
+INSTALL_DIR=$INSTALL_DIR
+REPO_URL=$REPO_URL
+REF=$REF
+REGISTRY=$REGISTRY
+BUILD=$BUILD
+EOF
+chown root:root "$CONF.tmp"; chmod 644 "$CONF.tmp"; mv -f "$CONF.tmp" "$CONF"
+
+cat > "$UNIT_DIR/drb-update.service" <<EOF
+[Unit]
+Description=DRB node updater (node-26#15)
+After=network-online.target docker.service
+Wants=network-online.target
+Requires=docker.service
+
+[Service]
+Type=oneshot
+ExecStart=/bin/bash $INSTALL_DIR/scripts/drb-update.sh
+TimeoutStartSec=$([ "$BUILD" = 1 ] && echo 3h || echo 45min)
+Nice=10
+EOF
+
+cat > "$UNIT_DIR/drb-update.timer" <<'EOF'
+[Unit]
+Description=DRB node updater — periodic HTTPS check-in with C2
+
+[Timer]
+OnBootSec=3min
+OnUnitActiveSec=15min
+RandomizedDelaySec=2min
+Persistent=true
+
+[Install]
+WantedBy=timers.target
+EOF
+
+cat > "$UNIT_DIR/drb-update.path" <<EOF
+[Unit]
+Description=DRB node updater — MQTT node_update sentinel
+
+[Path]
+PathExists=$INSTALL_DIR/configs/update-requested
+Unit=drb-update.service
+
+[Install]
+WantedBy=multi-user.target
+EOF
+
+systemctl daemon-reload
+systemctl enable --now drb-update.timer drb-update.path >/dev/null 2>&1 \
+  || die "could not enable drb-update.timer/path"
+printf 'install-updater: host updater enabled (timer every ~15 min + MQTT sentinel), tracking %s @ %s\n' \
+  "$REPO_URL" "$REF"