API Documentation

This API is still in development and may change without notice.

ControllerEngine

Bases: BaseEngine

The main engine class for the CUEMS system.

An object of this class runs all the inner logical part of communications with: - The WebSocket system - The Ossia System - The MTC System - The NodeEngine local and remote instances - The NNG communication system

It is responsible for

• Monitoring the NodeEngine local and remote instances
• Restarting the NodeEngine local and remote instances
• Updating the NodeEngine local and remote instances
• Handling the NodeEngine local and remote instances failures
• Handling the NNG communication system
• Handling the WebSocket system
• Handling the Ossia System
• Handling the MTC master system
• Handling the NodeConf system

Source code in src/cuemsengine/ControllerEngine.py

class ControllerEngine(BaseEngine):
    """
    The main engine class for the CUEMS system.

    An object of this class runs all the inner logical part of communications
    with:
      - The WebSocket system
      - The Ossia System
      - The MTC System
      - The NodeEngine local and remote instances
      - The NNG communication system

    It is responsible for:
      - Monitoring the NodeEngine local and remote instances
      - Restarting the NodeEngine local and remote instances
      - Updating the NodeEngine local and remote instances
      - Handling the NodeEngine local and remote instances failures
      - Handling the NNG communication system
      - Handling the WebSocket system
      - Handling the Ossia System
      - Handling the MTC master system
      - Handling the NodeConf system
    """

    # Controller→UI WebSocket throttle for cue percentage updates.
    # State transitions (0, 1, 100) always bypass this and broadcast
    # immediately.
    # Only in-progress percentage values (2-99) are throttled.
    # Two-tier throttle: Tier 1 is node-side (CUE_STATUS_UPDATE_HZ in
    # loop_cue.py);
    # Tier 2 is here, capping WS broadcasts even when multiple nodes send
    # updates
    # in quick succession.
    CUE_BROADCAST_MIN_INTERVAL = 0.25  # seconds — max 4 Hz to UI per cue

    def __init__(self, **kwargs):
        # Must be set before super().__init__() because BaseEngine sets
        # self.timecode = None which triggers on_timecode_change() via the
        # property setter, and that method reads these attributes.
        # last whole-second value broadcast to UI
        self._last_timecode_second: int = -1
        # Per-cue status dict: maps cue uuid → int status value.
        # Values: 0=unplayed, 1-99=playing (1 until percentage enabled),
        # 100=played, -1=error
        self.cue_status: dict[str, int] = {}
        # Per-cue enabled status: maps cue uuid → bool.
        # Initialised from XML on load_project, updated by show-time toggles.
        # Resets to XML values on reload; persists across stop/go.
        self.cue_enabled_status: dict[str, bool] = {}
        # Per-cue last-broadcast timestamps for WS throttle (Tier 2).
        self._cue_broadcast_timestamps: dict[str, float] = {}
        # Per-mixer-channel volume state: maps
        # "{node_uuid}/{output_index}/{channel}"
        # to float 0.0-1.0. Channel is "master" or a stringified index.
        # Persists
        # across project loads; resets only on engine restart. All access is on
        # the asyncio event loop (WS receive handler + _on_ws_client_connect),
        # so no lock is needed — keep it that way.
        self.mixer_status: dict[str, float] = {}

        # Per-cue names of the loaded script, {uuid: name}. Rebuilt at every
        # load and broadcast AFTER the load status, replayed to late joiners,
        # emptied on unload. The late-join dump reads it before the first load,
        # so it must exist from __init__.
        self.cue_names: dict[str, str] = {}

        # Cluster-state tracking. Populated at load time by
        # _resolve_cluster_state
        # (chunk 3); used to gate the armed=yes flip on all required nodes
        # having
        # reported armed_ready. _pong_responses is populated by the comms
        # thread
        # via status_operation_callback as pong replies arrive. All set
        # mutations
        # and the read in _probe_cluster_liveness happen under _cluster_lock.
        self._adopted_nodes: set[str] = set()
        self._required_nodes: set[str] = set()
        self._armed_nodes: set[str] = set()
        self._pong_responses: set[str] = set()
        self._pong_expected: set[str] = set()
        self._pong_event = threading.Event()
        self._cluster_lock = threading.Lock()
        # What the last load found wrong with the cluster, kept so it can be
        # pushed at load time AND replayed to a browser that connects later
        # (with boot auto-load the UI is normally opened after the load).
        # Initialised here on purpose: _on_ws_client_connect reads it, and a
        # client connecting before the first load of a freshly restarted engine
        # would otherwise raise AttributeError mid-coroutine and take the rest
        # of the late-join dump down with it. _load_id is bumped on every
        # resolve so a UI can tell an operator's retry from a reconnect replay.
        self._load_diagnosis: dict | None = None
        self._load_id: int = 0
        # One-shot watchdog: fires N seconds after _resolve_cluster_state if
        # _armed_nodes still doesn't cover _required_nodes (e.g. a node ponged
        # alive but then died mid-rsync). Logs an error listing the pending
        # nodes. Does NOT force armed=yes — operator decides.
        self._arm_watchdog: threading.Timer | None = None

        # Run counter for cross-node Auto follows (869fc8ytz). Every GO ships
        # it; STOP, load and unload bump it BEFORE they forward anything; a
        # node's follow message is relayed only if it carries the current
        # value. Started from the clock so a restarted controller never
        # reuses a value a node still holds. The lock makes "bump" and
        # "check, then forward a follow" atomic, so a relayed follow is
        # always queued to the nodes ahead of the STOP that ends its run.
        # Never held across a wait or the cluster probe.
        self._run_seq: int = time.time_ns() // 1_000_000
        self._run_seq_lock = threading.Lock()

        super().__init__(**kwargs)
        self.set_editor_request("")
        self.set_node_operation_callback()

    def start(self):
        self.create_timecode()
        self.set_comms()
        # Always re-detect after create_timecode(): the MtcMaster sender port
        # ("MtcMaster:MTCPort") only appears in the ALSA port list AFTER the
        # sender is created.  Connecting the listener directly to that port is
        # the most reliable loopback path; any earlier detection would have
        # picked a wrong/fallback port (e.g. rtpmidid:Announcements).
        Logger.info("Re-detecting MIDI port after MTC sender creation...")
        if self.mtc_listener is not None:
            self.mtc_listener._MtcListener__open_port(None)
            self.mtc_listener.start()
        super().start()

    def set_status(self, property: str, value: str, strict: bool = False) -> None:
        """
        Set status and push to UI via WebSocket when running, armed, or load.
        """
        super().set_status(property, value, strict)
        if property in ("running", "armed", "load", "nextcue"):
            self._broadcast_status(property, value)

    @logged
    def set_comms(self):
        # Start communicators with WebSocket handler on port 9190
        self.set_communicators()

    def set_communicators(self):
        Logger.info("Setting up Communicators")

        # Get NNG hub port from config (must match NodeEngine)
        if hasattr(self, "cm") and self.cm and hasattr(self.cm, "node_conf"):
            nng_hub_port = self.cm.node_conf.get("nng_hub_port", 9093)
            # Use port 9190 for WebSocket OSC - we start BEFORE pyossia to
            # claim this port
            # This allows UI to send commands via Apache's /realtime proxy to
            # ws://127.0.0.1:9190
            websocket_osc_port = self.cm.node_conf.get("oscquery_ws_port", 9190)
            node_id = self.cm.node_conf.get("uuid", "controller")
        else:
            nng_hub_port = 9093
            websocket_osc_port = 9190  # Take port 9190 for WebSocket OSC
            node_id = "controller"

        # LISTENER binds to all interfaces (0.0.0.0) so it does not depend on
        # the
        # avahi link-local address (169.254.x.x) being assigned before startup.
        # NodeEngine (DIALER) still targets the specific controller_url IP.
        nng_hub_address = f"tcp://0.0.0.0:{nng_hub_port}"

        Logger.info(f"NNG Hub address: {nng_hub_address}")

        # WebSocket OSC configuration for receiving commands from UI
        # Uses port 9190 (same as Apache /realtime proxy target) to receive
        # OSC commands directly. Started BEFORE pyossia to claim the port.
        websocket_osc_config = {
            "host": "0.0.0.0",
            "port": websocket_osc_port,
            "node_id": node_id,
        }
        Logger.info(f"WebSocket OSC port: {websocket_osc_port}")

        self.communications_thread = ControllerCommunications(
            nng_hub_address=nng_hub_address,
            editor_callback=self.editor_command_callback,
            node_operation_callback=self.node_operation_callback,
            websocket_osc_config=websocket_osc_config,
        )

        # Register command handlers for WebSocket OSC
        self._register_osc_command_handlers()
        self.communications_thread.set_on_client_connect(self._on_ws_client_connect)

        self.communications_thread.start()

        # Wait for NNG thread to initialize (prevents race condition in
        # nni_random)
        from time import sleep

        max_wait = 5.0  # seconds
        wait_interval = 0.1
        waited = 0.0
        while waited < max_wait:
            if (
                self.communications_thread.is_alive()
                and self.communications_thread.event_loop is not None
            ):
                Logger.info(f"NNG communications thread ready after {waited:.1f}s")
                break
            sleep(wait_interval)
            waited += wait_interval
        else:
            Logger.warning(f"NNG communications thread not ready after {max_wait}s")

    def _register_osc_command_handlers(self):
        """Register OSC command handlers for WebSocket OSC receiving.

        These handlers are called when commands are received from the UI via
        WebSocket OSC. Commands are also forwarded to NodeEngine via NNG.
        """
        # Command handlers - same as used in _command_poll_loop
        self.communications_thread.register_command_handler(
            "/engine/command/go", self.go_script, forward_to_nodes=False
        )
        self.communications_thread.register_command_handler(
            "/engine/command/load", self.deploy_project, forward_to_nodes=False
        )
        self.communications_thread.register_command_handler(
            "/engine/command/stop", self.stop_script, forward_to_nodes=False
        )
        self.communications_thread.register_command_handler(
            "/engine/command/setnextcue",
            self._setnextcue_handler,
            forward_to_nodes=False,
        )
        self.communications_thread.register_command_handler(
            "/engine/command/cue_enabled",
            self._cue_enabled_handler,
            forward_to_nodes=False,
        )

        # Register wildcard handler for player messages (engine format)
        self.communications_thread.register_osc_handler(
            "/engine/players/*", self._handle_player_osc_message
        )

        self._register_node_osc_handlers()

        Logger.info("OSC command handlers registered for WebSocket receiving")

    def _register_node_osc_handlers(self) -> None:
        """Register the direct player OSC route for every adopted node.

        UI sends /{node_uuid}/<type>/... for both controller and worker nodes;
        without per-node registration the WS dispatcher silently drops the
        message and the NNG forward never happens. The set deduplicates so the
        controller's own UUID isn't registered twice (it appears in both
        node_conf and network_map['node_list']).

        Called at startup and again after a nodelist_modify, so a node adopted
        from the UI gets its route without an engine restart. Re-registering is
        safe: register_osc_handler is dict-keyed, so a repeat overwrites.

        ⚠ get_nodes_by_adoption() mutates online/adopted from str to bool IN
        PLACE, so it only survives one pass over a given dict. Call this either
        at startup or immediately after ConfigManager.load_network_map() has
        installed a fresh dict — see _reload_network_map(). A REMOVE leaves the
        old route registered on purpose: traffic from a non-adopted node is
        already filtered by _adopted_nodes at the callbacks.
        """
        node_uuids: set[str] = set()
        own_uuid = (
            self.cm.node_conf.get("uuid", "") if hasattr(self, "cm") and self.cm else ""
        )
        if own_uuid:
            node_uuids.add(own_uuid)
        try:
            if self.cm and self.cm.network_map:
                adopted, _new = NetworkMap.get_nodes_by_adoption(self.cm.network_map)
                for entry in adopted:
                    nuuid = (entry.get("node") or {}).get("uuid")
                    if nuuid:
                        node_uuids.add(nuuid)
        except Exception as e:
            Logger.warning(f"Could not enumerate node UUIDs from network_map: {e}")

        for nuuid in node_uuids:
            self.communications_thread.register_osc_handler(
                f"/{nuuid}/*", self._handle_direct_player_osc_message
            )
            Logger.info(f"Registered direct player OSC handler for /{nuuid}/*")

    def _handle_direct_player_osc_message(self, address: str, args: list):
        """Handle direct player OSC messages from UI (/<node_uuid>/<type>/...).

        These are forwarded directly to the local node's player handlers.
        """
        value = args[0] if args else None

        # Parse: /<node_uuid>/<type>/<...>
        # parts[0] is node_uuid, parts[1] is type (audiomixer, jadeo, etc.)
        parts = address.strip("/").split("/")
        if len(parts) < 2:
            Logger.warning(f"Invalid direct player OSC address: {address}")
            return

        Logger.debug(f"Direct player OSC: {address} = {repr(value)}")

        # Shadow audio mixer volume so /realtime clients can recover state on
        # reconnect or page reload. Address shape:
        #   /{node_uuid}/audio/mixer/{output_index}/{channel}/volume
        # NaN/Inf are dropped to avoid polluting the dict.
        if (
            len(parts) >= 6
            and parts[1] == "audio"
            and parts[2] == "mixer"
            and parts[-1] == "volume"
            and value is not None
        ):
            try:
                vol = float(value)
            except (TypeError, ValueError):
                vol = None
            if vol is not None and math.isfinite(vol):
                node_uuid_part, output_index, channel = (
                    parts[0],
                    parts[3],
                    parts[4],
                )
                key = f"{node_uuid_part}/{output_index}/{channel}"
                self.mixer_status[key] = vol
                self._broadcast_status(
                    f"audio/mixer/{node_uuid_part}/{output_index}/{channel}/volume",
                    vol,
                )

        # Forward to NodeEngine via NNG as player_control
        operation = NodeOperation(
            type=OperationType.COMMAND,
            action=ActionType.UPDATE,
            sender=(
                self.cm.node_conf.get("uuid", "controller")
                if hasattr(self, "cm") and self.cm
                else "controller"
            ),
            target="player_control",
            data={"address": address, "value": value},
        )

        try:
            import asyncio

            asyncio.run_coroutine_threadsafe(
                self.communications_thread.nng_hub.send_operation(operation),
                self.communications_thread.event_loop,
            )
            Logger.debug(
                f"Forwarded direct player OSC to nodes: {address} = {repr(value)}"
            )
        except Exception as e:
            Logger.error(f"Error forwarding direct player OSC to nodes: {e}")

    def _handle_player_osc_message(self, address: str, args: list):
        """Handle player-related OSC messages from UI.

        These are forwarded to NodeEngine via NNG for player control
        (video, audio mixer, DMX, etc.)
        """
        # Forward to NodeEngine via NNG
        value = args[0] if args else None

        # Create a COMMAND operation for player control
        operation = NodeOperation(
            type=OperationType.COMMAND,
            action=ActionType.UPDATE,
            sender=(
                self.cm.node_conf.get("uuid", "controller")
                if hasattr(self, "cm") and self.cm
                else "controller"
            ),
            target="player_control",
            data={"address": address, "value": value},
        )

        try:
            import asyncio

            asyncio.run_coroutine_threadsafe(
                self.communications_thread.nng_hub.send_operation(operation),
                self.communications_thread.event_loop,
            )
            Logger.debug(f"Forwarded player OSC to nodes: {address} = {repr(value)}")
        except Exception as e:
            Logger.error(f"Error forwarding player OSC to nodes: {e}")

    def _forward_load_to_nodes(self, project_name: str) -> None:
        """Forward a load command to NodeEngine via NNG."""
        self._forward_command_to_nodes("/engine/command/load", project_name)

    def stop(self):
        self.stop_comms()
        super().stop()

    @logged
    def stop_comms(self):
        if self.with_mtc:
            self.stop_timecode()
        if hasattr(self, "communications_thread"):
            self.communications_thread.stop()

    #########################
    # Timecode
    #########################
    def create_timecode(self):
        if self.with_mtc:
            self.mtcmaster = libmtcmaster.MTCSender_create()
        else:
            Logger.info("Midi TimeCode requires with_mtc to be True.")

    def start_timecode(self):
        if self.with_mtc:
            libmtcmaster.MTCSender_play(self.mtcmaster)
            Logger.info("Midi TimeCode started.")
        else:
            Logger.info("Midi TimeCode requires with_mtc to be True.")

    def stop_timecode(self):
        if self.with_mtc:
            libmtcmaster.MTCSender_stop(self.mtcmaster)
            Logger.info("Midi TimeCode stopped.")
        else:
            Logger.info("Midi TimeCode requires with_mtc to be True.")

    #########################
    # Operation callbacks
    #########################
    def set_node_operation_callback(self):
        self.node_operation_callback = {
            OperationType.PLAYER: self.player_operation_callback,
            OperationType.CUE: self.cue_operation_callback,
            OperationType.STATUS: self.status_operation_callback,
        }

    def player_operation_callback(self, operation: NodeOperation):
        """
        Callback invoked when players are received from nodes.

        Parameters:
        - operation: NodeOperation with sender, target (player_id), and action
        """
        Logger.info(f"Player operation received: {operation}")

    def cue_operation_callback(self, operation: NodeOperation):
        """Callback invoked when cues are received from nodes.

        Handles three action types:
        - ADD: cue started playing on a node → status 1, broadcast immediately
        - REMOVE: cue finished playing on a node → status 100, broadcast
          - immediately
        - UPDATE: percentage progress from a node (future) → throttled
          - broadcast
        """
        Logger.info(f"Cue operation received: {operation}")
        cue_id = operation.data.get("id") if operation.data else None

        # Drop operations for cues not belonging to the current project.
        # This prevents stale REMOVE/ADD notifications from the NodeEngine
        # (sent when it disarms the previous project) from being broadcast
        # to the UI as unknown UUIDs.
        if cue_id and cue_id not in self.cue_status:
            Logger.debug(
                f"Ignoring cue operation for unknown/stale cue_id {cue_id}"
                f"(action={operation.action})"
            )
            return

        # No cue starts while the show is not running: go_script sets running
        # before it forwards the GO. An ADD here is a node reporting a start it
        # dispatched just before a STOP reached it (a stop_project cue on a
        # multi-node cluster, 869fhjdq5); without this gate it would show the
        # cue at 1 after the reset and its disarm's REMOVE would then make it
        # 100, with a spurious cue_started/cue_ended for the Companion relay.
        if (
            operation.action in (ActionType.ADD, ActionType.UPDATE)
            and self.get_status("running") != "yes"
        ):
            Logger.debug(
                f"Ignoring cue {operation.action} for {cue_id}: not running "
                "(a late start report after a STOP)"
            )
            return

        if operation.action == ActionType.ADD:
            # Cue started playing: mark as playing (1) and broadcast
            # immediately.
            if cue_id:
                self.cue_status[cue_id] = 1
                self._broadcast_cue_status(cue_id, 1, force=True)
            try:
                self.status.currentcue = [
                    operation.data["id"],
                    operation.data["offset"],
                ]
                Logger.debug(f"Current cue updated: {self.status.currentcue}")
            except Exception as e:
                Logger.error(f"Error updating currentcue: {e}")

        elif operation.action == ActionType.REMOVE:
            # Cue finished playing: mark as played (100) and broadcast
            # immediately.
            # Only transition to 100 if the cue was actually playing (status ==
            # 1).
            # REMOVEs that arrive while status is 0 (e.g. NodeEngine disarming
            # the
            # previous project after a reload) are stale and must be silently
            # dropped.
            if cue_id:
                if self.cue_status.get(cue_id) == 1:
                    self.cue_status[cue_id] = 100
                    self._broadcast_cue_status(cue_id, 100, force=True)
                else:
                    Logger.debug(
                        f"Ignoring stale REMOVE for cue {cue_id}"
                        f"(status={self.cue_status.get(cue_id)}, expected 1)"
                    )
            self.status.remove_currentcue(operation.data["id"])
            Logger.debug(f"Cue removed from currentcue: {operation.data['id']}")

        elif operation.action == ActionType.UPDATE:
            # Future: percentage progress updates from loop_cue() during
            # playback.
            # Throttled by _broadcast_cue_status (Tier 2 / controller-side).
            # The node-side Tier 1 throttle (CUE_STATUS_UPDATE_HZ) limits NNG
            # traffic.
            if cue_id:
                pct = operation.data.get("percentage", 1)
                self.cue_status[cue_id] = pct
                self._broadcast_cue_status(cue_id, pct)  # throttled
            Logger.debug(
                f"Cue percentage update: {cue_id} ="
                f"{operation.data.get('percentage')}"
            )

        else:
            Logger.warning(f"Unknown cue action: {operation.action}")

    def status_operation_callback(self, operation: NodeOperation):
        """Callback invoked when status updates are received from nodes.

        Handles stop_project, armed_ready, nextcue, cue_enabled, follow and
        pong notifications.
        """
        Logger.info(f"Status operation received: {operation}")
        if operation.target == "pong":
            # Cluster liveness probe reply. Runs on the AsyncCommsThread event
            # loop; _probe_cluster_liveness on the main thread waits on the
            # event.
            with self._cluster_lock:
                self._pong_responses.add(operation.sender)
                if self._pong_responses >= self._pong_expected:
                    self._pong_event.set()
            Logger.debug(f"Pong from {operation.sender}")
            return
        if operation.target == "stop_project":
            # The authored end of the show (869fhjdq5): a node revealed a
            # stop_project ActionCue. First valid message wins; see the handler.
            self._handle_stop_project(operation)
        elif operation.target == "armed_ready":
            if operation.data and operation.data.get("armed") == "yes":
                # Aggregate per-node: only flip armed=yes when all required
                # nodes have reported. Filter foreign senders to keep the
                # tracker bounded.
                sender = operation.sender
                if sender not in self._adopted_nodes:
                    Logger.debug(f"Ignoring armed_ready from non-adopted node {sender}")
                    return
                with self._cluster_lock:
                    self._armed_nodes.add(sender)
                    armed_now = set(self._armed_nodes)
                    required = set(self._required_nodes)
                Logger.info(f"Node {sender} armed ({len(armed_now)}/{len(required)})")
                if armed_now >= required and self.get_status("armed") != "yes":
                    if self.go_offset is None:
                        Logger.info(
                            "Re-arm after stop complete — restarting timecode"
                            "and enabling GO"
                        )
                        self.start_timecode()
                        self.go_offset = 0
                    else:
                        Logger.info("All required nodes armed — enabling GO")
                    self.set_status("armed", "yes")
                    self._cancel_arm_watchdog()
        elif operation.target == "nextcue":
            nextcue_id = operation.data.get("nextcue", "") if operation.data else ""
            self.set_status("nextcue", nextcue_id)
            Logger.info(f'Next cue updated: {nextcue_id or "(none)"}')
        elif operation.target == "cue_enabled":
            cue_id = operation.data.get("cue_id") if operation.data else None
            enabled = operation.data.get("enabled", True) if operation.data else True
            if cue_id and cue_id in self.cue_enabled_status:
                self.cue_enabled_status[cue_id] = enabled
                self._broadcast_cue_enabled(cue_id, enabled)
                Logger.info(f"Cue {cue_id} enabled status updated from node: {enabled}")
        elif operation.target == "audiomixer_status":
            # Authoritative mixer gain snapshot from a node (sent on mixer
            # startup / after reset_volumes). This is the ONLY node-sourced feed
            # of mixer_status; it lets the UI recover the TRUE gain rather than a
            # stale UI-write echo. jack-volume itself is write-only, so the node
            # reports what it commanded. Key shape matches the UI-write shadow:
            # "{node_uuid}/{output_index}/{channel}".
            node_uuid = operation.sender
            # Filter foreign/unadopted senders (mirrors armed_ready /
            # stop_project) so the shadow can't grow from nodes the cluster
            # never adopted. A pre-load startup snapshot from a not-yet-adopted
            # node is dropped, but the UI's unity default matches the post-reset
            # state, and the load-time reset snapshot (node adopted by then)
            # reconciles.
            if node_uuid not in self._adopted_nodes:
                Logger.debug(f"Ignoring audiomixer_status from non-adopted {node_uuid}")
                return
            data = operation.data or {}
            output_index = data.get("output_index", "0")
            entries = data.get("entries", {})
            if not isinstance(entries, dict):
                Logger.debug(f"audiomixer_status from {node_uuid} had non-dict entries")
                return
            updated = 0
            for channel, vol in entries.items():
                try:
                    v = float(vol)
                except (TypeError, ValueError):
                    continue
                if not math.isfinite(v):
                    continue
                key = f"{node_uuid}/{output_index}/{channel}"
                self.mixer_status[key] = v
                self._broadcast_status(
                    f"audio/mixer/{node_uuid}/{output_index}/{channel}/volume", v
                )
                updated += 1
            Logger.info(f"Mixer status from {node_uuid}: {updated} channel(s) updated")
        elif operation.target == "follow":
            self._relay_follow(operation)
        else:
            Logger.debug(f"Unknown status target: {operation.target}")

    def _bump_run_seq(self, why: str) -> None:
        """A run ends here (STOP, load, unload): follow messages sent for it
        are no longer relayed. Call BEFORE forwarding the command."""
        with self._run_seq_lock:
            self._run_seq += 1
            seq = self._run_seq
        Logger.debug(f"run_seq -> {seq} ({why})")

    def _relay_follow(self, operation: NodeOperation) -> None:
        """Relay a cross-node Auto follow message (announce or cancel) to
        every node (869fc8ytz). The controller keeps no follow state: the
        node that owns the source decides, the others act. Dropped when the
        sender is not adopted, a cue is not in the loaded project, the show
        is not running, or the message belongs to an earlier run."""
        data = operation.data if isinstance(operation.data, dict) else None
        if data is None:
            Logger.warning(f"Follow message without data from {operation.sender}")
            return
        what = (
            f"follow {data.get('op')} {data.get('source')} -> {data.get('target')}"
            f" at {data.get('seed_ms')} from {operation.sender}"
        )
        if operation.sender not in self._adopted_nodes:
            Logger.warning(f"Dropping {what}: sender is not an adopted node")
            return
        if (
            data.get("source") not in self.cue_status
            or data.get("target") not in self.cue_status
        ):
            Logger.warning(f"Dropping {what}: cue not in the loaded project")
            return
        with self._run_seq_lock:
            if (
                self.get_status("running") != "yes"
                or data.get("run_seq") != self._run_seq
            ):
                Logger.info(
                    f"Dropping {what}: not part of the current run "
                    f"(run_seq {data.get('run_seq')}, current {self._run_seq}, "
                    f"running={self.get_status('running')})"
                )
                return
            self._forward_command_to_nodes("/engine/command/follow", data)
        Logger.info(f"Relayed {what}")

    def _handle_stop_project(self, operation: NodeOperation) -> None:
        """A node revealed a stop_project ActionCue, the authored end of the
        show (ClickUp 869fhjdq5). Run the operator STOP for it.

        Validated like a follow message (_relay_follow): adopted sender, cue in
        the loaded project and of the right type, show running, current
        run_seq. ActionCues are local on every node, so every node sends one
        of these; the first valid one runs stop_script, which bumps run_seq,
        and the others are then dropped at INFO as expected duplicates.

        The lock is released BEFORE stop_script: it takes _run_seq_lock itself
        in _bump_run_seq. The check-then-call window is closed because this
        callback and the WebSocket STOP handler run on the same event loop,
        and stop_script's own `running != yes` guard covers either order.
        """
        data = operation.data if isinstance(operation.data, dict) else None
        if data is None:
            Logger.warning(f"stop_project message without data from {operation.sender}")
            return
        cue_id = str(data.get("cue", ""))
        sender = operation.sender
        what = (
            f"stop_project from {sender} (cue {cue_id}, run_seq {data.get('run_seq')})"
        )
        if sender not in self._adopted_nodes:
            Logger.warning(f"Dropping {what}: sender is not an adopted node")
            return
        if cue_id not in self.cue_status:
            Logger.warning(f"Dropping {what}: cue not in the loaded project")
            return
        cue = self.script.find(cue_id) if self.script is not None else None
        if not (
            isinstance(cue, ActionCue)
            and getattr(cue, "action_type", None) == "stop_project"
        ):
            Logger.warning(f"Dropping {what}: the cue is not a stop_project ActionCue")
            return
        with self._run_seq_lock:
            if (
                self.get_status("running") != "yes"
                or data.get("run_seq") != self._run_seq
            ):
                Logger.info(
                    f"Dropping {what}: not part of the current run "
                    f"(run_seq {data.get('run_seq')}, current {self._run_seq}, "
                    f"running={self.get_status('running')})"
                )
                return
        name = self.cue_names.get(cue_id, "")
        Logger.info(
            f'stop_project cue {cue_id} ("{name}") from node {sender}: '
            "stopping the show"
        )
        self.stop_script(None)

    #########################
    # Editor commands
    #########################

    def editor_command_callback(self, item: dict, context):
        Logger.debug(f"Received editor command: {item}, with context: {context}")
        _item_keys = item.keys()
        if "value" not in _item_keys:
            item["value"] = ""
        if "action_uuid" not in _item_keys:
            self.error_to_editor(context, "No action uuid submitted")
        self.set_editor_request(item["action_uuid"])

        if "type" in _item_keys:
            if item["type"] not in ["error", "initial_settings"]:

                self.set_editor_request("")
            self.error_to_editor(context, "Response not recognized")

        try:
            self.handle_editor_command(
                action=item["action"],
                value=item["value"],
                context=context,
                modify_action=item.get("modify_action"),
            )
        except Exception as e:
            Logger.error(f"{type(e)} handling editor command: {e}")

            request_uuid = self.get_editor_request()
            self.set_editor_request("")
            self.error_to_editor(
                context,
                value=f"Command {type(e)}: {e}",
                request_uuid=request_uuid,
            )

    def handle_editor_command(self, action, value, context=None, modify_action=None):
        # nodelist_modify cannot ride the dispatch table below: every entry is
        # invoked as command_dict[action](value, context) — a two-argument
        # contract with nowhere to put modify_action. Registering it there sends
        # modify_action=None to nodeconf, which answers
        # {'OK': False, 'error': 'Invalid modify_action: None'} to EVERY click.
        # So bind the message here and dispatch it directly.
        if action == "nodelist_modify":
            message = {
                "action": action,
                "value": value,
                "modify_action": modify_action,
            }
            if self.nodelist_modify(message, context):
                self.confirm_to_editor(context, type=action, value="OK")
                self.set_editor_request("")
            return

        command_dict = {
            "project_deploy": partial(self.load_project, deploy_only=True),
            "project_ready": self.load_project,
            "hw_discovery": self.hwdiscovery,
            "nodeconf": self.nodeconf,
            "go_script": self.go_script,
            "project_status": self.get_project_status,
            "project_unload": self.unload_project,
            "cluster_status": self.get_cluster_status,
        }
        if action in command_dict.keys():
            result = command_dict[action](value, context)
            if result:
                reply_value = result if isinstance(result, dict) else "OK"
                self.confirm_to_editor(context, type=action, value=reply_value)
                # Clear the editor request after successful confirmation
                self.set_editor_request("")

        else:
            raise ValueError(f"Command {action} not recognized")

    def confirm_to_editor(self, context, type=None, value=None):
        return_message = {
            "type": type,
            "value": value,
            "action_uuid": self.get_editor_request(),
        }
        Logger.debug(f"Sending confirm to editor: {return_message}")

        try:
            self.communications_thread.reply_to_editor(return_message, context)
        except Exception as e:
            Logger.error(f"{type(e)} confirming to editor: {e}")

    def error_to_editor(self, context, value=None, request_uuid=None, action=None):
        if not request_uuid:
            request_uuid = self.get_editor_request()
        return_message = {
            "type": "error",
            "value": value,
            "action_uuid": request_uuid,
        }
        if action:
            return_message["action"] = action
        Logger.debug(f"Sending error to editor: {return_message}")
        try:
            self.communications_thread.reply_to_editor(return_message, context)
        except Exception as e:
            Logger.error(f"{type(e)} sending error to editor: {e}")

    def set_editor_request(self, value):
        self._editor_request_uuid = value

    def get_editor_request(self):
        return self._editor_request_uuid

    #########################
    # External services
    #########################

    def hwdiscovery(self, message: dict, context=None) -> bool:
        Logger.debug(f"sending HW discovery request: {message}")
        try:
            reply = self.communications_thread.request_to_hwdiscovery(message)
            Logger.debug(f"Received HW discovery reply: {reply}")
            if "OK" in reply.values():
                return True
            else:
                return False
        except Exception as e:
            Logger.error(f"{type(e)} sending HW discovery request: {e}")
            return False

    def nodeconf(self, message: dict, context=None) -> bool:
        Logger.debug(f"sending nodeconf request: {message}")
        try:
            reply = self.communications_thread.request_to_nodeconf(message)
            Logger.debug(f"Received nodeconf reply: {reply}")
            if "OK" in reply.values():
                return True
            else:
                return False
        except Exception as e:
            Logger.error(f"{type(e)} sending nodeconf request: {e}")
            return False

    def _nodelist_refuse(self, context, msg: str) -> bool:
        """Refuse a nodelist_modify with one clean sentence for the operator.

        Every failure path in nodelist_modify goes through here and returns
        False. It must NOT raise: editor_command_callback's except clause also
        replies (prefixing "Command <class 'RuntimeError'>: ..."), and the
        editor wraps that again as "Engine reports error: ...". Raising after
        replying would also put two replies on a single NNG Rep context.

        Clears the pending editor request the same way the generic error path
        does (see editor_command_callback): capture the uuid, clear it, then
        reply with it explicitly, so no stale uuid is left behind for whatever
        reads get_editor_request() next.
        """
        Logger.error(f"nodelist_modify refused: {msg}")
        request_uuid = self.get_editor_request()
        self.set_editor_request("")
        self.error_to_editor(
            context, value=msg, request_uuid=request_uuid, action="nodelist_modify"
        )
        return False

    def nodelist_modify(self, message: dict, context=None) -> bool:
        """Adopt or un-adopt a node, then refresh our cached topology.

        Forwards {'action','value','modify_action'} to cuems-nodeconf over
        /tmp/nodeconf.ipc, which flips <adopted> in network_map.xml. On success
        we re-read that file, because our own copy was loaded once at startup
        (ConfigManager(load_all=True)) and would otherwise keep the GO gate
        blind to the change until an engine restart.

        Returns True only when nodeconf confirmed AND our reload succeeded; the
        caller confirms "OK" to the editor on True. Every False path has
        already sent an explanatory error.
        """
        node_uuid = message.get("value")
        modify_action = message.get("modify_action")
        Logger.info(f"nodelist_modify: {modify_action} node {node_uuid}")

        # Adoption is a setup-time operation. Refuse (never stop anything) when
        # a project is running, and also when one is merely loaded: the GO
        # gate's _required_nodes snapshot is computed once, in load_project, so
        # a REMOVE accepted now would leave it waiting on a node the operator
        # just removed until the next load.
        if self.get_status("running") == "yes":
            return self._nodelist_refuse(
                context,
                "Cannot modify the node list while a project is running. "
                "Stop playback first.",
            )
        if self.get_status("load"):
            return self._nodelist_refuse(
                context,
                "Cannot modify the node list while a project is loaded. "
                "Unload the project first.",
            )

        if not node_uuid or not isinstance(node_uuid, str):
            return self._nodelist_refuse(
                context, f"nodelist_modify needs a node uuid, got {node_uuid!r}"
            )
        if modify_action not in ("ADD", "REMOVE"):
            return self._nodelist_refuse(
                context,
                f"Invalid modify_action: {modify_action!r}. Must be 'ADD' or 'REMOVE'",
            )

        # Fail fast when the daemon is not there at all. Measured on the rig:
        # with the socket absent, request_to_nodeconf does NOT return quickly —
        # it burns the full 15 s IPC timeout and then raises. Editor commands
        # are serialized by the engine's single listener, so a second click
        # queues behind the first and can blow the editor's own 25 s timeout,
        # turning a clear refusal into "Engine did not respond". nodeconf ships
        # disabled on most of the fleet, so this is the common path, not a rare
        # one.
        if not os.path.exists(NODECONF_IPC_PATH):
            return self._nodelist_refuse(
                context,
                "The node configuration service (cuems-nodeconf) is not "
                "running on this controller, so the node list cannot be "
                "changed. Enable it with: systemctl enable --now cuems-nodeconf",
            )

        try:
            reply = self.communications_thread.request_to_nodeconf(
                message, timeout=NODECONF_TIMEOUT_S
            )
        except Exception as e:
            return self._nodelist_refuse(
                context, f"Could not reach cuems-nodeconf: {type(e).__name__}: {e}"
            )

        # Communicator.send_request swallows every exception and returns None —
        # an absent socket (nodeconf disabled, which is the fleet default) and a
        # peer that never answers both land here.
        if reply is None:
            return self._nodelist_refuse(
                context,
                "The node configuration service (cuems-nodeconf) is not "
                "responding. Is it enabled on this controller?",
            )
        if not isinstance(reply, dict):
            return self._nodelist_refuse(
                context, f"Unexpected reply from cuems-nodeconf: {reply!r}"
            )
        if not reply.get("OK", False):
            return self._nodelist_refuse(
                context, reply.get("error", "cuems-nodeconf reported an error")
            )

        Logger.info(
            f"nodelist_modify: cuems-nodeconf confirmed {modify_action} "
            f"for node {node_uuid}"
        )

        # nodeconf has already rewritten network_map.xml. If we cannot re-read
        # it the operation is a PARTIAL success: the map (and so the editor and
        # the UI) is correct, only this process is behind. Say exactly that
        # instead of confirming OK over a stale topology.
        try:
            self._reload_network_map()
        except Exception as e:
            Logger.error(f"{type(e).__name__} reloading network_map: {e}")
            return self._nodelist_refuse(
                context,
                f"Node {modify_action.lower()}ed, but the engine could not "
                f"reload the topology ({type(e).__name__}: {e}) — restart "
                f"cuems-controller-engine before loading a project.",
            )
        return True

    def _reload_network_map(self) -> None:
        """Re-read network_map.xml and re-register per-node OSC routes.

        Targeted on purpose: ConfigManager.load_network_map() reassigns
        self.cm.network_map to a NEW dict, so a concurrent reader sees either
        the old one or the new one, never a torn one. A full
        ConfigManager(load_all=True) would also re-read settings and mappings —
        that is the "restart both daemons" trap, not what we want here.

        ORDER MATTERS: _register_node_osc_handlers() calls
        NetworkMap.get_nodes_by_adoption(), which mutates online/adopted from
        str to bool IN PLACE and therefore survives only one pass over a given
        dict (same hazard _adopted_uuids_from_network_map documents). It is safe
        only on the freshly reloaded dict — never call it as a standalone
        "shortcut" refresh.
        """
        self.cm.load_network_map()
        Logger.info("network_map reloaded after a node list change")
        self._register_node_osc_handlers()

    #########################
    # Status Updates (stub - OSCQuery removed)
    #########################

    def set_oscquery_values(self, values: dict):
        """Stub for OSCQuery value setting - OSCQuery server has been removed.

        Status updates are now handled via internal state tracking.
        TODO: Implement WebSocket status push if UI needs real-time status.
        """
        for key, value in values.items():
            Logger.debug(f"Status update (no-op): {key} = {repr(value)}")

    def _collect_project_nodes(self, cuelist) -> set[str]:
        """Walk a cuelist and return the set of node UUIDs referenced by any
        cue's output_name.

        output_name format depends on cue type:
          - video / audio: "<36-char UUID>_<index>" (e.g.
            - "07131798-...-...d18f_0")
          - DMX:           "<UUID>" (no suffix)
        In both cases, output_name[:36] is the node UUID. We validate with a
        cheap UUID-shape check (hyphens at positions 8/13/18/23) so garbage
        output names don't leak non-UUID strings into the set.
        """
        from cuemsutils.cues import CueList

        nodes: set[str] = set()
        if not (hasattr(cuelist, "contents") and cuelist.contents):
            return nodes
        for item in cuelist.contents:
            if item is None:
                continue
            outputs = getattr(item, "outputs", None) or []
            for out in outputs:
                if not isinstance(out, dict):
                    continue
                name = out.get("output_name") or ""
                if len(name) >= 36:
                    head = name[:36]
                    if (
                        head[8] == "-"
                        and head[13] == "-"
                        and head[18] == "-"
                        and head[23] == "-"
                    ):
                        nodes.add(head)
            if isinstance(item, CueList):
                nodes.update(self._collect_project_nodes(item))
        return nodes

    def _collect_cue_ids(self, cuelist) -> list[str]:
        """
        Recursively collect all cue IDs from a cuelist (including nested
        CueLists).
        """
        from cuemsutils.cues import CueList

        ids = []
        if hasattr(cuelist, "contents") and cuelist.contents:
            for item in cuelist.contents:
                if item is None:
                    continue
                ids.append(item.id)
                if isinstance(item, CueList):
                    ids.extend(self._collect_cue_ids(item))
        return ids

    def _collect_cue_names(self, cuelist) -> dict[str, str]:
        """Recursively collect cue names from a cuelist, same walk as
        _collect_cue_ids (nested CueLists included, as items and recursed).

        Reads the name as a dict item, not through the Cue.name property: the
        property raises KeyError on a cue that has no name key. None becomes
        "" and anything else is str()-ed as is (never `or ""`, so a name an
        old cuems-utils coerced to 0/False still prints).
        """
        from cuemsutils.cues import CueList

        names = {}
        if hasattr(cuelist, "contents") and cuelist.contents:
            for item in cuelist.contents:
                if item is None:
                    continue
                name = item.get("name")
                names[item.id] = "" if name is None else str(name)
                if isinstance(item, CueList):
                    names.update(self._collect_cue_names(item))
        return names

    def _collect_cue_enabled(self, cuelist) -> dict[str, bool]:
        """Recursively collect cue enabled states from a cuelist."""
        from cuemsutils.cues import CueList

        result = {}
        if hasattr(cuelist, "contents") and cuelist.contents:
            for item in cuelist.contents:
                if item is None:
                    continue
                result[item.id] = item.enabled
                if isinstance(item, CueList):
                    result.update(self._collect_cue_enabled(item))
        return result

    def _broadcast_cue_name(self, cue_id: str, name: str) -> None:
        """
        Broadcast a cue's name to UI at /engine/status/cue_name/{uuid}.
        Sent once per load (no throttle), like cue_enabled.
        """
        if (
            hasattr(self, "communications_thread")
            and self.communications_thread
            and hasattr(self.communications_thread, "broadcast_osc")
        ):
            self.communications_thread.broadcast_osc(
                f"/engine/status/cue_name/{cue_id}", name
            )

    def _broadcast_cue_enabled(self, cue_id: str, enabled: bool) -> None:
        """
        Broadcast per-cue enabled status to UI at
        /engine/status/cue_enabled/{uuid}.
        """
        if (
            hasattr(self, "communications_thread")
            and self.communications_thread
            and hasattr(self.communications_thread, "broadcast_osc")
        ):
            self.communications_thread.broadcast_osc(
                f"/engine/status/cue_enabled/{cue_id}", 1 if enabled else 0
            )

    def _broadcast_cue_status(
        self, cue_id: str, value: int, force: bool = False
    ) -> None:
        """
        Broadcast per-cue status to UI via WebSocket OSC at
        /engine/status/cue/{uuid}.

        Values: 0=unplayed, 1-99=playing (1 until percentage is enabled),
        100=played, -1=error.

        State transitions (force=True: values 0, 1, 100) bypass throttle and
        broadcast
        immediately. In-progress percentage updates (2-99) are throttled
        per-cue to
        CUE_BROADCAST_MIN_INTERVAL to limit WS traffic even when multiple
        remote nodes
        send updates in quick succession (Tier 2 of the two-tier throttle
        strategy).
        """
        if not force:
            now = time.monotonic()
            last = self._cue_broadcast_timestamps.get(cue_id, 0)
            if now - last < self.CUE_BROADCAST_MIN_INTERVAL:
                return
            self._cue_broadcast_timestamps[cue_id] = now
        if (
            hasattr(self, "communications_thread")
            and self.communications_thread
            and hasattr(self.communications_thread, "broadcast_osc")
        ):
            self.communications_thread.broadcast_osc(
                f"/engine/status/cue/{cue_id}", value
            )

    def _broadcast_status(self, key: str, value) -> None:
        """Push status to UI via WebSocket OSC (realtime)."""
        if (
            hasattr(self, "communications_thread")
            and self.communications_thread
            and hasattr(self.communications_thread, "broadcast_osc")
        ):
            self.communications_thread.broadcast_osc(f"/engine/status/{key}", value)

    @staticmethod
    def _cluster_warning_payload(diagnosis: dict | None) -> str:
        """Serialize a load diagnosis for the OSC status channel.

        OSC cannot carry lists, so this travels as a JSON string — the same
        thing the codebase already does elsewhere on this channel. A cleared
        diagnosis serializes to empty lists rather than being withheld: a clean
        load after a bad one has to actively erase the previous warning, or it
        stays on the operator's screen forever.
        """
        d = diagnosis or {}
        return json.dumps(
            {
                "load_id": d.get("load_id", 0),
                "project": d.get("project", ""),
                "missing": d.get("missing", []),
                "unreachable": d.get("unreachable", []),
            }
        )

    def _broadcast_cluster_warning(self, diagnosis: dict | None) -> None:
        """Push the load diagnosis to every connected UI. Always sent."""
        self._broadcast_status(
            "cluster_warning", self._cluster_warning_payload(diagnosis)
        )

    def _clear_load_diagnosis(self) -> None:
        """Forget the last load's diagnosis and erase it from every UI.

        Called wherever a load ENDS — a new load starting, a load failing
        early, an unload — but deliberately NOT from _clear_playback_state(),
        which also runs on STOP: a stopped project is still loaded, and its
        missing nodes are still missing.
        """
        with self._cluster_lock:
            self._load_diagnosis = None
        self._broadcast_cluster_warning(None)

    async def _on_ws_client_connect(self, websocket) -> None:
        """Send full state dump to a newly connected WebSocket client."""
        from .osc.WebSocketOscHandler import build_osc_message

        # Engine status
        for key in ("running", "armed", "load", "nextcue"):
            val = self.get_status(key)
            if val is not None:
                data = build_osc_message(f"/engine/status/{key}", val)
                if data:
                    await websocket.send(data)

        # Last load's cluster diagnosis. Replayed here because with boot
        # auto-load the browser is normally opened well after the load, so the
        # push in _resolve_cluster_state has nobody listening.
        with self._cluster_lock:
            diagnosis = dict(self._load_diagnosis) if self._load_diagnosis else None
        data = build_osc_message(
            "/engine/status/cluster_warning",
            self._cluster_warning_payload(diagnosis),
        )
        if data:
            await websocket.send(data)

        # Per-cue playback status
        for cid, status in self.cue_status.items():
            data = build_osc_message(f"/engine/status/cue/{cid}", status)
            if data:
                await websocket.send(data)

        # Per-cue enabled status
        for cid, enabled in self.cue_enabled_status.items():
            data = build_osc_message(
                f"/engine/status/cue_enabled/{cid}", 1 if enabled else 0
            )
            if data:
                await websocket.send(data)

        # Per-cue names (after `load` above, like the live burst)
        for cid, name in self.cue_names.items():
            data = build_osc_message(f"/engine/status/cue_name/{cid}", name)
            if data:
                await websocket.send(data)

        # Per-mixer-channel volume status.
        # Scale note: each entry is one ~80-byte WS message. Acceptable up to
        # ~500 entries (~40 KB / ~500 ms over LAN). If a deployment ever
        # exceeds that (e.g. 8 nodes × 64 channels), switch to a single OSC
        # bundle via build_osc_bundle() instead of per-message sends.
        for key, vol in self.mixer_status.items():
            data = build_osc_message(f"/engine/status/audio/mixer/{key}/volume", vol)
            if data:
                await websocket.send(data)

        Logger.info("Late-join state dump sent to new WebSocket client")

    def on_timecode_change(self, value) -> None:
        """
        Broadcast timecode to UI as integer ms (whole seconds only), once per
        second.
        """
        try:
            ms = int(value) if value is not None else 0
        except (TypeError, ValueError):
            return
        current_second = ms // 1000
        if current_second != self._last_timecode_second:
            self._last_timecode_second = current_second
            self._broadcast_status("timecode", current_second * 1000)
            Logger.debug(f"Timecode broadcast {current_second}s")

    def _clear_playback_state(self):
        """
        Clear runtime playback tracking: timestamps, timecode, armed, nextcue.

        Also clears the per-node armed accumulator. The
        _required_nodes / _adopted_nodes snapshots stay — they belong to
        the loaded project, not playback state, and get recomputed on the
        next load_project. unload_project clears them explicitly.
        """
        self._cue_broadcast_timestamps.clear()
        self._last_timecode_second = -1
        self._broadcast_status("timecode", 0)
        self.set_status("armed", "no")
        self.set_status("nextcue", "")
        self.stop_timecode()
        # Clear THIS process's 24h MTC wrap accumulator on the project transition.
        # ControllerEngine inherits BaseEngine directly (NOT NodeEngine), so it
        # does not go through _load_project_inner — its own MtcListener must be
        # reset here, co-orchestrated with stop_timecode above. (Plan 4)
        if self.mtc_listener is not None:
            self.mtc_listener.reset_24h_state()
        with self._cluster_lock:
            self._armed_nodes.clear()
        self._cancel_arm_watchdog()

    #########################
    # Project management
    #########################

    def load_project(self, project_name, context=None, deploy_only=False):
        # Don't allow loading while script is running
        if self.get_status("running") == "yes":
            Logger.warning(
                f"Cannot load project {project_name} while script is running."
                f"Stop first."
            )
            return False

        Logger.info(f"Loading project {project_name}")
        self._bump_run_seq("load")
        self._clear_playback_state()
        self.reset_script()
        # Before anything can fail. load_project_config() and read_script()
        # both return False further down, and without this the previous
        # project's warning would stay on screen for a project that, to the
        # operator, just failed to load. A successful _resolve_cluster_state
        # overwrites it a few lines later.
        self._clear_load_diagnosis()
        # The same argument, applied to the load status. reset_script() drops
        # self.script, but nothing dropped `load`, so a failed load left the
        # engine holding no project while still broadcasting the PREVIOUS
        # project's name on /engine/status/load. That lie reached the transport
        # bar and power-bridge alike — project_loaded() stayed True, so
        # /setnextcue and /gocue were accepted against a project the engine no
        # longer had. The two signals now die together, which is what lets
        # get_project_status() trust either of them.
        # Deliberate cost: during a *successful* load the transport bar shows
        # "—" until the new name arrives, instead of the stale previous one.
        self.set_status("load", "")

        if deploy_only:
            Logger.info(f"Deploy only requested for {project_name}")
            return True

        try:
            self.cm.load_project_config(project_name)
        except Exception as e:
            Logger.error(f"Error loading project config: {e}")

            request_uuid = self.get_editor_request()
            self.set_editor_request("")
            self.error_to_editor(
                context,
                f"Project config error: {e}",
                request_uuid=request_uuid,
                action="project_ready",
            )
            return False

        try:
            self.read_script(project_name)
        except Exception as e:
            Logger.error(f"Error loading project script: {e}")

            request_uuid = self.get_editor_request()
            self.set_editor_request("")
            self.error_to_editor(
                context,
                f"Project script error: {e}",
                request_uuid=request_uuid,
                action="project_ready",
            )
            return False

        Logger.info(f"Script from {project_name} loaded")
        self.script.unix_name = project_name

        # Initialise per-cue status: every cue starts as unplayed (0).
        # Broadcasts one WS message per cue so the UI can populate its cue
        # list.
        self.cue_status = {cid: 0 for cid in self._collect_cue_ids(self.script.cuelist)}
        for cid in self.cue_status:
            self._broadcast_cue_status(cid, 0, force=True)
        Logger.info(f"Cue status initialised for {len(self.cue_status)} cues")

        # Initialise per-cue enabled status from XML (resets show-time
        # overrides).
        self.cue_enabled_status = self._collect_cue_enabled(self.script.cuelist)
        for cid, enabled in self.cue_enabled_status.items():
            self._broadcast_cue_enabled(cid, enabled)
        Logger.info(
            f"Cue enabled status initialised for {len(self.cue_enabled_status)} cues"
        )

        # Update internal status.
        # This broadcast carries the project's unix_name, NOT its uuid, and
        # that is a decision rather than a leftover: the frontend prints this
        # string in the transport bar and matches it against project.unix_name,
        # so switching it to a uuid shows the operator a raw uuid and breaks
        # that match. Consumers that need the uuid ask project_status, which
        # returns both (869dr2k2h, closed the additive way).
        self.set_status("load", project_name)

        # Per-cue names, deliberately AFTER the load status (unlike cue_status
        # and cue_enabled above): duplicated projects share cue uuids, so a
        # client treats a change of `load` as "new table" and refills it from
        # this burst. Keep this order.
        self.cue_names = self._collect_cue_names(self.script.cuelist)
        for cid, name in self.cue_names.items():
            self._broadcast_cue_name(cid, name)
        Logger.info(f"Cue names initialised for {len(self.cue_names)} cues")

        # Probe cluster, derive _required_nodes for GO gating, refresh <online>
        # in network_map. Done BEFORE _forward_load_to_nodes so the gating set
        # is in place by the time nodes start sending armed_ready.
        self._resolve_cluster_state()

        # Forward load command to NodeEngine via NNG (nodes will arm cues)
        self._forward_load_to_nodes(project_name)

        # Timecode starts on load; runs until next load or engine shutdown
        self.start_timecode()
        # Enable mtc_callback → on_timecode_change → broadcast
        self.go_offset = 0

        # armed=yes is NOT set here -- it's set when NodeEngine reports
        # armed_ready via status_operation_callback, ensuring cues are actually
        # armed before the UI shows GO as available

        # Confirm the project is loaded
        self.set_show_lock_file()
        Logger.info(f"Project {project_name} loaded")
        # Note: Don't clear editor_request here - handle_editor_command will
        # clear it after confirmation
        return True

    def deploy_project(self, project_name):
        self.load_project(project_name)

    def go_script(self, value, context=None):
        if self.get_status("armed") != "yes":
            Logger.warning("Cues not armed. GO not available.")
            return

        if not self.script:
            Logger.warning("No script loaded, cannot process GO command.")
            return

        self.set_status("running", "yes")

        # Ship OUR GO instant to every node instead of letting each one read
        # its own MTC whenever it happens to execute the command. Badajoz,
        # 2026-09-25: a node whose setnextcue pre-arm held _command_lock for
        # ~500ms read its own MTC ~520ms later than the other two boxes and
        # anchored its whole Auto-continue chain late -- MTC-following cannot
        # correct a wrong anchor, only agreeing on one avoids it. Nodes older
        # than this fix never read `value` on "go" (NodeEngine.go_script), so
        # this is safe to send unconditionally; a node running this fix falls
        # back to its own local MTC (NodeEngine._resolve_go_anchor) whenever
        # this dict is absent, malformed, or the lag looks wrong.
        # Only a LIVE reading may become the cluster's anchor: a listener that
        # never opened its port reads 0.0 forever, and shipping that made
        # every node fire 4.48s late on test2 (2026-09-25). Without a live
        # reading, send the GO without an anchor -- each node falls back to
        # its own MTC, exactly as before this fix.
        # Always a dict (869fc8ytz): it carries the run counter that
        # cross-node follow messages are checked against. Nodes read nothing
        # else from the GO value, so the caller's own `value` is not passed on.
        go_value = {"run_seq": self._run_seq}
        if self.mtc_listener is not None:
            if self.mtc_listener.is_receiving():
                go_value["go_mtc_ms"] = self.mtc_listener.main_tc.milliseconds_exact
            else:
                Logger.error(
                    "GO: controller MTC listener is not receiving -- sending GO "
                    "without a shared anchor; each node will anchor on its own "
                    "MTC. Check the controller's MIDI/MTC setup."
                )

        # Forward GO to NodeEngine via NNG (needed when called from editor;
        # when called from WebSocket the comms layer also forwards, but the
        # NodeEngine's run_command is idempotent so a double-call is harmless)
        self._forward_command_to_nodes("/engine/command/go", go_value)

        Logger.info("GO command processed")
        return True

    def _setnextcue_handler(self, value):
        """
        Handle setnextcue from UI — forward to NodeEngine which owns the
        pointer.
        """
        self._forward_command_to_nodes("/engine/command/setnextcue", value)

    def _cue_enabled_handler(self, value):
        """Handle cue_enabled toggle from UI.

        Value format: "<cue_id> <0|1>" (space-separated UUID and enabled flag).
        """
        if not value or not isinstance(value, str):
            Logger.warning(f"Invalid cue_enabled value: {repr(value)}")
            return

        parts = value.split(" ", 1)
        if len(parts) != 2 or parts[1] not in ("0", "1"):
            Logger.warning(
                f'Invalid cue_enabled format (expected "uuid 0|1"): {repr(value)}'
            )
            return

        cue_id, enabled_str = parts
        enabled = enabled_str == "1"

        if cue_id not in self.cue_enabled_status:
            Logger.warning(f"cue_enabled: unknown cue_id {cue_id}")
            return

        self.cue_enabled_status[cue_id] = enabled
        self._broadcast_cue_enabled(cue_id, enabled)
        self._forward_command_to_nodes("/engine/command/cue_enabled", value)
        Logger.info(f'Cue {cue_id} {"enabled" if enabled else "disabled"}')

    def _forward_command_to_nodes(self, address: str, value) -> None:
        """Forward a generic command to NodeEngine via NNG."""
        if not hasattr(self, "communications_thread") or not self.communications_thread:
            Logger.warning(
                "Cannot forward command to nodes: communications thread not available"
            )
            return

        parts = address.strip("/").split("/")
        command_name = parts[-1] if parts else address

        operation = NodeOperation(
            type=OperationType.COMMAND,
            action=ActionType.UPDATE,
            sender=(
                self.cm.node_conf.get("uuid", "controller")
                if hasattr(self, "cm") and self.cm
                else "controller"
            ),
            target=command_name,
            data={"value": value, "address": address},
        )

        try:
            asyncio.run_coroutine_threadsafe(
                self.communications_thread.nng_hub.send_operation(operation),
                self.communications_thread.event_loop,
            )
            Logger.debug(f"Forwarded command to nodes: {command_name} = {repr(value)}")
        except Exception as e:
            Logger.error(f"Error forwarding command to nodes: {e}")

    def _controller_uuid(self) -> str:
        if hasattr(self, "cm") and self.cm:
            return self.cm.node_conf.get("uuid", "controller")
        return "controller"

    def _adopted_uuids_from_network_map(self) -> set[str]:
        """Read adopted node UUIDs from the in-memory network_map.

        We avoid NetworkMap.get_nodes_by_adoption() because it mutates the dict
        via strtobool — only works on the first pass; subsequent calls raise
        because `adopted`/`online` are no longer string-typed.
        """
        out: set[str] = set()
        try:
            node_list = (self.cm.network_map or {}).get("node_list", [])
            for entry in node_list:
                if not isinstance(entry, dict):
                    continue
                node = entry.get("node") or {}
                uuid = node.get("uuid")
                adopted = node.get("adopted", False)
                if isinstance(adopted, str):
                    adopted = adopted.strip().lower() in ("true", "1", "yes")
                if adopted and uuid:
                    out.add(uuid)
        except Exception as e:
            Logger.warning(f"Could not read network_map: {e}")
        return out

    def _node_label(self, uuid: str) -> str:
        """Human-readable name for a node, for LOG LINES ONLY.

        A bare UUID in a journal is unreadable under pressure, so the log says
        `node01 (2b6f…)`. Wire payloads stay UUID-only: role_id/alias are
        mutable projections per the node-identity contract, and the browser
        already resolves them from initial_mappings. Two sources of truth for a
        name is exactly what that contract forbids.
        """
        try:
            node_list = (self.cm.network_map or {}).get("node_list", [])
            for entry in node_list:
                if not isinstance(entry, dict):
                    continue
                node = entry.get("node") or {}
                if node.get("uuid") != uuid:
                    continue
                for field in ("alias", "role_id", "hostname"):
                    name = node.get(field)
                    if name:
                        return f"{name} ({uuid[:8]}…)"
                break
        except Exception:
            pass
        return uuid

    def _probe_cluster_liveness(self, timeout: float = 1.5) -> set[str]:
        """Broadcast a ping to all nodes and collect pong replies.

        ⚠ NOT re-entrant, and not safe to run twice concurrently: the pong
        bookkeeping below (_pong_responses / _pong_expected / _pong_event) is
        shared instance state with no per-call correlation id, so two probes in
        flight would eat each other's replies. Nothing enforces that — it holds
        only because every caller arrives through the engine's single editor
        listener, which awaits one command's handler before accepting the next
        (see ControllerCommunications.editor_listener). If editor commands are
        ever processed concurrently, this needs a real lock or a correlation id
        first. get_cluster_status made this reachable from the UI, so the
        invariant now matters beyond project loads.

        The set of senders that respond within `timeout` is the authoritative
        "alive right now" view of the cluster. The controller's own UUID is
        always included (it never needs to ping itself).

        Returns the set of UUIDs that are alive.
        """
        controller_uuid = self._controller_uuid()
        adopted_uuids = self._adopted_uuids_from_network_map()

        expected = adopted_uuids - {controller_uuid}

        with self._cluster_lock:
            self._pong_responses.clear()
            self._pong_expected = set(expected)
        self._pong_event.clear()

        if not expected:
            Logger.debug("Cluster probe: no remote nodes to ping")
            return {controller_uuid}

        if not hasattr(self, "communications_thread") or not self.communications_thread:
            Logger.warning("Cluster probe skipped: communications thread not available")
            return {controller_uuid}

        ping_op = NodeOperation(
            type=OperationType.COMMAND,
            action=ActionType.UPDATE,
            sender=controller_uuid,
            target="ping",
            data={"value": None, "address": "/engine/cluster/ping"},
        )
        try:
            asyncio.run_coroutine_threadsafe(
                self.communications_thread.nng_hub.send_operation(ping_op),
                self.communications_thread.event_loop,
            )
        except Exception as e:
            Logger.warning(f"Could not broadcast cluster ping: {e}")
            return {controller_uuid}

        # Early-exit wait. The comms thread sets the event when all expected
        # pongs have arrived; otherwise we wake on timeout with whatever did.
        self._pong_event.wait(timeout=timeout)

        with self._cluster_lock:
            alive = set(self._pong_responses)
        alive.add(controller_uuid)
        Logger.debug(
            f"Cluster probe: expected={sorted(expected)} "
            f"alive_remote={sorted(alive - {controller_uuid})}"
        )
        return alive

    def _resolve_cluster_state(self) -> None:
        """Probe the cluster and decide which nodes the GO gate must wait for.

        Snapshots three sets at load time:
          - adopted: UUIDs from network_map.xml
          - alive:   UUIDs that ponged within the probe window
          - project: UUIDs referenced by any cue's output_name in self.script

        Computes self._required_nodes = adopted & alive & project, then
        always adds the controller's own UUID (its node-engine always
        processes the load and sends armed_ready, so this is reachable —
        and prevents the degenerate set() >= set() = True case from
        flipping armed=yes before any node has loaded).

        Categorizes each adopted node and logs info / warning / error.
        Resets _armed_nodes for the new project.
        """
        controller_uuid = self._controller_uuid()
        adopted = self._adopted_uuids_from_network_map()
        try:
            project = self._collect_project_nodes(self.script.cuelist)
        except Exception as e:
            Logger.warning(f"Could not collect project nodes: {e}")
            project = set()
        alive = self._probe_cluster_liveness()

        # Categorize for operator visibility.
        for uuid in sorted(adopted):
            in_alive = uuid in alive
            in_project = uuid in project
            if in_alive and in_project:
                continue  # the silent happy path — tracked via armed_ready
            if in_alive and not in_project:
                Logger.info(
                    f"node {self._node_label(uuid)} online but unused by this "
                    f"project"
                )
            elif not in_alive and in_project:
                # Say what the code below actually does. `required` is
                # (adopted & alive & project), so this node is EXCLUDED from
                # the gate: GO is enabled without it and its cues stay silent.
                # The old wording here claimed "GO blocked", which sent anyone
                # debugging a mute show from the journal the wrong way.
                Logger.error(
                    f"node {self._node_label(uuid)} is used by this project "
                    f"but did not respond to ping; its cues will NOT play. "
                    f"GO is NOT blocked — the node is excluded from the arm "
                    f"gate."
                )
            else:
                Logger.warning(
                    f"node {self._node_label(uuid)} is adopted but did not "
                    f"respond to ping; not required by this project — "
                    f"investigate why it is offline"
                )

        # The two categories worth telling the operator about. Warn loudly,
        # never block: `required` below is untouched by any of this.
        #
        # The controller is excluded from both. It is added to `required`
        # unconditionally, so it can never be a node whose cues silently do not
        # play — and it is not always flagged `adopted` in network_map.xml, so
        # without this it would be reported as "not in the cluster" on a
        # perfectly healthy load. A false alarm here is worse than no alarm:
        # it teaches operators to ignore the real one.
        missing = sorted(project - adopted - {controller_uuid})
        unreachable = sorted((project & adopted) - alive - {controller_uuid})

        # Project nodes that are NOT adopted at all — script is broken for
        # this cluster.
        for uuid in missing:
            Logger.warning(
                f"project references node {self._node_label(uuid)} which is "
                f"not in the cluster; cues for it will not fire"
            )

        required = (adopted & alive & project) | {controller_uuid}

        with self._cluster_lock:
            self._adopted_nodes = set(adopted)
            self._required_nodes = required
            self._armed_nodes.clear()
            self._load_id += 1
            self._load_diagnosis = {
                "load_id": self._load_id,
                "project": str(self.get_status("load") or ""),
                "missing": missing,
                "unreachable": unreachable,
            }
            diagnosis = dict(self._load_diagnosis)

        Logger.info(
            f"Cluster state resolved: required={sorted(required)} "
            f"alive={sorted(alive)} adopted={sorted(adopted)} "
            f"project={sorted(project)}"
        )

        # Tell the UI, even when both lists are empty — see
        # _cluster_warning_payload.
        self._broadcast_cluster_warning(diagnosis)

        # The probe's `alive` set is a runtime liveness snapshot (sub-second,
        # used here for GO gating). The <online> field in network_map.xml is
        # a different concept: nodeconf's startup-discovery flag, persisted
        # so that adopted-but-currently-absent nodes keep their identity
        # records instead of being dropped from the map. The engine must NOT
        # overwrite that with its runtime view. See CLAUDE.md "Node identity"
        # / "<online> field ownership" for the full rationale.

        self._arm_arm_watchdog()

    _ARM_WATCHDOG_S = 120.0

    def _arm_arm_watchdog(self, what: str = "Load") -> None:
        """(Re)start the watchdog timer that surfaces a stalled (re-)arm.

        Fires _ARM_WATCHDOG_S seconds after a load, or after a STOP (869fhjdq5:
        an unattended show that fails to re-arm would otherwise answer every
        /go with a silent 409), if not all required nodes have reported
        armed_ready by then. Logs an error naming the still-pending UUIDs so
        the operator sees what's wedged. `what` names the operation in that
        line: "Load stalled" or "Re-arm after STOP stalled".
        """
        self._cancel_arm_watchdog()
        self._arm_watchdog_what = what
        timer = threading.Timer(self._ARM_WATCHDOG_S, self._on_arm_watchdog_fire)
        timer.daemon = True
        self._arm_watchdog = timer
        timer.start()

    def _cancel_arm_watchdog(self) -> None:
        timer = self._arm_watchdog
        if timer is not None:
            timer.cancel()
            self._arm_watchdog = None

    def _on_arm_watchdog_fire(self) -> None:
        with self._cluster_lock:
            armed = set(self._armed_nodes)
            required = set(self._required_nodes)
        pending = required - armed
        if not pending:
            return
        what = getattr(self, "_arm_watchdog_what", "Load")
        Logger.error(
            f"{what} stalled: nodes still pending armed_ready after "
            f"{self._ARM_WATCHDOG_S:.0f}s: {sorted(pending)}"
        )

    def stop_script(self, value):
        """
        Handle STOP command - stop timecode, update status and forward to
        nodes.
        """
        if self.get_status("running") != "yes":
            Logger.info("Script not running, nothing to stop.")
            return

        self._bump_run_seq("stop")
        self.go_offset = None
        self.set_status("running", "no")
        self._clear_playback_state()
        # The nodes re-arm now; watch it as a load is watched (869fhjdq5).
        self._arm_arm_watchdog(what="Re-arm after STOP")

        # Reset all cue statuses to unplayed (0) and broadcast to UI.
        for cid in self.cue_status:
            self.cue_status[cid] = 0
            self._broadcast_cue_status(cid, 0, force=True)

        self._forward_command_to_nodes("/engine/command/stop", value)

        Logger.info("STOP command processed - timecode stopped; nodes will re-arm")
        return True

    def get_project_status(self, value, context=None):
        """Return the current project load/playback state.

        Three states, deliberately distinct:

        * `running` — a project is loaded AND playing.
        * `loaded`  — a project is loaded and NOT playing. This is the state
          the reply used to collapse into `none`, which made a loaded-but-
          stopped project indistinguishable from no project at all — and left
          this reply contradicting /engine/status/load, which kept naming the
          project all along.
        * `none`    — no project.

        `loaded` says nothing about readiness: the GO gate waits on `armed`,
        published separately on /engine/status/armed. A project whose nodes
        never answer sits in `loaded` for the whole 120 s arm watchdog and
        never becomes GO-able. Do not treat this field as a GO gate.

        `project_uuid` is populated in both `running` and `loaded`, and the
        invariant is one-way: **a status other than `none` always carries a
        non-empty uuid**. That is why `running` with no script — which the
        guards in load_project/unload_project should make unreachable —
        degrades to `none` here rather than to a `running` a client cannot
        act on.

        `project_unix_name` is the same string /engine/status/load broadcasts,
        returned here so a consumer that needs the uuid never has to make that
        broadcast carry one (869dr2k2h).

        The load state is read from two attributes that another thread can be
        mutating mid-load. They are ANDed, so a torn read falls to `none` —
        never to a `loaded` naming a project the engine does not hold.
        """
        running = self.get_status("running") == "yes"
        loaded = self.script is not None and bool(self.get_status("load"))

        if running:
            status = "running"
        elif loaded:
            status = "loaded"
        else:
            status = "none"

        if status == "none" or self.script is None:
            return {"status": "none", "project_uuid": "", "project_unix_name": ""}

        # unix_name is not a CuemsScript field — load_project attaches it by
        # hand — so any script built by another path simply lacks it.
        return {
            "status": status,
            "project_uuid": str(self.script.id),
            "project_unix_name": str(getattr(self.script, "unix_name", "") or ""),
        }

    def get_cluster_status(self, value, context=None) -> dict:
        """Return who is adopted and who answers right now.

        Two different signals, deliberately kept apart (the UI must never merge
        them):

        * `adopted` — from network_map.xml, i.e. what cuems-nodeconf knows.
        * `alive`   — this engine's sub-second ping/pong, the only thing the GO
          gate trusts. `<online>` in the XML is nodeconf's discovery view and
          is NOT this.

        `age_s` is how old the probe behind this answer is, computed here: the
        browser cannot interpret our monotonic clock.

        Read-only — it never loads, unloads, arms or stops anything, so it is
        safe while a show runs. Repeat calls inside CLUSTER_STATUS_CLAMP_S are
        served from cache so a polling panel cannot turn into a ping flood.
        Always returns a populated dict: the dispatch path confirms only on a
        truthy result, and an empty one would leave the editor waiting out its
        25 s timeout.
        """
        now = time.monotonic()
        cached = getattr(self, "_cluster_status_cache", None)
        if cached is not None and (now - cached[0]) < CLUSTER_STATUS_CLAMP_S:
            probed_at, payload = cached
        else:
            alive = self._probe_cluster_liveness()
            adopted = self._adopted_uuids_from_network_map()
            payload = {
                "alive": sorted(alive),
                "adopted": sorted(adopted),
                "controller": self._controller_uuid(),
            }
            probed_at = time.monotonic()
            self._cluster_status_cache = (probed_at, payload)

        # Read the diagnosis fresh, OUTSIDE the clamp. Folding it into the
        # cached payload would serve the pre-transition diagnosis to any poll
        # landing within CLUSTER_STATUS_CLAMP_S of a load, unload or reload —
        # a warning outliving its cause, which is the worst outcome this
        # surface has. The clamp exists to stop ping floods; a lock-guarded
        # dict read is not a ping. Same treatment as age_s.
        with self._cluster_lock:
            diagnosis = dict(self._load_diagnosis) if self._load_diagnosis else {}

        return dict(
            payload,
            age_s=round(time.monotonic() - probed_at, 3),
            missing=diagnosis.get("missing", []),
            unreachable=diagnosis.get("unreachable", []),
        )

    def unload_project(self, value, context=None):
        """Unload the current project. Rejects if playback is running."""
        if self.get_status("running") == "yes":
            raise RuntimeError("Cannot unload while running. Stop playback first.")
        self._bump_run_seq("unload")
        self._clear_playback_state()
        self.reset_script()
        self.cue_status = {}
        self.cue_enabled_status = {}
        self.cue_names = {}
        self.set_status("load", "")
        # No project loaded → no required/adopted snapshot. Without this, a
        # late armed_ready from a slow node could flip armed=yes on an
        # unloaded project.
        with self._cluster_lock:
            self._required_nodes.clear()
            self._adopted_nodes.clear()
        self._clear_load_diagnosis()
        self._forward_command_to_nodes("/engine/command/stop", value)
        Logger.info("Project unloaded")
        return True

cue_operation_callback(operation)

Callback invoked when cues are received from nodes.

Handles three action types: - ADD: cue started playing on a node → status 1, broadcast immediately - REMOVE: cue finished playing on a node → status 100, broadcast - immediately - UPDATE: percentage progress from a node (future) → throttled - broadcast

Source code in src/cuemsengine/ControllerEngine.py

def cue_operation_callback(self, operation: NodeOperation):
    """Callback invoked when cues are received from nodes.

    Handles three action types:
    - ADD: cue started playing on a node → status 1, broadcast immediately
    - REMOVE: cue finished playing on a node → status 100, broadcast
      - immediately
    - UPDATE: percentage progress from a node (future) → throttled
      - broadcast
    """
    Logger.info(f"Cue operation received: {operation}")
    cue_id = operation.data.get("id") if operation.data else None

    # Drop operations for cues not belonging to the current project.
    # This prevents stale REMOVE/ADD notifications from the NodeEngine
    # (sent when it disarms the previous project) from being broadcast
    # to the UI as unknown UUIDs.
    if cue_id and cue_id not in self.cue_status:
        Logger.debug(
            f"Ignoring cue operation for unknown/stale cue_id {cue_id}"
            f"(action={operation.action})"
        )
        return

    # No cue starts while the show is not running: go_script sets running
    # before it forwards the GO. An ADD here is a node reporting a start it
    # dispatched just before a STOP reached it (a stop_project cue on a
    # multi-node cluster, 869fhjdq5); without this gate it would show the
    # cue at 1 after the reset and its disarm's REMOVE would then make it
    # 100, with a spurious cue_started/cue_ended for the Companion relay.
    if (
        operation.action in (ActionType.ADD, ActionType.UPDATE)
        and self.get_status("running") != "yes"
    ):
        Logger.debug(
            f"Ignoring cue {operation.action} for {cue_id}: not running "
            "(a late start report after a STOP)"
        )
        return

    if operation.action == ActionType.ADD:
        # Cue started playing: mark as playing (1) and broadcast
        # immediately.
        if cue_id:
            self.cue_status[cue_id] = 1
            self._broadcast_cue_status(cue_id, 1, force=True)
        try:
            self.status.currentcue = [
                operation.data["id"],
                operation.data["offset"],
            ]
            Logger.debug(f"Current cue updated: {self.status.currentcue}")
        except Exception as e:
            Logger.error(f"Error updating currentcue: {e}")

    elif operation.action == ActionType.REMOVE:
        # Cue finished playing: mark as played (100) and broadcast
        # immediately.
        # Only transition to 100 if the cue was actually playing (status ==
        # 1).
        # REMOVEs that arrive while status is 0 (e.g. NodeEngine disarming
        # the
        # previous project after a reload) are stale and must be silently
        # dropped.
        if cue_id:
            if self.cue_status.get(cue_id) == 1:
                self.cue_status[cue_id] = 100
                self._broadcast_cue_status(cue_id, 100, force=True)
            else:
                Logger.debug(
                    f"Ignoring stale REMOVE for cue {cue_id}"
                    f"(status={self.cue_status.get(cue_id)}, expected 1)"
                )
        self.status.remove_currentcue(operation.data["id"])
        Logger.debug(f"Cue removed from currentcue: {operation.data['id']}")

    elif operation.action == ActionType.UPDATE:
        # Future: percentage progress updates from loop_cue() during
        # playback.
        # Throttled by _broadcast_cue_status (Tier 2 / controller-side).
        # The node-side Tier 1 throttle (CUE_STATUS_UPDATE_HZ) limits NNG
        # traffic.
        if cue_id:
            pct = operation.data.get("percentage", 1)
            self.cue_status[cue_id] = pct
            self._broadcast_cue_status(cue_id, pct)  # throttled
        Logger.debug(
            f"Cue percentage update: {cue_id} ="
            f"{operation.data.get('percentage')}"
        )

    else:
        Logger.warning(f"Unknown cue action: {operation.action}")

get_cluster_status(value, context=None)

Return who is adopted and who answers right now.

Two different signals, deliberately kept apart (the UI must never merge them):

• adopted — from network_map.xml, i.e. what cuems-nodeconf knows.
• alive — this engine's sub-second ping/pong, the only thing the GO gate trusts. <online> in the XML is nodeconf's discovery view and is NOT this.

age_s is how old the probe behind this answer is, computed here: the browser cannot interpret our monotonic clock.

Read-only — it never loads, unloads, arms or stops anything, so it is safe while a show runs. Repeat calls inside CLUSTER_STATUS_CLAMP_S are served from cache so a polling panel cannot turn into a ping flood. Always returns a populated dict: the dispatch path confirms only on a truthy result, and an empty one would leave the editor waiting out its 25 s timeout.

Source code in src/cuemsengine/ControllerEngine.py

def get_cluster_status(self, value, context=None) -> dict:
    """Return who is adopted and who answers right now.

    Two different signals, deliberately kept apart (the UI must never merge
    them):

    * `adopted` — from network_map.xml, i.e. what cuems-nodeconf knows.
    * `alive`   — this engine's sub-second ping/pong, the only thing the GO
      gate trusts. `<online>` in the XML is nodeconf's discovery view and
      is NOT this.

    `age_s` is how old the probe behind this answer is, computed here: the
    browser cannot interpret our monotonic clock.

    Read-only — it never loads, unloads, arms or stops anything, so it is
    safe while a show runs. Repeat calls inside CLUSTER_STATUS_CLAMP_S are
    served from cache so a polling panel cannot turn into a ping flood.
    Always returns a populated dict: the dispatch path confirms only on a
    truthy result, and an empty one would leave the editor waiting out its
    25 s timeout.
    """
    now = time.monotonic()
    cached = getattr(self, "_cluster_status_cache", None)
    if cached is not None and (now - cached[0]) < CLUSTER_STATUS_CLAMP_S:
        probed_at, payload = cached
    else:
        alive = self._probe_cluster_liveness()
        adopted = self._adopted_uuids_from_network_map()
        payload = {
            "alive": sorted(alive),
            "adopted": sorted(adopted),
            "controller": self._controller_uuid(),
        }
        probed_at = time.monotonic()
        self._cluster_status_cache = (probed_at, payload)

    # Read the diagnosis fresh, OUTSIDE the clamp. Folding it into the
    # cached payload would serve the pre-transition diagnosis to any poll
    # landing within CLUSTER_STATUS_CLAMP_S of a load, unload or reload —
    # a warning outliving its cause, which is the worst outcome this
    # surface has. The clamp exists to stop ping floods; a lock-guarded
    # dict read is not a ping. Same treatment as age_s.
    with self._cluster_lock:
        diagnosis = dict(self._load_diagnosis) if self._load_diagnosis else {}

    return dict(
        payload,
        age_s=round(time.monotonic() - probed_at, 3),
        missing=diagnosis.get("missing", []),
        unreachable=diagnosis.get("unreachable", []),
    )

get_project_status(value, context=None)

Return the current project load/playback state.

Three states, deliberately distinct:

• running — a project is loaded AND playing.
• loaded — a project is loaded and NOT playing. This is the state the reply used to collapse into none, which made a loaded-but- stopped project indistinguishable from no project at all — and left this reply contradicting /engine/status/load, which kept naming the project all along.
• none — no project.

loaded says nothing about readiness: the GO gate waits on armed, published separately on /engine/status/armed. A project whose nodes never answer sits in loaded for the whole 120 s arm watchdog and never becomes GO-able. Do not treat this field as a GO gate.

project_uuid is populated in both running and loaded, and the invariant is one-way: a status other than none always carries a non-empty uuid. That is why running with no script — which the guards in load_project/unload_project should make unreachable — degrades to none here rather than to a running a client cannot act on.

project_unix_name is the same string /engine/status/load broadcasts, returned here so a consumer that needs the uuid never has to make that broadcast carry one (869dr2k2h).

The load state is read from two attributes that another thread can be mutating mid-load. They are ANDed, so a torn read falls to none — never to a loaded naming a project the engine does not hold.

Source code in src/cuemsengine/ControllerEngine.py

def get_project_status(self, value, context=None):
    """Return the current project load/playback state.

    Three states, deliberately distinct:

    * `running` — a project is loaded AND playing.
    * `loaded`  — a project is loaded and NOT playing. This is the state
      the reply used to collapse into `none`, which made a loaded-but-
      stopped project indistinguishable from no project at all — and left
      this reply contradicting /engine/status/load, which kept naming the
      project all along.
    * `none`    — no project.

    `loaded` says nothing about readiness: the GO gate waits on `armed`,
    published separately on /engine/status/armed. A project whose nodes
    never answer sits in `loaded` for the whole 120 s arm watchdog and
    never becomes GO-able. Do not treat this field as a GO gate.

    `project_uuid` is populated in both `running` and `loaded`, and the
    invariant is one-way: **a status other than `none` always carries a
    non-empty uuid**. That is why `running` with no script — which the
    guards in load_project/unload_project should make unreachable —
    degrades to `none` here rather than to a `running` a client cannot
    act on.

    `project_unix_name` is the same string /engine/status/load broadcasts,
    returned here so a consumer that needs the uuid never has to make that
    broadcast carry one (869dr2k2h).

    The load state is read from two attributes that another thread can be
    mutating mid-load. They are ANDed, so a torn read falls to `none` —
    never to a `loaded` naming a project the engine does not hold.
    """
    running = self.get_status("running") == "yes"
    loaded = self.script is not None and bool(self.get_status("load"))

    if running:
        status = "running"
    elif loaded:
        status = "loaded"
    else:
        status = "none"

    if status == "none" or self.script is None:
        return {"status": "none", "project_uuid": "", "project_unix_name": ""}

    # unix_name is not a CuemsScript field — load_project attaches it by
    # hand — so any script built by another path simply lacks it.
    return {
        "status": status,
        "project_uuid": str(self.script.id),
        "project_unix_name": str(getattr(self.script, "unix_name", "") or ""),
    }

nodelist_modify(message, context=None)

Adopt or un-adopt a node, then refresh our cached topology.

Forwards {'action','value','modify_action'} to cuems-nodeconf over /tmp/nodeconf.ipc, which flips in network_map.xml. On success we re-read that file, because our own copy was loaded once at startup (ConfigManager(load_all=True)) and would otherwise keep the GO gate blind to the change until an engine restart.

Returns True only when nodeconf confirmed AND our reload succeeded; the caller confirms "OK" to the editor on True. Every False path has already sent an explanatory error.

Source code in src/cuemsengine/ControllerEngine.py

def nodelist_modify(self, message: dict, context=None) -> bool:
    """Adopt or un-adopt a node, then refresh our cached topology.

    Forwards {'action','value','modify_action'} to cuems-nodeconf over
    /tmp/nodeconf.ipc, which flips <adopted> in network_map.xml. On success
    we re-read that file, because our own copy was loaded once at startup
    (ConfigManager(load_all=True)) and would otherwise keep the GO gate
    blind to the change until an engine restart.

    Returns True only when nodeconf confirmed AND our reload succeeded; the
    caller confirms "OK" to the editor on True. Every False path has
    already sent an explanatory error.
    """
    node_uuid = message.get("value")
    modify_action = message.get("modify_action")
    Logger.info(f"nodelist_modify: {modify_action} node {node_uuid}")

    # Adoption is a setup-time operation. Refuse (never stop anything) when
    # a project is running, and also when one is merely loaded: the GO
    # gate's _required_nodes snapshot is computed once, in load_project, so
    # a REMOVE accepted now would leave it waiting on a node the operator
    # just removed until the next load.
    if self.get_status("running") == "yes":
        return self._nodelist_refuse(
            context,
            "Cannot modify the node list while a project is running. "
            "Stop playback first.",
        )
    if self.get_status("load"):
        return self._nodelist_refuse(
            context,
            "Cannot modify the node list while a project is loaded. "
            "Unload the project first.",
        )

    if not node_uuid or not isinstance(node_uuid, str):
        return self._nodelist_refuse(
            context, f"nodelist_modify needs a node uuid, got {node_uuid!r}"
        )
    if modify_action not in ("ADD", "REMOVE"):
        return self._nodelist_refuse(
            context,
            f"Invalid modify_action: {modify_action!r}. Must be 'ADD' or 'REMOVE'",
        )

    # Fail fast when the daemon is not there at all. Measured on the rig:
    # with the socket absent, request_to_nodeconf does NOT return quickly —
    # it burns the full 15 s IPC timeout and then raises. Editor commands
    # are serialized by the engine's single listener, so a second click
    # queues behind the first and can blow the editor's own 25 s timeout,
    # turning a clear refusal into "Engine did not respond". nodeconf ships
    # disabled on most of the fleet, so this is the common path, not a rare
    # one.
    if not os.path.exists(NODECONF_IPC_PATH):
        return self._nodelist_refuse(
            context,
            "The node configuration service (cuems-nodeconf) is not "
            "running on this controller, so the node list cannot be "
            "changed. Enable it with: systemctl enable --now cuems-nodeconf",
        )

    try:
        reply = self.communications_thread.request_to_nodeconf(
            message, timeout=NODECONF_TIMEOUT_S
        )
    except Exception as e:
        return self._nodelist_refuse(
            context, f"Could not reach cuems-nodeconf: {type(e).__name__}: {e}"
        )

    # Communicator.send_request swallows every exception and returns None —
    # an absent socket (nodeconf disabled, which is the fleet default) and a
    # peer that never answers both land here.
    if reply is None:
        return self._nodelist_refuse(
            context,
            "The node configuration service (cuems-nodeconf) is not "
            "responding. Is it enabled on this controller?",
        )
    if not isinstance(reply, dict):
        return self._nodelist_refuse(
            context, f"Unexpected reply from cuems-nodeconf: {reply!r}"
        )
    if not reply.get("OK", False):
        return self._nodelist_refuse(
            context, reply.get("error", "cuems-nodeconf reported an error")
        )

    Logger.info(
        f"nodelist_modify: cuems-nodeconf confirmed {modify_action} "
        f"for node {node_uuid}"
    )

    # nodeconf has already rewritten network_map.xml. If we cannot re-read
    # it the operation is a PARTIAL success: the map (and so the editor and
    # the UI) is correct, only this process is behind. Say exactly that
    # instead of confirming OK over a stale topology.
    try:
        self._reload_network_map()
    except Exception as e:
        Logger.error(f"{type(e).__name__} reloading network_map: {e}")
        return self._nodelist_refuse(
            context,
            f"Node {modify_action.lower()}ed, but the engine could not "
            f"reload the topology ({type(e).__name__}: {e}) — restart "
            f"cuems-controller-engine before loading a project.",
        )
    return True

on_timecode_change(value)

Broadcast timecode to UI as integer ms (whole seconds only), once per second.

Source code in src/cuemsengine/ControllerEngine.py

def on_timecode_change(self, value) -> None:
    """
    Broadcast timecode to UI as integer ms (whole seconds only), once per
    second.
    """
    try:
        ms = int(value) if value is not None else 0
    except (TypeError, ValueError):
        return
    current_second = ms // 1000
    if current_second != self._last_timecode_second:
        self._last_timecode_second = current_second
        self._broadcast_status("timecode", current_second * 1000)
        Logger.debug(f"Timecode broadcast {current_second}s")

player_operation_callback(operation)

Callback invoked when players are received from nodes.

Parameters: - operation: NodeOperation with sender, target (player_id), and action

Source code in src/cuemsengine/ControllerEngine.py

def player_operation_callback(self, operation: NodeOperation):
    """
    Callback invoked when players are received from nodes.

    Parameters:
    - operation: NodeOperation with sender, target (player_id), and action
    """
    Logger.info(f"Player operation received: {operation}")

set_oscquery_values(values)

Stub for OSCQuery value setting - OSCQuery server has been removed.

Status updates are now handled via internal state tracking. TODO: Implement WebSocket status push if UI needs real-time status.

Source code in src/cuemsengine/ControllerEngine.py

def set_oscquery_values(self, values: dict):
    """Stub for OSCQuery value setting - OSCQuery server has been removed.

    Status updates are now handled via internal state tracking.
    TODO: Implement WebSocket status push if UI needs real-time status.
    """
    for key, value in values.items():
        Logger.debug(f"Status update (no-op): {key} = {repr(value)}")

set_status(property, value, strict=False)

Set status and push to UI via WebSocket when running, armed, or load.

Source code in src/cuemsengine/ControllerEngine.py

def set_status(self, property: str, value: str, strict: bool = False) -> None:
    """
    Set status and push to UI via WebSocket when running, armed, or load.
    """
    super().set_status(property, value, strict)
    if property in ("running", "armed", "load", "nextcue"):
        self._broadcast_status(property, value)

status_operation_callback(operation)

Callback invoked when status updates are received from nodes.

Handles stop_project, armed_ready, nextcue, cue_enabled, follow and pong notifications.

Source code in src/cuemsengine/ControllerEngine.py

def status_operation_callback(self, operation: NodeOperation):
    """Callback invoked when status updates are received from nodes.

    Handles stop_project, armed_ready, nextcue, cue_enabled, follow and
    pong notifications.
    """
    Logger.info(f"Status operation received: {operation}")
    if operation.target == "pong":
        # Cluster liveness probe reply. Runs on the AsyncCommsThread event
        # loop; _probe_cluster_liveness on the main thread waits on the
        # event.
        with self._cluster_lock:
            self._pong_responses.add(operation.sender)
            if self._pong_responses >= self._pong_expected:
                self._pong_event.set()
        Logger.debug(f"Pong from {operation.sender}")
        return
    if operation.target == "stop_project":
        # The authored end of the show (869fhjdq5): a node revealed a
        # stop_project ActionCue. First valid message wins; see the handler.
        self._handle_stop_project(operation)
    elif operation.target == "armed_ready":
        if operation.data and operation.data.get("armed") == "yes":
            # Aggregate per-node: only flip armed=yes when all required
            # nodes have reported. Filter foreign senders to keep the
            # tracker bounded.
            sender = operation.sender
            if sender not in self._adopted_nodes:
                Logger.debug(f"Ignoring armed_ready from non-adopted node {sender}")
                return
            with self._cluster_lock:
                self._armed_nodes.add(sender)
                armed_now = set(self._armed_nodes)
                required = set(self._required_nodes)
            Logger.info(f"Node {sender} armed ({len(armed_now)}/{len(required)})")
            if armed_now >= required and self.get_status("armed") != "yes":
                if self.go_offset is None:
                    Logger.info(
                        "Re-arm after stop complete — restarting timecode"
                        "and enabling GO"
                    )
                    self.start_timecode()
                    self.go_offset = 0
                else:
                    Logger.info("All required nodes armed — enabling GO")
                self.set_status("armed", "yes")
                self._cancel_arm_watchdog()
    elif operation.target == "nextcue":
        nextcue_id = operation.data.get("nextcue", "") if operation.data else ""
        self.set_status("nextcue", nextcue_id)
        Logger.info(f'Next cue updated: {nextcue_id or "(none)"}')
    elif operation.target == "cue_enabled":
        cue_id = operation.data.get("cue_id") if operation.data else None
        enabled = operation.data.get("enabled", True) if operation.data else True
        if cue_id and cue_id in self.cue_enabled_status:
            self.cue_enabled_status[cue_id] = enabled
            self._broadcast_cue_enabled(cue_id, enabled)
            Logger.info(f"Cue {cue_id} enabled status updated from node: {enabled}")
    elif operation.target == "audiomixer_status":
        # Authoritative mixer gain snapshot from a node (sent on mixer
        # startup / after reset_volumes). This is the ONLY node-sourced feed
        # of mixer_status; it lets the UI recover the TRUE gain rather than a
        # stale UI-write echo. jack-volume itself is write-only, so the node
        # reports what it commanded. Key shape matches the UI-write shadow:
        # "{node_uuid}/{output_index}/{channel}".
        node_uuid = operation.sender
        # Filter foreign/unadopted senders (mirrors armed_ready /
        # stop_project) so the shadow can't grow from nodes the cluster
        # never adopted. A pre-load startup snapshot from a not-yet-adopted
        # node is dropped, but the UI's unity default matches the post-reset
        # state, and the load-time reset snapshot (node adopted by then)
        # reconciles.
        if node_uuid not in self._adopted_nodes:
            Logger.debug(f"Ignoring audiomixer_status from non-adopted {node_uuid}")
            return
        data = operation.data or {}
        output_index = data.get("output_index", "0")
        entries = data.get("entries", {})
        if not isinstance(entries, dict):
            Logger.debug(f"audiomixer_status from {node_uuid} had non-dict entries")
            return
        updated = 0
        for channel, vol in entries.items():
            try:
                v = float(vol)
            except (TypeError, ValueError):
                continue
            if not math.isfinite(v):
                continue
            key = f"{node_uuid}/{output_index}/{channel}"
            self.mixer_status[key] = v
            self._broadcast_status(
                f"audio/mixer/{node_uuid}/{output_index}/{channel}/volume", v
            )
            updated += 1
        Logger.info(f"Mixer status from {node_uuid}: {updated} channel(s) updated")
    elif operation.target == "follow":
        self._relay_follow(operation)
    else:
        Logger.debug(f"Unknown status target: {operation.target}")

stop_script(value)

Handle STOP command - stop timecode, update status and forward to nodes.

Source code in src/cuemsengine/ControllerEngine.py

def stop_script(self, value):
    """
    Handle STOP command - stop timecode, update status and forward to
    nodes.
    """
    if self.get_status("running") != "yes":
        Logger.info("Script not running, nothing to stop.")
        return

    self._bump_run_seq("stop")
    self.go_offset = None
    self.set_status("running", "no")
    self._clear_playback_state()
    # The nodes re-arm now; watch it as a load is watched (869fhjdq5).
    self._arm_arm_watchdog(what="Re-arm after STOP")

    # Reset all cue statuses to unplayed (0) and broadcast to UI.
    for cid in self.cue_status:
        self.cue_status[cid] = 0
        self._broadcast_cue_status(cid, 0, force=True)

    self._forward_command_to_nodes("/engine/command/stop", value)

    Logger.info("STOP command processed - timecode stopped; nodes will re-arm")
    return True

unload_project(value, context=None)

Unload the current project. Rejects if playback is running.

Source code in src/cuemsengine/ControllerEngine.py

def unload_project(self, value, context=None):
    """Unload the current project. Rejects if playback is running."""
    if self.get_status("running") == "yes":
        raise RuntimeError("Cannot unload while running. Stop playback first.")
    self._bump_run_seq("unload")
    self._clear_playback_state()
    self.reset_script()
    self.cue_status = {}
    self.cue_enabled_status = {}
    self.cue_names = {}
    self.set_status("load", "")
    # No project loaded → no required/adopted snapshot. Without this, a
    # late armed_ready from a slow node could flip armed=yes on an
    # unloaded project.
    with self._cluster_lock:
        self._required_nodes.clear()
        self._adopted_nodes.clear()
    self._clear_load_diagnosis()
    self._forward_command_to_nodes("/engine/command/stop", value)
    Logger.info("Project unloaded")
    return True

NodeEngine

Bases: BaseEngine

This engine manages players for each node

Communicates with the ControllerEngine via OSCQuery

Interacts with Player objects via OSC

It is responsible for

• Starting and stopping players
• Monitoring player status
• Restarting players
• Updating player configurations
• Handling player failures
• Providing a clean interface for starting and stopping players
• Providing a clean interface for monitoring player status

Source code in src/cuemsengine/NodeEngine.py

class NodeEngine(BaseEngine):
    """
    This engine manages players for each node

    Communicates with the ControllerEngine via OSCQuery

    Interacts with Player objects via OSC

    It is responsible for:
      - Starting and stopping players
      - Monitoring player status
      - Restarting players
      - Updating player configurations
      - Handling player failures
      - Providing a clean interface for starting and stopping players
      - Providing a clean interface for monitoring player status

    """

    # Set in _setup_nng_command_callback; None until then (and in unit tests
    # that build the engine without its comms).
    _follow_relay = None

    # stop_project (869fhjdq5): after this node reports an authored end of the
    # show, how long it waits for the controller's STOP before it logs that
    # the show is still running. Log only; the node never stops itself.
    _STOP_PROJECT_WATCHDOG_S = 3.0
    _stop_project_watchdog: threading.Timer | None = None

    # GO anchor tolerance (Badajoz 2026-09-25 fix): how far the controller's
    # shipped GO instant may disagree with this node's own MTC before
    # _resolve_go_anchor distrusts it and falls back to a live local read.
    # Positive: the controller may be slightly AHEAD of this node's own read
    # (its go_script runs microseconds before this node computes the lag).
    # Negative: this node may lag the controller by up to 10s before the
    # value is treated as stale/misrouted rather than "node was just slow".
    _GO_ANCHOR_FUTURE_TOL_MS = 200.0
    _GO_ANCHOR_MAX_LAG_MS = 10_000.0

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._command_lock = threading.Lock()
        self._loading_lock = threading.Lock()
        self._loading = False
        self._project_generation: int = 0
        # Bumped on EVERY set_next_cue call, even reselecting the same cue.
        # Lets a PreArm background thread's should_continue notice a later
        # selection landed and stop advancing its own (now stale) walk —
        # 869f79ecc.
        self._selection_epoch: int = 0
        # with_cm=False (test shells) skips ConfigManager — no controller_ip/cm.
        if self.with_cm and hasattr(self, "cm") and hasattr(self, "controller_ip"):
            self.nng_hub_address = (
                f"tcp://{self.controller_ip}:{self.cm.node_conf['nng_hub_port']}"
            )
            # add_system_ports() / add_config_ports() moved to
            # BaseEngine.__init__ so the controller engine gets them too.
            self.deploy_manager = CuemsDeploy(
                library_path=self.cm.library_path,
                tmp_path=self.cm.tmp_path,
                controller_ip=self.controller_ip,  # set by set_config_manager()
            )
            PLAYER_HANDLER.add_media_folder(self.cm.library_path)
            PLAYER_HANDLER.set_player_endpoints_generator(
                self.add_player_endpoints,
                # TODO: Use node host from config
                prefix="/players",
            )
        else:
            self.nng_hub_address = None
            self.deploy_manager = None

    def start(self):
        CUE_HANDLER.set_nng_comms(self.nng_hub_address, self.cm.node_uuid)
        self.deploy_manager.loop = CUE_HANDLER.communications_thread.event_loop
        # Creates command dictionary and OSCQuery client
        self.set_oscquery_comms()
        # Creates player devices - must be before NNG callback
        self.set_players()
        # Set up NNG command receiving (after players ready)
        self._setup_nng_command_callback()
        if self.mtc_listener is not None:
            self.mtc_listener.start()
        super().start()

    def _setup_nng_command_callback(self):
        """
        Set up the callback for receiving commands via NNG from
        ControllerEngine.

        This provides push-based command delivery as an alternative to HTTP
        polling.
        Commands are received via the NNG bus and routed to the appropriate
        handlers.
        """
        if (
            hasattr(CUE_HANDLER, "communications_thread")
            and CUE_HANDLER.communications_thread
        ):
            CUE_HANDLER.communications_thread.set_command_callback(
                self._handle_nng_command
            )
            Logger.info("NNG command callback registered for NodeEngine")
            # Cross-node Auto follow (869fc8ytz): announce this node's
            # follows, apply the ones relayed from other nodes, in bus order.
            self._follow_relay = FollowRelay(
                CUE_HANDLER,
                node_id=self.cm.node_uuid,
                send=self._send_follow,
                get_script=lambda: self.script,
                get_mtc=lambda: getattr(self, "mtc_listener", None),
                get_project_gen=lambda: self._project_generation,
                first_local=self._first_local_enabled_in_go_chain,
            )
            CUE_HANDLER._follow = self._follow_relay
            CUE_HANDLER.communications_thread.set_receive_hook(
                self._on_command_received
            )
        else:
            Logger.warning(
                "CUE_HANDLER communications thread: "
                "not available for command callback"
            )

        from .cues.ActionHandler import ACTION_HANDLER

        ACTION_HANDLER.finalize_node_layer_bindings()
        ACTION_HANDLER.set_result_sink(self._action_result_sink)

    def _on_command_received(self, command_name: str, value) -> bool:
        """Receive hook: runs for every command in BUS order, before its
        command thread races the others for the command lock (869fc8ytz).

        A follow message is applied here and consumed. GO records the
        controller's run counter, STOP and load clear it and every pending
        follow -- here, so a follow relayed before a STOP can never be applied
        after it. Must not block."""
        relay = self._follow_relay
        if relay is None:
            return False
        if command_name == "follow":
            try:
                relay.on_message(value)
            except Exception as e:
                # Consumed anyway: a follow message must never reach the
                # command lock.
                Logger.error(f"Follow message failed: {e}")
            return True
        if command_name == "go":
            relay.begin_run(value)
        elif command_name in ("stop", "load"):
            relay.end_run(command_name)
            # The STOP (or load) a stop_project cue was waiting for.
            self._cancel_stop_project_watchdog()
        return False

    def _send_follow(self, data: dict) -> None:
        """Queue a follow message (announce or cancel) to the controller,
        which relays it to every node. Never waits."""
        from .comms.NodesHub import ActionType, NodeOperation, OperationType

        CUE_HANDLER.communications_thread.send_operation_nowait(
            NodeOperation(
                type=OperationType.STATUS,
                action=ActionType.UPDATE,
                sender=self.cm.node_uuid,
                target="follow",
                data=data,
            )
        )

    def _handle_nng_command(self, command_name: str, value, address: str = None):
        """Handle a command received via NNG from ControllerEngine.

        Args:
            command_name: The command name (e.g., 'go', 'load', 'stop',
            'player_control')
            value: The command value
            address: The original OSC address (optional)
        """
        Logger.info(f"NNG command received: {command_name} = {repr(value)}")

        if command_name == "player_control" and address:
            # Handle player control messages (mixer volumes, video controls,
            # etc.)
            self._handle_player_control_message(address, value)
        else:
            # Handle standard commands (go, load, stop)
            self.run_command(command_name, value)

    def _handle_player_control_message(self, address: str, value):
        """Handle player control messages received via NNG.

        Routes to appropriate player handlers based on the OSC address.
        Supports two formats:
        1. Engine format: /engine/players/<uuid>/<type>/...
        2. Direct format: /<uuid>/<type>/... (from UI)

        Args:
            address: The OSC address
            value: The value to set
        """
        parts = address.strip("/").split("/")

        # Determine format and extract node_uuid, player_type, path_parts
        if len(parts) >= 4 and parts[0] == "engine" and parts[1] == "players":
            # Engine format: /engine/players/<node_uuid>/<type>/...
            node_uuid = parts[2]
            player_type = parts[3]
            path_parts = parts[4:] if len(parts) > 4 else []
        elif len(parts) >= 2:
            # Direct format: /<node_uuid>/<type>/...
            node_uuid = parts[0]
            player_type = parts[1]
            path_parts = parts[2:] if len(parts) > 2 else []
        else:
            Logger.warning(f"Invalid player control address: {address}")
            return

        # Only handle messages for this node
        if node_uuid != self.cm.node_uuid:
            Logger.debug(f"Ignoring player message for other node: {node_uuid}")
            return

        Logger.debug(
            f"Handling player control: type={player_type}, path={path_parts},"
            f"value={value}"
        )

        # Route to appropriate handler based on player type
        if player_type == "video":
            redirect_video_cmd(path_parts, value)
        elif player_type == "audio":
            CUE_HANDLER.route_audio_message(path_parts, value)
        elif player_type == "dmx":
            CUE_HANDLER.route_dmx_message(path_parts, value)
        elif player_type == "audiomixer":
            # Legacy: pre-2026 OSC format /{uuid}/audiomixer/{channel}.
            # The current UI sends
            # /{uuid}/audio/mixer/{output_index}/{channel}/volume,
            # which routes via player_type == 'audio' →
            # CueHandler.route_audio_message().
            # Kept for backwards compatibility with any external tool still
            # using
            # the old format; new code should not target this path.
            # Direct audiomixer command: /<uuid>/audiomixer/<channel>
            # path_parts[0] is channel (e.g., '0', 'master')
            self._handle_audiomixer_command(path_parts, value)
        elif player_type == "jadeo":
            # Direct video command: /<uuid>/jadeo/<cmd>
            redirect_video_cmd(["jadeo"] + path_parts, value)
        else:
            Logger.debug(f"Unknown player type in control message: {player_type}")

    def _handle_audiomixer_command(self, path_parts: list, value):
        """Handle direct audiomixer OSC command.

        Args:
            path_parts: Remaining path parts after /<uuid>/audiomixer/
                       e.g., ['0'] for channel 0, ['master'] for master
            value: Volume value (0.0 to 1.0)
        """
        if not path_parts:
            Logger.warning("Empty audiomixer command path")
            return

        channel = path_parts[0]
        # jack-volume expects /audiomixer/<client_name>/<channel>
        mixer_cmd = f"/audiomixer/0_mixer/{channel}"

        try:
            PLAYER_HANDLER.get_audio_mixer_client().set_value(mixer_cmd, value)
            Logger.debug(f"Audiomixer command: {mixer_cmd} = {value}")
        except Exception as e:
            Logger.error(f"Error sending audiomixer command: {e}")

    @logged
    def stop(self):
        self.stop_requested = True
        self._cancel_stop_project_watchdog()
        self.stop_node_engine()
        self.stop_comms()
        super().stop()

    @logged
    def stop_comms(self):
        """Stop the NNG comms thread before the interpreter tears down.

        pynng registers an atexit hook that calls nng_fini(), destroying NNG's
        global state. CPython does not join daemon threads before running
        atexit hooks, so a comms thread still polling the bus at that point
        calls into freed NNG internals and aborts the process
        ("panic: pthread_mutex_lock: Invalid argument", ClickUp 869ed00ya).

        ControllerEngine does the equivalent in its own stop_comms(); without
        this, only the node engine core-dumped on every restart.
        """
        comms = getattr(CUE_HANDLER, "communications_thread", None)
        if comms is None:
            return
        try:
            comms.stop()
        except Exception as e:
            Logger.error(f"Error stopping NNG comms thread: {e}")

    def stop_node_engine(self):
        """Stop the NodeEngine elements"""
        CUE_HANDLER.disarm_all(reason="shutdown")
        self.stop_video_devs()

    def stop_video_devs(self):
        try:
            self.unload_video_devs()
            Logger.info("Video devs stopped")
        except Exception as e:
            Logger.warning(f"Exception raised when stopping video devs: {e}")

    def quit_video_devs(self):
        try:
            PLAYER_HANDLER.quit_videocomposer()
            Logger.info("Videocomposer quit successfully")
        except Exception as e:
            Logger.exception(e)

    def unload_video_devs(self):
        try:
            PLAYER_HANDLER.reset_videocomposer()
            Logger.info("Videocomposer reset successfully")
        except Exception as e:
            Logger.exception(e)

    #########################
    # OSCQuery logic
    #########################
    def add_player_endpoints(self, cue: Cue, prefix: str = "/players"):
        """Add player endpoints from a cue to the OSCQuery server

        Args:
            cue: The cue containing the player client
            prefix: Prefix to add to all endpoint paths (default: '/players')
        """
        if not hasattr(cue, "_osc") or cue._osc is None:
            Logger.warning(f"Cue {cue.id} has no OSC client, cannot add endpoints")
            return

        try:
            # Get endpoints from the player client
            endpoints = cue._osc.get_endpoints()
            if not endpoints:
                Logger.warning(f"No endpoints found for cue {cue.id}")
                return

            # Add prefix to all endpoints
            prefixed_endpoints = add_prefix_to_all(
                endpoints, f"{prefix}/{self.cm.node_uuid}/{cue.id}"
            )

            # Add endpoints to OSCQuery server
            if hasattr(self, "oscquery_server") and self.oscquery_server:
                self.oscquery_server.add_endpoints(prefixed_endpoints)
                Logger.debug(
                    f"Added {len(prefixed_endpoints)} endpoints for cue {cue.id}"
                )
            else:
                Logger.warning("OSCQuery server not initialized, cannot add endpoints")
        except Exception as e:
            Logger.error(f"Error adding player endpoints for cue {cue.id}: {e}")
            Logger.exception(e)

    def set_oscquery_comms(self):
        """Set up the command dictionary for the NodeEngine.

        Commands are received via NNG from ControllerEngine.
        OSCQuery client is no longer used since pyossia server was removed.
        """
        self.commands_dict = {
            "deploy": self.ready_project,
            "load": self.load_project,
            "loadcue": None,
            "go": self.go_script,
            "gocue": self.go_script,
            "pause": None,
            "resetall": None,
            "stop": self.stop_playback,
            "setnextcue": self.set_next_cue,
            "cue_enabled": self._handle_cue_enabled,
            "test": None,
            "unload": None,
            "update": None,
        }

    def route_message(self, parameter, value):
        # Exclude 'engine' common node
        path_elements = str(parameter.node).split("/")[2:]
        if path_elements[0] == "command":
            self.run_command(path_elements[1], value)
        elif path_elements[0] == "status":
            Logger.debug(f"Status update received: {path_elements[1]} = {repr(value)}")
        elif path_elements[0] == "players":
            # Exclude other nodes' players
            if path_elements[1] != self.cm.node_uuid:
                Logger.debug(
                    f"Ignoring player message for other node: {path_elements[1]}"
                )
                return
            # Route the message to the appropriate player handler
            if path_elements[2] == "video":
                redirect_video_cmd(path_elements[3:], value)
            if path_elements[2] == "audio":
                CUE_HANDLER.route_audio_message(path_elements[3:], value)
            if path_elements[2] == "dmx":
                CUE_HANDLER.route_dmx_message(path_elements[3:], value)
        else:
            Logger.debug(f"Recieved unused OSCQuery path: {str(parameter.node)}")
            return

    def run_command(self, command, value):
        with self._command_lock:
            Logger.debug(f"NodeEngine executing command: {command}({repr(value)})")
            if command in self.commands_dict.keys():
                handler = self.commands_dict[command]
                if handler is not None:
                    try:
                        handler(value)
                    finally:
                        if command in ("stop", "load") and self._follow_relay:
                            # The receive hook counted it received; until
                            # here this node's epochs still belonged to the
                            # run it ends (869fc8ytz).
                            self._follow_relay.stop_applied()
                    return True
                else:
                    Logger.warning(f"Command {command} has no handler")
                    return False
            else:
                Logger.error(f"Command {command} not found")
                return False

    #########################
    # Player logic
    #########################
    def set_players(self):
        # A player binary the orphan cleanups do not know by name escapes them
        # all; say so at start (869evtdf7).
        check_orphan_binary_names(self.cm.node_conf)
        self.set_video_players()
        self.set_audio_players()
        self.set_dmx_players()
        self.set_gradient_client()

    def set_gradient_client(self) -> None:
        """
        Wire GradientClient into PLAYER_HANDLER using settings from node_conf.
        """
        port = int(self.cm.node_conf["gradient_osc_port"])
        node_name = self._resolve_gradient_node_name()
        PLAYER_HANDLER.set_gradient_client(port=port, node_name=node_name)

    def _resolve_gradient_node_name(self) -> str:
        """
        Resolve the identifier gradient-motiond's node_name filter expects.

        gradient-motiond defaults its --node-name to the OS hostname (see
        node-identity-contract.md in cuems-common); cuems-nodeconf keeps the
        OS hostname in sync with network_map.xml's <role_id> and only
        persists <hostname> as a legacy override when the two diverge. The
        node's UUID is never what the daemon filters on.
        """
        node = self.cm.node_network_map or {}
        node_name = node.get("hostname") or node.get("role_id")
        if not node_name:
            node_name = self.cm.node_uuid
            Logger.warning(
                "gradient_node_name: no role_id/hostname in network_map for "
                f"node {self.cm.node_uuid}; falling back to node_uuid — "
                "gradient-motiond's node_name filter will likely reject fades"
            )
        return node_name

    # Audio functions
    def set_audio_players(self):
        """Set the audio players and audio mixer"""
        # Initialize the audio mixer for this node
        if self.cm.node_hw_outputs.get("audio_outputs"):
            audio_outputs = self.cm.node_hw_outputs["audio_outputs"]
            Logger.info(f"Initializing audio mixer with {len(audio_outputs)} outputs")

            mixer_id = "0"  # TODO: make this a unique identifier for the mixer
            # Start the audio mixer. The port draw is inside the try: it can
            # now raise if the pool is exhausted, and losing the mixer must
            # not abort the whole node start.
            # (assign_ports(cue=None) registers via add_config_ports itself —
            # the explicit second call that used to sit here was redundant.)
            try:
                mixer_ports = PORT_HANDLER.assign_ports(["audio_mixer"])
                PLAYER_HANDLER.start_audio_mixer(
                    audio_outputs=audio_outputs,
                    port=mixer_ports["audio_mixer"],
                    mixer_id=mixer_id,
                    path=self.cm.node_conf["audiomixer"]["path"],
                    args=self.cm.node_conf["audiomixer"]["args"],
                )
                Logger.info(f"Audio mixer started successfully for mixer {mixer_id}")
                # Register mixer with Controller via NNG
                try:
                    CUE_HANDLER.communications_thread.add_player(
                        f"audiomixer_{mixer_id}", None, timeout=0.1
                    )
                    Logger.info(f"Audio mixer {mixer_id} registered with Controller")
                except Exception as e:
                    Logger.warning(f"Could not register mixer with Controller: {e}")

                # Assert a known unity state at startup so the controller shadow
                # (hence the UI) reflects real gain from t=0, then snapshot it.
                try:
                    mixer_client = PLAYER_HANDLER.get_audio_mixer_client()
                    if mixer_client:
                        mixer_client.reset_volumes()
                    self._report_mixer_status()
                except Exception as e:
                    Logger.warning(f"Mixer startup state assert/report failed: {e}")
            except Exception as e:
                Logger.error(
                    f"Error starting audio mixer: {e}. "
                    f"Audio on this node will be SILENT."
                )
                Logger.exception(e)
        else:
            Logger.info(
                "No audio outputs detected, skipping audio mixer initialization"
            )

        # Build audio output lookup keyed by <id> (mirrors video output
        # pattern)
        audio_outputs = {}
        for port_type_dict in self.cm.node_mappings.get("audio", []):
            for port_type_list in port_type_dict.values():
                for port in port_type_list:
                    for _, output_data in port.items():
                        output_id = str(output_data.get("id", output_data["name"]))
                        mappings = output_data.get("mappings", [])
                        mapped_to = (
                            mappings[0]["mapped_to"]
                            if mappings
                            else output_data["name"]
                        )
                        audio_outputs[output_id] = {
                            "name": output_data["name"],
                            "mapped_to": mapped_to,
                        }
        PLAYER_HANDLER.set_audio_outputs(audio_outputs)

        # Set the audio player generator. Append --output-latency-ms
        # from settings.xml when the operator supplied an integer
        # override (isinstance int); "auto" or absent ⇒ audioplayer
        # runs its Phase-3 JACK-latency query path.
        audio_args = _append_output_latency_flag(
            self.cm.node_conf["audioplayer"]["args"],
            self.cm.node_conf["audioplayer"],
        )
        PLAYER_HANDLER.set_audio_output_generator(
            self.cm.node_conf["audioplayer"]["path"],
            audio_args,
        )

    def _report_mixer_status(self, output_index: str = "0") -> None:
        """Snapshot the local mixer's authoritative gain to the controller.

        Sent on SETTLE EVENTS only — mixer startup and after each
        reset_volumes — never per UI write. A per-write report would race the
        controller's synchronous optimistic populate (unordered NNG command
        threads) and make faders jump backward on a fast drag; live UI writes
        are already shadowed accurately controller-side, so the node only needs
        to report the resets the controller can't observe.
        """
        try:
            mixer_client = PLAYER_HANDLER.get_audio_mixer_client()
            if not mixer_client:
                return
            entries = mixer_client.snapshot()
            if not entries:
                return
            CUE_HANDLER.communications_thread.update_mixer_status(
                entries, output_index=output_index, timeout=0.1
            )
            Logger.debug(f"Reported mixer status to controller: {entries}")
        except Exception as e:
            Logger.warning(f"Could not report mixer status: {e}")

    # Video functions
    def set_video_players(self):
        """Set the video players"""
        Logger.info(f'Setting video players with: {self.cm.node_conf["videoplayer"]}')
        if not self.cm.node_hw_outputs["video_outputs"]:
            Logger.info("No video outputs detected.")
            return

        PLAYER_HANDLER.add_node_uuid(self.cm.node_uuid)
        vc_conf = self.cm.node_conf.get("videoplayer", {})
        osc_video_port = int(vc_conf.get("osc_port", VIDEOCOMPOSER_OSC_PORT_DEFAULT))
        # Declare BEFORE building the client: the client draws its own local
        # OSC port from the pool, and must not be handed the videocomposer's.
        # This is also the only registration of this port — it comes from the
        # nested videoplayer/osc_port key, which get_config_ports() (top-level
        # only) cannot see, so BaseEngine's hoisted call misses it.
        PORT_HANDLER.add_config_ports({"videocomposer": osc_video_port})
        try:
            PLAYER_HANDLER.set_video_client(osc_video_port)
        except Exception as e:
            Logger.error(
                f"Could not create the videocomposer OSC client: {e}. "
                f"Video on this node will not respond."
            )
            Logger.exception(e)
            return

        # Canvas geometry comes from /run/cuems/display.conf, written by
        # cuems-generate-display-conf (videocomposer's ExecStartPre). It's the
        # same file the videocomposer reads, so engine + VC agree on canvas
        # size and per-output regions without a handshake. The XML's optional
        # <canvas_region> is a UI-template hint (normalized [0,1]) and is
        # ignored here — engine never sources physical layout from XML.
        display_regions, (canvas_w, canvas_h) = read_display_conf()

        video_outputs = {}
        for port_type_dict in self.cm.node_mappings.get("video", []):
            for port_type_list in port_type_dict.values():
                for port in port_type_list:
                    for _, output_data in port.items():
                        output_id = str(output_data.get("id", output_data["name"]))
                        name = output_data["name"]
                        mappings = output_data.get("mappings", [])
                        mapped_to = mappings[0]["mapped_to"] if mappings else name
                        region = display_regions.get(mapped_to)
                        if region is None:
                            Logger.warning(
                                f"DISPLAY_MISMATCH: XML output id={output_id} "
                                f"name={name!r} maps to {mapped_to!r} which"
                                f"is "
                                f"not in display.conf; skipping. Available: "
                                f"{sorted(display_regions.keys())}"
                            )
                            continue
                        video_outputs[output_id] = {
                            "name": name,
                            "mapped_to": mapped_to,
                            "x": region["x"],
                            "y": region["y"],
                            "width": region["width"],
                            "height": region["height"],
                            "canvas_region": dict(region),
                        }
        PLAYER_HANDLER.start_video_outputs(
            video_outputs, canvas_override=(canvas_w, canvas_h)
        )

    # DMX functions
    def set_dmx_players(self):
        """Set the DMX player for this node and register its endpoints."""
        # Get node UUID for player naming
        node_uuid = self.cm.node_conf.get("uuid", "default_node")

        # Start the DMX player. The port draw is inside the try: it can now
        # raise if the pool is exhausted, and losing DMX must not abort the
        # whole node start. (assign_ports(cue=None) registers via
        # add_config_ports itself — the second call here was redundant.)
        try:
            dmx_ports = PORT_HANDLER.assign_ports(["dmx_player"])
            # Append --output-latency-ms from settings.xml when an
            # integer override is present. Dmx has no "auto" form —
            # absent ⇒ dmxplayer's 35 ms Phase-5A default stands.
            dmx_args = _append_output_latency_flag(
                self.cm.node_conf["dmxplayer"]["args"],
                self.cm.node_conf["dmxplayer"],
            )
            PLAYER_HANDLER.start_dmx_player(
                port=dmx_ports["dmx_player"],
                node_uuid=node_uuid,
                path=self.cm.node_conf["dmxplayer"]["path"],
                args=dmx_args,
            )
            try:
                CUE_HANDLER.communications_thread.add_player(
                    f"dmxplayer_{node_uuid}", None, timeout=0.1
                )
            except Exception:
                pass  # Ignore - NNG is for distributed nodes
            Logger.info(f"DMX player started successfully for node {node_uuid}")
        except Exception as e:
            Logger.error(f"Error starting DMX player: {e}")
            Logger.exception(e)
            return

    def quit_dmx_devs(self):
        """Quit the DMX player if it exists"""
        dmx_client = PLAYER_HANDLER.get_dmx_player_client()
        if dmx_client:
            try:
                dmx_client.set_value("/quit", 1)
            except Exception as e:
                Logger.exception(e)
        CUE_HANDLER.communications_thread.remove_player(
            f"dmxplayer_{self.cm.node_uuid}"
        )

    #########################
    # Project logic
    #########################
    def ready_project(self, project):
        """Prepare the project to be played.

        deploy_project() runs in _load_project_inner BEFORE this point,
        before the teardown of the previous project — so that a deploy
        failure aborts the load without destroying running state.
        Media deploy runs here because it is best-effort: a failure
        logs but does not abort (cached media may already be on disk).
        """
        self.cm.load_project_config(project)
        self.read_script(project)
        media_synced = self.deploy_media(project)
        self.check_own_media_sizes(media_synced)
        self.ensure_video_indexes()
        self.outputs_map = self.map_cue_outputs()
        PLAYER_HANDLER.set_outputs_map(self.outputs_map)
        PORT_HANDLER.clean_random_ports()

    def map_cue_outputs(self, cuelist: CueList = None):
        """Load the output mappings for the project"""
        outputs_map = {}
        if cuelist is None:
            cuelist = self.script.cuelist
        if cuelist.contents is None:
            Logger.warning("Cuelist contents is None, skipping map_cue_outputs")
            return outputs_map
        for cue in cuelist.contents:
            if isinstance(cue, CueList):
                outputs_map.update(self.map_cue_outputs(cue))
                continue
            elif not isinstance(cue, MediaCue):
                continue

            outputs = [
                x[1] for x in cue.get_all_output_names() if x[0] == self.cm.node_uuid
            ]
            if outputs:
                outputs_map[cue.id] = outputs
        Logger.debug(f"Outputs map: {outputs_map}")
        return outputs_map

    def load_project(self, project):
        """Load the project files to the node"""
        with self._loading_lock:
            if self._loading:
                Logger.warning(
                    f"Load already in progress, ignoring duplicate load of {project}"
                )
                return
            self._loading = True

        try:
            return self._load_project_inner(project)
        finally:
            with self._loading_lock:
                self._loading = False

    def _load_project_inner(self, project):
        # Don't allow loading while script is running
        if self.get_status("running") == "yes":
            Logger.warning(
                f"Cannot load project {project} while script is running. "
                f"Stop the script first."
            )
            return False

        # Deploy the critical project files (script.xml, mappings.xml,
        # settings.xml) BEFORE tearing down the previous project. If the
        # controller is unreachable we abort here with the previous
        # project still armed and usable — better than ending up with
        # everything stopped and no new project loaded.
        if not self.deploy_project(project):
            Logger.error(
                f"Project deploy FAILED for {project} — aborting load; "
                f"previous project remains unchanged"
            )
            return False

        self._gradient_cancel_all("project load")

        # Stop any running cue threads from the previous project first,
        # so they can't interfere with cleanup (same logic as stop_playback).
        CUE_HANDLER.stop_all_cues()

        # DMX: stop following MTC, blackout all universes.
        self._dmx_reset()

        # Video: reset videocomposer (remove all layers, cancel loads, reset
        # master).
        self.unload_video_devs()

        # Audio: reset mixer volumes, kill all players, clean up JACK.
        mixer_client = PLAYER_HANDLER.get_audio_mixer_client()
        if mixer_client:
            try:
                mixer_client.reset_volumes()
            except Exception as e:
                Logger.warning(f"JACK volume reset failed: {e}")
            # Report the reset so the UI reflects real (unity) gain, not the
            # stale value it may have shown from a previous UI write.
            self._report_mixer_status()
        PLAYER_HANDLER.kill_all_audio_players()
        PLAYER_HANDLER.kill_orphaned_audio_processes()
        PLAYER_HANDLER.cleanup_zombie_jack_clients()

        # Disarm all cues from the previous project.
        CUE_HANDLER.disarm_all(reason="load")

        # Clear the engine's 24h MTC wrap accumulator on this project transition,
        # co-orchestrated with the DMX blackout + videocomposer reset above (which
        # clear the C++ receivers' offsets via resetWrapOffset). A graceful
        # reload's small backward wire delta can't trip the wire-driven reset, so
        # do it explicitly here — else a project loaded after a >24h run starts
        # at hour 24+ and the reset_callback cascade stays silenced. (Plan 4)
        if self.mtc_listener is not None:
            self.mtc_listener.reset_24h_state()

        # Obtain the project files (this replaces self.script with new project)
        self.ready_project(project)

        # Name the video cues aimed at outputs this node doesn't drive, before
        # anything arms. Here and not in ready_project: that one is also the
        # `deploy` bus command, which can arrive mid-show.
        self.check_unrouted_video_outputs()

        # Prepare the script to be played (arms new cues)
        self.ready_script(reason="load")

        # Start cue dependencies
        # self.set_players()

        # Confirm the project is loaded
        self.set_show_lock_file()
        self.script.unix_name = project
        self.set_status("load", project)
        Logger.info(f"Project {project} loaded")

        # Notify Controller that arming is complete (GO button can go green)
        self._notify_armed_ready("arming after load")

        # Broadcast initial nextcue to UI
        self._broadcast_nextcue()

        return True

    def deploy_project(self, project):
        """Deploy the project files (script.xml, mappings.xml, settings.xml).

        Critical path: if these fail to sync, the local copy may be stale
        and arming cues against it is unsafe. Caller is expected to abort
        the load on False.
        """
        return self.deploy_manager.sync_files(project, "project")

    def deploy_media(self, project):
        """Deploy the media files (and their .idx sidecar indexes).

        Best-effort: a failure here is recoverable if media is already
        cached on disk. Returns False to surface the failure to logs,
        but the caller continues the load.
        """
        if not self.script:
            Logger.error("No script loaded")
            return False
        bare_names = self.script.get_own_media_filenames(config=self.cm)
        if len(bare_names) == 0:
            Logger.info("No media files to deploy")
            return True
        if not self.deploy_manager.sync_files(project, "media", bare_names):
            Logger.error(
                f"Media deploy failed for {project} — continuing with cached "
                f"files; cues whose media is missing locally will fail on GO"
            )
            return False
        return True

    def check_own_media_sizes(self, media_synced: bool = True) -> list:
        """Report this node's media files that differ from the project, or
        are missing (869fat84r D20, L2).

        The editor corrects the project before a load when a file on the
        controller changed; this is the last defence, for a copy that differs
        only here (a failed media sync, a file placed on this node only, an
        OSC load that bypassed the editor). Report only: the cue keeps the
        stored duration, so this node stays in step with the others. Never
        raises: the previous project is already torn down at this point.
        """
        try:
            found = own_media_size_mismatches(
                self.script.get_own_media(config=self.cm),
                self.script.find,
                PLAYER_HANDLER.media_path,
            )
        except Exception as e:
            Logger.error(
                f"Media size check skipped ({type(e).__name__}: {e}); the load goes on"
            )
            return []
        sync = "" if media_synced else " This load's media sync failed."
        for file_name, stored, actual in found:
            if actual is None:
                Logger.error(
                    f"Media file {file_name} is missing on this node: its cues will "
                    f"fail.{sync}"
                )
            else:
                Logger.error(
                    f"Media file {file_name} on this node is {actual} bytes; the "
                    f"project stores {stored}: it was replaced or not synced. The "
                    "stored duration is kept so the nodes stay in step; re-sync it, "
                    f"or re-upload it in the editor.{sync}"
                )
        return found

    def check_unrouted_video_outputs(self) -> list:
        """Name every enabled video cue of this node that points at an output
        this node doesn't drive (869fgq3xv). Log only.

        Such an output is never loaded (arm_videoCue), so the cue stays dark
        there; this says so at load, before the show, instead of on the
        screens. Also starts the load's once-only arm errors afresh. Uses the
        same PlayerHandler.partition_cue_outputs as the arm, so the two cannot
        disagree. Never raises: the previous project is already torn down.
        """
        PLAYER_HANDLER.reset_unrouted_warnings()
        found = []
        try:
            video_cues = []
            for cue_id, output_names in (self.outputs_map or {}).items():
                cue = self.script.find(cue_id)
                # outputs_map holds every MediaCue of this node, audio and
                # disabled cues included.
                if isinstance(cue, VideoCue) and getattr(cue, "enabled", True):
                    video_cues.append((cue, output_names))
            if not video_cues:
                return found

            if PLAYER_HANDLER.get_video_client() is None:
                # set_video_players returned early: every video cue fails at
                # arm ("No video client"), for a reason that is not the cue's.
                Logger.warning(
                    f"no video stack on this node: {len(video_cues)} video cue(s)"
                    " will not play here"
                )
                return found

            driven = PLAYER_HANDLER.driven_video_outputs()
            for cue, output_names in video_cues:
                _, unrouted = PLAYER_HANDLER.partition_cue_outputs(cue, output_names)
                if not unrouted:
                    continue
                found.append(
                    {
                        "cue_id": cue.id,
                        "cue_name": getattr(cue, "name", ""),
                        "outputs": [
                            {"output": name, "reason": reason}
                            for name, reason in unrouted
                        ],
                    }
                )
                cue_name = getattr(cue, "name", "")
                for output_name, reason in unrouted:
                    Logger.warning(
                        f'UNROUTED_VIDEO_OUTPUT: cue {cue.id} "{cue_name}"'
                        f' → output "{output_name}" not driven by this node ({reason});'
                        f" it will stay dark. Driven outputs: {driven}"
                    )
            if found:
                Logger.warning(
                    f"{len(found)} video cue(s) point at outputs this node doesn't"
                    " drive; see UNROUTED_VIDEO_OUTPUT (and DISPLAY_MISMATCH at"
                    " node start)"
                )
        except Exception as e:
            Logger.exception(
                f"Unrouted video output check skipped ({type(e).__name__}: {e});"
                " the load goes on"
            )
            return []
        return found

    def ensure_video_indexes(self):
        """
        Run cuems-videoindexer on any video files that are missing a .idx
        sidecar.

        This is a safety net for files that were copied manually or deployed to
        a
        node that never ran the editor upload hook. For normally-uploaded files
        the
        index was already created by the editor and this is a no-op.
        """
        if not self.script:
            return
        file_names = self.script.get_own_media_filenames(config=self.cm)
        video_exts = {".mp4", ".mov", ".avi", ".mkv", ".mpg"}
        unindexed = []
        for name in file_names:
            ext = os.path.splitext(name)[1].lower()
            if ext not in video_exts:
                continue
            full_path = PLAYER_HANDLER.media_path(name)
            idx_dir = os.path.join(os.path.dirname(full_path), "indexes")
            idx_path = os.path.join(idx_dir, os.path.basename(full_path) + ".idx")
            if not os.path.exists(idx_path):
                unindexed.append(full_path)
        if unindexed:
            Logger.info(
                f"ensure_video_indexes: indexing {len(unindexed)} video(s)"
                f"missing .idx: "
                f"{[os.path.basename(p) for p in unindexed]}"
            )
            try:
                result = subprocess.run(
                    ["cuems-videoindexer"] + unindexed,
                    timeout=600,
                    capture_output=True,
                    text=True,
                )
            except Exception as e:
                Logger.warning(f"ensure_video_indexes: indexer failed: {e}")
                return
            if result.returncode == 0:
                Logger.info("ensure_video_indexes: indexer ok (rc=0)")
                if result.stdout:
                    Logger.debug(
                        f"ensure_video_indexes stdout: {result.stdout.strip()}"
                    )
            else:
                Logger.warning(
                    "ensure_video_indexes: indexer returned"
                    f"rc={result.returncode}. "
                    f"stdout={result.stdout.strip()!r}"
                    f"stderr={result.stderr.strip()!r}"
                )

    #########################
    # Nextcue
    #########################
    def _broadcast_nextcue(self):
        """Send the current next_cue_pointer UUID to the Controller via NNG."""
        cue_id = self.next_cue_pointer.id if self.next_cue_pointer else ""
        try:
            CUE_HANDLER.communications_thread.update_nextcue(cue_id, timeout=0.1)
            Logger.debug(f'Broadcast nextcue: {cue_id or "(none)"}')
        except Exception as e:
            Logger.warning(f"Could not broadcast nextcue: {e}")

    def _arm_with_enabled_guard(self, cue, project_gen: int, arm_epoch: int = None):
        """
        Arm a cue, and disarm it if it was disabled while arming.

        Runs in a daemon thread (ReArm:<id>). `arm_epoch` is the STOP/load
        epoch captured by whoever spawned it (_apply_cue_enabled_side_effects):
        the arm belongs to it, so CUE_HANDLER.arm() refuses it if a STOP or a
        load happened before this thread got to run, and abandons it -- never
        publishing the cue, releasing the player it spawned -- if one happens
        while it is arming.

        This method therefore does NOT disarm on a project change any more
        (869f9wqpn). It used to, after arm() had already published the cue:
        by then the restarted run could be playing it, and after a reload of
        the same project the disarm hit the NEW object's player, because Cue
        equality is by id.
        """
        if self._project_generation != project_gen:
            Logger.info(f"Aborting arm of {cue.id} — project generation changed")
            return
        try:
            CUE_HANDLER.arm(cue, init=True, epoch=arm_epoch)
        except Exception as exc:
            Logger.error(f"Async re-arm of enabled cue {cue.id} failed: {exc}")
            return
        if self._project_generation != project_gen:
            # A STOP/load landed during the arm. arm() either abandoned it or
            # published it before the STOP began, in which case the STOP's
            # own disarm_all took it. Nothing to undo, nothing to rejoin.
            Logger.info(
                f"Project changed during the async arm of {cue.id} — left to "
                "the STOP/load that caused it"
            )
            return
        # If cue was disabled while we were arming, disarm now.
        if not cue.enabled and CUE_HANDLER.find_armed_cue(cue):
            CUE_HANDLER.disarm(cue, reason="disabled")
            Logger.info(f"Disarmed cue {cue.id} — disabled during async arm")
            return
        if not CUE_HANDLER.find_armed_cue(cue):
            # arm() refused it (stale request) or failed. rejoin_chain would
            # only run go()'s fallback: another wait, another arm.
            return
        # Armed and still enabled: if it belongs to a chain that is running
        # right now, put it back at its own slot. Checked here rather than
        # before the arm because arm() can take ~15 s — long enough for the
        # slot to pass, which rejoin_chain re-tests against live MTC.
        self._rejoin_running_chain(cue)

    def _action_result_sink(self, outcome: dict):
        """
        Custom result sink for ActionHandler — extends default with cue_enabled
        sync.
        """
        from .cues.ActionHandler import ACTION_HANDLER

        # Always run default behavior (sends action_cue_outcome via NNG)
        ACTION_HANDLER._default_result_sink(outcome)

        # If an enable/disable action was applied, notify Controller and apply
        # the local arm/disarm side effects (pre-arm the enabled cue so its GO
        # doesn't hit the not-loaded fallback re-arm; disarm a disabled idle
        # cue). 'applied_no_change' outcomes are deliberately excluded →
        # idempotent under rapid toggles.
        action_type = outcome.get("action_type")
        status = outcome.get("status")
        if action_type == "stop_project" and status == "applied":
            # The authored end of the show (869fhjdq5): tell the controller,
            # which runs its own STOP path. Nothing is stopped locally.
            self._notify_stop_project(outcome.get("origin_cue_id"))
            return
        if action_type in ("enable", "disable") and status == "applied":
            target_id = outcome.get("target_id")
            if target_id:
                enabled = action_type == "enable"
                # Notify FIRST: the Controller cache/UI update must never be
                # starved by a side-effect failure (an exception here would be
                # swallowed by _emit_outcome's blanket try/except).
                self._notify_cue_enabled(target_id, enabled)
                script = self.script
                if script is None:
                    # Project unloading concurrently — nothing to arm/disarm;
                    # in-flight ReArm threads self-cancel on generation change.
                    return
                cue = script.find(target_id)
                if cue is None:
                    Logger.warning(
                        f"cue_enabled side effects: cue {target_id} not found in script"
                    )
                    return
                try:
                    self._apply_cue_enabled_side_effects(cue, enabled)
                except Exception as exc:
                    Logger.error(
                        f"cue_enabled side effects failed for {target_id}: {exc}"
                    )

    def _gradient_cancel_all(self, context: str):
        """Cancel all in-flight gradient fades.

        Shared by _load_project_inner (project load) and stop_playback
        (STOP) — both must flush queued fades before stopping cue threads,
        so gradient OSC traffic never outlives the cues that started it.
        """
        gradient_client = PLAYER_HANDLER.get_gradient_client()
        if gradient_client:
            try:
                gradient_client.send_cancel_all()
            except Exception as exc:
                Logger.error(f"gradient send_cancel_all failed on {context}: {exc}")
        else:
            Logger.debug(
                f"gradient_client not initialised, skipping cancel_all on {context}"
            )

    def _dmx_reset(self):
        """Stop DMX MTC following and blackout all universes.

        Shared by _load_project_inner and stop_playback: both need the DMX
        playhead frozen and outputs zeroed before video/audio teardown.
        """
        dmx_client = PLAYER_HANDLER.get_dmx_player_client()
        if dmx_client:
            try:
                dmx_client.disable_mtcfollow()
            except Exception as e:
                Logger.warning(f"DMX disable mtcfollow failed: {e}")
            try:
                dmx_client.send_blackout()
            except Exception as e:
                Logger.warning(f"DMX blackout failed: {e}")

    def _notify_armed_ready(self, context: str):
        """Notify Controller that this node finished (re)arming, so the GO
        button can go green.

        Shared by _load_project_inner (arming after load) and stop_playback
        (re-arm after STOP).
        """
        from .comms.NodesHub import ActionType, NodeOperation, OperationType

        try:
            operation = NodeOperation(
                type=OperationType.STATUS,
                action=ActionType.UPDATE,
                sender=self.cm.node_uuid,
                target="armed_ready",
                data={"armed": "yes"},
            )
            CUE_HANDLER.communications_thread.send_operation(operation, timeout=0.1)
            Logger.debug(f"Notified Controller that {context} is complete")
        except Exception as e:
            Logger.warning(f"Could not notify Controller of armed_ready: {e}")

    def _notify_cue_enabled(self, cue_id: str, enabled: bool):
        """Send cue enabled status to Controller via NNG."""
        from .comms.NodesHub import ActionType, NodeOperation, OperationType

        try:
            operation = NodeOperation(
                type=OperationType.STATUS,
                action=ActionType.UPDATE,
                sender=(
                    self.cm.node_uuid if hasattr(self, "cm") and self.cm else "node"
                ),
                target="cue_enabled",
                data={"cue_id": cue_id, "enabled": enabled},
            )
            CUE_HANDLER.communications_thread.send_operation(operation, timeout=0.1)
        except Exception as e:
            Logger.warning(f"Could not notify cue_enabled: {e}")

    def _notify_stop_project(self, cue_id):
        """A stop_project ActionCue was revealed on this node (869fhjdq5).

        Send the controller a STATUS op carrying the cue and the current run;
        the controller validates it like a follow message (adopted sender,
        cue in project, running, current run_seq) and runs stop_script, which
        is the operator STOP. Every node reveals every ActionCue, so the
        controller gets one message per node and acts on the first; the rest
        are dropped there at INFO.

        Nothing is stopped here: one stop path, the controller's. If no STOP
        or load is received within _STOP_PROJECT_WATCHDOG_S the watchdog
        logs a WARNING and the show keeps running, which is the safe side.
        Skipped (INFO) when a STOP or load is already pending on this node, or
        when no run is in progress (run_seq is None between a STOP/load and
        the next GO): both mean the show is ending or ended anyway.
        """
        from .comms.NodesHub import ActionType, NodeOperation, OperationType

        relay = self._follow_relay
        if relay is None:
            Logger.warning(
                f"stop_project cue {cue_id}: no comms relay on this node, cannot "
                "notify the controller; the show keeps running"
            )
            return
        if relay.stop_pending:
            Logger.info(
                f"stop_project cue {cue_id}: a STOP or load is already pending on "
                "this node, not sending"
            )
            return
        run_seq = relay.run_seq
        if run_seq is None:
            Logger.info(
                f"stop_project cue {cue_id}: no run in progress (no GO since the "
                "last STOP or load), not sending"
            )
            return
        stops_at_send = relay.stops_received
        try:
            CUE_HANDLER.communications_thread.send_operation_nowait(
                NodeOperation(
                    type=OperationType.STATUS,
                    action=ActionType.UPDATE,
                    sender=self.cm.node_uuid,
                    target="stop_project",
                    data={"cue": str(cue_id), "run_seq": run_seq},
                )
            )
        except Exception as e:
            Logger.error(
                f"stop_project cue {cue_id}: could not notify the controller "
                f"({e}); the show keeps running"
            )
            return
        Logger.info(
            f"stop_project cue {cue_id}: notified the controller (run {run_seq})"
        )
        self._arm_stop_project_watchdog(cue_id, stops_at_send)

    def _arm_stop_project_watchdog(self, cue_id, stops_at_send: int) -> None:
        self._cancel_stop_project_watchdog()
        timer = threading.Timer(
            self._STOP_PROJECT_WATCHDOG_S,
            self._on_stop_project_watchdog_fire,
            args=(cue_id, stops_at_send),
        )
        timer.daemon = True
        self._stop_project_watchdog = timer
        timer.start()

    def _cancel_stop_project_watchdog(self) -> None:
        timer = self._stop_project_watchdog
        if timer is not None:
            timer.cancel()
            self._stop_project_watchdog = None

    def _on_stop_project_watchdog_fire(self, cue_id, stops_at_send: int) -> None:
        """Compares STOP/load RECEIPTS, not executions: a STOP queued behind a
        long command has still been received, and an operator STOP that got in
        first counts too. Log only."""
        relay = self._follow_relay
        if relay is not None and relay.stops_received > stops_at_send:
            return
        Logger.warning(
            f"stop_project cue {cue_id} sent, no STOP received from the "
            f"controller in {self._STOP_PROJECT_WATCHDOG_S:.0f} s; the show "
            "keeps running"
        )

    def set_next_cue(self, value):
        """
        Handle setnextcue command from the UI — override next_cue_pointer.

        Arms the SELECTED cue synchronously (a GO right after this needs it
        immediately), but extends the arm-ahead window in a background
        PreArm:<id> thread. Badajoz, 2026-09-25 (869f79ecc): that extension
        used to run here, synchronously, inside run_command's
        _command_lock. A cold-boot pre-arm took ~500ms; a GO sent right
        after (UI, power-bridge /gocue) waited for the lock, executed
        late, and read its own MTC ~520ms later than the other nodes —
        anchoring its whole Auto-continue chain late. See go_script /
        _resolve_go_anchor for the anchor half of that fix; this is the
        half that stops the delay from happening in the first place.
        """
        if not self.script:
            Logger.warning("No script loaded, cannot set next cue.")
            return
        cue = self.script.find(value)
        if cue:
            self.next_cue_pointer = cue
            self._selection_epoch += 1
            selection_epoch = self._selection_epoch
            project_gen = self._project_generation
            # Captured HERE, on the command thread: the lookahead belongs to
            # the STOP/load epoch in which it was asked for.
            arm_epoch = CUE_HANDLER.arm_epoch()
            if not CUE_HANDLER.find_armed_cue(cue):
                Logger.info(f"Re-arming cue {cue.id} selected as next cue")
                CUE_HANDLER.arm(cue, init=True)
            threading.Thread(
                target=self._prearm_thread,
                args=(
                    self._prearm_lookahead,
                    cue,
                    project_gen,
                    selection_epoch,
                    arm_epoch,
                ),
                daemon=True,
                name=f"PreArm:{cue.id}",
            ).start()
            self._broadcast_nextcue()
            Logger.info(f"Next cue overridden by UI: {value}")
        else:
            Logger.warning(f"setnextcue: cue {value} not found in script")

    def _prearm_thread(self, body, cue, *args):
        """Body of every PreArm:<id> thread: run `body(cue, *args)`, then log
        what the node holds armed -- also when the walk was abandoned or
        failed. Measurement only."""
        try:
            body(cue, *args)
        finally:
            CUE_HANDLER.log_armed_inventory(f"PreArm:{cue.id}")

    def _prearm_lookahead(self, cue, project_gen, selection_epoch, arm_epoch):
        """Background half of set_next_cue — see its docstring.

        should_continue re-checks BOTH the project generation (a STOP/load
        bumps it, same guard as _arm_with_enabled_guard) and this
        selection's own epoch (a later set_next_cue call bumps it even if
        it reselects the same cue), so a stale walk stops advancing as
        soon as either happens, without waiting for it to finish.

        arm_epoch is the STOP/load epoch captured by the command that spawned
        this thread. Every arm the walk makes belongs to it: CUE_HANDLER
        refuses or abandons an arm whose epoch has moved, so a cue this walk
        was arming across a STOP or a load is never published. That is why
        this thread disarms NOTHING afterwards (869f9wqpn) -- it used to, and
        by then the restarted run could already be playing the cue.

        Audio is pre-armed here like any other local cue. It was excluded
        at first (skip_arming_types=(AudioCue,)) for fear of a slow JACK
        port wait racing a GO for the same cue; measured instead
        (2026-09-25): the port appears 0-0.6 s after the engine starts
        waiting (Badajoz, taller, test2 node01), and a fully concurrent
        spawn of 16 audioplayers takes ~4.2 s -- while this walk adds at
        most one extra concurrent arm. Excluding audio cost ~320 ms of late
        audio on every run where audio follows local cues. If that ever has
        to be revisited, skip_arming_types is still the lever.
        """

        def should_continue():
            return (
                self._project_generation == project_gen
                and self._selection_epoch == selection_epoch
            )

        CUE_HANDLER._arm_ahead(
            cue, should_continue=should_continue, arm_epoch=arm_epoch
        )

    def _prearm_after_advance(self):
        """Pre-arm this node's next segment after a GO advanced the pointer
        with nothing local to play.

        Target = what the NEXT GO will dispatch on this node:
        `_first_local_enabled_in_go_chain(next_cue_pointer)` -- None when
        that GO is another node's too, in which case nothing is armed yet
        (the next advance re-checks), so this never preloads a cue that is
        many GOs away. The target AND its lookahead are armed on the same
        PreArm:<id> background thread set_next_cue uses, with the same
        project-generation / selection-epoch abort guards (a STOP, a load or
        a later selection abandons it).

        Nothing is armed here, synchronously: this runs inside run_command's
        _command_lock, and an audio target spawns a player and waits for its
        JACK ports -- a STOP or the next GO sent meanwhile would wait for the
        whole arm (869f79ecc removed exactly that from set_next_cue). A GO
        that lands mid-arm is safe: arm() makes it wait on the in-progress
        arm instead of arming the cue twice.

        What IS done here, on the command thread, is capturing the STOP/load
        epoch the arm belongs to. A STOP between this GO and the thread's
        first step must find the arm stale, not current.
        """
        target = self._first_local_enabled_in_go_chain(self.next_cue_pointer)
        if target is None:
            return
        self._selection_epoch += 1
        Logger.info(
            "Pointer advanced with no local cue; pre-arming next local cue "
            f"{target.id} ({type(target).__name__}) + lookahead"
        )
        threading.Thread(
            target=self._prearm_thread,
            args=(
                self._prearm_segment,
                target,
                self._project_generation,
                self._selection_epoch,
                CUE_HANDLER.arm_epoch(),
            ),
            daemon=True,
            name=f"PreArm:{target.id}",
        ).start()

    def _prearm_segment(self, target, project_gen, selection_epoch, arm_epoch):
        """PreArm:<id> body for _prearm_after_advance: arm the segment's
        first cue, then the lookahead from it.

        It never disarms. If a STOP or a load lands while the target is
        arming, CUE_HANDLER.arm() abandons that arm itself: the cue is never
        published, so the restarted run's GO arms it fresh and nothing can
        cut it afterwards (869f9wqpn; see _prearm_lookahead).
        """

        def should_continue():
            return (
                self._project_generation == project_gen
                and self._selection_epoch == selection_epoch
            )

        if not should_continue():
            return
        if not CUE_HANDLER.find_armed_cue(target):
            # A walk, so arm()'s own post_go / action-target recursion obeys
            # the same guards and the same epoch (869f79ecc).
            walk = _ArmWalk(should_continue=should_continue, epoch=arm_epoch)
            CUE_HANDLER.arm(target, init=True, walk=walk)
        # The epoch moves when a STOP BEGINS (stop_all_cues), before the
        # generation does (ready_script): check it too, or the walk would
        # start inside the teardown.
        if not should_continue() or CUE_HANDLER.arm_epoch() != arm_epoch:
            return
        self._prearm_lookahead(target, project_gen, selection_epoch, arm_epoch)

    def _handle_cue_enabled(self, value):
        """Handle cue_enabled toggle from Controller.

        Value format: "<cue_id> <0|1>" (space-separated UUID and enabled flag).
        """
        if not self.script:
            Logger.warning("No script loaded, cannot toggle cue enabled")
            return

        if not value or not isinstance(value, str):
            Logger.warning(f"Invalid cue_enabled value: {repr(value)}")
            return

        parts = value.split(" ", 1)
        if len(parts) != 2 or parts[1] not in ("0", "1"):
            Logger.warning(f"Invalid cue_enabled format: {repr(value)}")
            return

        cue_id, enabled_str = parts
        enabled = enabled_str == "1"

        cue = self.script.find(cue_id)
        if not cue:
            Logger.warning(f"cue_enabled: cue {cue_id} not found in script")
            return

        cue.enabled = enabled
        self._apply_cue_enabled_side_effects(cue, enabled)

        self._notify_cue_enabled(cue_id, enabled)
        Logger.info(f'Cue {cue_id} set to {"enabled" if enabled else "disabled"}')

    def _rejoin_running_chain(self, cue) -> None:
        """Put a cue enabled mid-pass back into its chain, if there is a
        timeline to anchor it on (an engine built without MTC has none)."""
        mtc = getattr(self, "mtc_listener", None)
        if mtc is None:
            return
        try:
            CUE_HANDLER.rejoin_chain(cue, mtc)
        except Exception as exc:
            Logger.error(f"Could not rejoin cue {cue.id} to its chain: {exc}")

    def _apply_cue_enabled_side_effects(self, cue, enabled: bool):
        """React to a cue's enabled-flag change: async re-arm on enable,
        disarm (if idle) on disable.

        Shared by _handle_cue_enabled (Controller-driven toggle) and
        _action_result_sink (enable/disable ActionCue executed on this node).
        Callers notify the Controller themselves — this method never does.
        """
        # A CueList target has no player of its own (arm_cue is a no-op for
        # containers): react on its first enabled child instead, mirroring
        # run_cueList/initial_cuelist_process's first-enabled-child walk.
        if isinstance(cue, CueList):
            # The chain walk stamps the object it walked past, which for a
            # chain-link CueList is the container itself — so its rejoin is
            # decided here, before handing the arm/disarm work to the child.
            if enabled:
                self._rejoin_running_chain(cue)
            else:
                CUE_HANDLER.cancel_parked(cue)
            child = next(
                (c for c in getattr(cue, "contents", None) or [] if c.enabled), None
            )
            if child is not None:
                self._apply_cue_enabled_side_effects(child, enabled)
            return

        if not enabled:
            # Auto continue dispatches the whole chain at the trigger, so a
            # cue disabled mid-pass may already be dispatched and parked on
            # its anchor — nothing downstream re-reads `enabled`, so it would
            # otherwise still fire. Cancel it (a cue already playing is left
            # alone) and remember its slot in case it is enabled again.
            was_parked = CUE_HANDLER.cancel_parked(cue)
            # Disarm only if armed and NOT currently playing (never cut live
            # playback). _playing is the lifecycle flag set by go() and
            # cleared by disarm()/stop_all_cues() — unlike _go_generation,
            # it goes False again when playback ends, so a played-then-
            # re-armed cue can still be disarmed on disable.
            # A cue we just cancelled mid-pass is deliberately NOT disarmed:
            # cancel_parked cleared _playing, which would make the guard below
            # fire, and re-enabling it before its slot has to put it straight
            # back — a disarm would cost an audio cue a ~15s re-arm and it
            # would miss the slot it was supposed to keep.
            if (
                not was_parked
                and CUE_HANDLER.find_armed_cue(cue)
                and not getattr(cue, "_playing", False)
            ):
                CUE_HANDLER.disarm(cue, reason="disabled")
                Logger.info(f"Disarmed disabled cue {cue.id}")
            # Recalculate next_cue_pointer if the disabled cue was next
            if self.next_cue_pointer and self.next_cue_pointer.id == cue.id:
                self.next_cue_pointer = cue.get_next_cue()
                self._broadcast_nextcue()
                next_id = self.next_cue_pointer.id if self.next_cue_pointer else "none"
                Logger.info(f"Next cue was disabled, advanced to {next_id}")
        else:
            # Re-arm in a daemon thread to avoid blocking the caller
            # (arm() is slow — media loading, process spawning).
            if cue._local and not CUE_HANDLER.find_armed_cue(cue):
                gen = self._project_generation
                # Captured by the spawner -- the command thread, or an
                # ActionCue's own thread (_action_result_sink) -- not by the
                # ReArm thread when it gets to run.
                threading.Thread(
                    target=self._arm_with_enabled_guard,
                    args=(cue, gen, CUE_HANDLER.arm_epoch()),
                    daemon=True,
                    name=f"ReArm:{cue.id}",
                ).start()
                Logger.info(f"Re-arming enabled cue {cue.id} (async)")
            else:
                # Already armed: nothing to wait for, so try to put it back
                # into a running chain right away. (When an arm IS needed the
                # rejoin runs at the end of it — the slot must be re-checked
                # against the time the arm actually took.)
                self._rejoin_running_chain(cue)

    #########################
    # Script logic
    #########################
    def ready_script(self, reason: str = "ready_script"):
        """Check if the script is ready to be played.

        ``reason`` is passed to disarm_all so its log names the caller
        (``load``, ``stop``).
        """
        if not self.script:
            Logger.warning("No script loaded, cannot process GO command.")
            return

        self.ongoing_cue = None
        self.next_cue_pointer = None
        self.go_offset = 0
        self._project_generation += 1  # Abort in-flight daemon arm threads
        self.unload_video_devs()
        CUE_HANDLER.disarm_all(reason=reason)

        # Reset mixer volumes to default when preparing script
        mixer_client = PLAYER_HANDLER.get_audio_mixer_client()
        if mixer_client:
            mixer_client.reset_volumes()
            # Report the reset so the UI reflects real (unity) gain.
            self._report_mixer_status()

        self.initial_cuelist_process()

        # Set initial nextcue to the first enabled cue in the script
        if self.script.cuelist.contents:
            first_enabled = None
            for c in self.script.cuelist.contents:
                if c.enabled:
                    first_enabled = c
                    break
            self.next_cue_pointer = first_enabled

        Logger.info(f"Script {self.script.name} loaded and ready to be played")
        # Measurement only: what the load / STOP re-arm left armed.
        CUE_HANDLER.log_armed_inventory(reason)

    def _resolve_go_anchor(self, value):
        """Resolve the GO_mtc anchor for this GO.

        Prefers the controller's own GO instant, shipped as
        value == {"go_mtc_ms": <float>} by ControllerEngine.go_script.
        Falls back to this node's own live MTC read (today's behaviour,
        and the only thing an older controller ever sends) whenever
        `value` doesn't carry a usable one, or the controller's instant
        looks implausible against this node's own MTC.

        Badajoz, 2026-09-25: a node whose setnextcue pre-arm held
        _command_lock for ~500ms read its own MTC ~520ms later than the
        other two boxes at GO time and anchored its whole Auto-continue
        chain late. MTC-following cannot correct a wrong anchor -- only
        agreeing on one does.
        """
        local = self.mtc_listener.main_tc.milliseconds_exact
        if isinstance(value, dict):
            ctrl = value.get("go_mtc_ms")
            if isinstance(ctrl, (int, float)) and not isinstance(ctrl, bool):
                lag = local - ctrl
                if -self._GO_ANCHOR_FUTURE_TOL_MS <= lag <= self._GO_ANCHOR_MAX_LAG_MS:
                    Logger.info(
                        f"GO anchored at controller instant {ctrl:.1f}ms "
                        f"(node lag {lag:.1f}ms)"
                    )
                    return ctrl
                Logger.warning(
                    f"GO anchor from controller ({ctrl:.1f}ms) implausible "
                    f"against local MTC ({local:.1f}ms, lag {lag:.1f}ms) -- "
                    "using local MTC instead"
                )
        return local

    def go_script(self, value):
        try:
            self._go_script(value)
        finally:
            # Measurement only: what this node holds armed once the GO is
            # done (dispatched, nothing local, or skipped). go()'s own
            # lookahead is synchronous, so it is included.
            if self.script:
                CUE_HANDLER.log_armed_inventory("GO")

    def _go_script(self, value):
        if not self.script:
            Logger.warning("No script loaded, cannot process GO command.")
            return

        if not self.with_mtc:
            Logger.warning("No MTC listener, cannot process GO command.")
            return

        # Determine the cue to go
        if not self.ongoing_cue:
            # First GO - use next_cue_pointer (may have been overridden by
            # setnextcue)
            cue_to_go = self.next_cue_pointer or self.script.cuelist.contents[0]
            Logger.info(f"GO command received. Starting script {self.script.name}")
        else:
            # Successive GO - advance to next cue
            if self.next_cue_pointer:
                cue_to_go = self.next_cue_pointer
                Logger.info(
                    f"GO command received. Advancing to next cue: {cue_to_go.id}"
                )
            else:
                # No next cue - script has finished. Do not stop timecode or
                # reset state.
                Logger.info("No more cues. Press STOP to restart.")
                return

        # Capture the GO instant ONCE. Reused as the walk seed (base for the Σ
        # timeline accumulator) AND as the drift baseline (go_offset).
        # Resolved from the controller's own instant when it shipped one
        # (see _resolve_go_anchor) -- every node then derives its anchors
        # from the SAME value, rather than each reading its own MTC
        # whenever it happens to execute this command.
        GO_mtc = self._resolve_go_anchor(value)

        # Walk the post_go='go' chain to the first LOCAL + ENABLED cue,
        # accumulating the timeline offset Σ of the cues we skip. Under Auto
        # continue that advance is zero — every cue in the chain arrives at
        # the trigger — so the walk is really about finding where THIS node's
        # own segment starts. A cue that breaks the chain (post_go != 'go') is
        # a hand-off point: stop and wait for the next GO. Disabled cues are
        # transparent, but they are remembered: enabling one before its slot
        # rejoins this same pass.
        original = cue_to_go
        sigma_ms = 0.0
        walked = 0
        skipped_disabled = []
        while cue_to_go is not None and not (cue_to_go._local and cue_to_go.enabled):
            if not cue_to_go.enabled:
                skipped_disabled.append(cue_to_go)
            if cue_to_go.post_go != "go":
                cue_to_go = None
                break
            if cue_to_go.enabled:
                sigma_ms += CUE_HANDLER._chain_advance_ms(cue_to_go)
            cue_to_go = getattr(cue_to_go, "_target_object", None)
            walked += 1
            if walked > 1024:
                Logger.error("GO chain walk hit safety limit; aborting")
                cue_to_go = None
                break

        if cue_to_go is None:
            # No local+enabled cue reachable in this GO's chain (spent on cues
            # owned by other nodes, or a standalone disabled cue). Advance our
            # next_cue_pointer in lockstep with the node that DID play; otherwise
            # this node wedges on a cue it can never play and every subsequent GO
            # re-evaluates it. (next_cue_pointer is a global sequence property —
            # depends only on post_go/enabled, never on locality.)
            # Nothing dispatches here, so nothing would mint a pass — stamp the
            # disabled cues anyway, or enabling one mid-show could never put it
            # back into this GO's chain.
            if skipped_disabled:
                CUE_HANDLER.stamp_pass(skipped_disabled, GO_mtc)
            self.next_cue_pointer = original.get_next_cue()
            self._broadcast_nextcue()
            Logger.info(
                f'No local+enabled cue in post_go="go" chain from {original.id}; '
                f"nothing to play on this node. Advanced next cue to "
                f'{self.next_cue_pointer.id if self.next_cue_pointer else "none"}'
            )
            # ...and arm what the NEXT GO will dispatch here, now: nothing
            # else would (load/STOP pre-arm stops at another node's pause,
            # the controller never forwards setnextcue on an advance, and
            # the UI / power-bridge GO is a bare GO). Medina sala1,
            # 2026-09-30: the node's first cue fired 240 ms late otherwise.
            self._prearm_after_advance()
            return

        if walked:
            Logger.info(
                f"GO: skipped {walked} cue(s) (Σ={sigma_ms:.1f}ms)."
                f"Starting from local cue {cue_to_go.id}"
            )

        if not CUE_HANDLER.find_armed_cue(cue_to_go):
            Logger.info(f"Cue {cue_to_go.id} not armed, re-arming before GO")
            # Asked before the arm: by the time arm() gives up, the arm it
            # waited on may have finished, and the skip line must not then
            # call a slow arm a failed one.
            was_in_flight = CUE_HANDLER.describe_arm_in_flight(cue_to_go)
            CUE_HANDLER.arm(cue_to_go, init=True)
            if not CUE_HANDLER.find_armed_cue(cue_to_go):
                self._skip_unarmed_cue(
                    cue_to_go, skipped_disabled, GO_mtc, was_in_flight
                )
                return

        # Update state
        self.set_status("running", "yes")
        self.ongoing_cue = cue_to_go

        # Start the cue at its arrival = GO_mtc + Σ(preceding cues). go_threaded
        # adds this cue's own prewait to derive the reveal anchor (start).
        main_thread = CUE_HANDLER.go(
            cue_to_go,
            self.mtc_listener,
            GO_mtc + sigma_ms,
            stamp_skipped=skipped_disabled,
        )
        if main_thread is None:
            # Either a STOP landed while go() was re-arming this cue (the
            # chain-epoch barrier then refuses the dispatch — that is the F4
            # fix working), or the cue was disabled/became non-local between
            # the walk and the dispatch. Say so instead of dereferencing None.
            Logger.error(f"Cue {cue_to_go.id} was not started; aborting GO")
            self.set_status("running", "no")
            self.ongoing_cue = None
            return
        Logger.info(f"Cue {cue_to_go.id} started on main thread: {main_thread.name}")

        # Update next cue pointer
        self.next_cue_pointer = self.ongoing_cue.get_next_cue()
        # Drift baseline; consumed by BaseEngine.timecode = mtc - go_offset.
        # The GO press instant itself (not the first local cue's arrival).
        self.go_offset = GO_mtc

        # Broadcast nextcue to UI
        self._broadcast_nextcue()

        Logger.info(
            f"Cue {cue_to_go.id} started. Next cue:"
            f'{self.next_cue_pointer.id if self.next_cue_pointer else "none"}'
        )

    def _skip_unarmed_cue(self, cue, skipped_disabled, go_mtc, was_in_flight):
        """A GO found its cue unarmed and could not get it armed: skip the
        cue on this node, keep the pointer in lockstep, and say so.

        Decided 2026-09-30 (Ion, 869f9wqpn): arm() waits up to 5 s for an arm
        another thread holds, and 5 s is plenty for a slow load -- one still
        in flight after that counts as failed for this GO. Returning here
        without moving the pointer, as this used to, made the node dispatch
        this cue on the NEXT GO and run one GO behind the controller until a
        STOP. Nothing is stopped and the running status is not touched.

        ongoing_cue is set as on the success path: if the pointer becomes
        None, the next GO must read "No more cues", not "first GO" (which
        would restart this node from the top of the script).
        """
        in_flight = CUE_HANDLER.describe_arm_in_flight(cue)
        if in_flight:
            why = f"its arm is still in flight, {in_flight}"
        elif was_in_flight:
            why = "the arm it waited on did not deliver the cue in time"
        else:
            why = "its arm failed"
        # Nothing dispatches here, so nothing would mint a pass: stamp the
        # disabled cues the walk passed, as the no-local branch does, or
        # enabling one mid-show could never put it back into this GO's chain.
        if skipped_disabled:
            CUE_HANDLER.stamp_pass(skipped_disabled, go_mtc)
        self.ongoing_cue = cue
        self.next_cue_pointer = cue.get_next_cue()
        self._broadcast_nextcue()
        next_id = self.next_cue_pointer.id if self.next_cue_pointer else "none"
        Logger.error(
            f"Cue {cue.id} ({type(cue).__name__}, "
            f'"{getattr(cue, "name", "")}") SKIPPED on this node: it could '
            f"not be armed in time for this GO ({why}). "
            f"Pointer advanced to {next_id}."
        )
        # The skip never reaches go()'s own lookahead: arm what the next GO
        # will dispatch here, or it would arm at GO time and start late.
        self._prearm_after_advance()

    def stop_playback(self, value=None):
        """Stop playback, full cleanup, then re-arm so GO is available again.

        Does the cleanup that ready_script() doesn't handle (DMX blackout,
        disconnect video, kill audio), then delegates reset + re-arm to
        ready_script(). Notifies Controller when armed (GO button green).
        """
        Logger.info("STOP command received. Stopping playback.")

        self.set_status("running", "no")

        self._gradient_cancel_all("stop")

        # Signal all running cue threads to stop immediately.
        # Must happen BEFORE blackout/reset so loop_cue threads don't
        # re-push DMX frames or send /visible after cleanup.
        CUE_HANDLER.stop_all_cues()
        sleep(0.05)  # 50ms — loop_cue polls every 20ms

        # DMX: disable MTC following first (freezes the playhead so queued
        # scenes can't fire), then blackout via OLA for instant visual reset.
        self._dmx_reset()

        # Unload all video layers (instant visual blackout)
        self.unload_video_devs()

        # Kill all audio players (ready_script does not do this)
        PLAYER_HANDLER.kill_all_audio_players()
        PLAYER_HANDLER.cleanup_zombie_jack_clients()

        # Reset state + disarm + volume reset + re-arm cues
        if self.script:
            self.ready_script(reason="stop")
            Logger.info(f"Project {self.script.name} reset and ready for GO.")

            # Notify Controller that re-arm is complete (GO button can go
            # green)
            self._notify_armed_ready("re-arm")

            # Broadcast nextcue (reset to first cue after stop)
            self._broadcast_nextcue()
        else:
            Logger.info("Playback stopped (no script loaded).")

        Logger.info("Playback stopped.")

add_player_endpoints(cue, prefix='/players')

Add player endpoints from a cue to the OSCQuery server

Parameters:

Name	Type	Description	Default
cue	Cue	The cue containing the player client	required
prefix	str	Prefix to add to all endpoint paths (default: '/players')	'/players'

Source code in src/cuemsengine/NodeEngine.py

def add_player_endpoints(self, cue: Cue, prefix: str = "/players"):
    """Add player endpoints from a cue to the OSCQuery server

    Args:
        cue: The cue containing the player client
        prefix: Prefix to add to all endpoint paths (default: '/players')
    """
    if not hasattr(cue, "_osc") or cue._osc is None:
        Logger.warning(f"Cue {cue.id} has no OSC client, cannot add endpoints")
        return

    try:
        # Get endpoints from the player client
        endpoints = cue._osc.get_endpoints()
        if not endpoints:
            Logger.warning(f"No endpoints found for cue {cue.id}")
            return

        # Add prefix to all endpoints
        prefixed_endpoints = add_prefix_to_all(
            endpoints, f"{prefix}/{self.cm.node_uuid}/{cue.id}"
        )

        # Add endpoints to OSCQuery server
        if hasattr(self, "oscquery_server") and self.oscquery_server:
            self.oscquery_server.add_endpoints(prefixed_endpoints)
            Logger.debug(
                f"Added {len(prefixed_endpoints)} endpoints for cue {cue.id}"
            )
        else:
            Logger.warning("OSCQuery server not initialized, cannot add endpoints")
    except Exception as e:
        Logger.error(f"Error adding player endpoints for cue {cue.id}: {e}")
        Logger.exception(e)

check_own_media_sizes(media_synced=True)

Report this node's media files that differ from the project, or are missing (869fat84r D20, L2).

The editor corrects the project before a load when a file on the controller changed; this is the last defence, for a copy that differs only here (a failed media sync, a file placed on this node only, an OSC load that bypassed the editor). Report only: the cue keeps the stored duration, so this node stays in step with the others. Never raises: the previous project is already torn down at this point.

Source code in src/cuemsengine/NodeEngine.py

def check_own_media_sizes(self, media_synced: bool = True) -> list:
    """Report this node's media files that differ from the project, or
    are missing (869fat84r D20, L2).

    The editor corrects the project before a load when a file on the
    controller changed; this is the last defence, for a copy that differs
    only here (a failed media sync, a file placed on this node only, an
    OSC load that bypassed the editor). Report only: the cue keeps the
    stored duration, so this node stays in step with the others. Never
    raises: the previous project is already torn down at this point.
    """
    try:
        found = own_media_size_mismatches(
            self.script.get_own_media(config=self.cm),
            self.script.find,
            PLAYER_HANDLER.media_path,
        )
    except Exception as e:
        Logger.error(
            f"Media size check skipped ({type(e).__name__}: {e}); the load goes on"
        )
        return []
    sync = "" if media_synced else " This load's media sync failed."
    for file_name, stored, actual in found:
        if actual is None:
            Logger.error(
                f"Media file {file_name} is missing on this node: its cues will "
                f"fail.{sync}"
            )
        else:
            Logger.error(
                f"Media file {file_name} on this node is {actual} bytes; the "
                f"project stores {stored}: it was replaced or not synced. The "
                "stored duration is kept so the nodes stay in step; re-sync it, "
                f"or re-upload it in the editor.{sync}"
            )
    return found

check_unrouted_video_outputs()

Name every enabled video cue of this node that points at an output this node doesn't drive (869fgq3xv). Log only.

Such an output is never loaded (arm_videoCue), so the cue stays dark there; this says so at load, before the show, instead of on the screens. Also starts the load's once-only arm errors afresh. Uses the same PlayerHandler.partition_cue_outputs as the arm, so the two cannot disagree. Never raises: the previous project is already torn down.

Source code in src/cuemsengine/NodeEngine.py

def check_unrouted_video_outputs(self) -> list:
    """Name every enabled video cue of this node that points at an output
    this node doesn't drive (869fgq3xv). Log only.

    Such an output is never loaded (arm_videoCue), so the cue stays dark
    there; this says so at load, before the show, instead of on the
    screens. Also starts the load's once-only arm errors afresh. Uses the
    same PlayerHandler.partition_cue_outputs as the arm, so the two cannot
    disagree. Never raises: the previous project is already torn down.
    """
    PLAYER_HANDLER.reset_unrouted_warnings()
    found = []
    try:
        video_cues = []
        for cue_id, output_names in (self.outputs_map or {}).items():
            cue = self.script.find(cue_id)
            # outputs_map holds every MediaCue of this node, audio and
            # disabled cues included.
            if isinstance(cue, VideoCue) and getattr(cue, "enabled", True):
                video_cues.append((cue, output_names))
        if not video_cues:
            return found

        if PLAYER_HANDLER.get_video_client() is None:
            # set_video_players returned early: every video cue fails at
            # arm ("No video client"), for a reason that is not the cue's.
            Logger.warning(
                f"no video stack on this node: {len(video_cues)} video cue(s)"
                " will not play here"
            )
            return found

        driven = PLAYER_HANDLER.driven_video_outputs()
        for cue, output_names in video_cues:
            _, unrouted = PLAYER_HANDLER.partition_cue_outputs(cue, output_names)
            if not unrouted:
                continue
            found.append(
                {
                    "cue_id": cue.id,
                    "cue_name": getattr(cue, "name", ""),
                    "outputs": [
                        {"output": name, "reason": reason}
                        for name, reason in unrouted
                    ],
                }
            )
            cue_name = getattr(cue, "name", "")
            for output_name, reason in unrouted:
                Logger.warning(
                    f'UNROUTED_VIDEO_OUTPUT: cue {cue.id} "{cue_name}"'
                    f' → output "{output_name}" not driven by this node ({reason});'
                    f" it will stay dark. Driven outputs: {driven}"
                )
        if found:
            Logger.warning(
                f"{len(found)} video cue(s) point at outputs this node doesn't"
                " drive; see UNROUTED_VIDEO_OUTPUT (and DISPLAY_MISMATCH at"
                " node start)"
            )
    except Exception as e:
        Logger.exception(
            f"Unrouted video output check skipped ({type(e).__name__}: {e});"
            " the load goes on"
        )
        return []
    return found

deploy_media(project)

Deploy the media files (and their .idx sidecar indexes).

Best-effort: a failure here is recoverable if media is already cached on disk. Returns False to surface the failure to logs, but the caller continues the load.

Source code in src/cuemsengine/NodeEngine.py

def deploy_media(self, project):
    """Deploy the media files (and their .idx sidecar indexes).

    Best-effort: a failure here is recoverable if media is already
    cached on disk. Returns False to surface the failure to logs,
    but the caller continues the load.
    """
    if not self.script:
        Logger.error("No script loaded")
        return False
    bare_names = self.script.get_own_media_filenames(config=self.cm)
    if len(bare_names) == 0:
        Logger.info("No media files to deploy")
        return True
    if not self.deploy_manager.sync_files(project, "media", bare_names):
        Logger.error(
            f"Media deploy failed for {project} — continuing with cached "
            f"files; cues whose media is missing locally will fail on GO"
        )
        return False
    return True

deploy_project(project)

Deploy the project files (script.xml, mappings.xml, settings.xml).

Critical path: if these fail to sync, the local copy may be stale and arming cues against it is unsafe. Caller is expected to abort the load on False.

Source code in src/cuemsengine/NodeEngine.py

def deploy_project(self, project):
    """Deploy the project files (script.xml, mappings.xml, settings.xml).

    Critical path: if these fail to sync, the local copy may be stale
    and arming cues against it is unsafe. Caller is expected to abort
    the load on False.
    """
    return self.deploy_manager.sync_files(project, "project")

ensure_video_indexes()

Run cuems-videoindexer on any video files that are missing a .idx sidecar.

This is a safety net for files that were copied manually or deployed to a node that never ran the editor upload hook. For normally-uploaded files the index was already created by the editor and this is a no-op.

Source code in src/cuemsengine/NodeEngine.py

def ensure_video_indexes(self):
    """
    Run cuems-videoindexer on any video files that are missing a .idx
    sidecar.

    This is a safety net for files that were copied manually or deployed to
    a
    node that never ran the editor upload hook. For normally-uploaded files
    the
    index was already created by the editor and this is a no-op.
    """
    if not self.script:
        return
    file_names = self.script.get_own_media_filenames(config=self.cm)
    video_exts = {".mp4", ".mov", ".avi", ".mkv", ".mpg"}
    unindexed = []
    for name in file_names:
        ext = os.path.splitext(name)[1].lower()
        if ext not in video_exts:
            continue
        full_path = PLAYER_HANDLER.media_path(name)
        idx_dir = os.path.join(os.path.dirname(full_path), "indexes")
        idx_path = os.path.join(idx_dir, os.path.basename(full_path) + ".idx")
        if not os.path.exists(idx_path):
            unindexed.append(full_path)
    if unindexed:
        Logger.info(
            f"ensure_video_indexes: indexing {len(unindexed)} video(s)"
            f"missing .idx: "
            f"{[os.path.basename(p) for p in unindexed]}"
        )
        try:
            result = subprocess.run(
                ["cuems-videoindexer"] + unindexed,
                timeout=600,
                capture_output=True,
                text=True,
            )
        except Exception as e:
            Logger.warning(f"ensure_video_indexes: indexer failed: {e}")
            return
        if result.returncode == 0:
            Logger.info("ensure_video_indexes: indexer ok (rc=0)")
            if result.stdout:
                Logger.debug(
                    f"ensure_video_indexes stdout: {result.stdout.strip()}"
                )
        else:
            Logger.warning(
                "ensure_video_indexes: indexer returned"
                f"rc={result.returncode}. "
                f"stdout={result.stdout.strip()!r}"
                f"stderr={result.stderr.strip()!r}"
            )

load_project(project)

Load the project files to the node

Source code in src/cuemsengine/NodeEngine.py

def load_project(self, project):
    """Load the project files to the node"""
    with self._loading_lock:
        if self._loading:
            Logger.warning(
                f"Load already in progress, ignoring duplicate load of {project}"
            )
            return
        self._loading = True

    try:
        return self._load_project_inner(project)
    finally:
        with self._loading_lock:
            self._loading = False

map_cue_outputs(cuelist=None)

Load the output mappings for the project

Source code in src/cuemsengine/NodeEngine.py

def map_cue_outputs(self, cuelist: CueList = None):
    """Load the output mappings for the project"""
    outputs_map = {}
    if cuelist is None:
        cuelist = self.script.cuelist
    if cuelist.contents is None:
        Logger.warning("Cuelist contents is None, skipping map_cue_outputs")
        return outputs_map
    for cue in cuelist.contents:
        if isinstance(cue, CueList):
            outputs_map.update(self.map_cue_outputs(cue))
            continue
        elif not isinstance(cue, MediaCue):
            continue

        outputs = [
            x[1] for x in cue.get_all_output_names() if x[0] == self.cm.node_uuid
        ]
        if outputs:
            outputs_map[cue.id] = outputs
    Logger.debug(f"Outputs map: {outputs_map}")
    return outputs_map

quit_dmx_devs()

Quit the DMX player if it exists

Source code in src/cuemsengine/NodeEngine.py

def quit_dmx_devs(self):
    """Quit the DMX player if it exists"""
    dmx_client = PLAYER_HANDLER.get_dmx_player_client()
    if dmx_client:
        try:
            dmx_client.set_value("/quit", 1)
        except Exception as e:
            Logger.exception(e)
    CUE_HANDLER.communications_thread.remove_player(
        f"dmxplayer_{self.cm.node_uuid}"
    )

ready_project(project)

Prepare the project to be played.

deploy_project() runs in _load_project_inner BEFORE this point, before the teardown of the previous project — so that a deploy failure aborts the load without destroying running state. Media deploy runs here because it is best-effort: a failure logs but does not abort (cached media may already be on disk).

Source code in src/cuemsengine/NodeEngine.py

def ready_project(self, project):
    """Prepare the project to be played.

    deploy_project() runs in _load_project_inner BEFORE this point,
    before the teardown of the previous project — so that a deploy
    failure aborts the load without destroying running state.
    Media deploy runs here because it is best-effort: a failure
    logs but does not abort (cached media may already be on disk).
    """
    self.cm.load_project_config(project)
    self.read_script(project)
    media_synced = self.deploy_media(project)
    self.check_own_media_sizes(media_synced)
    self.ensure_video_indexes()
    self.outputs_map = self.map_cue_outputs()
    PLAYER_HANDLER.set_outputs_map(self.outputs_map)
    PORT_HANDLER.clean_random_ports()

ready_script(reason='ready_script')

Check if the script is ready to be played.

reason is passed to disarm_all so its log names the caller (load, stop).

Source code in src/cuemsengine/NodeEngine.py

def ready_script(self, reason: str = "ready_script"):
    """Check if the script is ready to be played.

    ``reason`` is passed to disarm_all so its log names the caller
    (``load``, ``stop``).
    """
    if not self.script:
        Logger.warning("No script loaded, cannot process GO command.")
        return

    self.ongoing_cue = None
    self.next_cue_pointer = None
    self.go_offset = 0
    self._project_generation += 1  # Abort in-flight daemon arm threads
    self.unload_video_devs()
    CUE_HANDLER.disarm_all(reason=reason)

    # Reset mixer volumes to default when preparing script
    mixer_client = PLAYER_HANDLER.get_audio_mixer_client()
    if mixer_client:
        mixer_client.reset_volumes()
        # Report the reset so the UI reflects real (unity) gain.
        self._report_mixer_status()

    self.initial_cuelist_process()

    # Set initial nextcue to the first enabled cue in the script
    if self.script.cuelist.contents:
        first_enabled = None
        for c in self.script.cuelist.contents:
            if c.enabled:
                first_enabled = c
                break
        self.next_cue_pointer = first_enabled

    Logger.info(f"Script {self.script.name} loaded and ready to be played")
    # Measurement only: what the load / STOP re-arm left armed.
    CUE_HANDLER.log_armed_inventory(reason)

set_audio_players()

Set the audio players and audio mixer

Source code in src/cuemsengine/NodeEngine.py

def set_audio_players(self):
    """Set the audio players and audio mixer"""
    # Initialize the audio mixer for this node
    if self.cm.node_hw_outputs.get("audio_outputs"):
        audio_outputs = self.cm.node_hw_outputs["audio_outputs"]
        Logger.info(f"Initializing audio mixer with {len(audio_outputs)} outputs")

        mixer_id = "0"  # TODO: make this a unique identifier for the mixer
        # Start the audio mixer. The port draw is inside the try: it can
        # now raise if the pool is exhausted, and losing the mixer must
        # not abort the whole node start.
        # (assign_ports(cue=None) registers via add_config_ports itself —
        # the explicit second call that used to sit here was redundant.)
        try:
            mixer_ports = PORT_HANDLER.assign_ports(["audio_mixer"])
            PLAYER_HANDLER.start_audio_mixer(
                audio_outputs=audio_outputs,
                port=mixer_ports["audio_mixer"],
                mixer_id=mixer_id,
                path=self.cm.node_conf["audiomixer"]["path"],
                args=self.cm.node_conf["audiomixer"]["args"],
            )
            Logger.info(f"Audio mixer started successfully for mixer {mixer_id}")
            # Register mixer with Controller via NNG
            try:
                CUE_HANDLER.communications_thread.add_player(
                    f"audiomixer_{mixer_id}", None, timeout=0.1
                )
                Logger.info(f"Audio mixer {mixer_id} registered with Controller")
            except Exception as e:
                Logger.warning(f"Could not register mixer with Controller: {e}")

            # Assert a known unity state at startup so the controller shadow
            # (hence the UI) reflects real gain from t=0, then snapshot it.
            try:
                mixer_client = PLAYER_HANDLER.get_audio_mixer_client()
                if mixer_client:
                    mixer_client.reset_volumes()
                self._report_mixer_status()
            except Exception as e:
                Logger.warning(f"Mixer startup state assert/report failed: {e}")
        except Exception as e:
            Logger.error(
                f"Error starting audio mixer: {e}. "
                f"Audio on this node will be SILENT."
            )
            Logger.exception(e)
    else:
        Logger.info(
            "No audio outputs detected, skipping audio mixer initialization"
        )

    # Build audio output lookup keyed by <id> (mirrors video output
    # pattern)
    audio_outputs = {}
    for port_type_dict in self.cm.node_mappings.get("audio", []):
        for port_type_list in port_type_dict.values():
            for port in port_type_list:
                for _, output_data in port.items():
                    output_id = str(output_data.get("id", output_data["name"]))
                    mappings = output_data.get("mappings", [])
                    mapped_to = (
                        mappings[0]["mapped_to"]
                        if mappings
                        else output_data["name"]
                    )
                    audio_outputs[output_id] = {
                        "name": output_data["name"],
                        "mapped_to": mapped_to,
                    }
    PLAYER_HANDLER.set_audio_outputs(audio_outputs)

    # Set the audio player generator. Append --output-latency-ms
    # from settings.xml when the operator supplied an integer
    # override (isinstance int); "auto" or absent ⇒ audioplayer
    # runs its Phase-3 JACK-latency query path.
    audio_args = _append_output_latency_flag(
        self.cm.node_conf["audioplayer"]["args"],
        self.cm.node_conf["audioplayer"],
    )
    PLAYER_HANDLER.set_audio_output_generator(
        self.cm.node_conf["audioplayer"]["path"],
        audio_args,
    )

set_dmx_players()

Set the DMX player for this node and register its endpoints.

Source code in src/cuemsengine/NodeEngine.py

def set_dmx_players(self):
    """Set the DMX player for this node and register its endpoints."""
    # Get node UUID for player naming
    node_uuid = self.cm.node_conf.get("uuid", "default_node")

    # Start the DMX player. The port draw is inside the try: it can now
    # raise if the pool is exhausted, and losing DMX must not abort the
    # whole node start. (assign_ports(cue=None) registers via
    # add_config_ports itself — the second call here was redundant.)
    try:
        dmx_ports = PORT_HANDLER.assign_ports(["dmx_player"])
        # Append --output-latency-ms from settings.xml when an
        # integer override is present. Dmx has no "auto" form —
        # absent ⇒ dmxplayer's 35 ms Phase-5A default stands.
        dmx_args = _append_output_latency_flag(
            self.cm.node_conf["dmxplayer"]["args"],
            self.cm.node_conf["dmxplayer"],
        )
        PLAYER_HANDLER.start_dmx_player(
            port=dmx_ports["dmx_player"],
            node_uuid=node_uuid,
            path=self.cm.node_conf["dmxplayer"]["path"],
            args=dmx_args,
        )
        try:
            CUE_HANDLER.communications_thread.add_player(
                f"dmxplayer_{node_uuid}", None, timeout=0.1
            )
        except Exception:
            pass  # Ignore - NNG is for distributed nodes
        Logger.info(f"DMX player started successfully for node {node_uuid}")
    except Exception as e:
        Logger.error(f"Error starting DMX player: {e}")
        Logger.exception(e)
        return

set_gradient_client()

Wire GradientClient into PLAYER_HANDLER using settings from node_conf.

Source code in src/cuemsengine/NodeEngine.py

def set_gradient_client(self) -> None:
    """
    Wire GradientClient into PLAYER_HANDLER using settings from node_conf.
    """
    port = int(self.cm.node_conf["gradient_osc_port"])
    node_name = self._resolve_gradient_node_name()
    PLAYER_HANDLER.set_gradient_client(port=port, node_name=node_name)

set_next_cue(value)

Handle setnextcue command from the UI — override next_cue_pointer.

Arms the SELECTED cue synchronously (a GO right after this needs it immediately), but extends the arm-ahead window in a background PreArm: thread. Badajoz, 2026-09-25 (869f79ecc): that extension used to run here, synchronously, inside run_command's _command_lock. A cold-boot pre-arm took ~500ms; a GO sent right after (UI, power-bridge /gocue) waited for the lock, executed late, and read its own MTC ~520ms later than the other nodes — anchoring its whole Auto-continue chain late. See go_script / _resolve_go_anchor for the anchor half of that fix; this is the half that stops the delay from happening in the first place.

Source code in src/cuemsengine/NodeEngine.py

def set_next_cue(self, value):
    """
    Handle setnextcue command from the UI — override next_cue_pointer.

    Arms the SELECTED cue synchronously (a GO right after this needs it
    immediately), but extends the arm-ahead window in a background
    PreArm:<id> thread. Badajoz, 2026-09-25 (869f79ecc): that extension
    used to run here, synchronously, inside run_command's
    _command_lock. A cold-boot pre-arm took ~500ms; a GO sent right
    after (UI, power-bridge /gocue) waited for the lock, executed
    late, and read its own MTC ~520ms later than the other nodes —
    anchoring its whole Auto-continue chain late. See go_script /
    _resolve_go_anchor for the anchor half of that fix; this is the
    half that stops the delay from happening in the first place.
    """
    if not self.script:
        Logger.warning("No script loaded, cannot set next cue.")
        return
    cue = self.script.find(value)
    if cue:
        self.next_cue_pointer = cue
        self._selection_epoch += 1
        selection_epoch = self._selection_epoch
        project_gen = self._project_generation
        # Captured HERE, on the command thread: the lookahead belongs to
        # the STOP/load epoch in which it was asked for.
        arm_epoch = CUE_HANDLER.arm_epoch()
        if not CUE_HANDLER.find_armed_cue(cue):
            Logger.info(f"Re-arming cue {cue.id} selected as next cue")
            CUE_HANDLER.arm(cue, init=True)
        threading.Thread(
            target=self._prearm_thread,
            args=(
                self._prearm_lookahead,
                cue,
                project_gen,
                selection_epoch,
                arm_epoch,
            ),
            daemon=True,
            name=f"PreArm:{cue.id}",
        ).start()
        self._broadcast_nextcue()
        Logger.info(f"Next cue overridden by UI: {value}")
    else:
        Logger.warning(f"setnextcue: cue {value} not found in script")

set_oscquery_comms()

Set up the command dictionary for the NodeEngine.

Commands are received via NNG from ControllerEngine. OSCQuery client is no longer used since pyossia server was removed.

Source code in src/cuemsengine/NodeEngine.py

def set_oscquery_comms(self):
    """Set up the command dictionary for the NodeEngine.

    Commands are received via NNG from ControllerEngine.
    OSCQuery client is no longer used since pyossia server was removed.
    """
    self.commands_dict = {
        "deploy": self.ready_project,
        "load": self.load_project,
        "loadcue": None,
        "go": self.go_script,
        "gocue": self.go_script,
        "pause": None,
        "resetall": None,
        "stop": self.stop_playback,
        "setnextcue": self.set_next_cue,
        "cue_enabled": self._handle_cue_enabled,
        "test": None,
        "unload": None,
        "update": None,
    }

set_video_players()

Set the video players

Source code in src/cuemsengine/NodeEngine.py

def set_video_players(self):
    """Set the video players"""
    Logger.info(f'Setting video players with: {self.cm.node_conf["videoplayer"]}')
    if not self.cm.node_hw_outputs["video_outputs"]:
        Logger.info("No video outputs detected.")
        return

    PLAYER_HANDLER.add_node_uuid(self.cm.node_uuid)
    vc_conf = self.cm.node_conf.get("videoplayer", {})
    osc_video_port = int(vc_conf.get("osc_port", VIDEOCOMPOSER_OSC_PORT_DEFAULT))
    # Declare BEFORE building the client: the client draws its own local
    # OSC port from the pool, and must not be handed the videocomposer's.
    # This is also the only registration of this port — it comes from the
    # nested videoplayer/osc_port key, which get_config_ports() (top-level
    # only) cannot see, so BaseEngine's hoisted call misses it.
    PORT_HANDLER.add_config_ports({"videocomposer": osc_video_port})
    try:
        PLAYER_HANDLER.set_video_client(osc_video_port)
    except Exception as e:
        Logger.error(
            f"Could not create the videocomposer OSC client: {e}. "
            f"Video on this node will not respond."
        )
        Logger.exception(e)
        return

    # Canvas geometry comes from /run/cuems/display.conf, written by
    # cuems-generate-display-conf (videocomposer's ExecStartPre). It's the
    # same file the videocomposer reads, so engine + VC agree on canvas
    # size and per-output regions without a handshake. The XML's optional
    # <canvas_region> is a UI-template hint (normalized [0,1]) and is
    # ignored here — engine never sources physical layout from XML.
    display_regions, (canvas_w, canvas_h) = read_display_conf()

    video_outputs = {}
    for port_type_dict in self.cm.node_mappings.get("video", []):
        for port_type_list in port_type_dict.values():
            for port in port_type_list:
                for _, output_data in port.items():
                    output_id = str(output_data.get("id", output_data["name"]))
                    name = output_data["name"]
                    mappings = output_data.get("mappings", [])
                    mapped_to = mappings[0]["mapped_to"] if mappings else name
                    region = display_regions.get(mapped_to)
                    if region is None:
                        Logger.warning(
                            f"DISPLAY_MISMATCH: XML output id={output_id} "
                            f"name={name!r} maps to {mapped_to!r} which"
                            f"is "
                            f"not in display.conf; skipping. Available: "
                            f"{sorted(display_regions.keys())}"
                        )
                        continue
                    video_outputs[output_id] = {
                        "name": name,
                        "mapped_to": mapped_to,
                        "x": region["x"],
                        "y": region["y"],
                        "width": region["width"],
                        "height": region["height"],
                        "canvas_region": dict(region),
                    }
    PLAYER_HANDLER.start_video_outputs(
        video_outputs, canvas_override=(canvas_w, canvas_h)
    )

stop_comms()

Stop the NNG comms thread before the interpreter tears down.

pynng registers an atexit hook that calls nng_fini(), destroying NNG's global state. CPython does not join daemon threads before running atexit hooks, so a comms thread still polling the bus at that point calls into freed NNG internals and aborts the process ("panic: pthread_mutex_lock: Invalid argument", ClickUp 869ed00ya).

ControllerEngine does the equivalent in its own stop_comms(); without this, only the node engine core-dumped on every restart.

Source code in src/cuemsengine/NodeEngine.py

@logged
def stop_comms(self):
    """Stop the NNG comms thread before the interpreter tears down.

    pynng registers an atexit hook that calls nng_fini(), destroying NNG's
    global state. CPython does not join daemon threads before running
    atexit hooks, so a comms thread still polling the bus at that point
    calls into freed NNG internals and aborts the process
    ("panic: pthread_mutex_lock: Invalid argument", ClickUp 869ed00ya).

    ControllerEngine does the equivalent in its own stop_comms(); without
    this, only the node engine core-dumped on every restart.
    """
    comms = getattr(CUE_HANDLER, "communications_thread", None)
    if comms is None:
        return
    try:
        comms.stop()
    except Exception as e:
        Logger.error(f"Error stopping NNG comms thread: {e}")

stop_node_engine()

Stop the NodeEngine elements

Source code in src/cuemsengine/NodeEngine.py

def stop_node_engine(self):
    """Stop the NodeEngine elements"""
    CUE_HANDLER.disarm_all(reason="shutdown")
    self.stop_video_devs()

stop_playback(value=None)

Stop playback, full cleanup, then re-arm so GO is available again.

Does the cleanup that ready_script() doesn't handle (DMX blackout, disconnect video, kill audio), then delegates reset + re-arm to ready_script(). Notifies Controller when armed (GO button green).

Source code in src/cuemsengine/NodeEngine.py

def stop_playback(self, value=None):
    """Stop playback, full cleanup, then re-arm so GO is available again.

    Does the cleanup that ready_script() doesn't handle (DMX blackout,
    disconnect video, kill audio), then delegates reset + re-arm to
    ready_script(). Notifies Controller when armed (GO button green).
    """
    Logger.info("STOP command received. Stopping playback.")

    self.set_status("running", "no")

    self._gradient_cancel_all("stop")

    # Signal all running cue threads to stop immediately.
    # Must happen BEFORE blackout/reset so loop_cue threads don't
    # re-push DMX frames or send /visible after cleanup.
    CUE_HANDLER.stop_all_cues()
    sleep(0.05)  # 50ms — loop_cue polls every 20ms

    # DMX: disable MTC following first (freezes the playhead so queued
    # scenes can't fire), then blackout via OLA for instant visual reset.
    self._dmx_reset()

    # Unload all video layers (instant visual blackout)
    self.unload_video_devs()

    # Kill all audio players (ready_script does not do this)
    PLAYER_HANDLER.kill_all_audio_players()
    PLAYER_HANDLER.cleanup_zombie_jack_clients()

    # Reset state + disarm + volume reset + re-arm cues
    if self.script:
        self.ready_script(reason="stop")
        Logger.info(f"Project {self.script.name} reset and ready for GO.")

        # Notify Controller that re-arm is complete (GO button can go
        # green)
        self._notify_armed_ready("re-arm")

        # Broadcast nextcue (reset to first cue after stop)
        self._broadcast_nextcue()
    else:
        Logger.info("Playback stopped (no script loaded).")

    Logger.info("Playback stopped.")

own_media_size_mismatches(own_media, find_cue, media_path)

This node's media files that differ from the project (869fat84r D20, L2).

own_media is {cue_id: file_name} (CuemsScript.get_own_media), find_cue maps a cue id to its cue, media_path a file name to this node's path. Returns [(file_name, stored_size, actual_size)], once per file: actual_size is None for a missing file, and stored_size is None when the project stores none. A file whose project has no stored size (an older editor wrote it) is compared with nothing.

Source code in src/cuemsengine/NodeEngine.py

def own_media_size_mismatches(own_media: dict, find_cue, media_path) -> list:
    """This node's media files that differ from the project (869fat84r D20, L2).

    *own_media* is ``{cue_id: file_name}`` (``CuemsScript.get_own_media``),
    *find_cue* maps a cue id to its cue, *media_path* a file name to this
    node's path. Returns ``[(file_name, stored_size, actual_size)]``, once
    per file: ``actual_size`` is ``None`` for a missing file, and
    ``stored_size`` is ``None`` when the project stores none. A file whose
    project has no stored size (an older editor wrote it) is compared with
    nothing.
    """
    found = []
    seen = set()
    for cue_id, file_name in own_media.items():
        if not file_name or file_name in seen:
            continue
        seen.add(file_name)
        media = getattr(find_cue(cue_id), "media", None)
        stored = media.get("file_size") if hasattr(media, "get") else None
        if not _positive_int(stored):
            stored = None
        try:
            actual = os.stat(media_path(file_name)).st_size
        except OSError:
            found.append((file_name, stored, None))
            continue
        if stored is not None and actual != stored:
            found.append((file_name, stored, actual))
    return found

redirect_audio_cmd(path_parts, value)

Redirect the audio command to the audio player

Source code in src/cuemsengine/NodeEngine.py

def redirect_audio_cmd(path_parts: list[str], value: str) -> None:
    """Redirect the audio command to the audio player"""
    if path_parts[0] == "mixer":
        redirect_audio_mixer_cmd(path_parts[1:], value)
    elif path_parts[0] == "cue":
        redirect_audio_player_cmd(path_parts[1:], value)
    else:
        Logger.error(f"Invalid audio command: {path_parts}")
        return

redirect_audio_mixer_cmd(path_parts, value)

Redirect the audio mixer command to the audio mixer Follows the logic: /master/volume -> /audiomixer/0_mixer/master /0/volume -> /audiomixer/0_mixer/0 /1/volume -> /audiomixer/0_mixer/1 ... Args: path_parts: List of path parts value: Value to set

Source code in src/cuemsengine/NodeEngine.py

def redirect_audio_mixer_cmd(path_parts: list[str], value: str) -> None:
    """Redirect the audio mixer command to the audio mixer
     Follows the logic:
     <output_index>/master/volume -> /audiomixer/0_mixer/master
     <output_index>/0/volume -> /audiomixer/0_mixer/0
     <output_index>/1/volume -> /audiomixer/0_mixer/1
     ...
    Args:
        path_parts: List of path parts
        value: Value to set
    """
    output_index, channel, _ = path_parts
    mixer_cmd = f"/audiomixer/0_mixer/{channel}"
    PLAYER_HANDLER.get_audio_mixer_client().set_value(mixer_cmd, value)

redirect_audio_player_cmd(path_parts, value)

Redirect the audio mixer command to the audio mixer Follows the logic: /master/volume -> /volmaster /0/volume -> /vol0 /1/volume -> /vol1 ...

Parameters:

Name	Type	Description	Default
path_parts	list[str]	List of path parts	required
value	str	Value to set	required

Source code in src/cuemsengine/NodeEngine.py

def redirect_audio_player_cmd(path_parts: list[str], value: str) -> None:
    """Redirect the audio mixer command to the audio mixer
     Follows the logic:
     <cue_uuid>/master/volume -> /volmaster
     <cue_uuid>/0/volume -> /vol0
     <cue_uuid>/1/volume -> /vol1
     ...

    Args:
        path_parts: List of path parts
        value: Value to set
    """
    cue_uuid, channel, _ = path_parts
    audio_cmd = f"/vol{channel}"
    cue = CUE_HANDLER.get_armed_cue(cue_uuid)
    if not cue:
        Logger.error(f"Cue {cue_uuid} not found")
        return
    client: AudioClient = cue._osc
    client.set_value(audio_cmd, value)

redirect_dmx_cmd(path_parts, value)

Redirect the DMX command to the DMX player

Source code in src/cuemsengine/NodeEngine.py

def redirect_dmx_cmd(path_parts: list[str], value: str) -> None:
    """Redirect the DMX command to the DMX player"""
    dmx_index = path_parts.index("mixer") + 1  # +1 to skip the 'mixer' keyword
    dmx_cmd = "/" + "/".join(path_parts[dmx_index:])
    client: DmxClient = PLAYER_HANDLER.get_dmx_player_client()
    client.set_value(dmx_cmd, value)

redirect_video_cmd(path_parts, value)

Redirect the video command to the video client

Source code in src/cuemsengine/NodeEngine.py

def redirect_video_cmd(path_parts: list[str], value: str) -> None:
    """Redirect the video command to the video client"""
    videocomposer_index = path_parts.index("videocomposer")
    videocomposer_cmd = "/" + "/".join(path_parts[videocomposer_index:])
    client: VideoClient = PLAYER_HANDLER.get_video_client()
    client.set_value(videocomposer_cmd, value)