Initial push

1 files changed, 2213 insertions, 0 deletions

diff --git a/stacks/daemon/max25d.py b/stacks/daemon/max25d.py
new file mode 100755
index 0000000..c6c03f3
--- /dev/null
+++ b/stacks/daemon/max25d.py
@@ -0,0 +1,2213 @@
+#!/usr/bin/env python3
+"""
+max25d — MAX25-Stack-Lite daemon (localhost).
+
+Linux/*BSD/macOS: max25-terminal + CRDOP + BayCom/based on the same machine.
+"""
+from __future__ import annotations
+
+import argparse
+import configparser
+import os
+import re
+import select
+import signal
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional, Set
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from device_backends import (  # noqa: E402
+    DeviceBackend,
+    DeviceBackendConfig,
+    backend_serial_label,
+    baycom_ctl_device_id,
+    create_backend,
+    parse_device_spec,
+    registry_backend,
+    registry_tested,
+)
+from banlist import BanList, extract_ax25_source  # noqa: E402
+from reporting_quality import DataQualityTracker, RxOutcome, classify_rx_line  # noqa: E402
+from daemon_log import LOGGER, emit_startup_banner, emit_startup_complete  # noqa: E402
+from kiss_bridge import KissBridge  # noqa: E402 — tests patch this symbol
+from paths import (  # noqa: E402
+    ctl_path,
+    MAX25_BCPR_KISS_DEFAULT,
+    normalize_max25_bcpr_path,
+    resolve_baycom_ini,
+    resolve_layout,
+)
+from max25_platform import (  # noqa: E402
+    default_bans_file,
+    default_unix_socket,
+    max25d_supported,
+    platform_label,
+    supported_device_ids,
+)
+from privilege_drop import RunAsConfig, drop_privileges_or_exit, parse_run_as  # noqa: E402
+from modular_tcp_server import ModularTcpMainService, ModularTcpConfig, load_modular_tcp  # noqa: E402
+
+_EXE = Path(__file__).resolve()
+TREE, PREFIX = resolve_layout(_EXE)
+ROOT = TREE  # dev checkout root or MAX25_ROOT / install prefix
+
+DEFAULT_TCP_PORT = 7325
+DEFAULT_UNIX = default_unix_socket()
+M25_MAX_LINE_BUF = 65536
+CALLSIGN_RE = re.compile(r"^[A-Z0-9]{1,6}(-(1[0-5]|[0-9]))?$")
+RESERVED_DEVICE_KEYS = frozenset({"default", "enabled"})
+
+
+def _device_is_baycom(dev: DeviceBackendConfig) -> bool:
+    if dev.device_id.startswith("baycom"):
+        return True
+    spec = (dev.device_spec or "").strip()
+    if spec.startswith("baycom:"):
+        return True
+    return dev.backend_type == "baycom-kiss" and dev.hardware == "modems"
+
+
+def _device_is_pccom(dev: DeviceBackendConfig) -> bool:
+    if "pccom" in dev.device_id.lower():
+        return True
+    ini = (dev.baycom_ini or "").lower()
+    return "pccom" in ini
+
+
+
+def _device_is_max25_bcpr(dev: DeviceBackendConfig) -> bool:
+    # Product device id = max25e0 (+ forks max25e0:bcN); backend = max25-bcpr
+    if dev.device_id == "max25e0" or dev.device_id.startswith("max25e0:"):
+        return True
+    spec = (dev.device_spec or "").strip()
+    if spec.startswith("max25-bcpr:") or spec.startswith("bcpr:"):
+        return True
+    return dev.backend_type in ("max25-bcpr-kiss", "bcpr-kiss")
+
+
+def _device_is_tmodem(dev: DeviceBackendConfig) -> bool:
+    return dev.device_id == "tmodem"
+
+
+def _device_allowed_by_features(dev: DeviceBackendConfig, cfg: DaemonConfig) -> bool:
+    if _device_is_baycom(dev) and not cfg.feature_baycom:
+        LOGGER.warn(
+            f"device {dev.device_id}: BayCom disabled — set [features] baycom=yes",
+            area="config",
+        )
+        return False
+    if _device_is_pccom(dev) and not cfg.feature_pccom:
+        LOGGER.warn(
+            f"device {dev.device_id}: PC-COM disabled — set [features] pccom=yes",
+            area="config",
+        )
+        return False
+    if _device_is_max25_bcpr(dev) and not cfg.feature_max25_bcpr:
+        LOGGER.warn(
+            f"device {dev.device_id}: max25-bcpr disabled — set [features] max25_bcpr=yes",
+            area="config",
+        )
+        return False
+    if _device_is_tmodem(dev) and not cfg.feature_tmodem:
+        LOGGER.warn(
+            f"device {dev.device_id}: T-Modem disabled — set [features] tmodem=yes",
+            area="config",
+        )
+        return False
+    return True
+
+
+@dataclass
+class TotConfig:
+    """Software TOT for BayCom/based (max25-bcpr) — host policy, not radio TOT."""
+
+    enabled: bool = True
+    max_key_sec: int = 25
+    min_gap_sec: float = 1.5
+    max_consecutive: int = 3
+    max_bursts: int = 8
+    recover_sec: int = 300
+
+
+@dataclass
+class DaemonConfig:
+    mode: str = "standalone"
+    hardware: str = "tncs"
+    device: str = "tnc2c"
+    default_device: str = ""
+    devices: list[DeviceBackendConfig] = field(default_factory=list)
+    tcp_host: str = "127.0.0.1"
+    tcp_port: int = DEFAULT_TCP_PORT
+    unix_socket: str = DEFAULT_UNIX
+    tcp_password: str = ""
+    callerid: str = "CB-0"
+    callid: str = "QST"
+    ax25_ui: bool = True
+    auto_start: bool = True
+    serial_enabled: bool = True
+    serial_watch: bool = True
+    serial_watch_interval: int = 60
+    serial_repair_cooldown: int = 20
+    serial_watch_startup_grace: int = 45
+    stack_recover_only: bool = True
+    stack_retry_interval: int = 120
+    serial_bootwait_escalate: bool = True
+    serial_bootwait_escalate_after: int = 3
+    serial_bootwait_escalate_cooldown: int = 300
+    bans_file: str = field(default_factory=default_bans_file)
+    config_path: str = ""
+    feature_baycom: bool = True
+    feature_pccom: bool = True
+    feature_max25_bcpr: bool = True
+    feature_tmodem: bool = False
+    hybbx_release_attach: bool = False
+    run_user: str = ""
+    run_group: str = ""
+    run_uid: Optional[int] = None
+    run_gid: Optional[int] = None
+    report_error_transmissions: bool = True
+    report_voice_transmissions: bool = True
+    report_data_passes: int = 3
+    report_data_quality_min: int = 50
+    report_data_pass_seconds: int = 20
+    # Legacy single-device [serial] overrides (used when [devices] absent).
+    serial_device: str = ""
+    serial_baud: int = 0
+    serial_line: str = ""
+    serial_dtr_rts: str = ""
+    serial_kiss_entry: str = ""
+    modular_tcp: ModularTcpConfig = field(default_factory=ModularTcpConfig)
+    tot: TotConfig = field(default_factory=TotConfig)
+
+
+@dataclass
+class DeviceRuntime:
+    cfg: DeviceBackendConfig
+    backend: Optional[DeviceBackend] = None
+    stack_proc: Optional[subprocess.Popen] = None
+    stack_status: str = "stopped"
+    link_status: str = "n/a"
+    last_watch: float = 0.0
+    last_repair: float = 0.0
+    last_stack_retry: float = 0.0
+    prep_done: bool = False
+    inline_repair_failures: int = 0
+    last_bootwait_escalate: float = 0.0
+    quality: DataQualityTracker = field(default_factory=DataQualityTracker)
+    tot_paused: bool = False
+    tot_paused_until: float = 0.0
+    tot_trip_reason: str = ""
+
+
+@dataclass
+class DaemonState:
+    cfg: DaemonConfig
+    connected: bool = False
+    monitor_only: bool = False
+    selected_device: str = ""
+    devices: dict[str, DeviceRuntime] = field(default_factory=dict)
+    clients: Set[socket.socket] = field(default_factory=set)
+    bans: BanList = field(default_factory=BanList)
+    lock: threading.Lock = field(default_factory=threading.Lock)
+    started_at: float = 0.0
+
+
+def log(msg: str) -> None:
+    """Legacy log hook — structured stderr (human + machine readable)."""
+    LOGGER.emit_unstructured(msg)
+
+
+def valid_callsign(value: str) -> bool:
+    return bool(value and CALLSIGN_RE.match(value.upper()))
+
+
+def _truthy(value: str) -> bool:
+    return value.lower() in ("1", "yes", "true", "on")
+
+
+def _ini_int(
+    cp: configparser.ConfigParser,
+    section: str,
+    key: str,
+    default: int,
+    *,
+    min_value: int = 1,
+    max_value: int = 86400,
+) -> int:
+    if not cp.has_option(section, key):
+        return default
+    raw = cp.get(section, key)
+    try:
+        value = int(str(raw).strip())
+    except (TypeError, ValueError):
+        LOGGER.warn(
+            f"[{section}] {key}={raw!r} invalid — using {default}",
+            area="config",
+        )
+        return default
+    if value < min_value:
+        LOGGER.warn(
+            f"[{section}] {key}={value} below {min_value} — using {min_value}",
+            area="config",
+        )
+        return min_value
+    if value > max_value:
+        LOGGER.warn(
+            f"[{section}] {key}={value} above {max_value} — using {max_value}",
+            area="config",
+        )
+        return max_value
+    return value
+
+
+def _safe_port(raw: str, default: int) -> int:
+    try:
+        port = int(str(raw).strip())
+    except (TypeError, ValueError):
+        return default
+    if port < 1 or port > 65535:
+        return default
+    return port
+
+
+def _serial_overrides_from_section(cp: configparser.ConfigParser, section: str) -> dict[str, str]:
+    if not cp.has_section(section):
+        return {}
+    out: dict[str, str] = {}
+    for key in ("device", "baud", "line", "dtr_rts", "kiss_entry"):
+        if cp.has_option(section, key):
+            out[key] = cp.get(section, key)
+    return out
+
+
+def _apply_serial_overrides(dev: DeviceBackendConfig, overrides: dict[str, str]) -> None:
+    if overrides.get("device"):
+        dev.serial_device = overrides["device"]
+    if overrides.get("baud"):
+        dev.serial_baud = int(overrides["baud"])
+    if overrides.get("line"):
+        dev.serial_line = overrides["line"]
+    if overrides.get("dtr_rts"):
+        dev.serial_dtr_rts = overrides["dtr_rts"]
+    if overrides.get("kiss_entry"):
+        dev.serial_kiss_entry = overrides["kiss_entry"]
+
+
+def parse_devices(cp: configparser.ConfigParser, cfg: DaemonConfig) -> list[DeviceBackendConfig]:
+    """Build device list from [devices] or legacy [daemon] device= + [serial]."""
+    defaults = {"hardware": cfg.hardware}
+    if cp.has_section("devices"):
+        default_id = cp.get("devices", "default", fallback="").strip()
+        enabled_raw = cp.get("devices", "enabled", fallback="").strip()
+        enabled_set: Optional[set[str]] = None
+        if enabled_raw:
+            enabled_set = {x.strip() for x in enabled_raw.split(",") if x.strip()}
+
+        entries: list[DeviceBackendConfig] = []
+        for key in cp.options("devices"):
+            if key.lower() in RESERVED_DEVICE_KEYS:
+                continue
+            device_id = key.strip()
+            spec = cp.get("devices", key, fallback="").strip()
+            dev = parse_device_spec(device_id, spec, cp, defaults)
+            if enabled_set is not None:
+                dev.enabled = device_id in enabled_set
+            entries.append(dev)
+
+        if not entries:
+            LOGGER.warn(
+                "[devices] section empty — falling back to legacy single device",
+                area="config",
+            )
+        else:
+            if default_id:
+                cfg.default_device = default_id
+                if enabled_set is not None and default_id not in enabled_set:
+                    enabled_entries = [d for d in entries if d.enabled]
+                    if enabled_entries:
+                        fallback = enabled_entries[0].device_id
+                        LOGGER.warn(
+                            f"[devices] default={default_id} not enabled — using {fallback}",
+                            area="config",
+                        )
+                        cfg.default_device = fallback
+            elif cfg.device:
+                cfg.default_device = cfg.device
+            else:
+                cfg.default_device = entries[0].device_id
+            return entries
+
+    # Legacy single device
+    device_id = cfg.device or "tnc2c"
+    dev = parse_device_spec(device_id, "", cp, defaults)
+    legacy = _serial_overrides_from_section(cp, "serial")
+    if cfg.serial_device:
+        legacy.setdefault("device", cfg.serial_device)
+    if cfg.serial_baud:
+        legacy.setdefault("baud", str(cfg.serial_baud))
+    if cfg.serial_line:
+        legacy.setdefault("line", cfg.serial_line)
+    if cfg.serial_dtr_rts:
+        legacy.setdefault("dtr_rts", cfg.serial_dtr_rts)
+    if cfg.serial_kiss_entry:
+        legacy.setdefault("kiss_entry", cfg.serial_kiss_entry)
+    _apply_serial_overrides(dev, legacy)
+    cfg.default_device = device_id
+    return [dev]
+
+
+def _ini_float(
+    cp: configparser.ConfigParser,
+    section: str,
+    key: str,
+    default: float,
+    *,
+    min_value: float = 0.0,
+    max_value: float = 3600.0,
+) -> float:
+    if not cp.has_option(section, key):
+        return default
+    raw = cp.get(section, key).strip()
+    try:
+        value = float(raw)
+    except ValueError:
+        LOGGER.warn(f"invalid float {section}.{key}={raw!r} — using {default}", area="config")
+        return default
+    if value < min_value:
+        return min_value
+    if value > max_value:
+        return max_value
+    return value
+
+
+def load_config(path: Optional[Path]) -> DaemonConfig:
+    cfg = DaemonConfig()
+    if path is None:
+        for candidate in (
+            Path(os.environ.get("MAX25D_INI", "")),
+            Path("/etc/max25/max25d.ini"),
+            ROOT / "share/max25/max25d.ini.example",
+        ):
+            if candidate and candidate.is_file():
+                path = candidate
+                break
+    if path is None or not path.is_file():
+        LOGGER.warn(f"using built-in defaults (no ini at {path})", area="config")
+        cfg.devices = [parse_device_spec(cfg.device, "", configparser.ConfigParser(), {"hardware": cfg.hardware})]
+        cfg.default_device = cfg.device
+        cfg.config_path = ""
+        return cfg
+
+    cfg.config_path = str(path)
+
+    cp = configparser.ConfigParser(strict=False)
+    cp.read(path)
+    if cp.has_section("daemon"):
+        cfg.mode = cp.get("daemon", "mode", fallback=cfg.mode)
+        cfg.hardware = cp.get("daemon", "hardware", fallback=cfg.hardware)
+        cfg.device = cp.get("daemon", "device", fallback=cfg.device)
+        run_as = parse_run_as(cp)
+        cfg.run_user = run_as.user
+        cfg.run_group = run_as.group
+        cfg.run_uid = run_as.uid
+        cfg.run_gid = run_as.gid
+    if cp.has_section("network"):
+        cfg.tcp_host = cp.get("network", "tcp_host", fallback=cfg.tcp_host)
+        cfg.tcp_port = _ini_int(cp, "network", "tcp_port", cfg.tcp_port, min_value=1, max_value=65535)
+        cfg.unix_socket = cp.get("network", "unix_socket", fallback=cfg.unix_socket)
+        cfg.tcp_password = cp.get("network", "tcp_password", fallback=cfg.tcp_password)
+    if cp.has_section("modem"):
+        cfg.callerid = cp.get("modem", "callerid", fallback=cfg.callerid).upper()
+        cfg.callid = cp.get("modem", "callid", fallback=cfg.callid).upper()
+        cfg.ax25_ui = _truthy(cp.get("modem", "ax25_ui", fallback="yes"))
+        cfg.bans_file = cp.get("modem", "bans_file", fallback=cfg.bans_file)
+    if cp.has_section("stack"):
+        cfg.auto_start = _truthy(cp.get("stack", "auto_start", fallback="yes"))
+        if cp.has_option("stack", "serial_watch"):
+            cfg.serial_watch = _truthy(cp.get("stack", "serial_watch"))
+        cfg.serial_watch_interval = _ini_int(
+            cp, "stack", "serial_watch_interval", cfg.serial_watch_interval
+        )
+        cfg.serial_repair_cooldown = _ini_int(
+            cp, "stack", "serial_repair_cooldown", cfg.serial_repair_cooldown
+        )
+        cfg.serial_watch_startup_grace = _ini_int(
+            cp, "stack", "serial_watch_startup_grace", cfg.serial_watch_startup_grace
+        )
+        if cp.has_option("stack", "stack_recover_only"):
+            cfg.stack_recover_only = _truthy(cp.get("stack", "stack_recover_only"))
+        cfg.stack_retry_interval = _ini_int(
+            cp, "stack", "stack_retry_interval", cfg.stack_retry_interval
+        )
+        if cp.has_option("stack", "serial_bootwait_escalate"):
+            cfg.serial_bootwait_escalate = _truthy(cp.get("stack", "serial_bootwait_escalate"))
+        cfg.serial_bootwait_escalate_after = _ini_int(
+            cp, "stack", "serial_bootwait_escalate_after", cfg.serial_bootwait_escalate_after
+        )
+        cfg.serial_bootwait_escalate_cooldown = _ini_int(
+            cp,
+            "stack",
+            "serial_bootwait_escalate_cooldown",
+            cfg.serial_bootwait_escalate_cooldown,
+        )
+    if cp.has_section("serial"):
+        cfg.serial_device = cp.get("serial", "device", fallback="")
+        cfg.serial_baud = cp.getint("serial", "baud", fallback=0)
+        cfg.serial_line = cp.get("serial", "line", fallback="")
+        cfg.serial_dtr_rts = cp.get("serial", "dtr_rts", fallback="")
+        cfg.serial_kiss_entry = cp.get("serial", "kiss_entry", fallback="")
+
+    if cp.has_section("features"):
+        cfg.feature_baycom = _truthy(cp.get("features", "baycom", fallback="yes"))
+        cfg.feature_pccom = _truthy(cp.get("features", "pccom", fallback="yes"))
+        # Product key max25_bcpr; legacy bcpr= accepted
+        cfg.feature_max25_bcpr = _truthy(
+            cp.get("features", "max25_bcpr", fallback=cp.get("features", "bcpr", fallback="yes"))
+        )
+        cfg.feature_tmodem = _truthy(cp.get("features", "tmodem", fallback="no"))
+    if cp.has_section("hybbx"):
+        cfg.hybbx_release_attach = _truthy(
+            cp.get("hybbx", "release_attach", fallback="no")
+        )
+        if cp.has_option("hybbx", "attach"):
+            cfg.hybbx_release_attach = _truthy(cp.get("hybbx", "attach"))
+    cfg.report_error_transmissions = _truthy(
+        cp.get("reporting", "error_transmissions", fallback="yes")
+    )
+    cfg.report_voice_transmissions = _truthy(
+        cp.get("reporting", "voice_transmissions", fallback="yes")
+    )
+    cfg.report_data_passes = _ini_int(
+        cp, "reporting", "data_passes", 3, min_value=1, max_value=32
+    )
+    cfg.report_data_quality_min = _ini_int(
+        cp, "reporting", "data_quality_min", 50, min_value=1, max_value=100
+    )
+    cfg.report_data_pass_seconds = _ini_int(
+        cp, "reporting", "data_pass_seconds", 20, min_value=1, max_value=300
+    )
+    if cp.has_section("tot"):
+        cfg.tot.enabled = _truthy(cp.get("tot", "enabled", fallback="yes"))
+        cfg.tot.max_key_sec = _ini_int(
+            cp, "tot", "max_key_sec", cfg.tot.max_key_sec, min_value=1, max_value=600
+        )
+        if cp.has_option("tot", "max_key_ms"):
+            cfg.tot.max_key_sec = max(
+                1, _ini_int(cp, "tot", "max_key_ms", cfg.tot.max_key_sec * 1000) // 1000
+            )
+        cfg.tot.min_gap_sec = _ini_float(
+            cp, "tot", "min_gap_sec", cfg.tot.min_gap_sec, min_value=0.0, max_value=60.0
+        )
+        if cp.has_option("tot", "min_gap_ms"):
+            cfg.tot.min_gap_sec = _ini_int(cp, "tot", "min_gap_ms", 1500) / 1000.0
+        cfg.tot.max_consecutive = _ini_int(
+            cp, "tot", "max_consecutive", cfg.tot.max_consecutive, min_value=1, max_value=32
+        )
+        cfg.tot.max_bursts = _ini_int(
+            cp, "tot", "max_bursts", cfg.tot.max_bursts, min_value=1, max_value=64
+        )
+        cfg.tot.recover_sec = _ini_int(
+            cp, "tot", "recover_sec", cfg.tot.recover_sec, min_value=0, max_value=86400
+        )
+
+    cfg.devices = parse_devices(cp, cfg)
+    cfg.devices = [d for d in cfg.devices if _device_allowed_by_features(d, cfg)]
+    allowed = supported_device_ids()
+    if allowed:
+        filtered: list[DeviceBackendConfig] = []
+        for dev in cfg.devices:
+            if dev.device_id in allowed:
+                filtered.append(dev)
+            else:
+                LOGGER.warn(
+                    f"device {dev.device_id} not supported on {platform_label()} — skipped",
+                    area="config",
+                )
+        cfg.devices = filtered
+    if not cfg.devices:
+        LOGGER.warn("no devices remain after config/platform filter", area="config")
+    cfg.modular_tcp = load_modular_tcp(cp)
+    if not cfg.default_device and cfg.devices:
+        cfg.default_device = cfg.devices[0].device_id
+    cfg.device = cfg.default_device
+    enforce_lite_localhost(cfg)
+    return cfg
+
+
+def _is_loopback_tcp_host(host: str) -> bool:
+    """Lite allows TCP only on this machine."""
+    h = (host or "").strip().lower()
+    if not h:
+        return False
+    if h in ("127.0.0.1", "::1", "localhost"):
+        return True
+    # Accept IPv4-mapped IPv6 loopback forms if ever written in INI.
+    if h in ("::ffff:127.0.0.1", "[::1]"):
+        return True
+    return False
+
+
+def enforce_lite_localhost(cfg: DaemonConfig) -> None:
+    """Refuse LAN / wildcard TCP binds — one-PC product rule."""
+    if _is_loopback_tcp_host(cfg.tcp_host):
+        return
+    msg = (
+        f"max25d: Lite only allows TCP on this computer "
+        f"(127.0.0.1, ::1, or localhost) — got tcp_host={cfg.tcp_host!r}. "
+        "Fix your INI (use max25-lite-setup) or connect via the Unix socket."
+    )
+    log(msg)
+    raise SystemExit(1)
+
+
+def init_device_runtimes(state: DaemonState) -> None:
+    state.devices.clear()
+    for dev_cfg in state.cfg.devices:
+        if not dev_cfg.enabled:
+            continue
+        state.devices[dev_cfg.device_id] = DeviceRuntime(
+            cfg=dev_cfg,
+            quality=DataQualityTracker(
+                passes_required=state.cfg.report_data_passes,
+                min_good_percent=state.cfg.report_data_quality_min,
+                pass_window_sec=state.cfg.report_data_pass_seconds,
+            ),
+        )
+        if not registry_tested(dev_cfg.device_id):
+            LOGGER.warn(
+                f"backend={dev_cfg.backend_type or 'auto'} — not hardware-validated in CI",
+                area="devices",
+                device=dev_cfg.device_id,
+            )
+    if state.cfg.default_device in state.devices:
+        state.selected_device = state.cfg.default_device
+    elif state.devices:
+        state.selected_device = next(iter(state.devices))
+    else:
+        state.selected_device = state.cfg.default_device
+        LOGGER.warn("no enabled device runtimes — SEND/CONNECT unavailable", area="devices")
+
+
+def enabled_device_ids(state: DaemonState) -> list[str]:
+    return list(state.devices.keys())
+
+
+def device_hardware(state: DaemonState, dev_id: str) -> str:
+    rt = state.devices.get(dev_id)
+    if rt is None:
+        return state.cfg.hardware
+    return rt.cfg.hardware or state.cfg.hardware
+
+
+def device_backend_kind(state: DaemonState, dev_id: str) -> str:
+    rt = state.devices.get(dev_id)
+    if rt is None:
+        return "kiss-serial"
+    return rt.cfg.backend_type or "kiss-serial"
+
+
+def on_backend_rx(state: DaemonState, dev_id: str, line: str) -> None:
+    rt = state.devices.get(dev_id)
+    if rt is not None:
+        _sync_quality_tracker(state, rt)
+        rt.quality.record(classify_rx_line(line, state.cfg.callid))
+    src = extract_ax25_source(line)
+    if src and state.bans.is_banned(src):
+        return
+    log(f"rx {dev_id}: {line}")
+    broadcast(state, f"RX device={dev_id} {line}")
+
+
+def on_backend_invalid_frame(state: DaemonState, dev_id: str) -> None:
+    rt = state.devices.get(dev_id)
+    if rt is None:
+        return
+    _sync_quality_tracker(state, rt)
+    rt.quality.record_bad()
+    if state.cfg.report_error_transmissions:
+        broadcast(state, f"EVENT device={dev_id} error=invalid")
+
+
+VOICE_BACKEND_KINDS = frozenset({"crdop-tcp", "audio-dummy"})
+
+
+def device_has_voice_path(rt: DeviceRuntime) -> bool:
+    kind = rt.cfg.backend_type or registry_backend(rt.cfg.device_id)
+    if kind in VOICE_BACKEND_KINDS:
+        return True
+    hw = (rt.cfg.hardware or "").lower()
+    return hw in ("acoustic-bench", "soft-modems")
+
+
+def device_link_status(rt: DeviceRuntime) -> str:
+    if rt.backend is not None:
+        return backend_serial_label(rt.backend)
+    return rt.link_status
+
+
+def link_status_is_healthy(status: str) -> bool:
+    if status in ("ready", "open", "n/a"):
+        return True
+    if status.startswith("error") or status in ("closed", "stopped"):
+        return False
+    return True
+
+
+def _sync_quality_tracker(state: DaemonState, rt: DeviceRuntime) -> None:
+    required = state.cfg.report_data_passes
+    minimum = state.cfg.report_data_quality_min
+    window = state.cfg.report_data_pass_seconds
+    if (
+        rt.quality.passes_required == required
+        and rt.quality.min_good_percent == minimum
+        and rt.quality.pass_window_sec == window
+        and rt.quality.passes.maxlen == required
+    ):
+        return
+    kept = list(rt.quality.passes)[-required:]
+    voice = rt.quality.voice_activity
+    current = rt.quality.current
+    rt.quality = DataQualityTracker(
+        passes_required=required,
+        min_good_percent=minimum,
+        pass_window_sec=window,
+        voice_activity=voice,
+    )
+    for item in kept:
+        rt.quality.passes.append(item)
+    if current.started_at != 0.0:
+        rt.quality.current = current
+
+
+def device_reporting_error(state: DaemonState, rt: DeviceRuntime) -> str:
+    if not state.cfg.report_error_transmissions:
+        return "invalid"
+    if not link_status_is_healthy(device_link_status(rt)):
+        return "invalid"
+    _sync_quality_tracker(state, rt)
+    if rt.quality.data_error_valid(reporting_enabled=True):
+        return "valid"
+    return "invalid"
+
+
+def aggregate_reporting_error(state: DaemonState) -> str:
+    if not state.devices:
+        return "invalid" if not state.cfg.report_error_transmissions else "valid"
+    for rt in state.devices.values():
+        if device_reporting_error(state, rt) == "invalid":
+            return "invalid"
+    return "valid"
+
+
+def aggregate_reporting_voice(state: DaemonState) -> str:
+    if not state.cfg.report_voice_transmissions:
+        return "invalid"
+    voice_rts = [rt for rt in state.devices.values() if device_has_voice_path(rt)]
+    if not voice_rts:
+        return "valid"
+    for rt in voice_rts:
+        if not link_status_is_healthy(device_link_status(rt)):
+            return "invalid"
+    return "valid"
+
+
+def device_reporting_voice(state: DaemonState, rt: DeviceRuntime) -> str:
+    if not device_has_voice_path(rt):
+        return "n/a"
+    _sync_quality_tracker(state, rt)
+    healthy = link_status_is_healthy(device_link_status(rt))
+    if rt.quality.voice_signal_valid(
+        reporting_enabled=state.cfg.report_voice_transmissions,
+        link_healthy=healthy,
+    ):
+        return "valid"
+    return "invalid"
+
+
+BACKEND_POLL_OPEN_STATUSES = frozenset(
+    {
+        "closed",
+        "error-open",
+        "error-connect",
+        "error-no-device",
+        "error-no-path",
+    }
+)
+
+SERIAL_REPAIR_STATUSES = frozenset(
+    {
+        "error-host",
+        "error-kiss",
+        "error-tx",
+        "error-io",
+        "error-config",
+    }
+)
+
+BACKEND_RETRY_STATUSES = BACKEND_POLL_OPEN_STATUSES | SERIAL_REPAIR_STATUSES
+
+HYBBX_ATTACH_MODES = frozenset({"hybbx-host", "hybbx-main", "hybbx-cohost"})
+
+
+def hybbx_release_attach(state: DaemonState) -> bool:
+    """HyBBX opens serial/KISS after max25d prep — max25d must not hold the fd."""
+    mode = (state.cfg.mode or "").strip().lower()
+    if mode in HYBBX_ATTACH_MODES:
+        return True
+    return bool(getattr(state.cfg, "hybbx_release_attach", False))
+
+
+def uses_inline_tnc_prep(state: DaemonState, dev_id: str) -> bool:
+    """kiss-serial owned by max25d — no boot-wait subprocess (avoids port conflict)."""
+    if not state.cfg.stack_recover_only:
+        return False
+    return device_backend_kind(state, dev_id) == "kiss-serial"
+
+
+def hybbx_host_hybbx_owns_serial(state: DaemonState, dev_id: str) -> bool:
+    """In HyBBX attach mode HyBBX opens serial/KISS after max25d prep."""
+    if not hybbx_release_attach(state):
+        return False
+    kind = device_backend_kind(state, dev_id)
+    # kiss-raw + max25-bcpr: HyBBX owns the KISS attach; max25d must not hold the PTY.
+    if kind in ("kiss-raw-serial", "max25-bcpr-kiss", "bcpr-kiss"):
+        return True
+    if kind != "kiss-serial":
+        return False
+    rt = state.devices.get(dev_id)
+    return rt is not None and rt.prep_done and rt.backend is None
+
+
+def prep_inline_serial_device(state: DaemonState, dev_id: str) -> None:
+    """Open serial and run initial recovery while holding DTR (no subprocess)."""
+    rt = state.devices[dev_id]
+    rt.stack_status = "ready"
+    LOGGER.info("inline prep — max25d owns serial recovery", area="stack", device=dev_id)
+    if not backend_enabled(state, dev_id):
+        return
+    if not open_backend(state, dev_id):
+        LOGGER.error(f"serial prep open failed status={rt.link_status}", area="serial", device=dev_id)
+        return
+    backend = rt.backend
+    stabilize = getattr(backend, "stabilize_session", None)
+    if stabilize is None:
+        return
+    ok = stabilize(state.cfg.callerid, force=False)
+    rt.prep_done = True
+    rt.link_status = backend.status
+    if ok:
+        LOGGER.ok("serial prep complete — terminal + KISS ready", area="serial", device=dev_id)
+        if hybbx_release_attach(state):
+            close_backend(state, dev_id)
+            rt.link_status = "ready"
+            LOGGER.info(
+                "HyBBX attach: serial released for HyBBX KISS attach",
+                area="serial",
+                device=dev_id,
+            )
+    elif (
+        backend.status == "error-host"
+        and state.cfg.serial_bootwait_escalate
+        and rt.stack_proc is None
+    ):
+        LOGGER.warn(
+            "inline ladder exhausted (error-host) — escalating to boot-wait + power-cycle hint",
+            area="serial",
+            device=dev_id,
+        )
+        rt.last_bootwait_escalate = time.time()
+        escalate_to_bootwait_stack(state, dev_id)
+    else:
+        LOGGER.warn(
+            f"prep deferred status={backend.status} — serial watch will retry",
+            area="serial",
+            device=dev_id,
+        )
+
+
+def escalate_to_bootwait_stack(state: DaemonState, dev_id: str) -> None:
+    """Release inline serial and run boot-wait subprocess (DTR + power-cycle rescue)."""
+    rt = state.devices[dev_id]
+    close_backend(state, dev_id)
+    rt.prep_done = False
+    ctl = ctl_path(ROOT, PREFIX, _EXE)
+    if not ctl.is_file():
+        rt.stack_status = "error-no-ctl"
+        log(f"serial watch: boot-wait escalate failed — no ctl ({dev_id})")
+        return
+    hw = device_hardware(state, dev_id)
+    args = [
+        str(ctl),
+        "start",
+        "--mode",
+        state.cfg.mode,
+        "--hardware",
+        hw,
+        "--device",
+        dev_id,
+    ]
+    env = os.environ.copy()
+    env["MAX25_MODE"] = state.cfg.mode
+    env.pop("MAX25_TNC_PREP", None)
+    workdir = str(ROOT if (ROOT / "plugins").is_dir() else (PREFIX or ROOT))
+    try:
+        proc = subprocess.Popen(
+            args,
+            cwd=workdir,
+            env=env,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            start_new_session=True,
+        )
+    except OSError as exc:
+        log(f"serial watch: boot-wait escalate failed ({dev_id}): {exc}")
+        rt.stack_status = "error"
+        return
+    rt.stack_proc = proc
+    rt.stack_status = "running"
+    log(
+        f"serial watch: escalating to boot-wait ({dev_id}) pid={proc.pid} "
+        "— power OFF TNC 10s then ON while script runs (DTR held high)"
+    )
+    broadcast(state, f"EVENT device={dev_id} serial=boot-wait-escalate")
+
+
+def backend_needs_open(backend: Optional[DeviceBackend]) -> bool:
+    if backend is None:
+        return True
+    return backend.status in BACKEND_POLL_OPEN_STATUSES
+
+
+def open_backend(state: DaemonState, dev_id: str) -> bool:
+    rt = state.devices.get(dev_id)
+    if rt is None or not backend_enabled(state, dev_id):
+        if rt is not None:
+            rt.link_status = "n/a"
+        return False
+    if hybbx_host_hybbx_owns_serial(state, dev_id):
+        rt.link_status = "ready"
+        rt.prep_done = True
+        rt.stack_status = "ready"
+        return True
+    if rt.backend is not None and rt.backend.status not in BACKEND_RETRY_STATUSES:
+        return rt.backend.status in ("open", "ready")
+    if (
+        rt.backend is not None
+        and rt.backend.status in SERIAL_REPAIR_STATUSES
+        and rt.prep_done
+    ):
+        return rt.backend.status in ("open", "ready", "error-host", "error-kiss")
+    if rt.backend is not None and rt.backend.status != "closed":
+        rt.backend.close()
+        rt.backend = None
+    backend = create_backend(
+        rt.cfg,
+        str(ROOT),
+        lambda line, d=dev_id: on_backend_rx(state, d, line),
+        log,
+        prefix=str(PREFIX) if PREFIX else None,
+        on_invalid=lambda d=dev_id: on_backend_invalid_frame(state, d),
+    )
+    if not backend.open():
+        rt.backend = backend
+        rt.link_status = backend.status
+        return False
+    rt.backend = backend
+    rt.link_status = backend.status
+    return True
+
+
+def close_backend(state: DaemonState, dev_id: str) -> None:
+    rt = state.devices.get(dev_id)
+    if rt is None or rt.backend is None:
+        return
+    rt.backend.close()
+    rt.link_status = rt.backend.status
+    rt.backend = None
+
+
+def attach_backend_session(state: DaemonState, dev_id: str) -> bool:
+    if not backend_enabled(state, dev_id):
+        return True
+    if hybbx_host_hybbx_owns_serial(state, dev_id):
+        return True
+    rt = state.devices[dev_id]
+    if backend_needs_open(rt.backend):
+        if not open_backend(state, dev_id):
+            return False
+    assert rt.backend is not None
+    ok = rt.backend.attach_session(state.cfg.callerid)
+    rt.link_status = rt.backend.status
+    return ok
+
+
+def detach_backend_session(state: DaemonState, dev_id: str) -> None:
+    rt = state.devices.get(dev_id)
+    if rt is None or rt.backend is None:
+        return
+    rt.backend.detach_session()
+    rt.link_status = rt.backend.status
+
+
+def attach_all_sessions(state: DaemonState) -> bool:
+    ok = True
+    for dev_id in enabled_device_ids(state):
+        if backend_enabled(state, dev_id):
+            if not attach_backend_session(state, dev_id):
+                ok = False
+    return ok
+
+
+def detach_all_sessions(state: DaemonState) -> None:
+    for dev_id in enabled_device_ids(state):
+        detach_backend_session(state, dev_id)
+
+
+def backend_enabled(state: DaemonState, dev_id: str) -> bool:
+    if not state.cfg.serial_enabled:
+        return False
+    rt = state.devices.get(dev_id)
+    if rt is None:
+        return False
+    if rt.tot_paused:
+        return False
+    kind = rt.cfg.backend_type
+    return kind in ("kiss-serial", "baycom-kiss", "max25-bcpr-kiss", "bcpr-kiss", "kiss-raw-serial", "crdop-tcp")
+
+
+def aggregate_stack_status(state: DaemonState) -> str:
+    if not state.devices:
+        return "stopped"
+    statuses = {rt.stack_status for rt in state.devices.values()}
+    if "running" in statuses:
+        return "running"
+    if any(s.startswith("error") for s in statuses):
+        return "error"
+    if statuses == {"ready"} or statuses == {"stopped"}:
+        return next(iter(statuses))
+    if "ready" in statuses:
+        return "ready"
+    return "running" if "running" in statuses else "stopped"
+
+
+def aggregate_link_status(state: DaemonState) -> str:
+    if not state.devices:
+        return "n/a"
+    if len(state.devices) == 1:
+        rt = next(iter(state.devices.values()))
+        return backend_serial_label(rt.backend) if rt.backend is not None else rt.link_status
+    parts: list[str] = []
+    for dev_id in sorted(state.devices):
+        rt = state.devices[dev_id]
+        st = backend_serial_label(rt.backend) if rt.backend is not None else rt.link_status
+        parts.append(f"{dev_id}={st}")
+    return ",".join(parts)
+
+
+def status_line(state: DaemonState) -> str:
+    c = state.cfg
+    dev_list = ",".join(enabled_device_ids(state))
+    selected = state.selected_device or c.default_device or c.device
+    return (
+        f"STATUS hardware={c.hardware} device={selected} devices={dev_list} "
+        f"mode={c.mode} callerid={c.callerid} callid={c.callid} "
+        f"ax25_ui={'on' if c.ax25_ui else 'off'} "
+        f"connected={'yes' if state.connected else 'no'} "
+        f"stack={aggregate_stack_status(state)} serial={aggregate_link_status(state)} "
+        f"error={aggregate_reporting_error(state)} "
+        f"voice={aggregate_reporting_voice(state)}"
+    )
+
+
+def broadcast(state: DaemonState, line: str, skip: Optional[socket.socket] = None) -> None:
+    payload = (line + "\n").encode("utf-8")
+    dead: list[socket.socket] = []
+    with state.lock:
+        for sock in state.clients:
+            if sock is skip:
+                continue
+            try:
+                sock.sendall(payload)
+            except OSError:
+                dead.append(sock)
+        for sock in dead:
+            state.clients.discard(sock)
+
+
+def send_line(sock: socket.socket, line: str) -> None:
+    sock.sendall((line + "\n").encode("utf-8"))
+
+
+def unix_path_is_live(path: str, *, timeout: float = 0.3) -> bool:
+    """True if path exists and accepts a Unix connect (live max25d listener)."""
+    if not path or not os.path.exists(path):
+        return False
+    probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    try:
+        probe.settimeout(timeout)
+        probe.connect(path)
+        return True
+    except OSError:
+        return False
+    finally:
+        try:
+            probe.close()
+        except OSError:
+            pass
+
+
+def unix_path_id(path: str) -> tuple[int, int] | None:
+    """Filesystem identity of a Unix socket path (dev, ino), or None."""
+    try:
+        st = os.stat(path)
+    except OSError:
+        return None
+    return (st.st_dev, st.st_ino)
+
+
+def unlink_unix_if_ours(path: str, bind_id: tuple[int, int] | None) -> None:
+    """Unlink path only when it still names the inode we bound.
+
+    AF_UNIX: fstat(listen_fd) is sockfs — must not compare to path st_ino.
+    A second max25d that unlinks+rebinds must not lose its path when the
+    first instance exits (orphaned listen FD + ENOENT for clients).
+    """
+    if not path or bind_id is None:
+        return
+    cur = unix_path_id(path)
+    if cur is None or cur != bind_id:
+        return
+    try:
+        os.unlink(path)
+    except FileNotFoundError:
+        pass
+    except OSError:
+        pass
+
+
+def resolve_max25_bcpr_ini(explicit: str = "") -> Path | None:
+    """Resolve max25-bcpr.ini for userspace SER12 (max25e0)."""
+    from pathlib import Path as _P
+    if explicit:
+        p = _P(explicit)
+        return p if p.is_file() else None
+    for cand in (
+        ROOT / "local" / "max25-bcpr.ini",
+        _P("/etc/max25/max25-bcpr.ini"),
+        _P("/etc/max25/bcpr.ini"),
+        ROOT / "local" / "bcpr.ini",
+        ROOT / "stacks" / "max25-bcpr" / "share" / "max25-bcpr.ini.example",
+        ROOT / "share" / "max25-bcpr" / "max25-bcpr.ini.example",
+    ):
+        if cand.is_file():
+            return cand
+    return None
+
+
+def resolve_max25_bcpr_ctl() -> Path | None:
+    from pathlib import Path as _P
+    for cand in (
+        ROOT / "stacks" / "max25-bcpr" / "tools" / "max25-bcpr-ctl",
+        _P("/usr/local/sbin/max25-bcpr-ctl"),
+        _P("/usr/sbin/max25-bcpr-ctl"),
+    ):
+        if cand.is_file():
+            return cand
+    return None
+
+
+def bcpr_bc_index(cfg: DeviceBackendConfig) -> int:
+    tag = (cfg.max25_bcpr_device or cfg.bcpr_device or "").strip()
+    if not tag and cfg.device_id.startswith("max25e0:"):
+        tag = cfg.device_id.split(":", 1)[1]
+    if not tag:
+        tag = "bc0"
+    if tag.startswith("bc") and tag[2:].isdigit():
+        return int(tag[2:])
+    return 0
+
+
+def read_bcpr_state_dir(ini_path: Path) -> str:
+    cp = configparser.ConfigParser()
+    cp.read(ini_path)
+    for sect in ("max25-bcpr", "bcpr"):
+        if cp.has_section(sect) and cp.has_option(sect, "state_dir"):
+            return cp.get(sect, "state_dir").strip() or "/tmp/max25-bcpr"
+    return "/tmp/max25-bcpr"
+
+
+def sync_tot_to_bcpr_ini(ini_path: Path, tot: TotConfig) -> None:
+    """Push max25d [tot] policy into max25-bcpr.ini before bcprd start."""
+    if not ini_path.is_file():
+        return
+    cp = configparser.ConfigParser()
+    cp.read(ini_path)
+    sect = "max25-bcpr"
+    if not cp.has_section(sect):
+        if cp.has_section("bcpr"):
+            sect = "bcpr"
+        else:
+            cp.add_section(sect)
+    cp.set(sect, "tot", "yes" if tot.enabled else "no")
+    cp.set(sect, "tot_max_key_sec", str(tot.max_key_sec))
+    cp.set(sect, "tot_min_gap_ms", str(int(tot.min_gap_sec * 1000.0)))
+    cp.set(sect, "tot_max_consecutive", str(tot.max_consecutive))
+    cp.set(sect, "tot_max_bursts", str(tot.max_bursts))
+    with ini_path.open("w", encoding="utf-8") as fh:
+        cp.write(fh)
+
+
+def tot_trip_path(state_dir: str, bc_index: int) -> Path:
+    return Path(state_dir) / f"tot-trip-bc{bc_index}"
+
+
+def parse_tot_trip_file(path: Path) -> dict[str, str]:
+    out: dict[str, str] = {}
+    try:
+        for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+            if "=" not in line:
+                continue
+            key, val = line.split("=", 1)
+            out[key.strip()] = val.strip()
+    except OSError:
+        pass
+    return out
+
+
+def device_tot_operational(state: DaemonState, dev_id: str) -> bool:
+    rt = state.devices.get(dev_id)
+    if rt is None:
+        return False
+    return not rt.tot_paused
+
+
+def handle_tot_trip(state: DaemonState, dev_id: str, reason: str) -> None:
+    rt = state.devices.get(dev_id)
+    if rt is None or rt.tot_paused:
+        return
+    log(f"TOT trip ({dev_id}) reason={reason} — pause, stop stack, reset bcpr")
+    rt.tot_paused = True
+    rt.tot_trip_reason = reason or "unknown"
+    recover = state.cfg.tot.recover_sec
+    rt.tot_paused_until = time.time() + recover if recover > 0 else 0.0
+    stop_device_stack(state, dev_id)
+    rt.stack_status = "tot-paused"
+    rt.link_status = "tot-paused"
+    rt.prep_done = False
+    broadcast(
+        state,
+        f"EVENT device={dev_id} tot=trip reason={rt.tot_trip_reason} stack=tot-paused",
+    )
+
+
+def poll_tot_trips(state: DaemonState) -> None:
+    if not state.cfg.tot.enabled:
+        return
+    for dev_id, rt in state.devices.items():
+        kind = rt.cfg.backend_type or ""
+        if kind not in ("max25-bcpr-kiss", "bcpr-kiss"):
+            continue
+        if rt.tot_paused:
+            continue
+        explicit = rt.cfg.max25_bcpr_ini or rt.cfg.bcpr_ini or ""
+        resolved = resolve_max25_bcpr_ini(explicit)
+        if resolved is None:
+            continue
+        state_dir = read_bcpr_state_dir(resolved)
+        trip = tot_trip_path(state_dir, bcpr_bc_index(rt.cfg))
+        if not trip.is_file():
+            continue
+        meta = parse_tot_trip_file(trip)
+        if meta.get("tripped") != "1":
+            continue
+        handle_tot_trip(state, dev_id, meta.get("reason", "unknown"))
+
+
+def poll_tot_recovery(state: DaemonState) -> None:
+    if not state.cfg.tot.enabled:
+        return
+    now = time.time()
+    for dev_id, rt in state.devices.items():
+        if not rt.tot_paused:
+            continue
+        kind = rt.cfg.backend_type or ""
+        if kind not in ("max25-bcpr-kiss", "bcpr-kiss"):
+            continue
+        recover = state.cfg.tot.recover_sec
+        if recover <= 0:
+            continue
+        if rt.tot_paused_until > 0 and now < rt.tot_paused_until:
+            continue
+        explicit = rt.cfg.max25_bcpr_ini or rt.cfg.bcpr_ini or ""
+        resolved = resolve_max25_bcpr_ini(explicit)
+        if resolved is not None:
+            trip = tot_trip_path(read_bcpr_state_dir(resolved), bcpr_bc_index(rt.cfg))
+            try:
+                trip.unlink(missing_ok=True)
+            except OSError:
+                pass
+        rt.tot_paused = False
+        rt.tot_trip_reason = ""
+        rt.tot_paused_until = 0.0
+        log(f"TOT recover ({dev_id}) — restarting max25-bcpr stack")
+        if state.cfg.auto_start and backend_enabled(state, dev_id):
+            start_device_stack_for_tot(state, dev_id)
+
+
+def start_device_stack_for_tot(state: DaemonState, dev_id: str) -> None:
+    """Restart bcpr stack after TOT cooldown (single device, no full start_stacks)."""
+    rt = state.devices[dev_id]
+    kind = rt.cfg.backend_type or ""
+    if kind in ("max25-bcpr-kiss", "bcpr-kiss"):
+        explicit = rt.cfg.max25_bcpr_ini or rt.cfg.bcpr_ini or ""
+        resolved = resolve_max25_bcpr_ini(explicit)
+        if resolved is None:
+            rt.stack_status = "error-no-ini"
+            return
+        sync_tot_to_bcpr_ini(resolved, state.cfg.tot)
+        start_max25_bcpr_stack(state, dev_id, resolved)
+
+
+def _read_subprocess_capture(fh, limit: int = 4000) -> str:
+    """Read captured ctl stdout/stderr (UTF-8, truncated tail)."""
+    try:
+        fh.seek(0)
+        raw = fh.read()
+    except OSError:
+        return ""
+    text = raw.decode("utf-8", errors="replace").strip()
+    if len(text) > limit:
+        return f"...\n{text[-limit:]}"
+    return text
+
+
+def start_max25_bcpr_stack(state: DaemonState, dev_id: str, ini: Path) -> None:
+    """Start max25-bcprd once via max25-bcpr-ctl for shared ini."""
+    rt = state.devices[dev_id]
+    ctl = resolve_max25_bcpr_ctl()
+    if ctl is None:
+        rt.stack_status = "error-no-ctl"
+        log(f"max25-bcpr-ctl not found ({dev_id})")
+        return
+    args = [str(ctl), "-c", str(ini), "start"]
+    env = os.environ.copy()
+    # Temp file (not PIPE): bcprd backgrounded by ctl must not block communicate().
+    # Capture ctl output; log tail only when rc!=0 or after ctl timeout.
+    proc: subprocess.Popen[bytes] | None = None
+    ctl_output = ""
+    try:
+        with tempfile.TemporaryFile(mode="w+b") as capfh:
+            try:
+                proc = subprocess.Popen(
+                    args,
+                    cwd=str(ROOT),
+                    env=env,
+                    stdin=subprocess.DEVNULL,
+                    stdout=capfh,
+                    stderr=subprocess.STDOUT,
+                    start_new_session=True,
+                )
+            except OSError as exc:
+                log(f"max25-bcpr start failed ({dev_id}): {exc}")
+                rt.stack_status = "error"
+                return
+            try:
+                proc.communicate(timeout=30)
+            except subprocess.TimeoutExpired:
+                proc.kill()
+                try:
+                    proc.communicate(timeout=2)
+                except (subprocess.TimeoutExpired, OSError):
+                    pass
+                ctl_output = _read_subprocess_capture(capfh)
+                # May still have spawned max25-bcprd — fall through to kiss probe.
+                log(f"max25-bcpr-ctl start timed out ({dev_id}) — probing live kiss/pid")
+                if ctl_output:
+                    log(f"max25-bcpr-ctl output ({dev_id}): {ctl_output}")
+            else:
+                ctl_output = _read_subprocess_capture(capfh)
+    except OSError as exc:
+        log(f"max25-bcpr start failed ({dev_id}): {exc}")
+        rt.stack_status = "error"
+        return
+    if proc is None:
+        rt.stack_status = "error"
+        return
+    if proc.returncode not in (0, None, -9, -15):
+        # -9/-15: we killed a hung ctl; still probe kiss below.
+        if proc.returncode > 0:
+            rt.stack_status = "error"
+            if ctl_output:
+                log(f"max25-bcpr-ctl start rc={proc.returncode} ({dev_id}): {ctl_output}")
+            else:
+                log(f"max25-bcpr-ctl start rc={proc.returncode} ({dev_id})")
+            return
+    # max25-bcpr-ctl itself exits; live daemon is max25-bcprd (pidfile under state_dir).
+    rt.stack_proc = None
+    rt.stack_status = "running"
+    kiss = normalize_max25_bcpr_path(
+        (rt.cfg.kiss_link or "").strip() or MAX25_BCPR_KISS_DEFAULT
+    )
+    deadline = time.time() + 5.0
+    while time.time() < deadline:
+        if os.path.exists(kiss):
+            break
+        time.sleep(0.1)
+    if not os.path.exists(kiss):
+        rt.stack_status = "error-no-kiss"
+        log(f"max25-bcpr kiss_link missing after start ({dev_id}: {kiss})")
+        return
+    addrs = f"ipv4={rt.cfg.ipv4 or '-'} ipv6={rt.cfg.ipv6 or '-'}"
+    log(f"max25-bcpr started ({dev_id}, ini={ini}, kiss={kiss}, {addrs})")
+    rt.stack_status = "ready"
+    # HyBBX attach: HyBBX opens kiss_link — do not hold the PTY here.
+    if hybbx_release_attach(state):
+        rt.link_status = "ready"
+        rt.prep_done = True
+        log(f"max25-bcpr ready — HyBBX owns KISS attach ({dev_id}: {kiss})")
+        return
+    # Standalone: open+hold KISS so max25-terminal TX works (stack_proc=None).
+    if backend_enabled(state, dev_id):
+        if open_backend(state, dev_id):
+            if attach_backend_session(state, dev_id):
+                # UI/datagram TX uses CONNECT as session arm — arm at start so
+                # max25-terminal SEND keys MCR without a separate CONNECT race.
+                state.connected = True
+                log(f"max25-bcpr KISS open ({dev_id}: {kiss})")
+            else:
+                log(f"max25-bcpr KISS open but attach failed ({dev_id})")
+        else:
+            log(f"max25-bcpr KISS open failed ({dev_id}) status={rt.link_status}")
+
+
+def start_device_stack(state: DaemonState, dev_id: str) -> None:
+    if uses_inline_tnc_prep(state, dev_id):
+        prep_inline_serial_device(state, dev_id)
+        return
+    rt = state.devices[dev_id]
+    ctl = ctl_path(ROOT, PREFIX, _EXE)
+    if not ctl.is_file():
+        rt.stack_status = "error-no-ctl"
+        return
+    hw = device_hardware(state, dev_id)
+    dev_cfg = rt.cfg
+    ctl_device = dev_id
+    if (dev_cfg.backend_type or "") == "baycom-kiss":
+        ctl_device = baycom_ctl_device_id(dev_cfg)
+    args = [
+        str(ctl),
+        "start",
+        "--mode",
+        state.cfg.mode,
+        "--hardware",
+        hw,
+        "--device",
+        ctl_device,
+    ]
+    kind = dev_cfg.backend_type or ""
+    if kind == "baycom-kiss":
+        explicit = dev_cfg.baycom_ini or ""
+        resolved = resolve_baycom_ini(dev_id, ROOT, PREFIX, explicit)
+        if resolved:
+            args.extend(["--baycom-ini", str(resolved)])
+    env = os.environ.copy()
+    env["MAX25_MODE"] = state.cfg.mode
+    if hw == "tncs" and state.cfg.stack_recover_only:
+        env["MAX25_TNC_PREP"] = "recover"
+    workdir = str(ROOT if (ROOT / "plugins").is_dir() else (PREFIX or ROOT))
+    try:
+        proc = subprocess.Popen(
+            args,
+            cwd=workdir,
+            env=env,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            start_new_session=True,
+        )
+    except OSError as exc:
+        log(f"stack start failed ({dev_id}): {exc}")
+        rt.stack_status = "error"
+        return
+    rt.stack_proc = proc
+    rt.stack_status = "running"
+    log(f"stack started pid={proc.pid} ({hw}/{dev_id})")
+
+
+def start_stacks(state: DaemonState) -> None:
+    """Start per-device stacks; one max25-bcpr-ctl per shared ini."""
+    started_baycom_ini: dict[str, str] = {}
+    started_max25_bcpr_ini: dict[str, str] = {}
+    for dev_id in enabled_device_ids(state):
+        rt = state.devices[dev_id]
+        kind = rt.cfg.backend_type or ""
+        if uses_inline_tnc_prep(state, dev_id):
+            prep_inline_serial_device(state, dev_id)
+            continue
+        if kind in ("max25-bcpr-kiss", "bcpr-kiss"):
+            explicit = rt.cfg.max25_bcpr_ini or rt.cfg.bcpr_ini or ""
+            resolved = resolve_max25_bcpr_ini(explicit)
+            ini_key = str(resolved) if resolved else ""
+            if ini_key and ini_key in started_max25_bcpr_ini:
+                primary = started_max25_bcpr_ini[ini_key]
+                primary_rt = state.devices[primary]
+                rt.stack_proc = primary_rt.stack_proc
+                rt.stack_status = primary_rt.stack_status
+                log(f"max25-bcpr stack shared with {primary} ({dev_id}, ini={ini_key})")
+                # Shared max25-bcprd — open this device's kiss only when max25d owns it.
+                if (
+                    not hybbx_release_attach(state)
+                    and rt.stack_status in ("ready", "running")
+                    and backend_enabled(state, dev_id)
+                ):
+                    if open_backend(state, dev_id):
+                        attach_backend_session(state, dev_id)
+                elif hybbx_release_attach(state):
+                    rt.link_status = "ready"
+                    rt.prep_done = True
+                continue
+            if resolved:
+                sync_tot_to_bcpr_ini(resolved, state.cfg.tot)
+                start_max25_bcpr_stack(state, dev_id, resolved)
+                started_max25_bcpr_ini[str(resolved)] = dev_id
+            else:
+                rt.stack_status = "error-no-ini"
+                log(f"max25-bcpr.ini not found ({dev_id})")
+            continue
+        if kind == "baycom-kiss":
+            explicit = rt.cfg.baycom_ini or ""
+            resolved = resolve_baycom_ini(dev_id, ROOT, PREFIX, explicit)
+            ini_key = str(resolved) if resolved else ""
+            if ini_key and ini_key in started_baycom_ini:
+                primary = started_baycom_ini[ini_key]
+                primary_rt = state.devices[primary]
+                rt.stack_proc = primary_rt.stack_proc
+                rt.stack_status = primary_rt.stack_status
+                log(f"stack shared with {primary} ({dev_id}, ini={ini_key})")
+                continue
+        start_device_stack(state, dev_id)
+        if kind == "baycom-kiss":
+            explicit = rt.cfg.baycom_ini or ""
+            resolved = resolve_baycom_ini(dev_id, ROOT, PREFIX, explicit)
+            if resolved:
+                started_baycom_ini[str(resolved)] = dev_id
+
+
+def stop_device_stack(state: DaemonState, dev_id: str) -> None:
+    close_backend(state, dev_id)
+    rt = state.devices[dev_id]
+    kind = rt.cfg.backend_type or ""
+    if kind in ("max25-bcpr-kiss", "bcpr-kiss"):
+        ctl = resolve_max25_bcpr_ctl()
+        explicit = rt.cfg.max25_bcpr_ini or rt.cfg.bcpr_ini or ""
+        resolved = resolve_max25_bcpr_ini(explicit)
+        if ctl is not None and resolved is not None:
+            subprocess.run(
+                [str(ctl), "-c", str(resolved), "stop"],
+                cwd=str(ROOT),
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+            )
+        rt.stack_proc = None
+        rt.stack_status = "stopped"
+        return
+    proc = rt.stack_proc
+    if proc is not None and proc.poll() is None:
+        try:
+            os.killpg(proc.pid, signal.SIGTERM)
+        except ProcessLookupError:
+            pass
+        except OSError:
+            proc.terminate()
+    rt.stack_proc = None
+    rt.stack_status = "stopped"
+    hw = device_hardware(state, dev_id)
+    ctl = ctl_path(ROOT, PREFIX, _EXE)
+    if ctl.is_file():
+        workdir = str(ROOT if (ROOT / "plugins").is_dir() else (PREFIX or ROOT))
+        stop_args = [str(ctl), "stop", "--hardware", hw, "--device", dev_id]
+        if kind == "baycom-kiss":
+            explicit = rt.cfg.baycom_ini or ""
+            resolved = resolve_baycom_ini(dev_id, ROOT, PREFIX, explicit)
+            if resolved:
+                stop_args.extend(["--baycom-ini", str(resolved)])
+        subprocess.run(
+            stop_args,
+            cwd=workdir,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+
+
+def stop_stacks(state: DaemonState) -> None:
+    for dev_id in list(state.devices):
+        stop_device_stack(state, dev_id)
+    log("all stacks stopped")
+
+
+def poll_device_stack(state: DaemonState, dev_id: str) -> None:
+    rt = state.devices[dev_id]
+    proc = rt.stack_proc
+    if proc is None:
+        return
+    rc = proc.poll()
+    if rc is None:
+        return
+    rt.stack_proc = None
+    if rc == 0:
+        rt.stack_status = "ready"
+        log(f"stack boot-wait finished ({dev_id}) rc={rc}")
+        if backend_enabled(state, dev_id):
+            open_backend(state, dev_id)
+            backend = rt.backend
+            stabilize = getattr(backend, "stabilize_session", None) if backend else None
+            if stabilize is not None:
+                ok = stabilize(state.cfg.callerid, force=False)
+                rt.prep_done = True
+                rt.link_status = backend.status
+                rt.inline_repair_failures = 0
+                if ok:
+                    log(f"serial post boot-wait OK ({dev_id})")
+                    if state.connected:
+                        attach_backend_session(state, dev_id)
+                else:
+                    log(
+                        f"serial post boot-wait deferred ({dev_id}) "
+                        f"status={backend.status}"
+                    )
+    else:
+        rt.stack_status = f"error-rc{rc}"
+        log(f"stack boot-wait failed ({dev_id}) rc={rc}")
+
+
+def poll_stacks(state: DaemonState) -> None:
+    for dev_id in enabled_device_ids(state):
+        poll_device_stack(state, dev_id)
+    retry_pending_backends(state)
+
+
+def retry_pending_backends(state: DaemonState) -> None:
+    """Re-attach KISS PTY/serial when stack is up but the link was not ready yet."""
+    for dev_id in enabled_device_ids(state):
+        if not backend_enabled(state, dev_id):
+            continue
+        rt = state.devices[dev_id]
+        if hybbx_host_hybbx_owns_serial(state, dev_id):
+            continue
+        # bcpr uses "running" until kiss open flips to "ready"; accept both.
+        if rt.stack_status not in ("ready", "stopped", "running"):
+            continue
+        if not backend_needs_open(rt.backend):
+            continue
+        if open_backend(state, dev_id):
+            attach_backend_session(state, dev_id)
+            if rt.stack_status == "running":
+                rt.stack_status = "ready"
+
+
+def poll_reporting_passes(state: DaemonState) -> None:
+    """Advance timed data-quality pass windows (default 20s each, 3 passes)."""
+    now = time.time()
+    for rt in state.devices.values():
+        _sync_quality_tracker(state, rt)
+        rt.quality.tick(now)
+
+
+def poll_serial_stability(state: DaemonState) -> None:
+    """Periodic TNC health probe + software recovery (no power cycle)."""
+    cfg = state.cfg
+    if not cfg.serial_watch:
+        return
+    now = time.time()
+    if state.started_at and now - state.started_at < cfg.serial_watch_startup_grace:
+        return
+    for dev_id in enabled_device_ids(state):
+        if device_backend_kind(state, dev_id) != "kiss-serial":
+            continue
+        if hybbx_host_hybbx_owns_serial(state, dev_id):
+            continue
+        rt = state.devices[dev_id]
+        if rt.stack_proc is not None and rt.stack_proc.poll() is None:
+            continue
+        if (
+            not uses_inline_tnc_prep(state, dev_id)
+            and cfg.stack_recover_only
+            and rt.stack_status.startswith("error")
+            and rt.stack_proc is None
+            and now - rt.last_stack_retry >= cfg.stack_retry_interval
+        ):
+            rt.last_stack_retry = now
+            log(f"serial watch: stack retry recover-only ({dev_id})")
+            start_device_stack(state, dev_id)
+            continue
+        if not backend_enabled(state, dev_id):
+            continue
+        backend = rt.backend
+        force = backend is not None and backend.status in SERIAL_REPAIR_STATUSES
+        due = now - rt.last_watch >= cfg.serial_watch_interval
+        if not force and not due:
+            continue
+        if backend is None:
+            if backend_needs_open(None):
+                open_backend(state, dev_id)
+                backend = rt.backend
+            if backend is None:
+                continue
+        if now - rt.last_repair < cfg.serial_repair_cooldown and not force:
+            continue
+        if backend.status == "ready" and not force:
+            if due:
+                rt.last_watch = now
+            continue
+        stabilize = getattr(backend, "stabilize_session", None)
+        if stabilize is None:
+            continue
+        rt.last_watch = now
+        rt.last_repair = now
+        ok = stabilize(state.cfg.callerid, force=force)
+        rt.link_status = backend.status
+        if ok:
+            rt.inline_repair_failures = 0
+            if force:
+                log(f"serial watch: repaired ({dev_id})")
+                broadcast(state, f"EVENT device={dev_id} serial=ready")
+        else:
+            log(f"serial watch: repair failed ({dev_id}) status={backend.status}")
+            if (
+                uses_inline_tnc_prep(state, dev_id)
+                and backend.status == "error-host"
+                and cfg.serial_bootwait_escalate
+            ):
+                rt.inline_repair_failures += 1
+                if (
+                    rt.inline_repair_failures >= cfg.serial_bootwait_escalate_after
+                    and now - rt.last_bootwait_escalate >= cfg.serial_bootwait_escalate_cooldown
+                ):
+                    rt.last_bootwait_escalate = now
+                    rt.inline_repair_failures = 0
+                    escalate_to_bootwait_stack(state, dev_id)
+                elif rt.inline_repair_failures >= cfg.serial_bootwait_escalate_after:
+                    log(
+                        f"serial watch: boot-wait escalate cooldown ({dev_id}) "
+                        f"— manual: stacks/tncs/{dev_id}-boot-wait.sh"
+                    )
+            if backend.status in ("error-io", "error-open", "error-no-device"):
+                close_backend(state, dev_id)
+                if open_backend(state, dev_id) and state.connected:
+                    attach_backend_session(state, dev_id)
+
+
+def format_tx(state: DaemonState, text: str) -> str:
+    if state.cfg.ax25_ui:
+        return f"[AX25 UI {state.cfg.callerid}>{state.cfg.callid}] {text}"
+    return text
+
+
+def resolve_selected_device(state: DaemonState) -> Optional[str]:
+    dev_id = state.selected_device
+    if dev_id in state.devices:
+        return dev_id
+    ids = enabled_device_ids(state)
+    return ids[0] if ids else None
+
+
+def device_line(state: DaemonState, dev_id: str) -> str:
+    rt = state.devices[dev_id]
+    link = backend_serial_label(rt.backend) if rt.backend is not None else rt.link_status
+    hw = device_hardware(state, dev_id)
+    backend = rt.cfg.backend_type or "auto"
+    enabled = "yes" if rt.cfg.enabled else "no"
+    return (
+        f"DEVICE id={dev_id} hardware={hw} backend={backend} serial={link} "
+        f"stack={rt.stack_status} enabled={enabled} "
+        f"error={device_reporting_error(state, rt)} "
+        f"voice={device_reporting_voice(state, rt)}"
+    )
+
+
+def handle_command(state: DaemonState, sock: socket.socket, line: str) -> None:
+    line = line.strip("\r\n")
+    if not line:
+        return
+    upper = line.upper()
+
+    if upper == "PING":
+        send_line(sock, "OK")
+        return
+
+    if upper == "GET STATUS":
+        send_line(sock, status_line(state))
+        send_line(sock, "OK")
+        return
+
+    if upper == "GET DEVICES":
+        for dev_id in sorted(state.devices):
+            send_line(sock, device_line(state, dev_id))
+        send_line(sock, "OK")
+        return
+
+    if upper.startswith("SET DEVICE ") or upper.startswith("SELECT DEVICE "):
+        prefix = "SET DEVICE " if upper.startswith("SET DEVICE ") else "SELECT DEVICE "
+        dev_id = line[len(prefix) :].strip()
+        if dev_id not in state.devices:
+            send_line(sock, f"ERR unknown device: {dev_id}")
+            return
+        state.selected_device = dev_id
+        state.cfg.device = dev_id
+        send_line(sock, "OK")
+        return
+
+    if upper.startswith("SET CALLERID "):
+        value = line[13:].strip().upper()
+        if not valid_callsign(value):
+            send_line(sock, "ERR invalid CALLERID")
+            return
+        state.cfg.callerid = value
+        send_line(sock, "OK")
+        return
+
+    if upper.startswith("SET CALLID "):
+        value = line[11:].strip().upper()
+        if not valid_callsign(value):
+            send_line(sock, "ERR invalid CALLID")
+            return
+        state.cfg.callid = value
+        send_line(sock, "OK")
+        return
+
+    if upper.startswith("SET AX25_UI "):
+        flag = line[12:].strip().lower()
+        if flag in ("on", "yes", "1", "true"):
+            state.cfg.ax25_ui = True
+        elif flag in ("off", "no", "0", "false"):
+            state.cfg.ax25_ui = False
+        else:
+            send_line(sock, "ERR ax25_ui on|off")
+            return
+        send_line(sock, "OK")
+        return
+
+    if upper == "CONNECT":
+        if not attach_all_sessions(state):
+            send_line(sock, "ERR link not ready")
+            return
+        state.connected = True
+        send_line(sock, "EVENT connected")
+        send_line(sock, "OK")
+        return
+
+    if upper == "DISCONNECT":
+        detach_all_sessions(state)
+        state.connected = False
+        send_line(sock, "EVENT disconnected")
+        send_line(sock, "OK")
+        return
+
+    if upper.startswith("MONITOR "):
+        flag = line[8:].strip().lower()
+        state.monitor_only = flag in ("on", "yes", "1", "true")
+        send_line(sock, "OK")
+        return
+
+    if upper.startswith("BAN "):
+        value = line[4:].strip().upper()
+        if not valid_callsign(value):
+            send_line(sock, "ERR invalid callsign")
+            return
+        try:
+            state.bans.add(value)
+        except OSError as exc:
+            send_line(sock, f"ERR ban save failed: {exc}")
+            return
+        send_line(sock, "OK")
+        return
+
+    if upper.startswith("UNBAN "):
+        value = line[6:].strip().upper()
+        if not valid_callsign(value):
+            send_line(sock, "ERR invalid callsign")
+            return
+        try:
+            if not state.bans.remove(value):
+                send_line(sock, "ERR not banned")
+                return
+        except OSError as exc:
+            send_line(sock, f"ERR ban save failed: {exc}")
+            return
+        send_line(sock, "OK")
+        return
+
+    if upper == "BANS":
+        for entry in state.bans.list():
+            send_line(sock, f"BAN {entry}")
+        send_line(sock, "OK")
+        return
+
+    if upper.startswith("SEND "):
+        if state.monitor_only:
+            send_line(sock, "ERR monitor-only")
+            return
+        # UI frames: auto-arm session if stack/KISS is up. Terminal Enter/F10→SEND
+        # must key PTT/MCR without requiring a prior CONNECT (L4 writes kiss
+        # directly and already keys; unix SEND must match).
+        if not state.connected:
+            if not attach_all_sessions(state):
+                send_line(sock, "ERR not connected")
+                return
+            state.connected = True
+            send_line(sock, "EVENT connected")
+        dev_id = resolve_selected_device(state)
+        if dev_id is None:
+            send_line(sock, "ERR no device configured")
+            return
+        payload = line[5:]
+        framed = format_tx(state, payload)
+        rt = state.devices[dev_id]
+        if backend_enabled(state, dev_id):
+            if rt.backend is None or rt.backend.status != "ready":
+                # Re-attach after DISCONNECT left kiss inactive but stack ready.
+                if not attach_backend_session(state, dev_id):
+                    send_line(sock, "ERR link not ready")
+                    return
+            if rt.backend is None or rt.backend.status != "ready":
+                send_line(sock, "ERR link not ready")
+                return
+            ok, display = rt.backend.transmit(
+                state.cfg.callerid,
+                state.cfg.callid,
+                payload,
+                state.cfg.ax25_ui,
+            )
+            if not ok and hasattr(rt.backend, "stabilize_session"):
+                log(f"serial watch: tx retry after repair ({dev_id})")
+                if rt.backend.stabilize_session(state.cfg.callerid, force=True):
+                    rt.link_status = rt.backend.status
+                    ok, display = rt.backend.transmit(
+                        state.cfg.callerid,
+                        state.cfg.callid,
+                        payload,
+                        state.cfg.ax25_ui,
+                    )
+            if not ok:
+                send_line(sock, f"ERR {display}")
+                return
+            framed = f"device={dev_id} {display}"
+        log(f"tx {dev_id}: {framed}")
+        send_line(sock, f"RX {framed}")
+        broadcast(state, f"RX {framed}", skip=sock)
+        send_line(sock, "OK")
+        return
+
+    send_line(sock, f"ERR unknown command: {line.split()[0]}")
+
+
+def tcp_auth_ok(sock: socket.socket, expected: str, timeout: float = 30.0) -> bool:
+    if not expected:
+        return True
+    send_line(sock, "AUTH required")
+    sock.settimeout(timeout)
+    buf = b""
+    try:
+        while True:
+            try:
+                chunk = sock.recv(4096)
+            except socket.timeout:
+                return False
+            if not chunk:
+                return False
+            buf += chunk
+            if len(buf) > M25_MAX_LINE_BUF:
+                return False
+            while b"\n" in buf:
+                raw, buf = buf.split(b"\n", 1)
+                try:
+                    line = raw.decode("utf-8").strip("\r")
+                except UnicodeDecodeError:
+                    return False
+                if not line:
+                    continue
+                if line.upper().startswith("AUTH "):
+                    supplied = line[5:]
+                    return supplied == expected
+                return False
+    finally:
+        sock.settimeout(300.0)
+
+
+def client_thread(state: DaemonState, sock: socket.socket, from_tcp: bool) -> None:
+    sock.settimeout(300.0)
+    buf = b""
+    try:
+        if from_tcp and state.cfg.tcp_password:
+            if not tcp_auth_ok(sock, state.cfg.tcp_password):
+                send_line(sock, "ERR auth failed")
+                return
+        send_line(sock, "OK")
+        send_line(sock, status_line(state))
+        while True:
+            try:
+                chunk = sock.recv(4096)
+            except socket.timeout:
+                continue
+            if not chunk:
+                break
+            buf += chunk
+            if len(buf) > M25_MAX_LINE_BUF:
+                send_line(sock, "ERR line too long")
+                break
+            while b"\n" in buf:
+                raw, buf = buf.split(b"\n", 1)
+                try:
+                    line = raw.decode("utf-8")
+                except UnicodeDecodeError:
+                    send_line(sock, "ERR invalid utf-8")
+                    continue
+                handle_command(state, sock, line)
+    except OSError:
+        pass
+    finally:
+        with state.lock:
+            state.clients.discard(sock)
+        try:
+            sock.close()
+        except OSError:
+            pass
+
+
+def serve(
+    state: DaemonState,
+    listeners: list[tuple[socket.socket, bool, tuple[int, int] | None]],
+) -> None:
+    running = True
+
+    def on_signal(_signum, _frame):
+        nonlocal running
+        running = False
+
+    signal.signal(signal.SIGTERM, on_signal)
+    signal.signal(signal.SIGINT, on_signal)
+
+    if state.cfg.auto_start:
+        start_stacks(state)
+
+    state.started_at = time.time()
+
+    device_lines: list[tuple[str, str, str]] = []
+    for dev_id in enabled_device_ids(state):
+        rt = state.devices[dev_id]
+        link = backend_serial_label(rt.backend) if rt.backend is not None else rt.link_status
+        device_lines.append((dev_id, rt.stack_status, link))
+
+    emit_startup_complete(
+        device_lines=device_lines,
+        tcp_host=state.cfg.tcp_host,
+        tcp_port=state.cfg.tcp_port,
+        unix_socket=state.cfg.unix_socket,
+    )
+
+    drop_privileges_or_exit(
+        RunAsConfig(
+            user=state.cfg.run_user,
+            group=state.cfg.run_group,
+            uid=state.cfg.run_uid,
+            gid=state.cfg.run_gid,
+        ),
+        log=lambda msg, area="privilege": LOGGER.info(msg, area=area),
+    )
+
+    while running:
+        poll_stacks(state)
+        poll_tot_trips(state)
+        poll_tot_recovery(state)
+        poll_reporting_passes(state)
+        poll_serial_stability(state)
+        socks = [lsock for lsock, _tcp, _bid in listeners]
+        rlist, _, _ = select.select(socks, [], [], 1.0)
+        for lsock, from_tcp, _bid in listeners:
+            if lsock not in rlist:
+                continue
+            try:
+                client, _addr = lsock.accept()
+            except OSError:
+                continue
+            client.setblocking(True)
+            with state.lock:
+                state.clients.add(client)
+            threading.Thread(
+                target=client_thread,
+                args=(state, client, from_tcp),
+                daemon=True,
+            ).start()
+
+    stop_stacks(state)
+    for lsock, from_tcp, bind_id in listeners:
+        if not from_tcp and state.cfg.unix_socket:
+            unlink_unix_if_ours(state.cfg.unix_socket, bind_id)
+        lsock.close()
+
+
+def make_listeners(
+    cfg: DaemonConfig,
+) -> list[tuple[socket.socket, bool, tuple[int, int] | None]]:
+    listeners: list[tuple[socket.socket, bool, tuple[int, int] | None]] = []
+
+    tcp = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    tcp.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    try:
+        tcp.bind((cfg.tcp_host, cfg.tcp_port))
+    except OSError as exc:
+        tcp.close()
+        log(f"TCP bind {cfg.tcp_host}:{cfg.tcp_port} failed: {exc}")
+        raise SystemExit(1) from exc
+    tcp.listen(32)
+    tcp.setblocking(False)
+    listeners.append((tcp, True, None))
+
+    if cfg.unix_socket:
+        sock_path = Path(cfg.unix_socket)
+        try:
+            sock_path.parent.mkdir(parents=True, exist_ok=True)
+        except OSError:
+            fallback = Path("/tmp/max25/modem.sock")
+            log(f"unix {cfg.unix_socket} unavailable, using {fallback}")
+            cfg.unix_socket = str(fallback)
+            sock_path = fallback
+            try:
+                sock_path.parent.mkdir(parents=True, exist_ok=True)
+            except OSError:
+                log("unix socket disabled (no writable path)")
+                cfg.unix_socket = ""
+                return listeners
+        # Never unlink a live peer path — that orphans the other max25d FD
+        # (ss still shows the name; clients get ENOENT).
+        if unix_path_is_live(cfg.unix_socket):
+            log(
+                f"unix socket {cfg.unix_socket} already live — not stealing "
+                "(refuse second max25d unix bind)"
+            )
+        else:
+            try:
+                os.unlink(cfg.unix_socket)
+            except FileNotFoundError:
+                pass
+            except OSError:
+                pass
+            unix = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            try:
+                unix.bind(cfg.unix_socket)
+            except OSError as exc:
+                log(f"unix socket {cfg.unix_socket} skipped ({exc})")
+                unix.close()
+            else:
+                try:
+                    os.chmod(cfg.unix_socket, 0o660)
+                except OSError:
+                    pass
+                unix.listen(32)
+                unix.setblocking(False)
+                listeners.append((unix, False, unix_path_id(cfg.unix_socket)))
+
+    return listeners
+
+
+def main(argv: Optional[list[str]] = None) -> int:
+    if not max25d_supported():
+        log(f"max25d is not supported on {sys.platform}")
+        return 1
+
+    parser = argparse.ArgumentParser(description=f"MAX25 daemon ({platform_label()})")
+    parser.add_argument(
+        "-c",
+        "--config",
+        type=Path,
+        default=None,
+        help="Path to max25d.ini",
+    )
+    parser.add_argument(
+        "--no-stack",
+        action="store_true",
+        help="Do not auto-start hardware stack",
+    )
+    parser.add_argument(
+        "--tcp-port",
+        type=int,
+        default=None,
+        help="Override TCP listen port",
+    )
+    parser.add_argument(
+        "--no-serial",
+        action="store_true",
+        help="Disable KISS serial bridge (loopback SEND only)",
+    )
+    parser.add_argument(
+        "--session",
+        choices=("tmux", "screen"),
+        metavar="BACKEND",
+        help="Re-exec via max25d-session (detach in tmux/screen); use max25d-session attach",
+    )
+    args = parser.parse_args(argv)
+
+    if args.session:
+        session_sh = ROOT / "scripts" / "max25d-session.sh"
+        if not session_sh.is_file():
+            session_sh = Path(PREFIX) / "bin" / "max25d-session" if PREFIX else session_sh
+        if not session_sh.is_file():
+            LOGGER.error(
+                "max25d-session not found — install scripts/max25d-session.sh or use tmux/screen manually",
+                area="session",
+            )
+            return 1
+        cmd = [str(session_sh), "start", f"--{args.session}"]
+        if args.config:
+            cmd.extend(["-c", str(args.config)])
+        os.execv(cmd[0], cmd)
+
+    cfg = load_config(args.config)
+    if args.no_stack:
+        cfg.auto_start = False
+    if args.tcp_port is not None:
+        cfg.tcp_port = args.tcp_port
+    if args.no_serial:
+        cfg.serial_enabled = False
+    enforce_lite_localhost(cfg)
+
+    if cfg.modular_tcp.enabled and cfg.modular_tcp.role == "main":
+        svc = ModularTcpMainService(cfg.modular_tcp, cfg.tcp_host, cfg.tcp_port, log)
+        svc.start()
+        log(
+            f"modular TCP/IP Servers Service — Main '{cfg.modular_tcp.service_name}' "
+            f"({len(cfg.modular_tcp.secondaries)} secondaries)"
+        )
+        try:
+            while True:
+                time.sleep(1.0)
+        except KeyboardInterrupt:
+            pass
+        finally:
+            svc.stop()
+        return 0
+
+    state = DaemonState(cfg=cfg, bans=BanList(cfg.bans_file))
+    init_device_runtimes(state)
+    emit_startup_banner(
+        config_path=cfg.config_path or None,
+        cfg=cfg,
+        devices=cfg.devices,
+        tested_fn=registry_tested,
+    )
+    listeners = make_listeners(cfg)
+    try:
+        serve(state, listeners)
+    except KeyboardInterrupt:
+        stop_stacks(state)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())