"""Main GUI composition root.

This module wires UI/controller mixins together and owns application-level
state shared across them (runtime services, readers, history buffers, timer).
"""

from __future__ import annotations

from collections import deque
from contextlib import suppress
from datetime import datetime
import html
import json
import logging
import os
from pathlib import Path
import sys
import traceback

from PyQt6.QtCore import QObject, QTimer, pyqtSignal
from PyQt6.QtGui import QTextCursor
from PyQt6.QtWidgets import QApplication, QMainWindow, QMessageBox

from python_app.gui.controllers.app_window_config_mixin import AppWindowConfigMixin
from python_app.gui.controllers.app_window_control_button_mixin import AppWindowControlButtonMixin
from python_app.gui.controllers.app_window_pipeline_mixin import AppWindowPipelineMixin
from python_app.gui.controllers.app_window_plot_mixin import AppWindowPlotMixin
from python_app.gui.controllers.app_window_recording_mixin import AppWindowRecordingMixin
from python_app.gui.controllers.app_window_preprocess_mixin import AppWindowPreprocessMixin
from python_app.gui.controllers.app_window_snapshot_mixin import AppWindowSnapshotMixin
from python_app.gui.controllers.app_window_ui_mixin import AppWindowUiMixin
from python_app.gui.controllers.app_window_web_mixin import AppWindowWebMixin
from python_app.gui.preprocess_dialog import PreprocessDialog
from python_app.models.dataset_model import ResultCollection, SweepCollection
from python_app.models.gui_profile_model import GuiProfileModel
from python_app.orchestration.config_writer import ConfigWriter
from python_app.orchestration.gui_session_state import GuiSessionState, GuiSessionStateStore
from python_app.orchestration.live_processing_config import ProcessingLiveConfigWriter
from python_app.orchestration.pipeline_metrics import PipelineMetrics
from python_app.orchestration.preprocess_assets import VISIBLE_PREPROCESS_ASSET_KEYS, preprocess_asset_model
from python_app.orchestration.process_supervisor import ProcessSupervisor
from python_app.orchestration.shm_reader import ShmRingReader
from python_app.logging_setup import (
    DEFAULT_LOG_LEVEL,
    add_handler,
    configure_logging,
    get_logger,
    set_log_level,
)
from python_app.storage.npz_store import NpzStore
from python_app.workflows.multi_radar_capture_workflow import MultiRadarSequentialCaptureSession
from python_app.workflows.radar_config_variants import RadarConfigScanSummary, RadarConfigVariant
from python_app.workflows.sequential_capture_workflow import SequentialCaptureSession


def _panel_extra(details: str | None, once_key: str | None) -> dict[str, object]:
    """Carry GUI-panel-only fields (details block, once-key dedup) on a log record."""
    return {"panel_details": details, "panel_once_key": once_key}


class _PanelLogBridge(QObject):
    """Marshals log records from any thread onto the GUI thread for panel rendering.

    A :class:`logging.Handler` can fire on a worker thread (readers, broadcaster),
    but the log widget may only be touched on the GUI thread; emitting this queued
    signal hands the record across safely (the GPIO-button pattern).
    """

    record = pyqtSignal(str, str, object, object)  # display level, message, details, once_key


class _QtLogPanelHandler(logging.Handler):
    """Logging handler that forwards application log records to the GUI log panel."""

    def __init__(self, bridge: _PanelLogBridge) -> None:
        super().__init__()
        self._bridge = bridge

    def emit(self, record: logging.LogRecord) -> None:
        """Forward one record to the panel bridge, mapping WARNING to the short 'WARN'."""
        try:
            display_level = "WARN" if record.levelname == "WARNING" else record.levelname
            self._bridge.record.emit(
                display_level,
                record.getMessage(),
                getattr(record, "panel_details", None),
                getattr(record, "panel_once_key", None),
            )
        except Exception:  # noqa: BLE001 - logging must never raise into the caller
            self.handleError(record)


class AppWindow(
    AppWindowUiMixin,
    AppWindowConfigMixin,
    AppWindowPreprocessMixin,
    AppWindowPlotMixin,
    AppWindowPipelineMixin,
    AppWindowSnapshotMixin,
    AppWindowRecordingMixin,
    AppWindowControlButtonMixin,
    AppWindowWebMixin,
    QMainWindow,
):
    """Top-level window coordinating GUI state and acquisition runtime."""

    def __init__(self, project_root: Path) -> None:
        """Initialize all app subsystems in deterministic order."""
        super().__init__()

        self._init_paths(project_root)
        self._init_logging()
        self._init_runtime_services()
        self._init_config_profile_state()
        self._init_reader_handles()
        self._init_preprocess_state()
        self._init_capture_state()
        self._init_history_state()
        self._init_recording_state()
        self._init_runtime_limits()
        self._init_polling_timer()
        self._init_control_button_state()
        self._bootstrap_ui_runtime()

    def _init_paths(self, project_root: Path) -> None:
        """Initialize static project paths and startup log queue."""
        self._project_root = project_root
        self._root_profile_path = project_root / "run_config.json"
        self._active_profile_path = self._root_profile_path
        # Run-config profiles the web UI can browse and load (it has no file dialog).
        self._run_configs_dir = project_root / "run_configs"
        self._pending_startup_log_entries: list[tuple[str, str, str | None]] = []
        # Guards closeEvent against re-entrant teardown (e.g. a second signal).
        self._closing = False

    def _init_logging(self) -> None:
        """Configure the application logger and the bridge that feeds the GUI panel.

        Installs the rotating-file (``runtime/logs/radar.log``) and stderr handlers on
        the ``python_app`` logger so every module's logs — and the GUI's own ``_log*``
        calls — share one level-controlled, rotated pipeline. The verbosity floor starts
        at the default and is replaced with the configured level once run_config loads;
        the GUI panel is wired in once its widget exists (see ``_attach_log_panel``).
        """
        log_dir = self._project_root / "python_app/runtime/logs"
        configure_logging(level=DEFAULT_LOG_LEVEL, log_dir=log_dir, console=True)
        self._gui_logger = get_logger("gui")
        self._log_panel_bridge = _PanelLogBridge()
        self._log_panel_bridge.record.connect(self._on_log_record)

    def _attach_log_panel(self) -> None:
        """Route application log records into the on-screen panel (widget now exists)."""
        add_handler(_QtLogPanelHandler(self._log_panel_bridge))

    def _on_log_record(self, level: str, text: str, details: object, once_key: object) -> None:
        """Render one forwarded log record in the panel (always on the GUI thread)."""
        if not hasattr(self, "_log_box"):
            return
        self._append_log_entry(
            level,
            text,
            details=details if isinstance(details, str) else None,
            once_key=once_key if isinstance(once_key, str) else None,
        )

    def _init_runtime_services(self) -> None:
        """Initialize long-lived service objects used by mixins."""
        runtime_dir = self._project_root / "python_app/runtime"
        self._runtime_dir = runtime_dir
        self._store = NpzStore(self._project_root / "python_app/data")
        self._config_writer = ConfigWriter(runtime_dir)
        self._supervisor = ProcessSupervisor(self._project_root)
        self._live_config_writer = ProcessingLiveConfigWriter(runtime_dir / "processing_live.json")
        self._gui_session_state_store = GuiSessionStateStore(runtime_dir / "gui_session_state.json")
        # `log_sink` is attached after the runtime log widget exists.
        self._pipeline_metrics = PipelineMetrics(
            report_every=self._resolve_metrics_report_every()
        )

    @staticmethod
    def _resolve_metrics_report_every() -> int:
        """Read the metrics flush threshold from env, falling back to 50."""
        raw = os.environ.get("RADAR_SYSTEM_METRICS_REPORT_EVERY", "").strip()
        if not raw:
            return 50
        try:
            value = int(raw)
        except ValueError:
            value = 0
        if value >= 1:
            return value
        # A non-empty but invalid value is an operator mistake — say so instead of
        # silently swallowing it (stderr is captured by journald in headless mode).
        print(
            f"[radar] Ignoring invalid RADAR_SYSTEM_METRICS_REPORT_EVERY={raw!r}; using default 50.",
            file=sys.stderr,
            flush=True,
        )
        return 50

    def _init_config_profile_state(self) -> None:
        """Resolve the startup profile path, load the active profile, and queue any
        fallback notices for replay once the log panel exists.

        On failure to load a non-root profile, falls back to the root run_config.json;
        a failure to load the root profile itself is fatal and re-raised.
        """
        active_profile_path = self._resolve_startup_profile_path()
        try:
            profile = GuiProfileModel.load_from_path(active_profile_path)
        except Exception as exc:
            if active_profile_path == self._root_profile_path:
                raise
            self._queue_startup_log_entry(
                "WARN",
                "Failed to load the last selected config profile; falling back to root run_config.json.",
                details=self._exception_details(exc),
            )
            profile = GuiProfileModel.load_from_path(self._root_profile_path)
            active_profile_path = self._root_profile_path

        self._active_profile_path = active_profile_path
        self._defaults_config = profile.run_config.clone()
        set_log_level(self._defaults_config.logging.level)
        if profile.gui is not None:
            self._gui_defaults = profile.gui
        else:
            self._gui_defaults = self._default_gui_state_for_config(self._defaults_config)
            if active_profile_path != self._root_profile_path:
                self._queue_startup_log_entry(
                    "INFO",
                    f"Loaded legacy run config without GUI defaults: {active_profile_path}",
                )
        self._remember_active_profile_path(active_profile_path, startup=True)

    def _init_reader_handles(self) -> None:
        """Initialize SHM readers as detached (not connected) handles."""
        self._raw_reader: ShmRingReader | None = None
        self._pre_reader: ShmRingReader | None = None
        self._result_reader: ShmRingReader | None = None

    def _init_preprocess_state(self) -> None:
        """Initialize preprocessing dialog and selected set names."""
        self._preprocess_dialog: PreprocessDialog | None = None
        self._preprocess_set_name = str(self._gui_defaults.preprocess_dialog.set_name)
        self._preprocess_radar_config_dir = str(self._gui_defaults.preprocess_dialog.radar_config_dir)
        self._preprocess_use_all_radar_configs = bool(self._gui_defaults.preprocess_dialog.use_all_radar_configs)
        self._preprocess_median_sweep_count = max(
            1, int(self._gui_defaults.preprocess_dialog.median_sweep_count)
        )
        self._preprocess_radar_variants: list[RadarConfigVariant] = []
        self._preprocess_radar_scan_summary = RadarConfigScanSummary(
            directory_path=self._preprocess_radar_config_dir,
            json_file_count=0,
            valid_variant_count=0,
            skipped_file_count=0,
            duplicate_variant_count=0,
            issues=(),
        )
        self._selected_preprocess_sets = {
            key: str(preprocess_asset_model(self._defaults_config, key).set_name)
            for key in VISIBLE_PREPROCESS_ASSET_KEYS
        }
        self._selected_preprocess_radar_key = self._radar_key(self._defaults_config)

    def _init_capture_state(self) -> None:
        """Initialize one-shot capture and sequence-control flags."""
        self._capture_session: SequentialCaptureSession | MultiRadarSequentialCaptureSession | None = None
        self._resume_pipeline_after_capture = False
        self._single_capture_active = False
        self._single_capture_start_ns: int | None = None
        self._single_capture_seen_raw = False
        self._single_capture_target_collection_id: int | None = None

    def _init_history_state(self) -> None:
        """Initialize runtime history buffers and render-cache state."""
        bscan_history_limit = self._history_limit_from_config()
        save_history_limit = self._save_history_limit_from_config()
        self._raw_history: deque[SweepCollection] = deque(maxlen=save_history_limit)
        self._pre_history: deque[SweepCollection] = deque(maxlen=save_history_limit)
        self._result_history: deque[ResultCollection] = deque(maxlen=save_history_limit)

        # Sequence id must survive GUI restarts so history commands stay monotonic.
        self._history_command_seq = self._load_history_command_seq(self._live_config_writer.path)
        self._bscan_history_limit = bscan_history_limit
        self._bscan_history_by_combo = {}
        self._bscan_depth_axis_by_combo = {}
        self._bscan_history_floor_collection_id = 0
        self._bscan_render_signature = None
        self._gpr_lookup_table = None
        self._gpr_image_item = None
        self._gpr_tx_item = None
        self._gpr_rx_item = None
        self._gpr_points_item = None
        self._gpr_region_centers_item = None
        self._gpr_point_labels = []
        self._gpr_region_center_labels = []
        self._gpr_region_mask_items = []
        self._gpr_region_contours = []
        self._gpr_geometry_signature = None
        self._gpr_selected_geometry = None
        self._phase_viewbox = None
        self._history_run_signature = None
        self._processor_run_signature = None
        self._active_processing_mode = "pass_through"
        self._radar_limits: dict[str, float | int] | None = None

    def _history_limit_from_config(self) -> int:
        """Return B-scan render history limit derived from configured ring capacities."""
        return self._history_limit_for_config(self._defaults_config)

    def _save_history_limit_from_config(self) -> int:
        """Return maxlen for GUI snapshot-save deques (independent of ring capacities)."""
        return self._save_history_limit_for_config(self._defaults_config)

    def _init_runtime_limits(self) -> None:
        """Initialize read/drain loop limits used by polling and snapshot code."""
        self._max_pop_per_poll = 256
        self._max_pop_per_snapshot_drain = 4096
        self._last_reader_error_signature: tuple[str, str] | None = None
        self._logged_once_keys: set[str] = set()

    def _init_polling_timer(self) -> None:
        """Create periodic timer that polls SHM rings for new data."""
        self._timer = QTimer(self)
        self._timer.setInterval(50)
        self._timer.timeout.connect(self._poll_rings)

    def _bootstrap_ui_runtime(self) -> None:
        """Build the UI and apply initial runtime-bound state once widgets exist.

        This is the first point at which the log panel is wired in, so it also
        replays any startup entries buffered during the headless init phase.
        """
        self._build_ui()
        self._attach_log_panel()
        self._flush_pending_startup_log_entries()
        self._log(f"Active config profile: {self._active_profile_path}")
        self._refresh_preprocess_summary_labels()
        self._apply_initial_radar_limits()
        self._on_processing_mode_changed(self._processing_mode.currentText())
        self._write_live_processing_config()
        # Defer the log sink wiring until the runtime log widget exists.
        self._pipeline_metrics.set_log_sink(self._log)
        self._timer.start()
        self._maybe_auto_start_pipeline()
        self._start_control_button_watcher()
        self._init_web_ui()
        if self._is_truthy_env("RADAR_SYSTEM_HEADLESS"):
            self._install_headless_watchdog()
        self._log_debug(
            f"Window bootstrap complete (mode={self._active_processing_mode}, "
            f"ring poll interval={self._timer.interval()}ms)."
        )

    def _resolve_startup_profile_path(self) -> Path:
        """Resolve active profile path from session-state or root fallback path."""
        env_profile_path = os.environ.get("RADAR_SYSTEM_PROFILE", "").strip()
        if env_profile_path:
            profile_path = Path(env_profile_path).expanduser()
            if not profile_path.is_absolute():
                profile_path = (self._project_root / profile_path).resolve(strict=False)
            return profile_path

        try:
            session_state = self._gui_session_state_store.load()
        except Exception as exc:
            self._queue_startup_log_entry(
                "WARN",
                "Failed to read GUI session-state; using root run_config.json.",
                details=self._exception_details(exc),
            )
            return self._root_profile_path

        raw_path = session_state.last_profile_path.strip()
        if not raw_path:
            return self._root_profile_path

        profile_path = Path(raw_path).expanduser()
        if not profile_path.is_absolute():
            profile_path = (self._project_root / profile_path).resolve(strict=False)
        return profile_path

    def _maybe_auto_start_pipeline(self) -> None:
        """Schedule pipeline start when requested by launcher environment.

        With `RADAR_SYSTEM_AUTO_APPLY_RADAR=1` the launcher also reproduces the
        "Apply Radar" click before "Start". This is the headless deployment
        recipe: the GUI configures the device exactly as a human operator
        would, then starts the capture pipeline.
        """
        auto_start = self._is_truthy_env("RADAR_SYSTEM_AUTO_START")
        if not auto_start:
            return
        self._log("Auto-start requested by launcher.")
        if self._is_truthy_env("RADAR_SYSTEM_AUTO_APPLY_RADAR"):
            QTimer.singleShot(500, self._auto_apply_radar_then_start)
        else:
            QTimer.singleShot(500, self._auto_start_pipeline_step)

    def _auto_apply_radar_then_start(self) -> None:
        """Apply current radar settings then start the pipeline (headless boot)."""
        try:
            self._apply_radar_settings()
        except Exception as exc:  # noqa: BLE001
            self._log_exception("Auto apply-radar failed", exc, level="WARN")
        # Hand control back to the event loop so widget updates from
        # _apply_radar_settings can flush, then wait one second before the
        # start takes over so the device settles after apply-radar.
        QTimer.singleShot(1000, self._auto_start_pipeline_step)

    def _auto_start_pipeline_step(self) -> None:
        """Run the launcher-requested pipeline start.

        In headless mode a start that does not bring the pipeline up is fatal: we
        exit non-zero so `systemd Restart=on-failure` restarts the unit instead of
        leaving an idle daemon producing nothing. (The producer itself waits for
        the device forever, so a live-but-deviceless producer counts as running.)
        """
        self._start_run()
        if self._is_truthy_env("RADAR_SYSTEM_HEADLESS") and not self._supervisor.is_running():
            self._headless_fatal("Headless auto-start did not bring the pipeline up")

    def _install_headless_watchdog(self) -> None:
        """Self-heal a headless daemon: if a managed pipeline process crashes (exits
        without us stopping it), exit non-zero so the service restarts clean.

        Intentional stops drop processes from the supervisor first, so a normal
        stop/start or tmp-reference transition never trips this.
        """
        self._headless_watchdog = QTimer(self)
        self._headless_watchdog.setInterval(2000)
        self._headless_watchdog.timeout.connect(self._headless_watchdog_tick)
        self._headless_watchdog.start()
        self._log_debug("Headless watchdog armed (interval=2000ms).")

    def _headless_watchdog_tick(self) -> None:
        """Escalate any unexpected managed-process exit to a fatal headless restart."""
        crashed = [
            report
            for report in self._supervisor.collect_exit_reports()
            if not report.expected_clean_exit
        ]
        if crashed:
            names = ", ".join(report.name for report in crashed)
            details = "\n\n".join(report.format() for report in crashed)
            self._headless_fatal(f"Pipeline process exited unexpectedly: {names}", details=details)

    def _headless_fatal(self, reason: str, *, details: str | None = None) -> None:
        """Log loudly to stderr and exit non-zero so systemd restarts the service.

        Headless deployments have no operator and the in-app log only reaches an
        offscreen widget, so a dead pipeline would otherwise go unnoticed.
        """
        self._log_error(reason, details=details)
        print(f"[radar] FATAL (headless): {reason}", file=sys.stderr, flush=True)
        if details:
            print(details, file=sys.stderr, flush=True)
        app = QApplication.instance()
        if app is not None:
            app.exit(1)

    @staticmethod
    def _is_truthy_env(name: str) -> bool:
        """Return True when an environment variable is set to a truthy literal."""
        return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}

    def _normalize_profile_path(self, path: Path) -> Path:
        """Return normalized absolute profile path."""
        return path.expanduser().resolve(strict=False)

    def _remember_active_profile_path(self, path: Path, *, startup: bool = False) -> None:
        """Persist the last successfully used config profile path to GUI session-state.

        A write failure is non-fatal: it is queued during startup or logged as a
        warning afterwards, since it only affects which profile reopens next launch.
        """
        normalized_path = self._normalize_profile_path(path)
        self._active_profile_path = normalized_path
        try:
            self._gui_session_state_store.write(GuiSessionState(last_profile_path=str(normalized_path)))
        except Exception as exc:
            if startup:
                self._queue_startup_log_entry(
                    "WARN",
                    "Failed to update GUI session-state with the active config profile path.",
                    details=self._exception_details(exc),
                )
            else:
                self._log_exception(
                    "Failed to update GUI session-state with the active config profile path",
                    exc,
                    level="WARN",
                )

    def _queue_startup_log_entry(self, level: str, text: str, *, details: str | None = None) -> None:
        """Queue startup log entry until log widget exists."""
        self._pending_startup_log_entries.append((level.upper(), text, details))

    def _flush_pending_startup_log_entries(self) -> None:
        """Replay queued startup entries through the logger now that every sink exists.

        Entries logged before the panel widget existed were buffered; routing them
        through the logger here delivers them to the file/console and the panel at once.
        """
        levels = {"WARN": logging.WARNING, "ERROR": logging.ERROR}
        for level, text, details in self._pending_startup_log_entries:
            self._gui_logger.log(levels.get(level, logging.INFO), text, extra=_panel_extra(details, None))
        self._pending_startup_log_entries.clear()

    def _apply_initial_radar_limits(self) -> None:
        """Apply the startup radar-limits strategy according to the selected radar mode.

        In ``native`` mode the limits are queried from the connected device; otherwise
        the UI is populated with no device-imposed limits.
        """
        if self._defaults_config.radar.driver_mode == "native":
            self._log_debug("Querying radar limits from device (native driver mode).")
            self._refresh_radar_limits_from_device()
            return
        self._log_debug(
            f"Skipping device radar-limit query (driver mode={self._defaults_config.radar.driver_mode})."
        )
        self._apply_radar_limits_to_ui(None)

    @staticmethod
    def _escape_log_text(text: str) -> str:
        """Escape log text for insertion into rich-text log widget."""
        return html.escape(text).replace("\n", "<br>")

    @staticmethod
    def _exception_summary(exc: Exception) -> str:
        """Build compact one-line exception summary."""
        message = str(exc).strip()
        if message:
            return f"{type(exc).__name__}: {message}"
        return type(exc).__name__

    @staticmethod
    def _exception_details(exc: Exception) -> str:
        """Return full chained traceback for error dialogs and log details."""
        return "".join(traceback.TracebackException.from_exception(exc).format(chain=True)).strip()

    def _append_log_entry(
        self,
        level: str,
        text: str,
        *,
        details: str | None = None,
        once_key: str | None = None,
    ) -> None:
        """Append formatted log entry with timestamp and optional details."""
        if once_key is not None:
            if once_key in self._logged_once_keys:
                return
            self._logged_once_keys.add(once_key)

        level_upper = level.upper()
        palette = {
            "DEBUG": ("#6c7b8d", "#52627a", "#8a97a8"),
            "INFO": ("#1d5fbf", "#1f2937", "#526277"),
            "WARN": ("#9a5b00", "#5c4300", "#7a6640"),
            "ERROR": ("#c43d4d", "#6b1f2a", "#8b5d66"),
        }
        accent_color, message_color, detail_color = palette.get(level_upper, palette["INFO"])
        timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
        header_html = (
            f"<span style='color:{accent_color}; font-weight:700;'>{html.escape(level_upper)}</span>"
            f" <span style='color:#60758d;'>{html.escape(timestamp)}</span>"
            f" <span style='color:{message_color};'>{self._escape_log_text(text)}</span>"
        )

        body_parts = [header_html]
        if details:
            body_parts.append(
                "<pre style='margin:3px 0 0 16px; color:"
                f"{detail_color};'>{html.escape(details)}</pre>"
            )

        entry_html = "<div style='margin:0 0 6px 0;'>" + "".join(body_parts) + "</div>"
        cursor = self._log_box.textCursor()
        cursor.movePosition(QTextCursor.MoveOperation.End)
        self._log_box.setTextCursor(cursor)
        self._log_box.insertHtml(entry_html)
        self._log_box.insertPlainText("\n")
        self._log_box.ensureCursorVisible()

        if level_upper == "ERROR" and hasattr(self, "_status_label"):
            self._status_label.setText("Status: error")

    def _on_log_level_selected(self, level_text: str) -> None:
        """Apply the chosen log level immediately (sub-level logs stop being generated).

        The value lives in run_config like any other field: it is recorded in the active
        config here and saved with the config through the normal path — no separate write.
        """
        level = level_text.strip().lower()
        set_log_level(level)
        self._defaults_config.logging.level = level
        self._log(f"Log level set to {level.upper()}.")

    def _log_debug(self, text: str, *, once_key: str | None = None) -> None:
        """Log a diagnostic message (emitted only while the level is DEBUG)."""
        self._gui_logger.debug(text, extra=_panel_extra(None, once_key))

    def _log(self, text: str, *, once_key: str | None = None) -> None:
        """Log an informational message to the panel, file, and console."""
        self._gui_logger.info(text, extra=_panel_extra(None, once_key))

    def _log_warning(self, text: str, *, details: str | None = None, once_key: str | None = None) -> None:
        """Log a warning to the panel, file, and console."""
        self._gui_logger.warning(text, extra=_panel_extra(details, once_key))

    def _log_error(self, text: str, *, details: str | None = None, once_key: str | None = None) -> None:
        """Log an error to the panel, file, and console."""
        self._gui_logger.error(text, extra=_panel_extra(details, once_key))

    def _log_exception(self, context: str, exc: Exception, *, level: str = "ERROR") -> tuple[str, str]:
        """Log exception with detailed traceback and return `(message, details)`."""
        message = f"{context}: {self._exception_summary(exc)}"
        details = self._exception_details(exc)
        if level.upper() == "WARN":
            self._log_warning(message, details=details)
        else:
            self._log_error(message, details=details)
        return message, details

    def _process_state_details(self) -> str:
        """Return formatted summary of managed pipeline process state."""
        if not hasattr(self, "_supervisor"):
            return "Managed processes: unavailable"
        pid_map = self._supervisor.pids()
        if not pid_map:
            return "Managed processes: none"
        return "Managed processes:\n" + "\n".join(
            f"- {name}: pid={pid}"
            for name, pid in sorted(pid_map.items())
        )

    def _runtime_history_details(self) -> str:
        """Return formatted summary of buffered runtime history counts."""
        return (
            "Runtime history:\n"
            f"- raw={len(getattr(self, '_raw_history', []))}\n"
            f"- preprocessed={len(getattr(self, '_pre_history', []))}\n"
            f"- results={len(getattr(self, '_result_history', []))}"
        )

    def _capture_state_details(self) -> str:
        """Return formatted summary of active preprocess capture state."""
        session = getattr(self, "_capture_session", None)
        if session is None:
            return "Capture session: none"
        state = session.state()
        lines = [
            "Capture session:",
            f"- kind={state.kind}",
            f"- progress={state.captured_count}/{state.total_count}",
        ]
        if state.current_combo is not None:
            lines.append(
                f"- current_combo=input={state.current_combo.input}, output={state.current_combo.output}"
            )
        return "\n".join(lines)

    @staticmethod
    def _load_history_command_seq(config_path: Path) -> int:
        """Load previously used live-command sequence from runtime config file."""
        try:
            payload = json.loads(config_path.read_text(encoding="utf-8"))
        except Exception:  # noqa: BLE001
            # Missing or malformed file should not block startup.
            return 0

        raw_value = payload.get("history_command_seq", 0)
        if isinstance(raw_value, bool):
            return 0
        if isinstance(raw_value, (int, float)):
            return max(0, int(raw_value))
        return 0

    def _capture_web_action_error(self, message: str) -> bool:
        """Record the first error of an in-flight web action so the browser can show it.

        Returns ``True`` if a web-triggered action is currently running (see
        ``AppWindowWebMixin._run_web_action``). Callers use that to skip the blocking
        desktop modal for web errors — the browser shows the message instead, and the
        operator at the browser must not have to dismiss a popup on the (often headless)
        host before the HTTP response returns. Desktop-only errors are unaffected.
        """
        capture = getattr(self, "_web_action_error_capture", None)
        if capture is None:
            return False
        if not capture:
            capture.append(message)
        return True

    def _show_error(self, message: str, *, details: str | None = None) -> None:
        """Log and present an error in a modal dialog with optional detail text."""
        self._log_error(message, details=details)
        if self._capture_web_action_error(message):
            return  # web-triggered: surfaced to the browser; no blocking desktop modal
        if self._is_truthy_env("RADAR_SYSTEM_HEADLESS"):
            return
        dialog = QMessageBox(self)
        dialog.setIcon(QMessageBox.Icon.Critical)
        dialog.setWindowTitle("Error")
        dialog.setText(message)
        if details:
            dialog.setDetailedText(details)
        dialog.exec()

    def _show_exception(self, context: str, exc: Exception) -> None:
        """Log full exception details and show modal dialog with expandable traceback."""
        message, details = self._log_exception(context, exc, level="ERROR")
        if self._capture_web_action_error(message):
            return  # web-triggered: surfaced to the browser; no blocking desktop modal
        if self._is_truthy_env("RADAR_SYSTEM_HEADLESS"):
            return
        dialog = QMessageBox(self)
        dialog.setIcon(QMessageBox.Icon.Critical)
        dialog.setWindowTitle("Error")
        dialog.setText(message)
        dialog.setDetailedText(details)
        dialog.exec()

    def closeEvent(self, event) -> None:  # noqa: N802
        """Tear down workers, readers, and dialogs before the window is destroyed.

        Guarded against re-entrancy so a second close signal (or a ``window.close()``
        after the event loop has already returned) does not run teardown twice.
        """
        if self._closing:
            # Re-entrant close (second signal, or window.close() after the event loop
            # already returned): teardown is in progress or done — do nothing more.
            super().closeEvent(event)
            return
        self._closing = True
        self._log("Window closing; shutting down runtime.")
        try:
            # 0) Stop the web server first so a late request cannot start work.
            self._shutdown_web_ui()
            # 0) Stop the disk-recording writer thread so it is not orphaned.
            self._shutdown_recording()
            # 0) Stop the GPIO button watcher so a late press cannot start work.
            self._stop_control_button_watcher()
            self._resume_pipeline_after_capture = False
            # 1) Abort active capture first (releases exclusive hardware resources).
            self._abort_capture_sequence(resume_pipeline=False)
            # 2) Stop all managed processes/readers.
            self._stop_all_processes()
            # 3) Close auxiliary dialog windows.
            if self._preprocess_dialog is not None:
                self._preprocess_dialog.close()
        finally:
            self._log_debug("Window teardown finished.")
            super().closeEvent(event)