path: root/stacks/daemon/max25d.py

#!/usr/bin/env python3
"""
max25d — MainAX25-Stack daemon.

Linux/KLinux: full hardware stack. FreeBSD: server + CRDOP/OSS (modular TCP/IP service).
"""
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 = "0.0.0.0"
    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
    return cfg


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

    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())