#!/usr/bin/python3
# @file
# @ingroup uatools_tools
# @copyright
#   SPDX-FileCopyrightText: 2026 European Southern Observatory (ESO)
#   SPDX-License-Identifier: LGPL-3.0-only

"""
UaSubscription - live OPC UA subscription viewer, aligned with UaPlot.

Two operating modes share one live table (one row per subscribed node):

  1. Driven by UaShell (PASSIVE / push mode): launched with --socket, UaShell
     owns the OPC UA connection and pushes value-change updates over a local
     socket (QLocalServer), one JSON object per line. This is the original
     behaviour and is preserved unchanged.

  2. Standalone (ACTIVE mode): launched with --uri (or after the user supplies
     one), UaSubscription owns its OWN asyncua client (UaSubscriptionWorker),
     can browse the address space, Add/Remove subscriptions live, load/save
     sessions, and record to CSV — mirroring UaPlot.

The two modes coexist: both feed the same _upsert(), keyed by a normalized
node id, so there is never more than one row per logical node.

The implementation deliberately DUPLICATES the relevant UaPlot machinery
(worker, dialogs, recorder, header, browse cache) rather than importing it,
to keep the two tools decoupled.

Wire protocol for the push path (newline-delimited JSON, mirroring UaPlot):
  controller -> viewer:
    {"command":"ping","id":"1"}
    {"command":"update","node":"ns=2;s=...","name":"State",
     "value":"...","type":"String","quality":"Good","ts":"12:01:02.345"}
    {"command":"remove","node":"ns=2;s=..."}
    {"command":"clear"}
    {"command":"get_subscriptions","id":"2"}   # read back the table
    {"command":"quit"}
  viewer -> controller:
    {"type":"pong","reply_to":"1"}
    {"type":"ack","reply_to":"..."}
    {"type":"subscriptions","rows":[{"node":...,"name":...,"value":...,
     "quality":...,"min":...,"max":...,"mean":...,"count":...}, ...]}
"""

import argparse
import asyncio
import csv
import json
import os
import socket
import sys
import time
from collections import deque
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Dict, List, Optional, Tuple

try:
    from PyQt6.QtCore import (
        QObject, Qt, QThread, QTimer, pyqtSignal,
    )
    from PyQt6.QtNetwork import QLocalServer, QLocalSocket
    from PyQt6.QtWidgets import (
        QApplication, QMainWindow, QTableWidget, QTableWidgetItem,
        QWidget, QVBoxLayout, QHBoxLayout, QLabel, QHeaderView,
        QPushButton, QToolButton, QMenu, QDialog, QDialogButtonBox,
        QLineEdit, QComboBox, QCheckBox, QGroupBox, QFormLayout,
        QFileDialog, QMessageBox, QInputDialog, QProgressDialog,
        QListWidget, QListWidgetItem, QAbstractItemView, QSizePolicy,
        QPlainTextEdit, QSpinBox, QTextEdit,
    )
    from PyQt6.QtGui import (
        QColor, QFont, QPalette, QShortcut, QKeySequence,
        QDoubleValidator, QIntValidator, QPixmap,
    )
except ImportError:
    sys.stderr.write(
        "UaSubscription requires PyQt6, which is not installed.\n")
    sys.exit(2)

# asyncua is only needed for STANDALONE mode (own worker). The UaShell-driven
# push path must keep working even when asyncua is absent, so the import is
# optional and gates worker creation only.
try:
    from asyncua import Client, ua
    try:
        from asyncua.common.subscription import SubHandler as _SubHandler
    except Exception:
        try:
            from asyncua.common.subscription_handler import SubHandler as _SubHandler
        except Exception:
            class _SubHandler:
                def datachange_notification(self, node, val, data):
                    pass

                def event_notification(self, event):
                    pass
    _HAVE_ASYNCUA = True
except Exception:
    Client = None
    ua = None

    class _SubHandler:
        pass
    _HAVE_ASYNCUA = False


# Column layout for the live table. The four stat columns (Min/Max/Mean/Count)
# mirror UaExplorer's subscription widget and are hidden unless stats are
# enabled in Preferences.
COLUMNS = ["Node ID", "Name", "Value", "Type", "Quality", "Timestamp",
           "Min", "Max", "Mean", "Count"]
(COL_NODE, COL_NAME, COL_VALUE, COL_TYPE, COL_QUALITY, COL_TS,
 COL_MIN, COL_MAX, COL_MEAN, COL_COUNT) = range(10)


# ---------------------------------------------------------------------------
# Theme subsystem — VERBATIM copy of UaPlot/UaExplorer (kept in sync by hand).
# ---------------------------------------------------------------------------

@dataclass
class Theme:
    name: str
    window_bg: str
    input_bg: str
    panel_bg: str
    panel_alt_bg: str
    panel_hover_bg: str
    panel_pressed_bg: str
    panel_disabled_bg: str
    text_primary: str
    text_muted: str
    text_disabled: str
    text_on_accent: str
    border: str
    border_strong: str
    border_subtle: str
    accent: str
    accent_hover: str
    section_title: str
    node_object: str
    node_variable: str
    node_method: str
    status_ok: str
    status_warn: str
    status_error: str
    record_idle_bg: str
    record_idle_border: str
    record_active_bg: str
    record_active_border: str


LIGHT_THEME = Theme(
    name="Light", window_bg="#e8e8e8", input_bg="#ffffff", panel_bg="#dcdcdc",
    panel_alt_bg="#d4d8db", panel_hover_bg="#cfcfcf", panel_pressed_bg="#bfbfbf",
    panel_disabled_bg="#d8d8d8", text_primary="#202020", text_muted="#666666",
    text_disabled="#aaaaaa", text_on_accent="#ffffff", border="#888888",
    border_strong="#555555", border_subtle="#bbbbbb", accent="#3a6ea5",
    accent_hover="#2c5380", section_title="#2c4a5e", node_object="#b35a00",
    node_variable="#1a6e3f", node_method="#1a3f8f", status_ok="#2E7D32",
    status_warn="#C68400", status_error="#C62828", record_idle_bg="#b30000",
    record_idle_border="#7a0000", record_active_bg="#ff6666",
    record_active_border="#7a0000",
)

DARK_THEME = Theme(
    name="Dark", window_bg="#1f2226", input_bg="#15181c", panel_bg="#2b2f34",
    panel_alt_bg="#363c44", panel_hover_bg="#3a4048", panel_pressed_bg="#222629",
    panel_disabled_bg="#262a2e", text_primary="#e6e6e6", text_muted="#a8b0b8",
    text_disabled="#6a7079", text_on_accent="#ffffff", border="#4a4f55",
    border_strong="#6a7079", border_subtle="#3a3f44", accent="#5b9bd5",
    accent_hover="#80b6e6", section_title="#9fb8c8", node_object="#e08a3a",
    node_variable="#5fbd80", node_method="#7da6e6", status_ok="#5fbd80",
    status_warn="#e6a93b", status_error="#e06c6c", record_idle_bg="#a02020",
    record_idle_border="#601818", record_active_bg="#e25555",
    record_active_border="#601818",
)

ATACAMA_THEME = Theme(
    name="Atacama", window_bg="#c9a07a", input_bg="#b88a5f", panel_bg="#b88a5f",
    panel_alt_bg="#a37047", panel_hover_bg="#a87b50", panel_pressed_bg="#8b5a35",
    panel_disabled_bg="#b09075", text_primary="#2a1607", text_muted="#5a3622",
    text_disabled="#7a5945", text_on_accent="#ffffff", border="#5a3622",
    border_strong="#3a2010", border_subtle="#8b5a35", accent="#8a2e10",
    accent_hover="#6a1f08", section_title="#3a1a08", node_object="#8a2e10",
    node_variable="#4a5520", node_method="#3a1408", status_ok="#4a5520",
    status_warn="#a86a0a", status_error="#8a2e10", record_idle_bg="#8a2810",
    record_idle_border="#5a1808", record_active_bg="#c44a28",
    record_active_border="#5a1808",
)

THEMES: Dict[str, Theme] = {
    LIGHT_THEME.name: LIGHT_THEME,
    DARK_THEME.name: DARK_THEME,
    ATACAMA_THEME.name: ATACAMA_THEME,
}


def resolve_theme(name: Optional[str]) -> Theme:
    if not name:
        return LIGHT_THEME
    for key, theme in THEMES.items():
        if key.lower() == name.lower():
            return theme
    return LIGHT_THEME


def build_qpalette_for_theme(theme: Theme) -> QPalette:
    pal = QPalette()
    pal.setColor(QPalette.ColorRole.Window, QColor(theme.window_bg))
    pal.setColor(QPalette.ColorRole.WindowText, QColor(theme.text_primary))
    pal.setColor(QPalette.ColorRole.Base, QColor(theme.input_bg))
    pal.setColor(QPalette.ColorRole.AlternateBase, QColor(theme.panel_alt_bg))
    pal.setColor(QPalette.ColorRole.Text, QColor(theme.text_primary))
    pal.setColor(QPalette.ColorRole.Button, QColor(theme.panel_bg))
    pal.setColor(QPalette.ColorRole.ButtonText, QColor(theme.text_primary))
    pal.setColor(QPalette.ColorRole.Highlight, QColor(theme.accent))
    pal.setColor(QPalette.ColorRole.HighlightedText, QColor(theme.text_on_accent))
    pal.setColor(QPalette.ColorRole.ToolTipBase, QColor(theme.panel_alt_bg))
    pal.setColor(QPalette.ColorRole.ToolTipText, QColor(theme.text_primary))
    pal.setColor(QPalette.ColorRole.Link, QColor(theme.accent))
    pal.setColor(QPalette.ColorRole.PlaceholderText, QColor(theme.text_muted))
    pal.setColor(QPalette.ColorGroup.Disabled, QPalette.ColorRole.Text,
                 QColor(theme.text_disabled))
    pal.setColor(QPalette.ColorGroup.Disabled, QPalette.ColorRole.ButtonText,
                 QColor(theme.text_disabled))
    return pal


def apply_theme_to_application(theme: Theme) -> None:
    app = QApplication.instance()
    if app is None:
        return
    try:
        app.setStyle("Fusion")
    except Exception:
        pass
    app.setPalette(build_qpalette_for_theme(theme))


def _now_ts() -> float:
    return time.time()


def _log(msg: str):
    if os.environ.get("UASUBSCRIPTION_DEBUG"):
        print(f"[UaSubscription] {msg}", file=sys.stderr, flush=True)


def _filter_terms(text: str) -> List[str]:
    """Split a quick-filter string into lowercase AND terms on '+'."""
    return [t for t in (s.strip().lower() for s in text.split("+")) if t]


def _ensure_csv_suffix(name: str) -> str:
    """Force a .csv extension (foo -> foo.csv, foo.dat -> foo.csv)."""
    stem = Path(name).stem.strip()
    if not stem:
        return "rec.csv"
    return f"{stem}.csv"


def _format_sample(v: object) -> str:
    if isinstance(v, bool):
        return "1" if v else "0"
    if isinstance(v, (int, float)):
        return repr(v)
    return str(v)


def _ipc_channel_safe(name: str) -> str:
    """Make a channel name safe for QLocalServer (Windows named pipes)."""
    safe = "".join(c if c.isalnum() or c == "_" else "_" for c in name)
    return safe[:64] if len(safe) > 64 else safe


def find_file(name: str, parent: Optional[Path] = None) -> Optional[Path]:
    """Locate ``name`` per the workspace File Location Rules (mirrors UaPlot).

    CFGPATH -> parent-relative -> INTROOT/PREFIX -> standalone fallback. The
    shared IfwOpcUaTools logo lives in uaexplorer/resource/, picked up by the
    walk-up fallback when running from source. Returns the resolved Path or
    None."""
    expanded = os.path.expanduser(os.path.expandvars(name))
    candidate = Path(expanded)
    if candidate.is_absolute():
        return candidate.resolve() if candidate.is_file() else None
    rel = Path(expanded)
    for root in [p for p in os.environ.get("CFGPATH", "").split(os.pathsep) if p]:
        hit = Path(root) / rel
        if hit.is_file():
            return hit.resolve()
    if parent is not None:
        parent_dir = parent if parent.is_dir() else parent.parent
        hit = parent_dir / rel
        if hit.is_file():
            return hit.resolve()
    for env_name in ("INTROOT", "PREFIX"):
        root = os.environ.get(env_name, "")
        if root and (Path(root) / rel).is_file():
            return (Path(root) / rel).resolve()
    here = Path(__file__).resolve().parent
    for fb in (here, here.parent / "resource", here.parent,
               here.parent.parent / "uaexplorer" / "resource"):
        hit = fb / rel
        if hit.is_file():
            return hit.resolve()
    return None


class SubscriptionDataHandler(_SubHandler):
    """Forward asyncua subscription updates back to the worker thread."""

    def __init__(self, callback: Callable[[str, object, float], None]):
        super().__init__()
        self._callback = callback

    def datachange_notification(self, node, val, data):
        timestamp = _now_ts()
        try:
            notif = getattr(data, "MonitoredItemNotification", None)
            dv = getattr(notif, "Value", None) if notif else None
            if dv and getattr(dv, "SourceTimestamp", None):
                timestamp = dv.SourceTimestamp.timestamp()
        except Exception:
            pass
        try:
            node_id = str(node.nodeid)
        except Exception:
            node_id = ""
        self._callback(node_id, val, timestamp)


class UaSubscriptionWorker(QThread):
    """Background asyncio worker — LOCAL copy of UaPlot's UaPlotWorker.

    Connects to an OPC UA server, subscribes (or polls) watched nodes, browses
    the address space on request, and emits data/connection/browse signals back
    to the GUI thread. All asyncua I/O happens on this thread's own loop; the
    GUI thread only ever touches it via run_coroutine_threadsafe / signals."""

    data_update = pyqtSignal(str, object, float)
    connected = pyqtSignal(str)
    disconnected = pyqtSignal()
    error = pyqtSignal(str)
    browse_result = pyqtSignal(object)
    browse_error = pyqtSignal(str)

    def __init__(self, uri: str):
        super().__init__()
        self.uri = uri
        self._loop: Optional[asyncio.AbstractEventLoop] = None
        self._stop_requested = False
        self._watched_nodes: set = set()
        self._active_handles: Dict[str, object] = {}
        self.client = None
        self.subscription = None
        self.monitor_mode = "subscription"
        self.polling_interval_ms = 500
        self._poll_task = None
        # The in-flight address-space browse, tracked so it can be cancelled on
        # shutdown — otherwise the loop tears down with it still pending
        # ("Task was destroyed but it is pending!").
        self._browse_future = None
        self._node_norm_cache: Dict[str, str] = {}

    def set_uri(self, uri: str):
        self.uri = uri

    def set_monitor_mode(self, mode: str, polling_interval_ms: int):
        self.monitor_mode = mode if mode in ("polling", "subscription") else "subscription"
        if polling_interval_ms > 0:
            self.polling_interval_ms = polling_interval_ms

    def add_node(self, node_id: str):
        self._watched_nodes.add(node_id)
        if self._loop:
            asyncio.run_coroutine_threadsafe(self._subscribe_node(node_id), self._loop)

    def remove_node(self, node_id: str):
        self._watched_nodes.discard(node_id)
        if self._loop:
            asyncio.run_coroutine_threadsafe(self._unsubscribe_node(node_id), self._loop)

    async def _unsubscribe_node(self, node_id: str):
        handle = self._active_handles.pop(node_id, None)
        if handle is None or not self.subscription:
            return
        try:
            await self.subscription.unsubscribe(handle)
        except Exception as exc:
            _log(f"[_unsubscribe_node] Failed for {node_id}: {exc}")

    def request_browse(self) -> bool:
        if not self._loop:
            return False
        # Track the future so _cleanup() can cancel an in-flight browse on
        # shutdown (avoids "Task was destroyed but it is pending!").
        self._browse_future = asyncio.run_coroutine_threadsafe(
            self._browse_address_space(), self._loop)
        return True

    _BROWSE_MAX_NODES = 20000
    _BROWSE_MAX_DEPTH = 12

    async def _browse_address_space(self):
        if not self.client:
            self.browse_error.emit("Not connected to the server yet.")
            return
        results: List[Tuple[str, str]] = []
        visited: set = set()
        truncated = False
        try:
            root = self.client.nodes.objects
            stack: List[Tuple[object, str, int]] = [(root, "Objects", 0)]
            while stack:
                if len(results) >= self._BROWSE_MAX_NODES:
                    truncated = True
                    break
                node, path, depth = stack.pop()
                nid = node.nodeid.to_string()
                if nid in visited:
                    continue
                visited.add(nid)
                try:
                    nclass = await node.read_node_class()
                except Exception:
                    continue
                if nclass == ua.NodeClass.Variable:
                    results.append((nid, path))
                # Don't descend into Method nodes: their children are the
                # InputArguments/OutputArguments Variables (Argument structs),
                # which are not meaningful subscription/plot targets — they'd
                # otherwise show up as bogus "values" to add.
                if nclass == ua.NodeClass.Method:
                    continue
                if depth >= self._BROWSE_MAX_DEPTH:
                    continue
                try:
                    children = await node.get_children()
                except Exception:
                    children = []
                for child in children:
                    try:
                        bn = await child.read_browse_name()
                        seg = bn.Name
                    except Exception:
                        seg = child.nodeid.to_string()
                    stack.append((child, f"{path}/{seg}", depth + 1))
        except Exception as exc:
            self.browse_error.emit(f"Browse failed: {exc}")
            return
        seen: set = set()
        deduped: List[Tuple[str, str]] = []
        for nid, path in sorted(results, key=lambda r: r[1].lower()):
            if nid in seen:
                continue
            seen.add(nid)
            deduped.append((nid, path))
        if truncated:
            _log(f"[_browse_address_space] truncated at {self._BROWSE_MAX_NODES}")
        self.browse_result.emit(deduped)

    def run(self):
        self._loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self._loop)
        try:
            self._loop.run_until_complete(self._run_loop())
        finally:
            self._loop.close()

    def stop(self):
        # Flag only — never loop.stop() (aborts pending awaits on idle loop).
        self._stop_requested = True

    async def _run_loop(self):
        while not self._stop_requested:
            if not self.uri:
                await asyncio.sleep(0.2)
                continue
            try:
                self.client = Client(self.uri)
                await self.client.connect()
                if self.monitor_mode == "subscription":
                    handler = SubscriptionDataHandler(self._handle_update)
                    self.subscription = await self.client.create_subscription(250, handler)
                    await asyncio.sleep(0.15)
                    await self._resubscribe_all()
                else:
                    await asyncio.sleep(0.15)
                    await self._start_polling()
                self.connected.emit(self.uri)
                while not self._stop_requested:
                    if self.monitor_mode == "subscription" and self.subscription:
                        pending = [n for n in list(self._watched_nodes)
                                   if n not in self._active_handles]
                        for node_id in pending:
                            await self._subscribe_node(node_id)
                    await asyncio.sleep(0.3)
            except Exception as exc:
                _log(f"[worker] error {exc}")
                self.error.emit(str(exc))
            finally:
                await self._cleanup()
                self.disconnected.emit()
                if not self._stop_requested:
                    await asyncio.sleep(1.0)

    async def _resubscribe_all(self):
        for node_id in list(self._watched_nodes):
            await self._subscribe_node(node_id)

    async def _subscribe_node(self, node_id: str):
        if not self.client:
            return
        if self.monitor_mode == "subscription":
            if not self.subscription or node_id in self._active_handles:
                return
            try:
                node = self.client.get_node(node_id)
                handle = await self.subscription.subscribe_data_change(node, queuesize=1)
                self._active_handles[node_id] = handle
            except Exception as exc:
                self.error.emit(f"Subscribe failed for {node_id}: {exc}")
        else:
            return

    async def _start_polling(self):
        if self._poll_task:
            try:
                self._poll_task.cancel()
            except Exception:
                pass
        self._poll_task = asyncio.create_task(self._poll_loop())

    async def _poll_loop(self):
        while not self._stop_requested and self.monitor_mode == "polling":
            if not self.client:
                break
            nodes = list(self._watched_nodes)
            for node_id in nodes:
                try:
                    node = self.client.get_node(node_id)
                    val = await node.read_value()
                    self._handle_update(node_id, val, _now_ts())
                except Exception as exc:
                    self.error.emit(f"Polling failed for {node_id}: {exc}")
            await asyncio.sleep(self.polling_interval_ms / 1000.0)

    async def _cleanup(self):
        if self._poll_task:
            try:
                self._poll_task.cancel()
            except Exception:
                pass
            self._poll_task = None
        # Cancel an in-flight address-space browse so the loop doesn't tear down
        # with a pending task ("Task was destroyed but it is pending!").
        if self._browse_future is not None:
            try:
                self._browse_future.cancel()
            except Exception:
                pass
            self._browse_future = None
        if self.subscription:
            try:
                await self.subscription.delete()
            except Exception:
                pass
            self.subscription = None
        if self.client:
            try:
                await self.client.disconnect()
            except Exception:
                pass
            self.client = None
        self._active_handles.clear()

    def _handle_update(self, node_id: str, value: object, timestamp: float):
        norm = self._normalize_node_id(node_id)
        self.data_update.emit(norm, value, timestamp)

    def _normalize_node_id(self, node_id: str) -> str:
        if not node_id:
            return ""
        if node_id in self._node_norm_cache:
            return self._node_norm_cache[node_id]
        text = str(node_id)
        if hasattr(node_id, "to_string"):
            try:
                text = node_id.to_string()
            except Exception:
                text = str(node_id)
        if text.startswith("ns="):
            self._node_norm_cache[node_id] = text
            return text
        if "Identifier=" in text and "NamespaceIndex=" in text:
            try:
                ident = text.split("Identifier=")[1].split(",")[0].strip().strip("'\"")
                ns_part = text.split("NamespaceIndex=")[1].split(",")[0].strip()
                if ident and ns_part:
                    norm = f"ns={ns_part};s={ident}"
                    self._node_norm_cache[node_id] = norm
                    return norm
            except Exception:
                pass
        self._node_norm_cache[node_id] = text
        return text


class _AddSubscriptionDialog(QDialog):
    """Pick variable nodes (from a live browse) to subscribe to. Flat filtered
    list with a '+'-AND quick filter and multi-select."""

    def __init__(self, nodes: List[Tuple[str, str]], existing: set, parent=None):
        super().__init__(parent)
        self.setWindowTitle("Add Subscription")
        self.setWindowFlags(
            Qt.WindowType.Window
            | Qt.WindowType.CustomizeWindowHint
            | Qt.WindowType.WindowTitleHint
            | Qt.WindowType.WindowSystemMenuHint
            | Qt.WindowType.WindowMinimizeButtonHint
            | Qt.WindowType.WindowMaximizeButtonHint
            | Qt.WindowType.WindowCloseButtonHint
        )
        layout = QVBoxLayout(self)

        self.filter_input = QLineEdit()
        self.filter_input.setPlaceholderText("Filter (use + for AND, e.g. sens+temp)")
        self.filter_input.textChanged.connect(self._apply_filter)
        layout.addWidget(self.filter_input)

        self.list = QListWidget()
        self.list.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
        layout.addWidget(self.list, 1)

        for node_id, path in nodes:
            already = node_id in existing
            text = f"{path}  [{node_id}]"
            if already:
                text += "  (already subscribed)"
            item = QListWidgetItem(text)
            item.setData(Qt.ItemDataRole.UserRole, node_id)
            # Display name = the LEAF segment of the path (e.g. "Temperature"),
            # matching what UaShell pushes into the Name column. The full path
            # + node id stay visible in the list row above for disambiguation,
            # and the table's Node ID column tells identically-named nodes apart.
            leaf = path.rsplit("/", 1)[-1] if path else ""
            item.setData(Qt.ItemDataRole.UserRole + 1, leaf or node_id)
            if already:
                item.setForeground(QColor("#888"))
            self.list.addItem(item)

        if self.list.count() == 0:
            self.list.addItem("(no variable nodes found)")
            self.list.setEnabled(False)

        bbox = QDialogButtonBox(
            QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel
        )
        bbox.accepted.connect(self.accept)
        bbox.rejected.connect(self.reject)
        layout.addWidget(bbox)

        self.resize(640, 480)
        self.filter_input.setFocus()

    def _apply_filter(self, text: str):
        terms = _filter_terms(text)
        for i in range(self.list.count()):
            item = self.list.item(i)
            if item.data(Qt.ItemDataRole.UserRole) is None:
                continue
            hay = item.text().lower()
            item.setHidden(bool(terms) and not all(t in hay for t in terms))

    def selected_nodes(self) -> List[Tuple[str, str]]:
        out: List[Tuple[str, str]] = []
        for item in self.list.selectedItems():
            node_id = item.data(Qt.ItemDataRole.UserRole)
            if node_id is None or item.isHidden():
                continue
            label = item.data(Qt.ItemDataRole.UserRole + 1) or node_id
            out.append((str(node_id), str(label)))
        return out


class _RemoveSubscriptionDialog(QDialog):
    """List current subscriptions (display name + node id), multi-select."""

    def __init__(self, subs: List[Tuple[str, str]], parent=None):
        super().__init__(parent)
        self.setWindowTitle("Remove Subscription")
        self.setWindowFlags(
            Qt.WindowType.Window
            | Qt.WindowType.CustomizeWindowHint
            | Qt.WindowType.WindowTitleHint
            | Qt.WindowType.WindowSystemMenuHint
            | Qt.WindowType.WindowCloseButtonHint
        )
        layout = QVBoxLayout(self)
        layout.addWidget(QLabel("Select subscriptions to remove:"))

        self.list = QListWidget()
        self.list.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
        layout.addWidget(self.list, 1)
        for node_id, label in subs:
            shown = f"{label}  [{node_id}]" if label and label != node_id else node_id
            item = QListWidgetItem(shown)
            item.setData(Qt.ItemDataRole.UserRole, node_id)
            self.list.addItem(item)

        bbox = QDialogButtonBox(
            QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel
        )
        ok_btn = bbox.button(QDialogButtonBox.StandardButton.Ok)
        if ok_btn is not None:
            ok_btn.setText("Remove")
        bbox.accepted.connect(self.accept)
        bbox.rejected.connect(self.reject)
        layout.addWidget(bbox)
        self.resize(480, 360)

    def selected_nodes(self) -> List[str]:
        return [
            str(item.data(Qt.ItemDataRole.UserRole))
            for item in self.list.selectedItems()
            if item.data(Qt.ItemDataRole.UserRole) is not None
        ]


class _RecordingViewerDialog(QDialog):
    """Read-only raw viewer for a recording CSV (comment header + rows)."""

    def __init__(self, initial: Path, parent=None):
        super().__init__(parent)
        self.setWindowTitle("View Recording")
        self.setSizeGripEnabled(True)
        self.setWindowFlags(
            Qt.WindowType.Window
            | Qt.WindowType.CustomizeWindowHint
            | Qt.WindowType.WindowTitleHint
            | Qt.WindowType.WindowSystemMenuHint
            | Qt.WindowType.WindowMinimizeButtonHint
            | Qt.WindowType.WindowMaximizeButtonHint
            | Qt.WindowType.WindowCloseButtonHint
        )
        layout = QVBoxLayout(self)
        picker_row = QHBoxLayout()
        picker_row.addWidget(QLabel("File:"))
        self.path_label = QLineEdit(str(initial))
        self.path_label.setReadOnly(True)
        picker_row.addWidget(self.path_label, 1)
        browse_btn = QPushButton("Browse…")
        browse_btn.clicked.connect(self._on_browse)
        picker_row.addWidget(browse_btn)
        layout.addLayout(picker_row)
        self.text = QPlainTextEdit()
        self.text.setReadOnly(True)
        mono = QFont("Monospace")
        mono.setStyleHint(QFont.StyleHint.TypeWriter)
        self.text.setFont(mono)
        self.text.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap)
        layout.addWidget(self.text, 1)
        bbox = QDialogButtonBox(QDialogButtonBox.StandardButton.Close)
        bbox.rejected.connect(self.reject)
        bbox.accepted.connect(self.accept)
        layout.addWidget(bbox)
        self.resize(800, 560)
        self._load_file(initial)

    def _load_file(self, path: Path):
        try:
            content = path.read_text(encoding="utf-8")
        except Exception as exc:
            QMessageBox.warning(self, "View Recording", f"Could not read file:\n{exc}")
            return
        self.path_label.setText(str(path))
        self.text.setPlainText(content)
        cursor = self.text.textCursor()
        cursor.movePosition(cursor.MoveOperation.Start)
        self.text.setTextCursor(cursor)

    def _on_browse(self):
        start_dir = str(Path(self.path_label.text()).parent) if self.path_label.text() else str(Path.home())
        dialog = QFileDialog(self, "View Recording", start_dir)
        dialog.setFileMode(QFileDialog.FileMode.ExistingFile)
        dialog.setNameFilters(["CSV files (*.csv)"])
        dialog.setOption(QFileDialog.Option.DontUseNativeDialog, True)
        if dialog.exec() != QDialog.DialogCode.Accepted:
            return
        files = dialog.selectedFiles()
        if files:
            self._load_file(Path(files[0]))


class _SubscriptionRecorder(QObject):
    """Periodic snapshotter — LOCAL copy of UaPlot's _PlotRecorder, adapted to
    snapshot the live subscription table instead of plot curves.

    Each tick writes the latest value of every captured node (read from a
    ``latest_provider()`` callable supplied by the window: node_id -> str).
    The capture set + labels are frozen at start. RFC-4180 CSV with a
    ``#``-comment header, identical format to UaPlot/UaExplorer recordings."""

    def __init__(
        self,
        latest_provider: Callable[[], Tuple[List[str], Dict[str, str], Dict[str, str]]],
        on_started: Callable[[Path], None],
        on_stopped: Callable[[Optional[str]], None],
        on_tick: Callable[[int], None],
        parent: Optional[QObject] = None,
    ):
        super().__init__(parent)
        # latest_provider() -> (ordered node-ids, node->label, node->value)
        self._provider = latest_provider
        self._on_started = on_started
        self._on_stopped = on_stopped
        self._on_tick = on_tick
        self._timer = QTimer(self)
        self._timer.timeout.connect(self._tick)
        self._file = None
        self._writer = None
        self._keys: List[str] = []
        self._row_count = 0
        self._mode = "Until stopped"
        self._limit = 0
        self._t_start = 0.0
        self.path: Optional[Path] = None

    @property
    def is_running(self) -> bool:
        return self._timer.isActive()

    def start(self, rate_hz: float, mode: str, limit: int, directory: Path,
              filename: str, overwrite: bool, uri: str = "") -> Tuple[bool, Optional[str]]:
        if self.is_running:
            return False, "Already recording."
        keys, labels_map, _ = self._provider()
        if not keys:
            return False, "No subscriptions to record."
        labels = [labels_map.get(k, k) for k in keys]

        directory.mkdir(parents=True, exist_ok=True)
        if overwrite and filename.strip():
            target = directory / _ensure_csv_suffix(filename.strip())
        else:
            stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            base = Path(filename.strip() or "rec").stem
            target = directory / f"{base}_{stamp}.csv"

        try:
            self._file = open(target, "w", encoding="utf-8", newline="")
            now_utc = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
            for line in (
                "# Generated by UaSubscription",
                f"# Date (UTC): {now_utc}",
                f"# Server URI: {uri or 'N/A'}",
                f"# Recording rate (Hz): {rate_hz}",
                f"# Recording mode: {mode}",
                f"# Recording limit: {limit}",
                f"# Output file: {target}",
            ):
                self._file.write(line + "\n")
            self._writer = csv.writer(self._file, lineterminator="\n")
            self._writer.writerow(["timestamp"] + labels)
            self._file.flush()
        except Exception as exc:
            return False, f"Could not open output file: {exc}"

        self._keys = keys
        self._row_count = 0
        self._mode = mode
        self._limit = max(0, int(limit))
        self._t_start = time.time()
        self.path = target
        rate = max(0.01, float(rate_hz))
        interval_ms = max(1, int(1000.0 / rate))
        self._timer.start(interval_ms)
        self._on_started(target)
        return True, None

    def stop(self, reason: Optional[str] = None):
        if not self.is_running and self._file is None:
            return
        self._timer.stop()
        if self._file is not None:
            try:
                self._file.close()
            except Exception:
                pass
            self._file = None
        self._writer = None
        self._on_stopped(reason)

    def _tick(self):
        if self._file is None:
            return
        _, _, latest = self._provider()
        cells = [datetime.now().isoformat(timespec="milliseconds")]
        for key in self._keys:
            cells.append(latest.get(key, ""))
        try:
            self._writer.writerow(cells)
            self._file.flush()
        except Exception as exc:
            self.stop(reason=f"Write error: {exc}")
            return
        self._row_count += 1
        self._on_tick(self._row_count)
        if self._mode == "Samples" and self._limit > 0 and self._row_count >= self._limit:
            self.stop(reason="Sample limit reached")
        elif self._mode == "Period of time" and self._limit > 0:
            if (time.time() - self._t_start) >= self._limit:
                self.stop(reason="Time limit reached")


class CommandServer(QObject):
    """QLocalServer wrapper that parses newline-delimited JSON commands.

    Unchanged from the original passive-viewer protocol used by UaShell."""

    def __init__(self, channel: str,
                 handler: Callable[[Dict[str, object]], Dict[str, object]],
                 parent=None):
        super().__init__(parent)
        self.channel = channel
        self.handler = handler
        self._buffers = {}
        self.server = QLocalServer(self)
        QLocalServer.removeServer(channel)
        try:
            self.server.setSocketOptions(QLocalServer.SocketOption.UserAccessOption)
        except Exception:
            pass
        self.server.newConnection.connect(self._on_new_connection)
        if not self.server.listen(channel):
            raise RuntimeError(
                f"Could not listen on {channel}: {self.server.errorString()}")

    def _on_new_connection(self):
        while self.server.hasPendingConnections():
            sock = self.server.nextPendingConnection()
            self._buffers[id(sock)] = ""
            sock.readyRead.connect(lambda s=sock: self._on_ready_read(s))
            sock.disconnected.connect(lambda s=sock: self._buffers.pop(id(s), None))

    def _on_ready_read(self, sock: "QLocalSocket"):
        key = id(sock)
        buf = self._buffers.get(key, "")
        buf += bytes(sock.readAll()).decode(errors="replace")
        while "\n" in buf:
            line, buf = buf.split("\n", 1)
            line = line.strip()
            if not line:
                continue
            try:
                payload = json.loads(line)
            except Exception:
                continue
            response = {}
            try:
                response = self.handler(payload) or {}
            except Exception as exc:
                response = {"type": "error", "message": str(exc)}
            if "id" in payload:
                if "reply_to" not in response:
                    response["reply_to"] = payload["id"]
                try:
                    sock.write((json.dumps(response) + "\n").encode())
                    sock.flush()
                except Exception:
                    pass
        self._buffers[key] = buf


class SubscriptionWindow(QMainWindow):
    """Main viewer: a live table (one row per subscribed node), aligned with
    UaPlot — panel menu, Server/Name[/Session] header, Add/Remove buttons,
    recording control. Dual-source: UaShell push (CommandServer) and/or its own
    asyncua worker feed the same table.

    ``standalone`` controls the parent-vs-standalone policy: when launched by
    UaExplorer/UaShell the launching tool owns the connection, so changing the
    URI and loading sessions are DISABLED; both are only offered for a plain,
    standalone launch."""

    def __init__(self, uri: str = "", channel: str = "",
                 theme: Optional[Theme] = None, server_name: str = "",
                 title: str = "OPC UA Subscriptions", standalone: bool = True,
                 connect: bool = True):
        super().__init__()
        self.setWindowTitle(title)
        self.uri = uri
        self.channel = channel
        self.standalone = standalone
        # When False, ``uri`` is display-only (push mode): show it in the
        # header but never open our own connection.
        self._connect = connect

        # node_id -> table row index
        self._rows: Dict[str, int] = {}
        # node_id -> latest value string (for the recorder)
        self._latest: Dict[str, str] = {}
        # node_id -> display name (set at subscribe-time; survives value ticks)
        self._node_names: Dict[str, str] = {}
        # Per-row stats (Min/Max/Mean/Count), mirroring UaExplorer's
        # subscription widget. Columns hidden unless enabled in Preferences.
        # window_size = sliding buffer (deque maxlen); only numeric samples are
        # buffered (booleans/strings render "-").
        self.subscription_stats = {
            "enabled": False, "show_min": True, "show_max": True,
            "show_mean": True, "show_count": True, "window_size": 1000,
            "decimals": 3,
        }
        self._stats_buffers: Dict[str, deque] = {}
        self._stats_total_counts: Dict[str, int] = {}
        # Subscription-table font size (pt), configurable in Preferences.
        self._table_font_size = 9

        # Worker / connect / browse state.
        self.worker: Optional[UaSubscriptionWorker] = None
        self._worker_connected = False
        self._pending_add = False
        self._browse_progress: Optional[QProgressDialog] = None
        self._browse_cache: Optional[List[Tuple[str, str]]] = None
        self._browse_cache_uri: Optional[str] = None
        self._browse_in_flight = False

        # Per-URI friendly names + recording settings + session state.
        self.server_names: Dict[str, str] = {}
        self.recording_settings = {
            "visible": False, "rate_hz": 1.0, "mode": "Until stopped",
            "limit": 60, "filename": "", "directory": "", "overwrite": False,
        }
        self._recorder: Optional[_SubscriptionRecorder] = None
        self._last_session_path: Optional[str] = None
        self._session_loaded = False
        self._stored_geometry: Optional[bytes] = None

        self._theme_explicit = theme is not None
        self.theme: Theme = theme or LIGHT_THEME

        override = os.environ.get("UASUBSCRIPTION_HOME", "").strip()
        if override:
            self.settings_dir = Path(os.path.expanduser(os.path.expandvars(override)))
        else:
            self.settings_dir = Path.home() / ".uatools" / "UaSubscription"
        self.settings_file = self.settings_dir / "settings.json"
        self.sessions_dir = self.settings_dir / "sessions"
        # Default recordings dir = $HOME, same as UaPlot (recordings are user
        # data, not app state). View/Save open here until the user picks else.
        if not self.recording_settings["directory"]:
            self.recording_settings["directory"] = str(Path.home())

        try:
            candidate = Path(sys.argv[0]).resolve()
            self._startup_script = candidate if candidate.exists() else Path(__file__).resolve()
        except Exception:
            self._startup_script = Path(__file__).resolve()

        # Push path (UaShell) — only bind the socket when a channel was given.
        self._server = None
        if channel:
            self._server = CommandServer(channel, self._handle_command, self)

        self._load_settings()
        # A name supplied by the launcher wins over the persisted one.
        if server_name and uri:
            if self.server_names.get(uri, "") != server_name:
                self.server_names[uri] = server_name
                self._save_settings()

        self._build_ui()

        # Start our own worker only when we're meant to connect (a real --uri,
        # i.e. standalone). In push mode (--socket / --display-uri) the URI is
        # display-only and UaShell's pushes are the sole live source, so we
        # never open a second connection / double-subscribe.
        if _HAVE_ASYNCUA and self.uri and self._connect:
            self._start_worker()

    # ---- settings ---------------------------------------------------------

    def _load_settings(self):
        try:
            if not self.settings_file.is_file():
                return
            data = json.loads(self.settings_file.read_text(encoding="utf-8"))
        except Exception:
            return
        rec = data.get("recording_settings")
        if isinstance(rec, dict):
            for k in ("visible", "rate_hz", "mode", "limit", "filename",
                      "directory", "overwrite"):
                if k in rec:
                    self.recording_settings[k] = rec[k]
        names = data.get("server_names")
        if isinstance(names, dict):
            self.server_names = {
                str(k): str(v).strip() for k, v in names.items()
                if isinstance(k, str) and isinstance(v, str) and str(v).strip()
            }
        stats = data.get("subscription_stats")
        if isinstance(stats, dict):
            for k in ("enabled", "show_min", "show_max", "show_mean",
                      "show_count", "window_size", "decimals"):
                if k in stats:
                    self.subscription_stats[k] = stats[k]
        try:
            self._table_font_size = int(data.get("table_font_size", self._table_font_size))
        except (TypeError, ValueError):
            pass
        stored_theme = data.get("theme")
        if stored_theme and not self._theme_explicit:
            resolved = THEMES.get(stored_theme)
            if resolved is not None:
                self.theme = resolved
        last_session = data.get("last_session")
        if isinstance(last_session, str) and last_session:
            self._last_session_path = last_session
        geometry_hex = data.get("geometry")
        if isinstance(geometry_hex, str) and geometry_hex:
            try:
                self._stored_geometry = bytes.fromhex(geometry_hex)
            except Exception:
                self._stored_geometry = None

    def _save_settings(self):
        try:
            self.settings_dir.mkdir(parents=True, exist_ok=True)
            try:
                geometry_hex = bytes(self.saveGeometry().toHex().data()).decode("ascii")
            except Exception:
                geometry_hex = ""
            payload = {
                "theme": self.theme.name,
                "recording_settings": dict(self.recording_settings),
                "server_names": dict(self.server_names),
                "subscription_stats": dict(self.subscription_stats),
                "table_font_size": int(self._table_font_size),
                "geometry": geometry_hex,
                "last_session": self._last_session_path or "",
            }
            tmp = self.settings_file.with_suffix(".tmp")
            tmp.write_text(json.dumps(payload, indent=2))
            tmp.replace(self.settings_file)
        except Exception:
            pass

    # ---- UI build ---------------------------------------------------------

    def _build_ui(self):
        central = QWidget(self)
        self.setCentralWidget(central)
        layout = QVBoxLayout(central)

        base_font = QApplication.font()
        base_pt = base_font.pointSizeF() if base_font.pointSizeF() > 0 else 10.0
        header_font = QFont(base_font)
        header_font.setPointSizeF(base_pt * 1.4)

        # --- header strip ---
        header = QHBoxLayout()

        self.menu_button = QToolButton()
        self.menu_button.setText("=")
        self.menu_button.setToolButtonStyle(Qt.ToolButtonStyle.ToolButtonTextOnly)
        self.menu_button.setAutoRaise(True)
        self.menu_button.setStyleSheet(self._menu_button_stylesheet())
        self.menu_button.setPopupMode(QToolButton.ToolButtonPopupMode.InstantPopup)
        self.menu_button.setMenu(self._build_menu())
        header.addWidget(self.menu_button)

        server_label = QLabel("Server:")
        server_label.setFont(header_font)
        header.addWidget(server_label)
        self.uri_label = QLabel(self.uri or "-")
        self.uri_label.setFont(header_font)
        self.uri_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
        header.addWidget(self.uri_label)

        sep_label = QLabel(" / ")
        sep_label.setFont(header_font)
        header.addWidget(sep_label)

        name_label = QLabel("Name:")
        name_label.setFont(header_font)
        header.addWidget(name_label)
        self.name_edit = QLineEdit()
        self.name_edit.setPlaceholderText("-")
        self.name_edit.setToolTip(
            "Friendly name for this server URI. Saved per URI; the same\n"
            "URI on a future run will show the same name automatically.")
        self.name_edit.setMaximumWidth(220)
        self.name_edit.setFont(header_font)
        self.name_edit.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
        self.name_edit.setFrame(False)
        self.name_edit.setTextMargins(0, 0, 0, 0)
        self.name_edit.setStyleSheet(self._name_edit_stylesheet())
        self.name_edit.editingFinished.connect(self._on_server_name_edited)
        self.name_edit.textChanged.connect(lambda _=None: self._resize_name_edit_to_content())
        header.addWidget(self.name_edit)
        self._refresh_server_name_display()
        self._resize_name_edit_to_content()

        # Session segment (standalone only — parent-driven viewers can't load).
        self.session_sep_label = QLabel(" / ")
        self.session_sep_label.setFont(header_font)
        header.addWidget(self.session_sep_label)
        self.session_caption_label = QLabel("Session:")
        self.session_caption_label.setFont(header_font)
        header.addWidget(self.session_caption_label)
        self.session_name_label = QLabel("")
        self.session_name_label.setFont(header_font)
        self.session_name_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
        header.addWidget(self.session_name_label)
        self._refresh_session_display()

        header.addStretch(1)

        # Add / Remove subscription buttons on the right. Active only in
        # standalone mode — when UaShell drives the viewer it owns the
        # subscription set, so the buttons are disabled with a hint.
        self.add_btn = QPushButton("Add")
        self.add_btn.clicked.connect(self._on_add_subscription)
        self.add_btn.setEnabled(self.standalone)
        header.addWidget(self.add_btn)
        self.remove_btn = QPushButton("Remove")
        self.remove_btn.clicked.connect(self._on_remove_subscription)
        self.remove_btn.setEnabled(self.standalone)
        header.addWidget(self.remove_btn)
        if not self.standalone:
            tip = "Subscriptions are managed from UaShell while it drives this viewer."
            self.add_btn.setToolTip(tip)
            self.remove_btn.setToolTip(tip)

        layout.addLayout(header)

        # --- count label ---
        self._count_label = QLabel("Subscriptions: 0")
        layout.addWidget(self._count_label)

        # --- recording row ---
        self.recording_row = self._build_recording_row()
        self.recording_row.setVisible(bool(self.recording_settings.get("visible")))
        layout.addWidget(self.recording_row)

        # --- table ---
        self.table = QTableWidget(0, len(COLUMNS), self)
        self.table.setHorizontalHeaderLabels(COLUMNS)
        self.table.verticalHeader().setVisible(False)
        table_font = QFont()
        table_font.setPointSize(int(self._table_font_size))
        self.table.setFont(table_font)
        self.table.horizontalHeader().setFont(table_font)
        # Don't bold the header of the column holding the current/selected cell
        # (Qt does this by default; it made "Node ID" go bold after a load).
        self.table.horizontalHeader().setHighlightSections(False)
        h = self.table.horizontalHeader()
        h.setSectionResizeMode(COL_NODE, QHeaderView.ResizeMode.ResizeToContents)
        h.setSectionResizeMode(COL_NAME, QHeaderView.ResizeMode.ResizeToContents)
        h.setSectionResizeMode(COL_VALUE, QHeaderView.ResizeMode.Stretch)
        h.setSectionResizeMode(COL_TYPE, QHeaderView.ResizeMode.ResizeToContents)
        h.setSectionResizeMode(COL_QUALITY, QHeaderView.ResizeMode.ResizeToContents)
        h.setSectionResizeMode(COL_TS, QHeaderView.ResizeMode.ResizeToContents)
        for col in (COL_MIN, COL_MAX, COL_MEAN, COL_COUNT):
            h.setSectionResizeMode(col, QHeaderView.ResizeMode.ResizeToContents)
        self._apply_stats_visibility()
        # Full-row selection + multi-select so the user can pick rows and
        # remove them with the Delete key or a right-click menu (standalone:
        # actually unsubscribes; push mode: just drops the row from the view).
        self.table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
        self.table.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
        self.table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
        self.table.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
        self.table.customContextMenuRequested.connect(self._on_table_context_menu)
        QShortcut(QKeySequence(Qt.Key.Key_Delete), self.table,
                  activated=self._remove_selected_rows)
        layout.addWidget(self.table)

        if self._stored_geometry:
            try:
                from PyQt6.QtCore import QByteArray
                self.restoreGeometry(QByteArray(self._stored_geometry))
            except Exception:
                self.resize(1000, 600)
        else:
            self.resize(1000, 600)

        QShortcut(QKeySequence("Ctrl+Shift+R"), self, activated=self.restart_application)

    def _menu_button_stylesheet(self) -> str:
        # Tight horizontal padding (was 8px each side -> too much space around
        # the "=" glyph); UaPlot uses none. 2px keeps it from looking cramped.
        return ("QToolButton { border: none; padding: 2px 2px; font-size: 16px; }"
                "QToolButton::menu-indicator { image: none; width: 0px; }")

    def _name_edit_stylesheet(self) -> str:
        return (f"QLineEdit {{ background: transparent; padding: 0px; "
                f"color: {self.theme.text_primary}; }}")

    def _build_menu(self) -> QMenu:
        menu = QMenu(self)
        # Load Session is standalone-only (a parent owns the connection).
        save_action = menu.addAction("Save Session…")
        save_action.triggered.connect(self._save_session_dialog)
        load_action = menu.addAction("Load Session…")
        load_action.triggered.connect(self._load_session_dialog)
        load_action.setEnabled(self.standalone)
        if not self.standalone:
            load_action.setToolTip("Available only when UaSubscription is started standalone")
        menu.addSeparator()
        about_action = menu.addAction("About")
        about_action.triggered.connect(self._show_about)
        prefs_action = menu.addAction("Preferences")
        prefs_action.triggered.connect(self._show_preferences)
        menu.addSeparator()
        self.rec_action = menu.addAction("Show Recording Controls")
        self.rec_action.setCheckable(True)
        self.rec_action.setChecked(bool(self.recording_settings.get("visible")))
        self.rec_action.toggled.connect(self._toggle_recording_controls)
        menu.addSeparator()
        restart_action = menu.addAction("Restart\tCtrl+Shift+R")
        restart_action.triggered.connect(self.restart_application)
        exit_action = menu.addAction("Exit")
        exit_action.triggered.connect(self.close)
        return menu

    # ---- header helpers ---------------------------------------------------

    def _refresh_server_name_display(self):
        if not hasattr(self, "name_edit"):
            return
        name = str(self.server_names.get(self.uri, "")).strip() if self.uri else ""
        self.name_edit.blockSignals(True)
        self.name_edit.setText(name)
        self.name_edit.blockSignals(False)
        self._resize_name_edit_to_content()

    def _resize_name_edit_to_content(self):
        if not hasattr(self, "name_edit"):
            return
        # Build metrics from the widget's FONT (QFontMetrics(font)) rather than
        # self.name_edit.fontMetrics(): before the widget is shown — which is
        # the case when launched from UaShell with --server-name — fontMetrics()
        # can report a near-zero advance and collapse the field to one char even
        # though a name is present. QFontMetrics(font) is reliable pre-show.
        from PyQt6.QtGui import QFontMetrics
        fm = QFontMetrics(self.name_edit.font())
        text = self.name_edit.text() or self.name_edit.placeholderText() or "-"
        m = self.name_edit.textMargins()
        adv = fm.horizontalAdvance(text)
        # Cap against a CONSTANT, never against the widget's live
        # maximumWidth(): setFixedWidth() sets max==min, so reading
        # maximumWidth() here would clamp to whatever a previous (e.g. the
        # short "-" placeholder) call set — permanently collapsing the field to
        # one char and never letting it grow back. That was the bug.
        width = min(adv + m.left() + m.right() + 8, self._NAME_EDIT_MAX_W)
        final = max(width, 24)
        # Only touch the widget when the width actually changes — otherwise
        # setFixedWidth re-triggers a resize/relayout that calls us again,
        # producing an infinite repaint loop.
        if self.name_edit.width() != final:
            self.name_edit.setFixedWidth(final)

    # Hard cap for the Name field width (was setMaximumWidth(220); kept as a
    # constant so the resize math can't be poisoned by setFixedWidth).
    _NAME_EDIT_MAX_W = 220

    def _refresh_session_display(self):
        if not hasattr(self, "session_name_label"):
            return
        stem = ""
        if self._session_loaded and self._last_session_path:
            stem = Path(self._last_session_path).stem.strip()
        visible = bool(stem)
        self.session_sep_label.setVisible(visible)
        self.session_caption_label.setVisible(visible)
        self.session_name_label.setVisible(visible)
        self.session_name_label.setText(stem)

    def _on_server_name_edited(self):
        if not self.uri:
            self._refresh_server_name_display()
            return
        name = self.name_edit.text().strip()
        if name:
            self.server_names[self.uri] = name
        else:
            self.server_names.pop(self.uri, None)
        self._save_settings()
        self._resize_name_edit_to_content()

    # ---- worker / connect / browse ---------------------------------------

    def _set_uri(self, uri: str):
        # Standalone-only: a parent-driven viewer must not change the server.
        if not self.standalone:
            return
        if uri and uri != self.uri:
            self.uri = uri
            self.uri_label.setText(uri)
            self._refresh_server_name_display()
            self._invalidate_browse_cache()
            self._start_worker()

    def _start_worker(self):
        if not _HAVE_ASYNCUA:
            return
        if self.worker:
            self.worker.stop()
            self.worker.wait()
        self._worker_connected = False
        self._browse_in_flight = False
        self.worker = UaSubscriptionWorker(self.uri)
        self.worker.data_update.connect(self._on_data_update)
        self.worker.error.connect(self._on_worker_error)
        self.worker.connected.connect(self._on_worker_connected)
        self.worker.disconnected.connect(self._on_worker_disconnected)
        self.worker.browse_result.connect(self._on_browse_result)
        self.worker.browse_error.connect(self._on_browse_error)
        self.worker.start()

    def _on_worker_error(self, message: str):
        _log(f"[worker error] {message}")
        self.statusBar().showMessage(message, 5000)

    def _on_worker_disconnected(self):
        self._worker_connected = False

    def _on_worker_connected(self, uri: str):
        self._worker_connected = True
        self._start_background_browse()
        if self._pending_add and self.worker and not self._browse_in_flight:
            self._browse_in_flight = True
            self.worker.request_browse()

    def _browse_cache_valid(self) -> bool:
        return (self._browse_cache is not None
                and self._browse_cache_uri == self.uri and bool(self.uri))

    def _invalidate_browse_cache(self):
        self._browse_cache = None
        self._browse_cache_uri = None

    def _start_background_browse(self):
        if not self.uri or self._browse_in_flight or self._browse_cache_valid():
            return
        if self._worker_connected and self.worker:
            self._browse_in_flight = True
            self.worker.request_browse()

    def _ensure_connected_for_browse(self) -> bool:
        """Ensure a URI to browse. In standalone mode, prompt for one if none
        is set. In parent-driven mode the URI is fixed by the launcher; if it
        somehow isn't set, refuse rather than prompt."""
        if self.uri:
            return True
        if not self.standalone:
            QMessageBox.information(
                self, "Add Subscription",
                "This viewer is driven by its launcher; connect from there.")
            return False
        uri, ok = QInputDialog.getText(
            self, "Connect to server",
            "OPC UA endpoint to browse for values:", text="opc.tcp://")
        if not ok or not uri.strip():
            return False
        self._set_uri(uri.strip())
        return True

    def _on_add_subscription(self):
        if not _HAVE_ASYNCUA:
            QMessageBox.warning(self, "Add Subscription",
                                "asyncua is not installed; live browse unavailable.")
            return
        if self._pending_add:
            return
        if not self._ensure_connected_for_browse():
            return
        if self._browse_cache_valid():
            self._open_add_dialog(self._browse_cache)
            return
        self._pending_add = True
        self._browse_progress = QProgressDialog(
            "Browsing server address space…", "Cancel", 0, 0, self)
        self._browse_progress.setWindowTitle("Add Subscription")
        self._browse_progress.setMinimumDuration(0)
        self._browse_progress.canceled.connect(self._cancel_pending_browse)
        self._browse_progress.show()
        QTimer.singleShot(10000, self._browse_timeout)
        if self._worker_connected and self.worker and not self._browse_in_flight:
            self._browse_in_flight = True
            self.worker.request_browse()

    def _open_add_dialog(self, nodes):
        existing = set(self._rows.keys())
        dlg = _AddSubscriptionDialog(list(nodes or []), existing, self)
        if dlg.exec() != QDialog.DialogCode.Accepted:
            return
        for node_id, display_name in dlg.selected_nodes():
            norm = self._normalize_node_id(node_id)
            self._node_names[norm] = display_name
            # Placeholder row so it shows immediately; values fill in on ticks.
            self._upsert({"node": norm, "name": display_name, "value": "",
                          "type": "", "quality": "", "ts": ""})
            if self.worker:
                self.worker.add_node(norm)

    def _cancel_pending_browse(self):
        self._pending_add = False
        self._close_browse_progress()

    def _browse_timeout(self):
        if not self._pending_add:
            return
        self._cancel_pending_browse()
        QMessageBox.warning(self, "Add Subscription",
                            f"Could not browse the server (no response from {self.uri or 'server'}).")

    def _close_browse_progress(self):
        if self._browse_progress is not None:
            self._browse_progress.close()
            self._browse_progress = None

    def _on_browse_result(self, nodes):
        node_list = list(nodes or [])
        self._browse_cache = node_list
        self._browse_cache_uri = self.uri
        self._browse_in_flight = False
        pending = self._pending_add
        self._pending_add = False
        self._close_browse_progress()
        if pending:
            self._open_add_dialog(node_list)

    def _on_browse_error(self, message: str):
        self._browse_in_flight = False
        had_pending = self._pending_add
        self._pending_add = False
        self._close_browse_progress()
        if had_pending:
            QMessageBox.warning(self, "Add Subscription", message or "Browse failed.")

    def _on_remove_subscription(self):
        if not self._rows:
            QMessageBox.information(self, "Remove Subscription", "No subscriptions to remove.")
            return
        subs = [(nid, self._node_names.get(nid, "")) for nid in self._rows.keys()]
        dlg = _RemoveSubscriptionDialog(subs, self)
        if dlg.exec() != QDialog.DialogCode.Accepted:
            return
        self._remove_nodes(dlg.selected_nodes())

    def _remove_nodes(self, node_ids: List[str]):
        """Unsubscribe (if we own the connection) and drop the rows for the
        given node ids. Shared by the Remove dialog, the table context menu,
        and the Delete key."""
        for node_id in node_ids:
            norm = self._normalize_node_id(node_id)
            if self.worker:
                self.worker.remove_node(norm)
            self._remove(norm)
            self._node_names.pop(norm, None)
            self._latest.pop(norm, None)

    def _selected_row_nodes(self) -> List[str]:
        """Node ids of the currently selected table rows (from the Node ID
        column), de-duplicated and in row order."""
        nodes: List[str] = []
        seen = set()
        for idx in self.table.selectionModel().selectedRows() if self.table.selectionModel() else []:
            item = self.table.item(idx.row(), COL_NODE)
            if item is not None:
                nid = item.text()
                if nid and nid not in seen:
                    seen.add(nid)
                    nodes.append(nid)
        return nodes

    def _remove_selected_rows(self):
        # Removing only makes sense when we own the subscription set. In push
        # mode UaShell owns it and would re-push the row, so ignore.
        if not self.standalone:
            return
        nodes = self._selected_row_nodes()
        if nodes:
            self._remove_nodes(nodes)

    def _on_table_context_menu(self, pos):
        if not self.standalone:
            return
        nodes = self._selected_row_nodes()
        # If the right-click landed on a row that isn't part of the current
        # selection, act on that row instead.
        if not nodes:
            item = self.table.itemAt(pos)
            if item is not None:
                nid_item = self.table.item(item.row(), COL_NODE)
                if nid_item is not None and nid_item.text():
                    nodes = [nid_item.text()]
        if not nodes:
            return
        menu = QMenu(self)
        label = "Remove subscription" if len(nodes) == 1 else f"Remove {len(nodes)} subscriptions"
        act = menu.addAction(label)
        chosen = menu.exec(self.table.viewport().mapToGlobal(pos))
        if chosen is act:
            self._remove_nodes(nodes)

    # ---- live data --------------------------------------------------------

    @staticmethod
    def _infer_type(value: object) -> str:
        """Readable type name inferred from a Python value. The worker delivers
        only (node_id, value, ts) — it has no DataType — so in standalone mode
        we derive the Type column from the value's Python type."""
        if isinstance(value, bool):
            return "Boolean"
        if isinstance(value, int):
            return "Int64"
        if isinstance(value, float):
            return "Double"
        if isinstance(value, str):
            return "String"
        if isinstance(value, (bytes, bytearray)):
            return "ByteString"
        if isinstance(value, (list, tuple)):
            return "Array"
        return type(value).__name__

    def _on_data_update(self, node_id: str, value: object, timestamp: float):
        norm = self._normalize_node_id(node_id)
        try:
            ts = datetime.fromtimestamp(timestamp).strftime("%H:%M:%S.%f")[:-3]
        except Exception:
            ts = ""
        self._upsert({"node": norm, "name": self._node_names.get(norm, ""),
                      "value": str(value), "type": self._infer_type(value),
                      "quality": "Good", "ts": ts})

    # ---- recording row ----------------------------------------------------

    def _build_recording_row(self) -> QWidget:
        row_widget = QWidget()
        row = QHBoxLayout(row_widget)
        row.setContentsMargins(2, 2, 2, 2)
        row.setSpacing(6)
        row.addWidget(QLabel("Recording"))

        self.rec_rate_input = QLineEdit(self._fmt_rate(self.recording_settings.get("rate_hz") or 1.0))
        self.rec_rate_input.setFixedWidth(56)
        self.rec_rate_input.setMaxLength(5)
        rv = QDoubleValidator(0.1, 999.9, 1, self.rec_rate_input)
        rv.setNotation(QDoubleValidator.Notation.StandardNotation)
        self.rec_rate_input.setValidator(rv)
        self.rec_rate_input.setToolTip("Sample rate (Hz): how many CSV rows per second.")
        self.rec_rate_input.editingFinished.connect(self._on_rec_rate_edited)
        row.addWidget(self.rec_rate_input)
        row.addWidget(QLabel("Hz"))

        self.rec_mode_combo = QComboBox()
        self.rec_mode_combo.addItems(["Until stopped", "Period of time", "Samples"])
        cur_mode = str(self.recording_settings.get("mode") or "Until stopped")
        if cur_mode in ("Until stopped", "Period of time", "Samples"):
            self.rec_mode_combo.setCurrentText(cur_mode)
        self.rec_mode_combo.currentTextChanged.connect(self._on_rec_mode_changed)
        row.addWidget(self.rec_mode_combo)

        self.rec_limit_label = QLabel("Limit")
        row.addWidget(self.rec_limit_label)
        self.rec_limit_input = QLineEdit(str(int(self.recording_settings.get("limit") or 60)))
        self.rec_limit_input.setFixedWidth(56)
        self.rec_limit_input.setMaxLength(4)
        self.rec_limit_input.setValidator(QIntValidator(1, 9999, self.rec_limit_input))
        self.rec_limit_input.editingFinished.connect(self._on_rec_limit_edited)
        row.addWidget(self.rec_limit_input)
        self.rec_limit_unit_label = QLabel("")
        row.addWidget(self.rec_limit_unit_label)

        row.addWidget(QLabel("File"))
        self.rec_filename_edit = QLineEdit(str(self.recording_settings.get("filename") or ""))
        self.rec_filename_edit.setPlaceholderText("auto-timestamped")
        self.rec_filename_edit.editingFinished.connect(
            lambda: self._update_rec_setting("filename", self.rec_filename_edit.text()))
        row.addWidget(self.rec_filename_edit, 1)
        self.rec_browse_btn = QPushButton("…")
        self.rec_browse_btn.setFixedWidth(28)
        self.rec_browse_btn.clicked.connect(self._on_rec_browse)
        row.addWidget(self.rec_browse_btn)

        self.rec_overwrite_check = QCheckBox("Overwrite")
        self.rec_overwrite_check.setChecked(bool(self.recording_settings.get("overwrite")))
        self.rec_overwrite_check.toggled.connect(
            lambda v: self._update_rec_setting("overwrite", bool(v)))
        row.addWidget(self.rec_overwrite_check)

        self.rec_start_btn = QPushButton("Start")
        self.rec_start_btn.clicked.connect(self._on_rec_start)
        row.addWidget(self.rec_start_btn)
        self.rec_stop_btn = QPushButton("Stop")
        self.rec_stop_btn.setEnabled(False)
        self.rec_stop_btn.clicked.connect(self._on_rec_stop)
        row.addWidget(self.rec_stop_btn)
        self.rec_load_btn = QPushButton("View…")
        self.rec_load_btn.clicked.connect(self._on_rec_load)
        row.addWidget(self.rec_load_btn)

        self.rec_status_label = QLabel("● Idle")
        row.addWidget(self.rec_status_label)

        self._on_rec_mode_changed(self.rec_mode_combo.currentText())
        return row_widget

    def _on_rec_mode_changed(self, mode: str):
        needs_limit = mode in ("Period of time", "Samples")
        self.rec_limit_label.setVisible(needs_limit)
        self.rec_limit_input.setVisible(needs_limit)
        self.rec_limit_unit_label.setVisible(needs_limit)
        self.rec_limit_unit_label.setText(" s" if mode == "Period of time" else "")
        self._update_rec_setting("mode", mode)

    @staticmethod
    def _fmt_rate(value) -> str:
        try:
            return f"{float(value):.1f}"
        except (TypeError, ValueError):
            return "1.0"

    def _on_rec_rate_edited(self):
        rate = self._rec_rate_value()
        self.rec_rate_input.setText(self._fmt_rate(rate))
        self._update_rec_setting("rate_hz", rate)

    def _on_rec_limit_edited(self):
        self._update_rec_setting("limit", self._rec_limit_value())
        self.rec_limit_input.setText(str(self._rec_limit_value()))

    def _rec_rate_value(self) -> float:
        try:
            rate = float(self.rec_rate_input.text())
        except ValueError:
            rate = float(self.recording_settings.get("rate_hz") or 1.0)
        return min(999.9, max(0.1, rate))

    def _rec_limit_value(self) -> int:
        try:
            limit = int(self.rec_limit_input.text())
        except ValueError:
            limit = int(self.recording_settings.get("limit") or 60)
        return min(9999, max(1, limit))

    def _update_rec_setting(self, key: str, value):
        self.recording_settings[key] = value
        self._save_settings()

    def _on_rec_browse(self):
        rec_dir = Path(self.recording_settings.get("directory") or str(Path.home()))
        rec_dir.mkdir(parents=True, exist_ok=True)
        cur = self.recording_settings.get("filename") or ""
        start_path = str(rec_dir / cur) if cur else str(rec_dir)
        dialog = QFileDialog(self, "Recording output file", start_path)
        dialog.setAcceptMode(QFileDialog.AcceptMode.AcceptSave)
        dialog.setNameFilters(["CSV files (*.csv)"])
        dialog.setDefaultSuffix("csv")
        dialog.setOption(QFileDialog.Option.DontUseNativeDialog, True)
        dialog.resize(720, 480)
        if dialog.exec() != QDialog.DialogCode.Accepted:
            return
        files = dialog.selectedFiles()
        if not files:
            return
        chosen = Path(files[0])
        csv_name = _ensure_csv_suffix(chosen.name)
        self.recording_settings["directory"] = str(chosen.parent)
        self.recording_settings["filename"] = csv_name
        self.rec_filename_edit.setText(csv_name)
        self._save_settings()

    def _on_rec_load(self):
        rec_dir = Path(self.recording_settings.get("directory") or str(Path.home()))
        try:
            rec_dir.mkdir(parents=True, exist_ok=True)
        except Exception:
            pass
        start_path = str(rec_dir)
        # Prefer the just-recorded file, else the configured filename if present.
        if self._recorder and self._recorder.path:
            start_path = str(self._recorder.path)
        elif self.recording_settings.get("filename"):
            cand = rec_dir / str(self.recording_settings["filename"])
            if cand.exists():
                start_path = str(cand)
        dialog = QFileDialog(self, "View Recording", start_path)
        dialog.setFileMode(QFileDialog.FileMode.ExistingFile)
        dialog.setNameFilters(["CSV files (*.csv)"])
        dialog.setOption(QFileDialog.Option.DontUseNativeDialog, True)
        if dialog.exec() != QDialog.DialogCode.Accepted:
            return
        files = dialog.selectedFiles()
        if files:
            _RecordingViewerDialog(Path(files[0]), parent=self).exec()

    def _recorder_snapshot(self):
        """Provider for _SubscriptionRecorder: (ordered keys, labels, values)."""
        keys = list(self._rows.keys())
        labels = {k: self._node_names.get(k, k) for k in keys}
        values = dict(self._latest)
        return keys, labels, values

    def _on_rec_start(self):
        if self._recorder is None:
            self._recorder = _SubscriptionRecorder(
                self._recorder_snapshot, self._on_rec_started_ui,
                self._on_rec_stopped_ui, self._on_rec_tick_ui, self)
        rec_dir = Path(self.recording_settings.get("directory") or str(Path.home()))
        ok, err = self._recorder.start(
            rate_hz=self._rec_rate_value(), mode=self.rec_mode_combo.currentText(),
            limit=self._rec_limit_value(), directory=rec_dir,
            filename=self.rec_filename_edit.text().strip(),
            overwrite=self.rec_overwrite_check.isChecked(), uri=self.uri)
        if not ok:
            QMessageBox.warning(self, "Recording", err or "Failed to start.")

    def _on_rec_stop(self):
        if self._recorder is not None:
            self._recorder.stop(reason="User stopped")

    def _on_rec_started_ui(self, path: Path):
        self.rec_start_btn.setEnabled(False)
        self.rec_stop_btn.setEnabled(True)
        for w in (self.rec_rate_input, self.rec_mode_combo, self.rec_limit_input,
                  self.rec_filename_edit, self.rec_browse_btn, self.rec_overwrite_check):
            w.setEnabled(False)
        self.rec_status_label.setText(f"● Recording — {path.name}")

    def _on_rec_stopped_ui(self, reason: Optional[str]):
        self.rec_start_btn.setEnabled(True)
        self.rec_stop_btn.setEnabled(False)
        for w in (self.rec_rate_input, self.rec_mode_combo, self.rec_limit_input,
                  self.rec_filename_edit, self.rec_browse_btn, self.rec_overwrite_check):
            w.setEnabled(True)
        suffix = f" — {reason}" if reason else ""
        rows = getattr(self._recorder, "_row_count", 0) if self._recorder else 0
        self.rec_status_label.setText(f"● Idle ({rows} rows{suffix})")

    def _on_rec_tick_ui(self, row_count: int):
        name = self._recorder.path.name if self._recorder and self._recorder.path else ""
        self.rec_status_label.setText(f"● Recording — {name} ({row_count} rows)")

    def _toggle_recording_controls(self, enabled: bool):
        self.recording_row.setVisible(bool(enabled))
        self.recording_settings["visible"] = bool(enabled)
        self._save_settings()

    def _stop_recording_if_running(self, reason: str):
        if self._recorder is not None and self._recorder.is_running:
            self._recorder.stop(reason=reason)

    # ---- table (_upsert/_remove/_clear) -----------------------------------

    def _handle_command(self, payload: Dict[str, object]) -> Dict[str, object]:
        command = str(payload.get("command", "")).lower()
        if command == "ping":
            return {"type": "pong"}
        if command == "update":
            self._upsert(payload)
            return {"type": "ack"}
        if command == "remove":
            self._remove(self._normalize_node_id(str(payload.get("node", ""))))
            return {"type": "ack"}
        if command == "clear":
            self._clear()
            return {"type": "ack"}
        if command == "get_subscriptions":
            return {"type": "subscriptions", "rows": self._snapshot_rows()}
        if command == "quit":
            QTimer.singleShot(100, self.close)
            return {"type": "ack"}
        return {"type": "error", "message": f"Unknown command: {command}"}

    def _snapshot_rows(self) -> List[Dict[str, str]]:
        """Read back the current table as a list of row dicts. Used by a
        controller / integration tests to verify state (the table is otherwise
        write-only over IPC). Row order matches the table."""
        def cell(row: int, col: int) -> str:
            item = self.table.item(row, col)
            return item.text() if item is not None else ""
        rows: List[Dict[str, str]] = []
        # Iterate in table-row order so the result is stable.
        ordered = sorted(self._rows.items(), key=lambda kv: kv[1])
        for node, row in ordered:
            rows.append({
                "node": node,
                "name": cell(row, COL_NAME),
                "value": cell(row, COL_VALUE),
                "type": cell(row, COL_TYPE),
                "quality": cell(row, COL_QUALITY),
                "ts": cell(row, COL_TS),
                "min": cell(row, COL_MIN),
                "max": cell(row, COL_MAX),
                "mean": cell(row, COL_MEAN),
                "count": cell(row, COL_COUNT),
            })
        return rows

    def _normalize_node_id(self, node_id: str) -> str:
        if not node_id:
            return ""
        text = str(node_id)
        if text.startswith("ns="):
            return text
        if "Identifier=" in text and "NamespaceIndex=" in text:
            try:
                ident = text.split("Identifier=")[1].split(",")[0].strip().strip("'\"")
                ns_part = text.split("NamespaceIndex=")[1].split(",")[0].strip()
                if ident and ns_part:
                    return f"ns={ns_part};s={ident}"
            except Exception:
                pass
        return text

    def _upsert(self, data: Dict[str, object]):
        node = self._normalize_node_id(str(data.get("node", "")))
        if not node:
            return
        name = str(data.get("name", ""))
        value = str(data.get("value", ""))
        dtype = str(data.get("type", ""))
        quality = str(data.get("quality", ""))
        ts = str(data.get("ts", ""))

        # Remember a non-blank name and the latest value (for Remove + recorder).
        if name:
            self._node_names[node] = name
        if value != "" or node not in self._latest:
            self._latest[node] = value

        row = self._rows.get(node)
        if row is None:
            row = self.table.rowCount()
            self.table.insertRow(row)
            self._rows[node] = row
            # A row added by Add/session before its first datachange has no
            # value yet — show a muted "pending" in Value/Quality so it reads
            # as "subscribing", not broken. The first real update overwrites it.
            pending = (value == "")
            disp_value = "pending" if pending else value
            disp_quality = "pending" if pending else quality
            cells = [node, name or self._node_names.get(node, ""),
                     disp_value, dtype, disp_quality, ts]
            for col, text in enumerate(cells):
                item = QTableWidgetItem(text)
                if pending and col in (COL_VALUE, COL_QUALITY):
                    item.setForeground(QColor(self.theme.text_muted))
                self.table.setItem(row, col, item)
            self._count_label.setText(f"Subscriptions: {len(self._rows)}")
        else:
            # Update cells, but don't clobber a known Name with a blank.
            updates = {COL_VALUE: value, COL_TS: ts}
            if name:
                updates[COL_NAME] = name
            if dtype:
                updates[COL_TYPE] = dtype
            if quality:
                updates[COL_QUALITY] = quality
            for col, text in updates.items():
                item = self.table.item(row, col)
                if item is None:
                    self.table.setItem(row, col, QTableWidgetItem(text))
                else:
                    item.setText(text)
            # Real data has arrived: clear the muted "pending" tint on Value
            # (Quality is re-coloured below; Value uses the normal text color).
            value_item = self.table.item(row, COL_VALUE)
            if value_item is not None:
                value_item.setForeground(QColor(self.theme.text_primary))

        quality_item = self.table.item(row, COL_QUALITY)
        if quality_item is not None:
            qtext = quality_item.text()
            # "pending" is the muted placeholder, not a real quality value.
            if qtext == "pending":
                pass  # leave the muted pending tint until a real update
            else:
                good = (qtext == "Good" or qtext == "")
                # Theme-aware: a hardcoded black was invisible on dark themes
                # (text present but unreadable -> looked like empty Quality).
                quality_item.setForeground(
                    QColor(self.theme.text_primary) if good
                    else QColor(self.theme.status_error))

        # Feed numeric samples into the per-node stats buffer and refresh the
        # stat cells (only meaningful when a real value arrived).
        if value not in ("", "pending"):
            self._feed_stats(node, value)
            if self.subscription_stats.get("enabled"):
                self._refresh_stats_cells_for_row(row, node)
        self.table.viewport().update()

    def _feed_stats(self, node: str, value_str: str):
        """Append a numeric sample to the node's sliding buffer. Booleans and
        non-numeric values are skipped (their stat cells show '-')."""
        low = value_str.strip().lower()
        if low in ("true", "false"):
            return  # booleans: min/max/mean meaningless
        try:
            numeric = float(value_str)
        except (TypeError, ValueError):
            return
        win = int(self.subscription_stats.get("window_size") or 1000)
        buf = self._stats_buffers.get(node)
        if buf is None or buf.maxlen != win:
            buf = deque(buf or (), maxlen=win)
            self._stats_buffers[node] = buf
        buf.append(numeric)
        self._stats_total_counts[node] = self._stats_total_counts.get(node, 0) + 1

    def _fmt_stat(self, v: float) -> str:
        try:
            decimals = int(self.subscription_stats.get("decimals", 3))
        except (TypeError, ValueError):
            decimals = 3
        decimals = max(0, min(decimals, 12))
        try:
            return f"{float(v):.{decimals}f}"
        except (TypeError, ValueError):
            return "-"

    def _refresh_stats_cells_for_row(self, row: int, node: str):
        buf = self._stats_buffers.get(node)
        if buf and len(buf) > 0:
            mn, mx = min(buf), max(buf)
            mean = sum(buf) / len(buf)
            cnt = self._stats_total_counts.get(node, len(buf))
            self.table.setItem(row, COL_MIN, QTableWidgetItem(self._fmt_stat(mn)))
            self.table.setItem(row, COL_MAX, QTableWidgetItem(self._fmt_stat(mx)))
            self.table.setItem(row, COL_MEAN, QTableWidgetItem(self._fmt_stat(mean)))
            self.table.setItem(row, COL_COUNT, QTableWidgetItem(str(cnt)))
        else:
            for col in (COL_MIN, COL_MAX, COL_MEAN, COL_COUNT):
                self.table.setItem(row, col, QTableWidgetItem("-"))

    def _apply_stats_visibility(self):
        enabled = bool(self.subscription_stats.get("enabled"))
        for key, col in (("show_min", COL_MIN), ("show_max", COL_MAX),
                         ("show_mean", COL_MEAN), ("show_count", COL_COUNT)):
            hidden = not (enabled and bool(self.subscription_stats.get(key, True)))
            self.table.setColumnHidden(col, hidden)

    def _refresh_all_stats(self):
        for node, row in self._rows.items():
            self._refresh_stats_cells_for_row(row, node)

    def set_subscription_stats(self, config: Dict[str, object]):
        old_window = int(self.subscription_stats.get("window_size") or 1000)
        self.subscription_stats.update(config)
        new_window = int(self.subscription_stats.get("window_size") or 1000)
        if new_window != old_window:
            for nid, buf in list(self._stats_buffers.items()):
                self._stats_buffers[nid] = deque(buf, maxlen=new_window)
        self._apply_stats_visibility()
        if self.subscription_stats.get("enabled"):
            self._refresh_all_stats()
        self._save_settings()

    def _remove(self, node: str):
        row = self._rows.pop(node, None)
        if row is None:
            return
        self.table.removeRow(row)
        self._stats_buffers.pop(node, None)
        self._stats_total_counts.pop(node, None)
        self._reindex()
        self._count_label.setText(f"Subscriptions: {len(self._rows)}")

    def _clear(self):
        self.table.setRowCount(0)
        self._rows.clear()
        self._node_names.clear()
        self._latest.clear()
        self._stats_buffers.clear()
        self._stats_total_counts.clear()
        self._count_label.setText("Subscriptions: 0")

    def _reindex(self):
        self._rows.clear()
        for row in range(self.table.rowCount()):
            item = self.table.item(row, COL_NODE)
            if item is not None:
                self._rows[item.text()] = row

    # ---- sessions (standalone only) --------------------------------------

    def _build_session_payload(self) -> dict:
        monitor_mode = self.worker.monitor_mode if self.worker else "subscription"
        polling = self.worker.polling_interval_ms if self.worker else 500
        subs = []
        for node_id in self._rows.keys():
            subs.append({
                "node_id": node_id,
                "display_name": self._node_names.get(node_id, node_id),
                "monitor_mode": monitor_mode,
                "polling_interval_ms": polling,
            })
        return {
            "version": "1.1",
            "timestamp": datetime.now().isoformat(),
            "source": "UaSubscription",
            "uri": self.uri or "",
            "server_name": str(self.server_names.get(self.uri, "")).strip() if self.uri else "",
            "theme": self.theme.name,
            "recording_settings": dict(self.recording_settings),
            "monitor_mode": monitor_mode,
            "polling_interval_ms": polling,
            "subscriptions": subs,
            "plots": [],
        }

    def _save_session_dialog(self):
        self.sessions_dir.mkdir(parents=True, exist_ok=True)
        dialog = QFileDialog(self, "Save Session", str(self.sessions_dir))
        dialog.setAcceptMode(QFileDialog.AcceptMode.AcceptSave)
        dialog.setNameFilters(["Session Files (*.json)"])
        dialog.setDefaultSuffix("json")
        dialog.setOption(QFileDialog.Option.DontUseNativeDialog, True)
        if dialog.exec() != QDialog.DialogCode.Accepted:
            return
        files = dialog.selectedFiles()
        if not files:
            return
        filepath = files[0]
        if not filepath.endswith(".json"):
            filepath += ".json"
        try:
            tmp = Path(filepath).with_suffix(".tmp")
            tmp.write_text(json.dumps(self._build_session_payload(), indent=2))
            tmp.replace(filepath)
            self._last_session_path = filepath
            self._session_loaded = True
            self._refresh_session_display()
            self._save_settings()
        except Exception as exc:
            QMessageBox.critical(self, "Save Session", f"Failed to save session:\n{exc}")

    def _load_session_dialog(self):
        if not self.standalone:
            return
        start = str(self.sessions_dir) if self.sessions_dir.exists() else str(Path.home())
        dialog = QFileDialog(self, "Load Session", start)
        dialog.setFileMode(QFileDialog.FileMode.ExistingFile)
        dialog.setNameFilters(["Session Files (*.json)"])
        dialog.setOption(QFileDialog.Option.DontUseNativeDialog, True)
        if dialog.exec() != QDialog.DialogCode.Accepted:
            return
        files = dialog.selectedFiles()
        if files:
            self._load_session_from_file(files[0])

    def _load_session_from_file(self, filepath: str):
        try:
            session = json.loads(Path(filepath).read_text(encoding="utf-8"))
        except Exception as exc:
            QMessageBox.critical(self, "Load Session", f"Failed to read session:\n{exc}")
            return
        self._stop_recording_if_running(reason="Session loaded")
        self._clear()

        # Theme + recording settings.
        stored_theme = session.get("theme")
        if stored_theme and stored_theme in THEMES:
            self.set_theme(THEMES[stored_theme])
        rec = session.get("recording_settings")
        if isinstance(rec, dict):
            for k in ("visible", "rate_hz", "mode", "limit", "filename",
                      "directory", "overwrite"):
                if k in rec:
                    self.recording_settings[k] = rec[k]
            self._sync_recording_row_from_settings()

        # URI -> (re)connect the worker. Restore the friendly name too so the
        # header shows "Name: <name>" rather than collapsing to the "-"
        # placeholder (the session carries the name; server_names may not).
        uri = str(session.get("uri") or "")
        if uri:
            # Prefer the name saved in the session; fall back to any name we
            # already hold for this URI (covers older sessions saved before the
            # session carried server_name). Only an empty result shows "-".
            sname = str(session.get("server_name") or "").strip() \
                or str(self.server_names.get(uri, "")).strip()
            if sname:
                self.server_names[uri] = sname
            self._set_uri(uri)
            self._refresh_server_name_display()

        for entry in session.get("subscriptions") or []:
            node_id = self._normalize_node_id(str(entry.get("node_id") or ""))
            if not node_id:
                continue
            display = str(entry.get("display_name") or node_id)
            self._node_names[node_id] = display
            self._upsert({"node": node_id, "name": display, "value": "",
                          "type": "", "quality": "", "ts": ""})
            if self.worker:
                self.worker.add_node(node_id)

        self._last_session_path = filepath
        self._session_loaded = True
        self._refresh_session_display()
        # Re-fit the Name field AFTER the Session segment becomes visible: that
        # reflows the header, and the name field must be re-sized in the new
        # layout context or it can collapse to one char on reload. The deferred
        # pass inside _resize handles any residual settle.
        self._refresh_server_name_display()
        self._save_settings()

    def _sync_recording_row_from_settings(self):
        if not hasattr(self, "recording_row"):
            return
        visible = bool(self.recording_settings.get("visible"))
        self.recording_row.setVisible(visible)
        if hasattr(self, "rec_action"):
            self.rec_action.blockSignals(True)
            self.rec_action.setChecked(visible)
            self.rec_action.blockSignals(False)
        pairs = [
            (self.rec_rate_input, "setText", self._fmt_rate(self.recording_settings.get("rate_hz") or 1.0)),
            (self.rec_mode_combo, "setCurrentText", str(self.recording_settings.get("mode") or "Until stopped")),
            (self.rec_limit_input, "setText", str(int(self.recording_settings.get("limit") or 60))),
            (self.rec_filename_edit, "setText", str(self.recording_settings.get("filename") or "")),
            (self.rec_overwrite_check, "setChecked", bool(self.recording_settings.get("overwrite"))),
        ]
        for widget, setter, value in pairs:
            widget.blockSignals(True)
            try:
                getattr(widget, setter)(value)
            except Exception:
                pass
            widget.blockSignals(False)

    # ---- theme / about / prefs / restart ---------------------------------

    def set_theme(self, theme: Theme):
        self.theme = theme
        apply_theme_to_application(theme)
        if hasattr(self, "menu_button"):
            self.menu_button.setStyleSheet(self._menu_button_stylesheet())
        if hasattr(self, "name_edit"):
            self.name_edit.setStyleSheet(self._name_edit_stylesheet())
            # Applying a stylesheet / restyling the app can reset the field's
            # font metrics — re-fit so the Name doesn't collapse to one char.
            self._resize_name_edit_to_content()
        self._save_settings()

    def _show_about(self):
        """About dialog — same shape as UaPlot's: scaling logo on top (if it
        can be located), then a selectable monospaced summary, with screen
        clamping so it can't open off-screen on short displays."""
        class _ScalingLogo(QLabel):
            def __init__(self, source: QPixmap, parent=None):
                super().__init__(parent)
                self._source = source
                self.setAlignment(Qt.AlignmentFlag.AlignCenter)
                self.setMinimumHeight(1)
                policy = QSizePolicy(QSizePolicy.Policy.Expanding,
                                    QSizePolicy.Policy.Preferred)
                policy.setHeightForWidth(True)
                self.setSizePolicy(policy)
                self._rescale()

            def _rescale(self):
                width = max(self.width(), 1)
                if self._source.isNull():
                    return
                target_h = self.heightForWidth(width)
                scaled = self._source.scaled(
                    width, target_h, Qt.AspectRatioMode.KeepAspectRatio,
                    Qt.TransformationMode.SmoothTransformation)
                self.setPixmap(scaled)
                self.setFixedHeight(target_h)

            def resizeEvent(self, event):  # noqa: N802
                self._rescale()
                super().resizeEvent(event)

            def hasHeightForWidth(self):  # noqa: N802
                return True

            def heightForWidth(self, w):  # noqa: N802
                if self._source.isNull() or self._source.width() == 0:
                    return 0
                return int(w * self._source.height() / self._source.width())

        LBL = 25  # same column width as UaPlot/UaExplorer's About
        summary = (
            "ESO IFW — UA Tools\n"
            "\n"
            "UA Tools is the ESO Instrument Framework collection of utilities\n"
            "for working with OPC UA servers used in instrument control. It\n"
            "provides client-side tools for browsing namespaces, exercising\n"
            "methods, recording data, scripting, and live plotting — used both\n"
            "during the development of OPC UA control servers and for\n"
            "diagnostics on deployed instruments. Preparing first release\n"
            "under the UA Tools project name.\n"
            "\n"
            "UA Subscription\n"
            "\n"
            "A PyQt6 live subscription viewer for OPC UA variables. Can be\n"
            "launched on demand by UA Shell (which owns the subscription and\n"
            "pushes value updates over local IPC), or run standalone with a\n"
            "saved session — connecting to the server itself to browse, add /\n"
            "remove subscriptions, and record.\n"
            "\n"
            "Capabilities\n"
            "\n"
            "  Live table\n"
            "    - One row per subscribed node: value, type, quality,\n"
            "      timestamp, and optional Min / Max / Mean / Count stats\n"
            "    - Configurable table font size\n"
            "\n"
            "  Data sources\n"
            "    - Pushed by UA Shell over IPC, OR own subscription / polling\n"
            "      worker when run standalone\n"
            "    - Add / Remove subscriptions via a live address-space browse\n"
            "      (standalone)\n"
            "\n"
            "  Recording\n"
            "    - Snapshot the table to RFC-4180 CSV with a comment header\n"
            "    - Rate / mode / limit / file, same format as UA Plot\n"
            "\n"
            "  Sessions (standalone)\n"
            "    - Save / Load: URI, theme, recording settings, subscriptions\n"
            "    - Session JSON shares UA Explorer / UA Plot shape\n"
            "\n"
            "  Appearance\n"
            "    - Three themes: Light / Dark / Atacama, switchable from\n"
            "      Preferences; persisted per user\n"
            "\n"
            "Storage\n"
            "\n"
            f"{'Working/home directory:':<{LBL}}$HOME/.uatools/UaSubscription\n"
            f"{'Settings file:':<{LBL}}$HOME/.uatools/UaSubscription/settings.json\n"
            f"{'Sessions:':<{LBL}}$HOME/.uatools/UaSubscription/sessions/\n"
            "\n"
            "Runtime\n"
            "\n"
            f"{'Server URI:':<{LBL}}{self.uri or '(none)'}\n"
            f"{'IPC channel:':<{LBL}}{self.channel or '(none)'}\n"
            f"{'Mode:':<{LBL}}{'standalone' if self.standalone else 'driven by launcher'}\n"
            f"{'Active theme:':<{LBL}}{self.theme.name}\n"
            "\n"
            "Keyboard\n"
            "\n"
            f"{'Restart panel:':<{LBL}}Ctrl+Shift+R\n"
            "\n"
            "Version\n"
            "\n"
            f"{'Source file:':<{LBL}}{Path(__file__).resolve()}"
        )

        dlg = QDialog(self)
        dlg.setWindowTitle("About UA Subscription")
        dlg.setSizeGripEnabled(True)
        layout = QVBoxLayout(dlg)

        logo_widget = None
        logo_path = find_file(
            "image/uatools/EsoOpcUaToolsLogo.png",
            parent=Path(__file__).resolve().parent.parent / "resource",
        )
        if logo_path is not None:
            pixmap = QPixmap(str(logo_path))
            if not pixmap.isNull():
                logo_widget = _ScalingLogo(pixmap)
                layout.addWidget(logo_widget)

        text = QTextEdit()
        text.setReadOnly(True)
        mono_font = QFont("Monospace")
        mono_font.setStyleHint(QFont.StyleHint.TypeWriter)
        text.setFont(mono_font)
        text.setPlainText(summary)
        text.setTextInteractionFlags(
            Qt.TextInteractionFlag.TextSelectableByMouse
            | Qt.TextInteractionFlag.TextSelectableByKeyboard)
        text.setMinimumSize(480, 200)
        layout.addWidget(text, 1)

        dlg.resize(560, 760)

        bbox = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok)
        bbox.accepted.connect(dlg.accept)
        layout.addWidget(bbox)

        # Clamp to the available screen before exec so the dialog can't open
        # off-screen on short displays; re-center on the parent (mirrors UaPlot).
        screen = self.screen() if hasattr(self, "screen") else None
        if screen is not None:
            avail = screen.availableGeometry()
            max_h = max(280, avail.height() - 60)
            max_w = max(360, avail.width() - 60)
            if logo_widget is not None and max_h < 760:
                logo_widget.setMaximumHeight(max(60, int(max_h * 0.25)))
            cur = dlg.size()
            clamp_w = min(cur.width(), max_w)
            clamp_h = min(cur.height(), max_h)
            if clamp_w != cur.width() or clamp_h != cur.height():
                dlg.resize(clamp_w, clamp_h)
            parent_geo = self.frameGeometry()
            x = max(avail.left(), min(parent_geo.center().x() - clamp_w // 2,
                                      avail.right() - clamp_w))
            y = max(avail.top(), min(parent_geo.center().y() - clamp_h // 2,
                                     avail.bottom() - clamp_h))
            dlg.move(x, y)

        dlg.exec()

    def _set_table_font_size(self, size: int):
        size = max(6, min(int(size), 32))
        if size == self._table_font_size:
            return
        self._table_font_size = size
        f = QFont()
        f.setPointSize(size)
        self.table.setFont(f)
        self.table.horizontalHeader().setFont(f)
        self._save_settings()

    def _show_preferences(self):
        dlg = QDialog(self)
        dlg.setWindowTitle("Preferences")
        layout = QVBoxLayout(dlg)

        # --- Appearance ---
        appearance = QGroupBox("Appearance")
        appearance_form = QFormLayout()
        theme_combo = QComboBox()
        theme_combo.addItems(list(THEMES.keys()))
        theme_combo.setCurrentText(self.theme.name)
        appearance_form.addRow("Theme:", theme_combo)
        # Table font size lives here too.
        font_spin = QSpinBox()
        font_spin.setRange(6, 32)
        font_spin.setValue(int(self._table_font_size))
        font_spin.setToolTip("Font size (pt) for the subscription table.")
        appearance_form.addRow("Table font size:", font_spin)
        appearance.setLayout(appearance_form)
        layout.addWidget(appearance)

        # --- Recording (mirrors UaPlot's Preferences Recording group) ---
        rec_group = QGroupBox("Recording")
        rec_form = QFormLayout()
        rec_rate_input = QLineEdit(self._fmt_rate(self.recording_settings.get("rate_hz") or 1.0))
        rec_rate_input.setMaxLength(5)
        _rv = QDoubleValidator(0.1, 999.9, 1, rec_rate_input)
        _rv.setNotation(QDoubleValidator.Notation.StandardNotation)
        rec_rate_input.setValidator(_rv)
        rec_form.addRow("Rate (Hz):", rec_rate_input)
        rec_mode_combo = QComboBox()
        rec_mode_combo.addItems(["Until stopped", "Period of time", "Samples"])
        rec_mode_combo.setCurrentText(str(self.recording_settings.get("mode") or "Until stopped"))
        rec_form.addRow("Mode:", rec_mode_combo)
        rec_limit_input = QLineEdit(str(int(self.recording_settings.get("limit") or 60)))
        rec_limit_input.setMaxLength(4)
        rec_limit_input.setValidator(QIntValidator(1, 9999, rec_limit_input))
        rec_form.addRow("Limit:", rec_limit_input)
        rec_file_edit = QLineEdit(str(self.recording_settings.get("filename") or ""))
        rec_file_edit.setPlaceholderText("auto-timestamped")
        rec_form.addRow("File:", rec_file_edit)
        rec_dir_edit = QLineEdit(str(self.recording_settings.get("directory") or ""))
        rec_form.addRow("Directory:", rec_dir_edit)
        rec_overwrite_check = QCheckBox("Overwrite (use exact filename, no timestamp)")
        rec_overwrite_check.setChecked(bool(self.recording_settings.get("overwrite")))
        rec_form.addRow("", rec_overwrite_check)
        rec_visible_check = QCheckBox("Show recording controls in main window")
        rec_visible_check.setChecked(bool(self.recording_settings.get("visible")))
        rec_form.addRow("", rec_visible_check)
        rec_group.setLayout(rec_form)
        layout.addWidget(rec_group)

        # --- Statistics (mirrors UaExplorer's subscription stats) ---
        stats_group = QGroupBox("Statistics")
        stats_form = QFormLayout()
        cur = self.subscription_stats
        stats_enabled_check = QCheckBox("Show statistics columns in the table")
        stats_enabled_check.setChecked(bool(cur.get("enabled")))
        stats_form.addRow("", stats_enabled_check)
        stat_min = QCheckBox("Min"); stat_min.setChecked(bool(cur.get("show_min", True)))
        stat_max = QCheckBox("Max"); stat_max.setChecked(bool(cur.get("show_max", True)))
        stat_mean = QCheckBox("Mean"); stat_mean.setChecked(bool(cur.get("show_mean", True)))
        stat_count = QCheckBox("Count"); stat_count.setChecked(bool(cur.get("show_count", True)))
        per_stat = QHBoxLayout()
        per_stat.setSpacing(12)
        for w in (stat_min, stat_max, stat_mean, stat_count):
            per_stat.addWidget(w)
        per_stat.addStretch(1)
        per_stat_box = QWidget()
        per_stat_box.setLayout(per_stat)
        stats_form.addRow("Show:", per_stat_box)
        stats_window_spin = QSpinBox()
        stats_window_spin.setRange(10, 1000000)
        stats_window_spin.setSingleStep(100)
        stats_window_spin.setValue(int(cur.get("window_size") or 1000))
        stats_window_spin.setToolTip("Sliding window of recent samples kept per row for the stats.")
        stats_form.addRow("Window size (samples):", stats_window_spin)
        stats_decimals_spin = QSpinBox()
        stats_decimals_spin.setRange(0, 12)
        stats_decimals_spin.setValue(int(cur.get("decimals", 3)))
        stats_form.addRow("Decimals:", stats_decimals_spin)
        stats_group.setLayout(stats_form)
        layout.addWidget(stats_group)

        bbox = QDialogButtonBox(
            QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel)
        bbox.accepted.connect(dlg.accept)
        bbox.rejected.connect(dlg.reject)
        layout.addWidget(bbox)

        if dlg.exec() != QDialog.DialogCode.Accepted:
            return

        # Apply theme.
        chosen = THEMES.get(theme_combo.currentText())
        if chosen is not None and chosen.name != self.theme.name:
            self._theme_explicit = True
            self.set_theme(chosen)
        # Apply table font size.
        self._set_table_font_size(int(font_spin.value()))
        # Apply recording settings.
        def _clamp_rate(t):
            try:
                return min(999.9, max(0.1, float(t)))
            except ValueError:
                return float(self.recording_settings.get("rate_hz") or 1.0)
        def _clamp_limit(t):
            try:
                return min(9999, max(1, int(t)))
            except ValueError:
                return int(self.recording_settings.get("limit") or 60)
        self.recording_settings.update({
            "rate_hz": _clamp_rate(rec_rate_input.text()),
            "mode": rec_mode_combo.currentText(),
            "limit": _clamp_limit(rec_limit_input.text()),
            "filename": rec_file_edit.text().strip(),
            "directory": rec_dir_edit.text().strip() or str(Path.home()),
            "overwrite": bool(rec_overwrite_check.isChecked()),
            "visible": bool(rec_visible_check.isChecked()),
        })
        self._sync_recording_row_from_settings()
        # Apply statistics config (resizes buffers + refreshes cells + persists).
        self.set_subscription_stats({
            "enabled": bool(stats_enabled_check.isChecked()),
            "show_min": bool(stat_min.isChecked()),
            "show_max": bool(stat_max.isChecked()),
            "show_mean": bool(stat_mean.isChecked()),
            "show_count": bool(stat_count.isChecked()),
            "window_size": int(stats_window_spin.value()),
            "decimals": int(stats_decimals_spin.value()),
        })
        self._save_settings()

    def restart_application(self):
        self._stop_recording_if_running(reason="UaSubscription restarting")
        try:
            self._save_settings()
        except Exception:
            pass
        if self.worker:
            try:
                self.worker.stop()
                self.worker.wait()
            except Exception:
                pass
        try:
            script = str(self._startup_script)
            args = [sys.executable, script] if script.endswith(".py") else [script]
            # Preserve the original argv tail (channel/uri/theme/...), add --restart.
            tail = [a for a in sys.argv[1:] if a != "--restart"]
            os.execv(args[0], args + tail + ["--restart"])
        except Exception as exc:
            _log(f"restart failed: {exc}")

    def closeEvent(self, event):
        self._stop_recording_if_running(reason="UaSubscription closing")
        self._save_settings()
        if self.worker:
            try:
                self.worker.stop()
                self.worker.wait()
            except Exception:
                pass
        return super().closeEvent(event)


def parse_args():
    parser = argparse.ArgumentParser(
        description="UaSubscription - live OPC UA subscription viewer")
    parser.add_argument("--socket", default="",
                        help="Absolute local-socket PATH for the UaShell push path")
    parser.add_argument("--channel",
                        default=f"uasubscription_{socket.gethostname()}_{os.getpid()}",
                        help="QLocalSocket channel name (push path)")
    parser.add_argument("--uri", default="", help="OPC UA server URI (standalone mode)")
    parser.add_argument("--display-uri", dest="display_uri", default="",
                        help="Show this URI in the header WITHOUT connecting "
                             "(used by UaShell push mode, which owns the connection)")
    parser.add_argument("--server-name", dest="server_name", default="",
                        help="Friendly name for the URI, propagated by the launcher")
    parser.add_argument("--theme", default="Light",
                        help="UI theme (Light, Dark, Atacama)")
    parser.add_argument("--title", default="OPC UA Subscriptions", help="Window title")
    parser.add_argument("--restart", action="store_true", help=argparse.SUPPRESS)
    # Test hook: bind the CommandServer on this socket WITHOUT affecting the
    # standalone/connect determination. Lets integration tests query the table
    # (get_subscriptions) in any mode — including standalone session-load —
    # which a plain --socket can't, since --socket also flips off standalone.
    parser.add_argument("--ipc-socket", dest="ipc_socket", default="",
                        help=argparse.SUPPRESS)
    return parser.parse_args()


def main():
    args = parse_args()
    # Quiet asyncua's client-internal WARNINGs (e.g. "Future for request id N is
    # already done", emitted when a response settles after the request future was
    # already resolved on disconnect). They are harmless lib internals, not user-
    # facing — a viewer should not spew them to the console.
    import logging
    logging.getLogger("asyncua").setLevel(logging.ERROR)
    # Bind a CommandServer (UaShell push path) when a socket PATH is given, OR
    # when the test-only --ipc-socket hook is set.
    channel = args.socket or args.ipc_socket or ""
    # The launcher signals "not standalone" by passing --socket (UaShell push)
    # OR --uri (a parent driving a live connection). --ipc-socket is a pure
    # query hook and does NOT count. Standalone = neither --socket nor --uri.
    standalone = not (bool(args.socket) or bool(args.uri))
    # --display-uri shows a URI in the header WITHOUT connecting (push mode owns
    # the connection). A real --uri both displays AND connects (standalone).
    header_uri = args.uri or args.display_uri
    connect = bool(args.uri)  # only a true --uri starts our own worker

    theme = resolve_theme(args.theme)
    app = QApplication(sys.argv)
    apply_theme_to_application(theme)

    window = SubscriptionWindow(
        uri=header_uri, channel=channel, theme=theme,
        server_name=args.server_name, title=args.title,
        standalone=standalone, connect=connect)

    # Restart auto-load: only standalone, only with no explicit URI.
    if args.restart and standalone and not args.uri and window._last_session_path:
        candidate = Path(window._last_session_path)
        if candidate.is_file():
            try:
                window._load_session_from_file(str(candidate))
            except Exception as exc:
                _log(f"auto-load of last session failed: {exc}")

    window.show()
    sys.exit(app.exec())


if __name__ == "__main__":
    main()
