Cue Architecture

ActionHandler (action_handler.py)

Owns all ActionCue processing — validation, dispatch, hooks, and result delivery.

• Hook phases: before_dispatch, after_dispatch, wrap_dispatch
• Registration layers: cue_layer (from CueHandler), node_layer (from NodeEngine)
• Result sink: injectable callable; defaults to NNG NodeOperation.STATUS via NodeCommunications.send_operation

See action-handler-extensibility contract for integration details.

Dedicated action-cue execution, extension hooks, and optional result sink.

ActionHandler

Owns ActionCue validation, default handlers, hooks, and result delivery.

Source code in src/cuemsengine/cues/ActionHandler.py

class ActionHandler:
    """
    Owns ActionCue validation, default handlers, hooks, and result delivery.
    """

    def __init__(self) -> None:
        self._cue_handler: CueOrchestrator | None = None
        self._lock = threading.Lock()
        self._hooks: dict[
            tuple[str, str, frozenset[str]], Callable[[ActionHookContext], Any]
        ] = {}
        self._result_sink: Callable[[dict], None] | None = None
        self._emit_enabled: bool = True

    # ---- binding ----

    def bind_cue_handler(self, cue_handler: CueOrchestrator) -> None:
        """Bind the singleton cue orchestrator (arm, go, armed lookups)."""
        self._cue_handler = cue_handler

    def set_result_sink(self, sink: Callable[[dict], None] | None) -> None:
        """
        Replace result delivery; None restores default (NNG via comms thread).
        """
        with self._lock:
            self._result_sink = sink

    def set_emit_enabled(self, enabled: bool) -> None:
        """When False, suppress outcome emission (useful in tests)."""
        with self._lock:
            self._emit_enabled = enabled

    def clear_action_extensions(self) -> None:
        """Remove all hooks and custom sink (for isolated tests)."""
        with self._lock:
            self._hooks.clear()
            self._result_sink = None
            self._emit_enabled = True

    # ---- registration ----

    def register_action_hook(
        self,
        phase: HookPhase,
        fn: Callable[[ActionHookContext], Any],
        *,
        source: RegistrationLayer = "cue_layer",
        action_types: frozenset[str] | None = None,
    ) -> None:
        """
        Register a hook; last registration wins for the same (phase, source,
        filter).
        """
        filter_key = action_types if action_types is not None else _ALL_ACTIONS
        key = (phase, source, filter_key)
        with self._lock:
            self._hooks[key] = fn

    def unregister_action_hook(
        self,
        phase: HookPhase,
        *,
        source: RegistrationLayer,
        action_types: frozenset[str] | None = None,
    ) -> None:
        filter_key = action_types if action_types is not None else _ALL_ACTIONS
        key = (phase, source, filter_key)
        with self._lock:
            self._hooks.pop(key, None)

    def finalize_node_layer_bindings(self) -> None:
        """
        Call from NodeEngine after comms are ready (extension point; default
        no-op).
        """
        return

    # ---- hook resolution ----

    def _matching_hooks(
        self, phase: HookPhase, action_type: str
    ) -> list[tuple[str, Callable[[ActionHookContext], Any]]]:
        """Return (layer, fn) pairs: cue_layer first, then node_layer."""
        with self._lock:
            items = list(self._hooks.items())
        cue_hooks: list[tuple[str, Callable[[ActionHookContext], Any]]] = []
        node_hooks: list[tuple[str, Callable[[ActionHookContext], Any]]] = []
        for (ph, layer, filter_key), fn in items:
            if ph != phase or not _filter_matches(action_type, filter_key):
                continue
            if layer == "cue_layer":
                cue_hooks.append((layer, fn))
            else:
                node_hooks.append((layer, fn))
        return cue_hooks + node_hooks

    def _wrap_for_action(
        self, layer: RegistrationLayer, action_type: str
    ) -> Callable[..., Any] | None:
        with self._lock:
            best_specific: Callable[..., Any] | None = None
            best_all: Callable[..., Any] | None = None
            for (ph, src, filter_key), fn in self._hooks.items():
                if ph != "wrap_dispatch" or src != layer:
                    continue
                if not filter_key:
                    best_all = fn
                elif action_type in filter_key:
                    best_specific = fn
            return best_specific if best_specific is not None else best_all

    # ---- result delivery ----

    def _emit_outcome(self, outcome: dict) -> None:
        with self._lock:
            sink = self._result_sink
            emit = self._emit_enabled
        if not emit:
            return
        if sink is not None:
            try:
                sink(outcome)
            except Exception as exc:
                Logger.error(f"Custom action result sink raised: {exc}")
            return
        self._default_result_sink(outcome)

    def _default_result_sink(self, outcome: dict) -> None:
        ch = self._cue_handler
        if ch is None:
            return
        ct: NodeCommunications | None = getattr(ch, "communications_thread", None)
        if ct is None:
            return
        try:
            op = NodeOperation(
                type=OperationType.STATUS,
                action=ActionType.UPDATE,
                sender=ct.node_id,
                target="action_cue_outcome",
                data=dict(outcome),
            )
            ct.send_operation(op, timeout=0.1)
        except Exception as exc:
            Logger.debug(f"Default action outcome emit skipped: {exc}")

    # ---- main dispatch ----

    def execute_action(
        self,
        cue: ActionCue,
        mtc: MtcListener,
        frozen_mtc_ms: float | None = None,
    ) -> dict:
        action_type = cue.action_type
        target = cue._action_target_object

        if action_type not in SUPPORTED_CUE_ACTIONS:
            reason = f"Unsupported action_type: {action_type!r}"
            Logger.warning(reason)
            out = self._action_result("rejected", action_type, None, reason)
            self._emit_outcome(out)
            return out

        if target is None and action_type not in PROJECT_SCOPE_ACTIONS:
            reason = (
                f"Missing target for {action_type} "
                f"(action_target={cue.action_target!r})"
            )
            Logger.warning(reason)
            out = self._action_result("rejected", action_type, None, reason)
            self._emit_outcome(out)
            return out

        target_id = getattr(target, "id", None)
        ctx = ActionHookContext(
            cue=cue,
            target=target,
            mtc=mtc,
            action_type=action_type,
            target_id=target_id,
            outcome=None,
            cue_handler=self._cue_handler,
            frozen_mtc_ms=frozen_mtc_ms,
        )

        # before_dispatch hooks
        for _layer, hook_fn in self._matching_hooks("before_dispatch", action_type):
            try:
                hook_fn(ctx)
            except Exception as exc:
                reason = f"before_dispatch hook raised {type(exc).__name__}: {exc}"
                Logger.error(reason)
                out = self._action_result("failed", action_type, target_id, reason)
                self._emit_outcome(out)
                return out

        handler = _ACTION_HANDLERS.get(action_type)
        if handler is None:
            reason = f"No handler registered for {action_type}"
            Logger.error(reason)
            out = self._action_result("failed", action_type, target_id, reason)
            self._emit_outcome(out)
            return out

        ch = self._cue_handler

        def run_default() -> dict:
            return handler(ch, cue, target, mtc, frozen_mtc_ms)

        def apply_wraps() -> dict:
            inner: Callable[[], dict] = run_default
            for layer in ("node_layer", "cue_layer"):
                wfn = self._wrap_for_action(layer, action_type)
                if wfn is None:
                    continue
                prev = inner

                def make_wrapped(
                    w: Callable[..., Any] = wfn, p: Callable[[], dict] = prev
                ) -> Callable[[], dict]:
                    def _w() -> dict:
                        return w(ctx, p)

                    return _w

                inner = make_wrapped()
            return inner()

        dispatch_exc: bool
        try:
            has_wrap = any(
                self._wrap_for_action(layer, action_type) is not None
                for layer in ("cue_layer", "node_layer")
            )
            if has_wrap:
                result = apply_wraps()
            else:
                result = run_default()
            dispatch_exc = False
        except Exception as exc:
            dispatch_exc = True
            reason = f"{action_type} on {target_id} raised {type(exc).__name__}: {exc}"
            Logger.error(reason)
            result = self._action_result("failed", action_type, target_id, reason)

        ctx.outcome = result

        # after_dispatch hooks (skipped if default handler raised)
        if not dispatch_exc:
            for _layer, hook_fn in self._matching_hooks("after_dispatch", action_type):
                try:
                    hook_fn(ctx)
                except Exception as exc:
                    reason = f"after_dispatch hook raised {type(exc).__name__}: {exc}"
                    Logger.error(reason)
                    result = self._action_result(
                        "failed", action_type, target_id, reason
                    )
                    ctx.outcome = result
                    break
            scope = target_id if target_id is not None else "project"
            Logger.info(
                f'Action {action_type} on {scope}: {result["status"]}'
                + (f' ({result["reason"]})' if result.get("reason") else "")
            )

        self._emit_outcome(result)
        return result

    @staticmethod
    def _action_result(
        status: str,
        action_type: str,
        target_id: str | None,
        reason: str | None = None,
        origin_cue_id: str | None = None,
    ) -> dict:
        out = {
            "status": status,
            "action_type": action_type,
            "target_id": target_id,
            "reason": reason,
        }
        if origin_cue_id is not None:
            # Project-scope actions have no target; the sink needs to know
            # WHICH cue fired (the controller names it in its journal line).
            out["origin_cue_id"] = origin_cue_id
        return out

bind_cue_handler(cue_handler)

Bind the singleton cue orchestrator (arm, go, armed lookups).

Source code in src/cuemsengine/cues/ActionHandler.py

def bind_cue_handler(self, cue_handler: CueOrchestrator) -> None:
    """Bind the singleton cue orchestrator (arm, go, armed lookups)."""
    self._cue_handler = cue_handler

clear_action_extensions()

Remove all hooks and custom sink (for isolated tests).

Source code in src/cuemsengine/cues/ActionHandler.py

def clear_action_extensions(self) -> None:
    """Remove all hooks and custom sink (for isolated tests)."""
    with self._lock:
        self._hooks.clear()
        self._result_sink = None
        self._emit_enabled = True

finalize_node_layer_bindings()

Call from NodeEngine after comms are ready (extension point; default no-op).

Source code in src/cuemsengine/cues/ActionHandler.py

def finalize_node_layer_bindings(self) -> None:
    """
    Call from NodeEngine after comms are ready (extension point; default
    no-op).
    """
    return

register_action_hook(phase, fn, *, source='cue_layer', action_types=None)

Register a hook; last registration wins for the same (phase, source, filter).

Source code in src/cuemsengine/cues/ActionHandler.py

def register_action_hook(
    self,
    phase: HookPhase,
    fn: Callable[[ActionHookContext], Any],
    *,
    source: RegistrationLayer = "cue_layer",
    action_types: frozenset[str] | None = None,
) -> None:
    """
    Register a hook; last registration wins for the same (phase, source,
    filter).
    """
    filter_key = action_types if action_types is not None else _ALL_ACTIONS
    key = (phase, source, filter_key)
    with self._lock:
        self._hooks[key] = fn

set_emit_enabled(enabled)

When False, suppress outcome emission (useful in tests).

Source code in src/cuemsengine/cues/ActionHandler.py

def set_emit_enabled(self, enabled: bool) -> None:
    """When False, suppress outcome emission (useful in tests)."""
    with self._lock:
        self._emit_enabled = enabled

set_result_sink(sink)

Replace result delivery; None restores default (NNG via comms thread).

Source code in src/cuemsengine/cues/ActionHandler.py

def set_result_sink(self, sink: Callable[[dict], None] | None) -> None:
    """
    Replace result delivery; None restores default (NNG via comms thread).
    """
    with self._lock:
        self._result_sink = sink

ActionHookContext dataclass

Context passed to extension hooks (stable field names for integrators).

Source code in src/cuemsengine/cues/ActionHandler.py

@dataclass
class ActionHookContext:
    """
    Context passed to extension hooks (stable field names for integrators).
    """

    cue: ActionCue
    target: Cue | None
    mtc: MtcListener
    action_type: str
    target_id: str | None
    outcome: dict | None = None
    cue_handler: CueOrchestrator | None = None
    frozen_mtc_ms: float | None = None

CueOrchestrator

Bases: Protocol

Subset of CueHandler needed by action dispatch (avoids circular import).

Source code in src/cuemsengine/cues/ActionHandler.py

class CueOrchestrator(Protocol):
    """Subset of CueHandler needed by action dispatch (avoids circular import)."""

    def arm(self, cue: Cue, init: bool = False) -> bool: ...

    def disarm(self, cue: Cue) -> bool: ...

    def go_from(
        self,
        start_cue: Cue,
        mtc: MtcListener,
        seed_ms: float | None = None,
    ) -> Thread | None: ...

    def cancel_pending_descendants(self, cue: Cue) -> int: ...

CueHandler

Singleton class responsible for handling Cue objects.

Holds a list of armed cues and manages video players. Thread-safe: internal state mutations are guarded by a Lock.

Source code in src/cuemsengine/cues/CueHandler.py

class CueHandler:
    """
    Singleton class responsible for handling Cue objects.

    Holds a list of armed cues and manages video players.
    Thread-safe: internal state mutations are guarded by a Lock.
    """

    _instance: "CueHandler | None" = None

    # Instance attributes (declared for IDE/type checker support)
    _armed_cues: list[Cue]
    _armed_cues_set: set[str]
    _lock: Lock
    communications_thread: NodeCommunications

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            # Initialize instance attributes
            cls._instance._armed_cues = []
            cls._instance._armed_cues_set = set()
            cls._instance._lock = Lock()
        return cls._instance

    # ---------------------------
    # Communications To Controller
    # ---------------------------
    def set_nng_comms(self, hub_address: str, node_id: str):
        """Set the communications infrastructure"""
        from time import sleep

        Logger.info(f"Starting communications for Node {node_id}")
        Logger.info(f"NNG Hub address: {hub_address}")
        self.communications_thread = NodeCommunications(
            hub_address=hub_address, node_id=node_id
        )
        self.communications_thread.start()

        # Wait for NNG thread to initialize (prevents race condition in
        # nni_random)
        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")

    # ---------------------------
    # Armed Cues List Methods
    # ---------------------------

    def add_armed_cue(self, cue: Cue) -> None:
        """Adds an armed cue to the list."""
        with self._lock:
            self._armed_cues.append(cue)
            self._armed_cues_set.add(cue.id)

    def get_armed_cues(self) -> list[Cue]:
        """Returns the list of armed cues."""
        with self._lock:
            return self._armed_cues

    def get_armed_cue(self, cue: Cue) -> Cue | None:
        """Returns the armed cue with the given uuid."""
        try:
            return self.get_armed_cues().index(cue)
        except ValueError:
            return None

    def find_armed_cue(self, cue: Cue) -> Cue | None:
        """Finds an armed cue with the given uuid."""
        with self._lock:
            return cue.id in self._armed_cues_set

    def remove_armed_cue(self, cue: Cue) -> bool:
        """Removes an armed cue from the list."""
        with self._lock:
            if cue.id in self._armed_cues_set:
                self._armed_cues.remove(cue)
                self._armed_cues_set.remove(cue.id)
                return True
        return False

    # ---------------------------
    # Cue Management
    # ---------------------------

    # Minimum effective duration (ms) for a cue to "count" as providing
    # enough time to arm subsequent cues during its playback.
    # Configurable per deployment. Default 1000ms covers 4K video decode.
    _ARM_WINDOW_THRESHOLD_MS = 1000

    # Maximum cues to walk ahead. Prevents runaway on pathological chains.
    _MAX_LOOKAHEAD_DEPTH = 15

    # Total time one arm() call may spend waiting for an arm of the same cue
    # that another thread holds. A ceiling on the WAIT only: a waiter that
    # wakes inside it and finds the cue still unarmed arms it itself.
    _ARM_WAIT_TIMEOUT_S = 5.0

    # Slack before a cue reached after its own anchor is reported as late.
    # One frame at 25 fps — below that it is dispatch jitter, not a mistimed
    # chain, and warning on it would flood the journal.
    _LATE_DISPATCH_TOLERANCE_MS = 40

    # How far ahead of its start a cue performs its HELD setup (run_cue).
    # Auto continue unrolls the whole chain from the trigger, so without this
    # every cue would run_cue at GO — for video that is N layers decoding
    # invisibly for the whole of their prewaits. Configurable per deployment;
    # 5s covers 4K decode-at-offset with room to spare.
    _RUN_AHEAD_MS = 5000

    # ---- chain epoch -----------------------------------------------------
    # Class-level DEFAULTS only: the first bump shadows them with an instance
    # attribute, so two handlers (the singleton and a test-built one) keep
    # independent counters, and a handler built via object.__new__ still reads
    # a sane value.
    #
    # _chain_epoch identifies one "pass" of a chain. It is bumped for every
    # fresh chain entry (manual GO, go_from re-entry, enable-rejoin) and
    # carried UNCHANGED through that pass's continuations, so a dispatch can
    # be told apart from a dispatch belonging to a newer pass. Per-cue
    # `_go_epoch` is the high-water mark of the newest pass that owns the cue.
    # Needed because dispatch now happens at chain entry: two cascades can
    # reach the same cue concurrently and `_go_generation` (last-writer-wins)
    # cannot order them.
    _chain_epoch = 0

    # The epoch a STOP claimed for itself. Every continuation of a pass that
    # started at or before it is refused — this is what makes a STOP landing
    # inside a cue's ~15s arm window stick (F4).
    _last_stop_chain_epoch = 0

    # ---- disarm epoch ----------------------------------------------------
    # Class-level default, like the two above. Bumped whenever everything a
    # cue was armed with is about to be torn down: in stop_all_cues() -- the
    # first step of a STOP and of a load, BEFORE the DMX/video reset and the
    # audio-player kill -- and again in disarm_all(), together with its
    # snapshot. An arm belongs to the epoch in which it was requested; one
    # whose epoch has moved is never published (see arm()).
    _disarm_epoch = 0

    # Cross-node Auto follow relay (cues.FollowRelay), attached by NodeEngine.
    # None (unit tests, an engine without comms): nothing is announced.
    _follow = None

    def arm_epoch(self) -> int:
        """The current STOP/load epoch. Whoever requests an arm on one thread
        and performs it on another captures this at request time and passes
        it along (arm(epoch=), _ArmWalk.epoch, _arm_ahead(arm_epoch=))."""
        return self._disarm_epoch

    @staticmethod
    def _effective_duration_ms(cue: Cue) -> float:
        """Effective time a cue occupies: prewait + body + postwait.

        prewait/postwait are always CTimecode (format_timecode returns
        CTimecode() for None/empty). CTimecode(0) is truthy but
        .milliseconds_exact returns 0.0.
        """
        pre = cue.prewait.milliseconds_exact
        post = cue.postwait.milliseconds_exact

        if isinstance(cue, CueList):
            # container — body is intentionally 0 for chain-anchoring Σ math.
            # A CueList mid-chain therefore contributes only its own pre+post;
            # its children's duration does not push following cues' anchors.
            # (CueLists rarely appear as post_go='go' targets; revisit if that
            # changes — compute body from children then.)
            body = 0
        elif isinstance(cue, (AudioCue, VideoCue)):
            try:
                body = (
                    CTimecode(cue.media.duration).milliseconds_exact if cue.media else 0
                )
            except Exception:
                body = 0
            if (
                body == 0
                and getattr(cue, "enabled", True)
                and not getattr(cue, "_body0_logged", False)
            ):
                # An enabled A/V cue with zero body feeds a zero-length slot into
                # the chain anchors — every following cue shifts earlier on THIS
                # node only (media missing/unreadable here but present elsewhere)
                # → silent cross-node desync. Surface it loudly (Fable 3.5).
                # Log ONCE per cue: this runs per arm pass / GO walk / chain hop,
                # so an unguarded log would flood the journal for one broken cue.
                Logger.error(
                    f"{type(cue).__name__} {cue.id} enabled but body==0 "
                    f"(media missing/unreadable?); chain anchor timing will be "
                    f"wrong on this node."
                )
                try:
                    cue._body0_logged = True
                except Exception:
                    pass
        elif isinstance(cue, DmxCue):
            # fadein_time/fadeout_time stored in MILLISECONDS (authoritative:
            # run_dmxCue in run_cue.py reads fadein_ms then fade_time = fadein_ms/1000).
            # fadeout_time exists in model but not yet implemented (always 0.0).
            fadein = getattr(cue, "fadein_time", 0) or 0
            fadeout = getattr(cue, "fadeout_time", 0) or 0
            body = fadein + fadeout  # already ms
        elif isinstance(cue, ActionCue):
            # play/stop/enable/disable/go_to = instant
            # TODO: use fade duration once fade_in/fade_out implemented
            body = 0
        else:
            body = 0

        return pre + body + post

    @staticmethod
    def _chain_advance_ms(cue: Cue) -> float:
        """Timeline advance a cue contributes to a post_go='go' (Auto continue)
        chain: **zero**.

        Auto continue triggers the WHOLE chain at once. Every cue in it shares
        one arrival — the chain trigger — and the only thing that separates the
        cues is each cue's own prewait:

            arrival(k) = chain_trigger
            start(k)   = chain_trigger + prewait(k)

        So a cue's own waits must not push the cues after it. `prewait` is a
        per-cue offset from the common trigger, and `postwait` does not gate a
        chain whose pointer has already advanced — it stays meaningful for
        Auto follow / Auto pause (a real tail after the body) and for this
        cue's own illumination window, but it contributes nothing here.

        Previously this returned `prewait + postwait`, which made each cue's
        prewait cascade into every following cue: prewaits authored as absolute
        offsets from GO (5/35/65/95/380 s) played as 5/40/105/200/580.
        Castillo Medina del Campo, 2026-08-14 — ClickUp 869ej3cc8.

        ⚠ Returning anything non-zero here is NOT just an arithmetic change.
        The chain is now dispatched at its trigger and every cue parks on
        `arrival + prewait`, so a non-zero advance would push a cue's anchor
        past the moment its own thread already started waiting on — and it
        would put the arrival of a rejoining cue (Plans/
        prewait-dispatch-reorder-phase2.md §6) out of step with the pass that
        stamped it. Re-audit that design before changing this value.

        Auto follow (post_go='go_at_end') breaks the chain walk and fires its
        next cue from loop_cue *after* the body, so it never reaches this
        accumulator — its body wait is real, not anchored. See CLAUDE.md
        "Cue play modes & pre/post-wait semantics".
        """
        return 0.0

    def _wait_mtc(
        self, cue: Cue, mtc: MtcListener, target_ms: float, go_gen: int = 0
    ) -> str:
        """Block until live MTC reaches target_ms; return 'reached' or 'stopped'.

        Shared timeline gate used both to illuminate a cue at its arrival and to
        reveal it at its start. Exits early on STOP (`_stop_requested`) or a
        newer GO/reload (`_go_generation` changed). Deliberately does NOT bail on
        a recoverable MTC stall — it self-recovers (fires late when timecode
        resumes). milliseconds_exact is wrap-accumulated → 24h-safe.
        """
        while mtc.main_tc.milliseconds_exact < target_ms:
            if (
                getattr(cue, "_stop_requested", False)
                or getattr(cue, "_go_generation", 0) != go_gen
            ):
                return "stopped"
            sleep(0.02)
        # Final flag check AFTER the loop: a STOP landing in the last poll
        # interval (~20ms) would otherwise be reported as 'reached' and let the
        # caller proceed to reveal/fire right after the operator stopped.
        if (
            getattr(cue, "_stop_requested", False)
            or getattr(cue, "_go_generation", 0) != go_gen
        ):
            return "stopped"
        return "reached"

    def _arm_ahead(
        self,
        start_cue: Cue,
        should_continue: Callable[[], bool] | None = None,
        skip_arming_types: tuple[type, ...] = (),
        arm_epoch: int | None = None,
    ) -> list[Cue]:
        """Arm ahead in the target chain until 2 cues with meaningful
        duration are armed. Short/zero-duration cues are armed but don't
        count. CueList targets are skipped (handled by
        initial_cuelist_process).

        Cues belonging to a DIFFERENT node (not `_local`) are also skipped,
        transparently — arm() is a no-op for them anyway, so counting them
        toward the budget (as this used to) starved THIS node's own segment
        of any lookahead at all on a mixed-node chain: node01/controller
        never pre-armed on Badajoz's actual chain shape, 2026-09-25,
        869f79ecc. Unlike the CueList/disabled skips, a non-local run does
        NOT consume `_MAX_LOOKAHEAD_DEPTH` — a whole other node's segment can
        be longer than that — so it is bounded instead by the same
        1024-step cycle guard `_next_local_fire` already uses against an
        all-remote chain.

        should_continue: an optional predicate, re-checked before each cue.
        Returning False stops the walk at that point without touching
        anything further. Used by the PreArm background thread
        (NodeEngine.set_next_cue) to abandon a stale selection or a
        reloaded project instead of racing the new state.

        skip_arming_types: cue types this call must not arm() or count —
        left armed for whichever single-threaded path reaches them later
        instead. Only AudioCue is ever passed here in production: two
        threads racing to arm the same slow audio cue is exactly the
        Badajoz-adjacent race this fix must not introduce (see
        NodeEngine.set_next_cue and AudioMixer.connect_player_to_outputs).
        Unlike the non-local skip above, this DOES consume the depth
        budget — it is still this node's own chain, just a type this call
        chooses not to arm.

        arm_epoch: the STOP/load epoch this walk was requested in (see
        arm_epoch()). Once it has moved the walk stops, and arm() refuses or
        abandons whatever it was still arming -- nothing is armed for a
        request that predates a STOP or a load.

        Returns the cues THIS call transitioned from unloaded to loaded —
        not merely already-armed ones, and not ones a concurrent call
        armed.
        """
        target = getattr(start_cue, "_target_object", None)
        counted = 0
        walked = 0
        total_steps = 0
        newly_armed: list[Cue] = []
        # Only a background walk (NodeEngine's PreArm) passes these; then its
        # rules reach arm()'s own recursion and newly_armed is the walk's
        # exact record. Every other caller keeps arm()'s original call shape.
        walk = (
            _ArmWalk(skip_arming_types, should_continue, newly_armed, arm_epoch)
            if (
                should_continue is not None
                or skip_arming_types
                or arm_epoch is not None
            )
            else None
        )

        while (
            isinstance(target, Cue)
            and counted < 2
            and walked < self._MAX_LOOKAHEAD_DEPTH
        ):
            total_steps += 1
            if total_steps > 1024:
                Logger.error(
                    f"_arm_ahead from {start_cue.id} hit the cross-node "
                    "safety limit (1024 steps); aborting"
                )
                break
            if should_continue is not None and not should_continue():
                Logger.info(
                    f"_arm_ahead from {start_cue.id} stopped early — caller "
                    "asked to abort (stale selection or reloaded project)"
                )
                break
            if arm_epoch is not None and arm_epoch != self._disarm_epoch:
                Logger.info(
                    f"_arm_ahead from {start_cue.id} stopped early — a STOP "
                    "or a load happened since it was requested"
                )
                break
            if isinstance(target, CueList):
                # CueLists are containers — skip, don't count
                target = getattr(target, "_target_object", None)
                walked += 1
                continue
            if not target.enabled:
                target = getattr(target, "_target_object", None)
                walked += 1
                continue
            if not getattr(target, "_local", False):
                # Another node's cue — transparent, doesn't spend the depth
                # budget (see docstring).
                target = getattr(target, "_target_object", None)
                continue
            if isinstance(target, skip_arming_types):
                target = getattr(target, "_target_object", None)
                walked += 1
                continue
            already_loaded = getattr(target, "loaded", False)
            if not already_loaded:
                if walk is not None:
                    self.arm(target, init=True, walk=walk)
                else:
                    self.arm(target, init=True)
                    if getattr(target, "loaded", False):
                        newly_armed.append(target)
            if self._effective_duration_ms(target) >= self._ARM_WINDOW_THRESHOLD_MS:
                counted += 1
            target = getattr(target, "_target_object", None)
            walked += 1

        if walked >= self._MAX_LOOKAHEAD_DEPTH and counted < 2:
            Logger.warning(
                f"_arm_ahead hit depth limit ({self._MAX_LOOKAHEAD_DEPTH}) "
                f"from cue {start_cue.id} with only {counted}/2 real-duration "
                f"cues found. Remaining cues will rely on safety-net re-arm."
            )

        return newly_armed

    def _arming_registry(self) -> dict:
        """Arms in flight, keyed by cue id. Call with self._lock held.

        Keyed by ID, not held on the cue object: what an arm creates is keyed
        by id (the tracked player, the JACK client name Audio_Player-<uuid>,
        the video layer ids) and Cue.__eq__/__hash__ are by id, so the old and
        the new object of one cue -- a project loaded twice -- must exclude
        each other. Created lazily: handlers built with object.__new__ (tests)
        never ran __new__.
        """
        registry = self.__dict__.get("_arming")
        if registry is None:
            registry = self.__dict__.setdefault("_arming", {})
        return registry

    def arm(
        self,
        cue: Cue,
        init=False,
        walk: _ArmWalk | None = None,
        epoch: int | None = None,
        wait_report: dict | None = None,
    ) -> bool:
        """Arms a cue by appending it to the armed_cues list.

        wait_report: optional dict. When this call waits for another thread's
        arm of the same cue, it sets wait_report["waited_s"] (total seconds
        waited) and wait_report["holder"] (that thread's name), so a caller
        can say what really happened instead of guessing from a check made
        before the call (869fbyjzx).

        walk: set only by a background _arm_ahead walk (NodeEngine's PreArm).
        Its excluded types and should_continue checkpoint also govern this
        call's own post_go / ActionCue-target recursion, and every cue this
        call chain actually arms is recorded in walk.armed (869f79ecc).

        One arm is in flight per cue id (_arming_registry). An init caller
        that finds one waits for it, up to _ARM_WAIT_TIMEOUT_S in total, and
        then re-evaluates: if the cue is still unarmed (the arm it waited on
        failed or was abandoned) it arms the cue itself instead of giving up.

        PUBLISH OR ABANDON. An arm belongs to the STOP/load epoch in which it
        was REQUESTED: `epoch` if given, else the walk's, else _disarm_epoch
        as read here at entry. It is never re-read. If it has moved:
        - before the cue is claimed (entry, or waking from a wait): return
          False and arm nothing -- even if the cue is loaded by now, because
          then the NEW epoch loaded it and a stale caller must not adopt it;
        - while arm_cue() was running: the arm is ABANDONED at the publish
          gate. The cue is never marked loaded and never enters the armed
          list, so no GO can pick it up; what this arm built (a player the
          STOP may already have killed, video layers) is released here,
          while this thread still holds the cue's claim.
        The gate also abandons a cue that was disabled while it was arming.
        So no thread ever has to disarm a cue after the fact, which is what
        used to cut a cue the next run was already playing (869f9wqpn).
        """
        if cue is None:
            return False
        if epoch is None:
            epoch = (
                walk.epoch
                if walk is not None and walk.epoch is not None
                else self._disarm_epoch
            )
        if walk is not None:
            if walk.skip_types and isinstance(cue, walk.skip_types):
                return False
            if walk.should_continue is not None and not walk.should_continue():
                return False

        wait_deadline = None
        waited_total = 0.0
        while True:
            needs_disarm = False
            claim = None
            pending = None
            stale = False

            with self._lock:
                registry = self._arming_registry()
                in_flight = registry.get(cue.id)
                if epoch != self._disarm_epoch:
                    # Checked FIRST, on entry and on every wake.
                    stale = True
                elif hasattr(cue, "loaded") and cue.loaded:
                    if not cue.enabled:
                        needs_disarm = True
                elif in_flight is not None:
                    if not init:
                        # Non-init callers just register; no need to wait
                        return False
                    # Another thread is arming — wait for it outside the lock
                    pending = in_flight
                    pending.waiters += 1
                elif not init:
                    if cue.id not in self._armed_cues_set:
                        self._armed_cues.append(cue)
                        self._armed_cues_set.add(cue.id)
                elif cue._local and cue.enabled:
                    # Claim the cue inside the lock to block concurrent arm
                    # attempts; released in the finally below (outside the
                    # lock — intentional: avoids holding it during arm_cue()).
                    claim = _ArmClaim(holder=current_thread().name, started=monotonic())
                    registry[cue.id] = claim

            if stale:
                Logger.info(
                    f"Not arming {type(cue).__name__} {cue.id}: a STOP or a "
                    "load happened since this arm was requested"
                )
                return False

            if pending is None:
                break

            if wait_deadline is None:
                wait_deadline = monotonic() + self._ARM_WAIT_TIMEOUT_S
            remaining = wait_deadline - monotonic()
            Logger.debug(
                f"Waiting for in-progress arm of {type(cue).__name__} {cue.id}"
            )
            wait_started = monotonic()
            woke = remaining > 0 and pending.event.wait(timeout=remaining)
            waited_total += monotonic() - wait_started
            if wait_report is not None:
                wait_report["waited_s"] = waited_total
                wait_report["holder"] = pending.holder
            with self._lock:
                pending.waiters -= 1
            if not woke:
                Logger.warning(
                    f"Timed out waiting for arm of {cue.id} (held by "
                    f"{pending.holder} for {monotonic() - pending.started:.1f}s)"
                )
                return (
                    bool(getattr(cue, "loaded", False)) and epoch == self._disarm_epoch
                )
            # Woken: loop and look again. Epoch moved -> give up; loaded ->
            # done; still unarmed -> this thread claims it and arms it itself.

        # Disarm disabled-but-loaded cues outside lock (disarm acquires lock)
        if needs_disarm:
            self.disarm(cue, reason="disabled")
            return False

        if claim is None:
            return not needs_disarm

        abandoned = None
        try:
            Logger.info(f"Arming {type(cue).__name__} {cue.id}")
            arm_started = monotonic()
            arm_cue(cue)
            # The publish gate. Same lock as the epoch bumps (stop_all_cues,
            # disarm_all): either this section comes first and the STOP finds
            # the cue in the armed list and disarms it itself, or the bump
            # comes first and the cue is never published.
            with self._lock:
                if epoch != self._disarm_epoch:
                    abandoned = "a STOP or a load happened while it was arming"
                elif not cue.enabled:
                    abandoned = "it was disabled while it was arming"
                else:
                    cue.loaded = True
                    # Measurement only (the armed-to-start and never-played
                    # lines): set at every publish, so a cue object reused
                    # across runs never carries the previous arm's stamps.
                    cue._armed_at = monotonic()
                    cue._ever_played = False
                    if cue.id not in self._armed_cues_set:
                        self._armed_cues.append(cue)
                        self._armed_cues_set.add(cue.id)
            if abandoned is not None:
                Logger.info(
                    f"Abandoned arm of {type(cue).__name__} {cue.id}: {abandoned}"
                )
                self._release_unpublished(cue, "abandoned_arm")
                return False
            # Engine side only: for audio that is the spawn and the JACK port
            # wait; for video it is the OSC sends, and the file open is the
            # videocomposer's own "AsyncVideoLoader: Loaded ... in Nms".
            Logger.info(
                f"Armed {type(cue).__name__} {cue.id} in "
                f"{cue._armed_at - arm_started:.3f} s"
            )
            if walk is not None:
                walk.armed.append(cue)
            if isinstance(cue, AudioCue):
                try:
                    self.communications_thread.add_player(
                        f"audioplayer_{cue.id}", None, timeout=0.1
                    )
                except Exception:
                    pass
        except Exception as e:
            # Fail loud, fail local. _arm_ahead() runs on the GO daemon
            # thread, so letting this propagate would kill a *playing* cue's
            # thread before loop_cue() and skip its generation-tracked
            # cleanup — far worse than one broken cue. Leave the cue
            # unloaded and let go()'s fallback re-arm retry it; if that also
            # fails, go() raises where an operator can see it.
            Logger.error(f"Failed to arm {type(cue).__name__} {cue.id}: {e}")
            Logger.exception(e)
            cue.loaded = False
            # ...and give back whatever it got as far as building (a spawned
            # player, registered layers), still under this cue's claim.
            self._release_unpublished(cue, "arm_failed")
            return False
        finally:
            # Always, and before the recursion below: a leaked claim would
            # block every later arm of this cue id, across reloads too, and a
            # claim held through the recursion would deadlock a go-chain cycle.
            with self._lock:
                registry = self._arming_registry()
                if registry.get(cue.id) is claim:
                    del registry[cue.id]
            claim.event.set()

        # Recursive arms — only reached if cue was actually armed.
        # The claim is released by now; the loaded guard prevents re-arm.
        if cue.post_go == "go" and cue._target_object:
            if cue._target_object.enabled:
                if walk is not None:
                    self.arm(cue._target_object, init, walk=walk, epoch=epoch)
                else:
                    self.arm(cue._target_object, init, epoch=epoch)

        # ActionCue(play) and FadeCue(fade_action) + target = 1 unit. Arm
        # target
        # so it's ready when the action fires (ActionCue has zero duration;
        # FadeCue
        # expects target_cue already armed before reading its OSC cache).
        if isinstance(cue, ActionCue) and cue._action_target_object:
            if cue.action_type in ("play", "fade_action"):
                if walk is not None:
                    self.arm(cue._action_target_object, init, walk=walk, epoch=epoch)
                else:
                    self.arm(cue._action_target_object, init, epoch=epoch)

        return True

    def armed_inventory(self) -> dict:
        """What this node holds armed right now: published (loaded) cues by
        type, the video layers they hold, and how many of them a GO owns
        (playing) versus are only held (idle). CueLists are left out: they
        hold nothing. Measurement only."""
        with self._lock:
            held = [
                c
                for c in self._armed_cues
                if getattr(c, "loaded", False) and not isinstance(c, CueList)
            ]
        inv = {
            "cues": len(held),
            "video": 0,
            "layers": 0,
            "audio": 0,
            "dmx": 0,
            "other": 0,
            "playing": 0,
            "idle": 0,
        }
        for cue in held:
            if isinstance(cue, VideoCue):
                inv["video"] += 1
                inv["layers"] += len(getattr(cue, "_layer_ids", None) or [])
            elif isinstance(cue, AudioCue):
                inv["audio"] += 1
            elif isinstance(cue, DmxCue):
                inv["dmx"] += 1
            else:
                inv["other"] += 1
            inv["playing" if getattr(cue, "_playing", False) else "idle"] += 1
        return inv

    def log_armed_inventory(self, where: str) -> None:
        """One INFO line with armed_inventory(). Never raises: it is called
        at the end of a GO and of a PreArm thread, which must not fail on a
        log line."""
        try:
            inv = self.armed_inventory()
            Logger.info(
                f"Armed inventory after {where}: {inv['cues']} cues "
                f"(video {inv['video']} cues / {inv['layers']} layers, "
                f"audio {inv['audio']} players, dmx {inv['dmx']}, "
                f"other {inv['other']}); playing {inv['playing']}; "
                f"idle {inv['idle']}"
            )
        except Exception as e:
            Logger.warning(f"Could not count the armed inventory ({where}): {e}")

    @staticmethod
    def _log_arm_wait(
        cue: Cue, where: str, in_flight: str | None, wait_report: dict
    ) -> None:
        """Report what arm() did about an arm in flight (869fbyjzx)."""
        waited = wait_report.get("waited_s")
        if waited is not None:
            Logger.info(
                f"Cue {cue.id} waited {waited:.2f}s at {where} for the arm in "
                f"progress (held by {wait_report.get('holder')})"
            )
        elif in_flight is not None:
            Logger.info(
                f"Cue {cue.id}: the arm in progress at {where} had finished "
                f"before this call looked; no wait"
            )

    def describe_arm_in_flight(self, cue: Cue) -> str | None:
        """Who is arming this cue right now and for how long, as text for a
        log line; None when no arm of it is in flight."""
        with self._lock:
            claim = self._arming_registry().get(cue.id)
        if claim is None:
            return None
        return f"held by {claim.holder} for {monotonic() - claim.started:.1f}s"

    def _release_unpublished(self, cue: Cue, reason: str) -> None:
        """Give back what an arm built for a cue it is not going to publish.

        Only ever called by the thread that holds the cue's arm claim, so
        nobody else can have created resources for that cue id meanwhile.
        Never raises: the caller is about to release the claim, and a claim
        that leaked here would block every later arm of this cue.
        """
        try:
            self._release_cue_resources(cue, reason, what="Released")
        except Exception as e:
            Logger.error(
                f"Could not release what the unpublished arm of {cue.id} built "
                f"({reason}): {e}"
            )

    def disarm(self, cue: Cue, reason: str = "unspecified") -> bool:
        """Disarms a cue by removing it from the armed_cues list.

        ``reason`` names why (``cue_end``, ``stop_action``, ``disabled``,
        ``project_changed``, ``load``, ``ready_script``, ``shutdown``) and is
        logged: a disarm, the automatic one at cue end included, must never
        be silent.
        """
        cue._playing = False
        if hasattr(cue, "loaded") and cue.loaded:
            self.remove_armed_cue(cue)
            cue.loaded = False
            armed_at = getattr(cue, "_armed_at", None)
            if (
                armed_at is not None
                and not getattr(cue, "_ever_played", False)
                and reason != "cue_end"
                and not isinstance(cue, CueList)
            ):
                # An arm that bought nothing: what a pre-arm policy costs.
                # Not a CueList: go() never dispatches one (the project's
                # root is armed at every load and STOP) and it holds nothing.
                Logger.info(
                    f"Cue {cue.id} disarmed after {monotonic() - armed_at:.1f} s "
                    f"armed, never played ({reason})"
                )
            try:
                if isinstance(cue, AudioCue):
                    self.communications_thread.remove_player(
                        f"audioplayer_{cue.id}", timeout=0.1
                    )
                self.communications_thread.remove_cue(cue.id, timeout=0.1)
            except Exception:
                pass

            self._release_cue_resources(cue, reason)
            return True

        return False

    def _release_cue_resources(
        self, cue: Cue, reason: str, what: str = "Disarmed"
    ) -> None:
        """Release what arm_cue() built for a cue: its video layers and its
        player. The resource half of disarm(), also used for an arm that is
        abandoned or fails before the cue is ever published (arm()).

        Deliberately touches neither `_playing`, nor the armed list, nor the
        controller's cue status: an unpublished cue has none of them.
        """
        if isinstance(cue, VideoCue):
            layer_ids = getattr(cue, "_layer_ids", [])
            client = getattr(cue, "_osc", None)
            unloaded, skipped = [], []
            if client and layer_ids:
                for layer_id in layer_ids:
                    # /videocomposer/reset (STOP, load, ready_script) runs
                    # before disarm_all on purpose (instant blackout) and
                    # already removed this layer and its endpoints.
                    if not PLAYER_HANDLER.is_layer_registered(layer_id):
                        skipped.append(layer_id)
                        continue
                    # Hiding is cosmetic (unload removes the layer anyway):
                    # a failure here must not skip the unload.
                    try:
                        client.set_value(f"/videocomposer/layer/{layer_id}/visible", 0)
                    except Exception as e:
                        Logger.warning(
                            f"Could not hide video layer {layer_id} (visible 0)"
                            f" of cue {cue.id} ({reason}): {e}"
                        )
                    try:
                        client.set_value("/videocomposer/layer/unload", layer_id)
                        client.remove_layer_endpoints(layer_id)
                        PLAYER_HANDLER.deregister_layer(layer_id)
                        unloaded.append(layer_id)
                    except Exception as e:
                        # Left registered: the next /reset still cleans it up.
                        Logger.warning(
                            f"Could not unload video layer {layer_id} of cue"
                            f" {cue.id} ({reason}): {e}"
                        )
            cue._layer_ids = []
            Logger.debug(
                f"{what} video cue {cue.id} ({reason}): unloaded {unloaded};"
                f" skipped {skipped} (no longer tracked: removed by reset or quit)"
                + ("" if client else "; no video client")
            )
        else:
            Logger.debug(f"{what} {type(cue).__name__} {cue.id} ({reason})")

        PLAYER_HANDLER.remove_cue_player(cue)

    def stop_all_cues(self) -> None:
        """Signal all armed cues to stop their playback loops.

        Also bumps each cue's generation counter so that any still-running
        go_threaded threads will see a mismatch and skip post-loop cleanup
        (disarm), which would otherwise undo the re-arm that follows.
        """
        with self._lock:
            # Claim an epoch for this STOP and record it: every continuation
            # of a pass that started at or before it is now refused, including
            # one currently blocked inside go()'s re-arm fallback whose cue is
            # not in _armed_cues yet (F4).
            self._chain_epoch = self._chain_epoch + 1
            self._last_stop_chain_epoch = self._chain_epoch
            # The teardown starts HERE: between this call and disarm_all()
            # the engine resets DMX and video and kills every audio player.
            # Moving the epoch in this same section means an arm publishes
            # either before it (then it is in _armed_cues, flagged just
            # below and disarmed later) or after it (then it is abandoned).
            self._disarm_epoch = self._disarm_epoch + 1
            for cue in self._armed_cues:
                cue._stop_requested = True
                cue._go_generation = getattr(cue, "_go_generation", 0) + 1
                cue._go_epoch = self._chain_epoch
                cue._playing = False

    def cancel_pending_descendants(self, cue: Cue) -> int:
        """Cancel the cues an Auto-continue chain dispatched behind `cue`.

        A chained cue only exists because its predecessor ran, so stopping a
        cue must stop everything scheduled behind it. Under Phase 2 the whole
        chain is dispatched at the trigger, so by now those cues are parked on
        their own anchors and nothing else would hold them back — the
        `_stop_requested` guard that used to sit in front of the dispatch has
        moved to chain entry.

        Never touches a cue that has already produced output (`_revealed`):
        that one is on stage and stays there. The walk continues PAST it,
        because the cues behind it still depend on the cue being stopped.
        Non-local cues are skipped too — the node that owns them runs this same
        walk on its own graph (ActionCues are local everywhere), so the union
        covers the cluster.

        Returns how many cues were cancelled.
        """
        cancelled = []
        with self._lock:
            # The pass this cue belongs to. Cues that a NEWER pass has since
            # claimed are not ours to cancel — a loop-back can already have
            # re-scheduled them against a fresh trigger.
            target_epoch = getattr(cue, "_go_epoch", 0)
            self._chain_epoch = self._chain_epoch + 1
            epoch = self._chain_epoch
            # Mark the target too: a dispatch of it currently blocked in its
            # own arm must not come back to life after this.
            cue._go_epoch = epoch

            node = (
                getattr(cue, "_target_object", None)
                if getattr(cue, "post_go", None) == "go"
                else None
            )
            # A circular project (A→B→C→A) would otherwise re-mark the same
            # cues on every revolution until the safety limit, reporting ~1024
            # cancellations and firing that many remove_cue sends.
            seen = set()
            while node is not None and id(node) not in seen:
                seen.add(id(node))
                is_local = getattr(node, "_local", False)
                owned = getattr(node, "_go_epoch", 0) <= target_epoch
                if is_local and owned and getattr(node, "enabled", False):
                    if not getattr(node, "_revealed", False):
                        node._stop_requested = True
                        node._go_generation = getattr(node, "_go_generation", 0) + 1
                        node._go_epoch = epoch
                        node._playing = False
                        cancelled.append(node)
                elif is_local and owned:
                    # Disabled: nothing to cancel, but claim it so enabling it
                    # later cannot rejoin a chain that was stopped.
                    node._go_epoch = epoch
                if getattr(node, "post_go", None) != "go":
                    # A chain break is a hand-off point, not a descendant.
                    break
                node = getattr(node, "_target_object", None)
                if len(seen) > 1024:
                    Logger.error(
                        "cancel_pending_descendants hit safety limit; aborting"
                    )
                    break

        # Outside the lock: a cancelled cue's thread leaves through the
        # generation guard, which sits upstream of the usual remove_cue, and
        # the whole chain has been lit since the trigger — so the highlight
        # has to be dropped here or the UI keeps showing dead cues as running.
        for node in cancelled:
            self._end_illumination(node)
        if cancelled:
            # str(): a cue's id is a Uuid, not a str, and join() would raise —
            # inside the one log line that tells an operator this happened.
            Logger.info(
                f"Stop of cue {cue.id} cancelled {len(cancelled)} scheduled "
                f"cue(s) behind it: {', '.join(str(n.id) for n in cancelled)}"
            )
        return len(cancelled)

    def cancel_parked(self, cue: Cue) -> bool:
        """Cancel a cue that was dispatched but has not produced output yet.

        Used when a cue is DISABLED after its chain was dispatched. Without
        this the cue would still fire: the whole chain leaves at the trigger
        and nothing downstream re-reads `enabled`. A cue that already
        revealed is left alone — disabling never cuts live playback.

        The cue keeps a stamp of the pass it was cancelled out of, so
        re-enabling it before its slot puts it back (rejoin_chain).
        """
        if getattr(cue, "_revealed", False) or not getattr(cue, "_playing", False):
            return False
        with self._lock:
            self._chain_epoch = self._chain_epoch + 1
            epoch = self._chain_epoch
            cue._stop_requested = True
            cue._go_generation = getattr(cue, "_go_generation", 0) + 1
            cue._go_epoch = epoch
            cue._playing = False
            # Remember where it would have played, so a re-enable can put it
            # back at the same anchor instead of losing the slot.
            arrival = getattr(cue, "_dispatch_arrival_ms", None)
            if arrival is not None:
                cue._chain_pass = (epoch, arrival)
        self._end_illumination(cue)
        Logger.info(f"Cue {cue.id} disabled before its slot — cancelled")
        return True

    def rejoin_chain(self, cue: Cue, mtc: MtcListener) -> bool:
        """Put a cue enabled mid-pass back into its running chain.

        The chain walk stamped this cue with the pass that would have
        dispatched it while it was disabled. If that pass is still alive and
        the cue's slot has not gone by, dispatch it at its original anchor so
        it plays exactly when it was always going to.

        Dispatched with unroll=False: the rest of the chain is already
        running, and re-dispatching it would churn cues that are on stage.

        Returns True if the cue was put back.
        """
        if not getattr(cue, "enabled", False) or not getattr(cue, "_local", False):
            return False
        stamp = getattr(cue, "_chain_pass", None)
        if not stamp:
            return False
        epoch, arrival_ms = stamp
        if arrival_ms is None:
            return False

        with self._lock:
            barrier = self._last_stop_chain_epoch
            if epoch <= self._last_stop_chain_epoch:
                Logger.info(
                    f"Cue {cue.id} enabled, but the chain it belonged to was "
                    f"stopped — it will play on the next GO"
                )
                return False
            if getattr(cue, "_go_epoch", 0) > epoch:
                # A stop action (or a newer pass) claimed this cue after the
                # stamp: a stop takes its scheduled descendants with it, and
                # that outranks re-enabling one of them.
                Logger.info(
                    f"Cue {cue.id} enabled, but a stop already cancelled this "
                    f"part of the chain — it will play on the next GO"
                )
                return False

        start_ms = arrival_ms + cue.prewait.milliseconds_exact
        now_ms = mtc.main_tc.milliseconds_exact
        if start_ms < now_ms - self._LATE_DISPATCH_TOLERANCE_MS:
            # Firing it now would put it on stage at the wrong moment, which
            # mid-show is worse than not playing it at all.
            Logger.warning(
                f"Cue {cue.id} enabled after its slot (start={start_ms:.0f}ms, "
                f"mtc={now_ms:.0f}ms) — NOT firing it late; it will play on "
                f"the next GO"
            )
            return False

        Logger.info(
            f"Cue {cue.id} enabled before its slot — rejoining the running "
            f"chain at {start_ms:.0f}ms"
        )
        # require_stop_epoch closes the window between the check above and the
        # dispatch: the lock is dropped in between, and a STOP landing there
        # must not be overridden by the fresh epoch this go() mints.
        return (
            self.go(cue, mtc, arrival_ms, unroll=False, require_stop_epoch=barrier)
            is not None
        )

    def disarm_all(self, reason: str = "unspecified") -> None:
        """Disarms all cues.

        Epoch bump, snapshot and clearing the armed list are ONE lock
        section. Clearing the list after the loop (as this used to) wiped
        any cue a current-epoch arm published while the loop was killing
        players: loaded but not in the list, so arm() said "already armed",
        find_armed_cue() said no, and the cue was unplayable until a reload.
        """
        self.stop_all_cues()
        with self._lock:
            self._disarm_epoch = self._disarm_epoch + 1
            cues_snapshot = list(self._armed_cues)
            self._armed_cues = []
            self._armed_cues_set.clear()
        for cue in cues_snapshot:
            self.disarm(cue, reason=reason)

    def get_next_cue(self, cue: Cue) -> Cue | None:
        """Returns the next cue to be played."""
        return cue._target_object if cue._target_object else None

    # ---------------------------
    # Cue Execution
    # ---------------------------

    def _warn_if_late(self, cue: Cue, mtc: MtcListener, start_ms: float) -> None:
        """Say it out loud when a cue is reached after its own anchor.

        Firing late is recoverable; firing late silently mistimes a show and
        nobody knows why. Tolerance is one frame — below that it is jitter.
        """
        now_ms = mtc.main_tc.milliseconds_exact
        if start_ms < now_ms - self._LATE_DISPATCH_TOLERANCE_MS:
            Logger.warning(
                f"Cue {cue.id} reached after its anchor — firing LATE by "
                f"{now_ms - start_ms:.0f}ms (start={start_ms:.0f}ms, "
                f"mtc={now_ms:.0f}ms). Its slot had already passed when the "
                f"cue got here (seed in the past, a re-arm that overran its "
                f"prewait, or an MTC jump)."
            )

    def _end_illumination(self, cue: Cue) -> None:
        """Drop the sequence-view highlight for a cue that will not play.

        Under Auto continue a cue is lit from the trigger, so any path that
        abandons it before its normal end must clear the highlight or the UI
        shows a dead cue as running.
        """
        try:
            self.communications_thread.remove_cue(cue.id, timeout=0.1)
        except Exception:
            pass

    def stamp_pass(self, skipped: list, arrival_ms: float) -> int:
        """Mint a chain epoch and stamp `skipped` with it, dispatching nothing.

        For the case where a chain walk finds NO local+enabled cue on this
        node (every local cue in the chain is currently disabled): go() is
        never called, so nothing would mint an epoch, and enabling one of
        those cues later could not rejoin the pass. Returns the epoch.
        """
        with self._lock:
            self._chain_epoch = self._chain_epoch + 1
            epoch = self._chain_epoch
            self._stamp_skipped_locked(skipped, epoch, arrival_ms)
        return epoch

    def _stamp_skipped_locked(
        self, skipped: list | None, epoch: int, arrival_ms: float
    ) -> None:
        """Record the pass that would have dispatched each skipped-disabled
        cue, so enabling it later can rejoin at its own slot.

        Caller must hold self._lock. A None arrival is not an anchor a rejoin
        could be pinned to (the thread derives its own from live MTC), so it is
        not stamped at all rather than stamped with something unusable.
        """
        if arrival_ms is None:
            return
        for cue in skipped or ():
            try:
                cue._chain_pass = (epoch, arrival_ms)
            except Exception:
                pass

    @logged
    def go(
        self,
        cue: Cue,
        mtc: MtcListener,
        frozen_mtc_ms: float = None,
        chain_epoch: int = None,
        unroll: bool = True,
        stamp_skipped: list = None,
        require_stop_epoch: int = None,
        arm_epoch: int = None,
        commit_gate: Callable[[Cue, int], bool] | None = None,
    ) -> Thread | None:
        """Starts a cue in a thread.

        Args:
            cue: The cue to start
            mtc: The MTC listener
            frozen_mtc_ms: Optional frozen MTC timestamp for sync with chained
            cues
            chain_epoch: None for a FRESH chain entry (manual GO, go_from
            re-entry, enable-rejoin) — mints a new epoch. Otherwise the epoch
            of the pass this dispatch belongs to, which is validated against
            the cue's own high-water mark and the STOP barrier.
            unroll: False stops go_threaded from dispatching the rest of the
            chain — used by the enable-rejoin, where the chain is already
            unrolled and re-dispatching it would churn live cues.
            stamp_skipped: cues the caller's chain walk skipped because they
            are disabled; stamped with this pass so a later enable can rejoin.
            arm_epoch: the STOP/load epoch this dispatch belongs to (see
            arm_epoch()). None for a fresh entry: read here. A continuation
            inherits its chain's. It goes into every arm this dispatch causes
            (the fallback re-arm, the lookahead, the cue's own thread), and a
            dispatch whose epoch has moved is refused.
            commit_gate: called inside the commit section, after every other
            refusal, with the cue and the generation about to be assigned;
            False (or an exception) refuses the dispatch. A cross-node follow
            uses it so that its cancel and its dispatch are decided in one
            place (869fc8ytz).

        Returns:
            Thread running the cue, or None if the cue is disabled, not local
            to this node (the node owning the target will run it via its own
            GO/post_go dispatch), or the dispatch was refused as stale/after a
            STOP.
        """
        if not cue.enabled:
            Logger.info(f"Cue {cue.id} is disabled, skipping execution")
            return None
        if not getattr(cue, "_local", True):
            # Non-local target: handled by the node where it IS local via that
            # node's own go_threaded → post_go chain. Trying to arm/run it
            # locally would fail at re-arm (no local player), raise inside the
            # caller's thread, and kill chained playback (master videocomposer
            # froze after loop 1 when post_go='go' targeted an audio cue local
            # to slave only).
            Logger.info(f"Cue {cue.id} is not local to this node, skipping execution")
            return None
        Logger.info(f"GO command received. Starting cue {cue.id}")

        if arm_epoch is None:
            arm_epoch = self._disarm_epoch
        is_continuation = chain_epoch is not None
        if not is_continuation:
            # Fresh entry: claim the next epoch. It is greater than every
            # recorded stop and every cue mark at this instant, so the checks
            # below can only refuse it if a STOP arrives LATER — while this
            # call is blocked in the re-arm fallback. That is the right
            # outcome: the operator's STOP came after their GO.
            with self._lock:
                self._chain_epoch = self._chain_epoch + 1
                chain_epoch = self._chain_epoch

        if not hasattr(cue, "loaded") or not cue.loaded:
            # 869fbyjzx: say which case this is. An arm still in flight (a
            # pre-arm, an audio player still starting) is not a failure; the
            # old "pre-arm may have failed" said it was. Best-effort: the arm
            # can start or finish between this look and arm() below, so the
            # wait itself is reported from what arm() actually did. True
            # fallbacks keep "not loaded at go() time" and "Re-arming as
            # fallback": the harnesses count them.
            # An arm in flight that a GO waits for is still a LATE cue when the
            # arm is slower than the GO gap (an audio arm is ~0.47 s cold):
            # arming before the hand-off ("Part B") is the only cure and is
            # not built; see AudioMixer.connect_player_to_outputs (869fbyjzx).
            in_flight = self.describe_arm_in_flight(cue)
            if is_continuation:
                Logger.info(
                    f"Cue {cue.id} not armed yet at go() time — dispatching; "
                    f"its own thread arms it or waits for the arm in progress"
                )
            elif in_flight is not None:
                Logger.info(
                    f"Cue {cue.id} is still being armed ({in_flight}) at go() "
                    f"time — waiting for the arm in progress"
                )
            else:
                Logger.warning(
                    f"Cue {cue.id} not loaded at go() time — no arm in flight "
                    f"(never armed, or an earlier arm failed or was "
                    f"abandoned). Re-arming as fallback."
                )
            # A continuation does NOT arm here: dispatch now happens at chain
            # entry, so this call runs on the PREVIOUS cue's thread, and an
            # audio arm can block ~15s on its JACK ports — delaying or killing
            # that cue's own reveal. go_threaded arms the cue on its own
            # thread instead, overlapping its own prewait.
            if not is_continuation:
                wait_report = {}
                self.arm(cue, init=True, epoch=arm_epoch, wait_report=wait_report)
                self._log_arm_wait(cue, "go() time", in_flight, wait_report)
                if not hasattr(cue, "loaded") or not cue.loaded:
                    if arm_epoch != self._disarm_epoch:
                        # Not an arm failure: the operator's STOP (or a load)
                        # arrived while this cue was being re-armed, and
                        # arm() refused or abandoned it. The STOP wins.
                        Logger.warning(
                            f"Refusing dispatch of cue {cue.id}: a STOP or a "
                            "load arrived while it was being re-armed"
                        )
                        return None
                    raise Exception(
                        f"{cue.__class__.__name__} {cue.id} not loaded to go"
                        f"(re-arm failed)"
                    )

        with self._lock:
            # Validate + commit atomically. Both checks are deliberately made
            # AFTER any arm above: that is what lets a STOP which landed
            # during a long arm window survive instead of being wiped by the
            # _stop_requested reset below (F4).
            if arm_epoch != self._disarm_epoch:
                # Whether or not the cue is loaded: if it is, the STOP's own
                # re-arm loaded it for the NEXT run, not for this dispatch.
                Logger.warning(
                    f"Refusing dispatch of cue {cue.id}: a STOP or a load "
                    "arrived while it was being started"
                )
                return None
            if (
                require_stop_epoch is not None
                and self._last_stop_chain_epoch != require_stop_epoch
            ):
                # The caller validated something against the STOP barrier and
                # then had to drop the lock (the enable-rejoin does). A STOP
                # landing in that window must win — otherwise the cue plays
                # after the operator stopped the show.
                Logger.warning(
                    f"Refusing dispatch of cue {cue.id}: a STOP arrived while "
                    f"it was being put back into its chain"
                )
                return None
            if chain_epoch <= self._last_stop_chain_epoch:
                # For a continuation: the chain was stopped. For a fresh
                # entry: a STOP landed while we were arming — it is newer than
                # this GO, so it wins (the caller must tolerate None).
                Logger.warning(
                    f"Refusing dispatch of cue {cue.id}: a STOP ended this "
                    f"chain (pass {chain_epoch} <= stop "
                    f"{self._last_stop_chain_epoch})"
                )
                return None
            if is_continuation:
                if getattr(cue, "_go_epoch", 0) >= chain_epoch:
                    # Either a newer pass already owns the cue (loop-back
                    # racing the original unroll), or this same pass already
                    # dispatched it — a post_go='go' cycle, which entry-time
                    # dispatch would otherwise spin at thread-spawn rate.
                    Logger.warning(
                        f"Refusing stale/duplicate chain dispatch of cue "
                        f"{cue.id} (cue pass {getattr(cue, '_go_epoch', 0)} >= "
                        f"dispatch pass {chain_epoch})"
                    )
                    return None

            go_gen = getattr(cue, "_go_generation", 0) + 1
            if commit_gate is not None:
                try:
                    allowed = commit_gate(cue, go_gen)
                except Exception as e:
                    Logger.error(f"Commit gate for cue {cue.id} failed: {e}")
                    allowed = False
                if not allowed:
                    Logger.info(
                        f"Refusing dispatch of cue {cue.id}: its cross-node "
                        "follow was cancelled or replaced"
                    )
                    return None
            # Only past every refusal: a dispatch that is refused must not
            # cancel the follow of the pass still playing.
            self._take_follow_on_restart_locked(cue)

            cue._go_epoch = chain_epoch
            # Recorded here, not in the spawned thread: a disable arriving in
            # between has to find this cue's anchor to stamp a rejoin with, and
            # in a loop project it would otherwise read the PREVIOUS cycle's.
            # None (manual GO / go_at_end) means "derived from live MTC by the
            # thread", which is not an anchor a rejoin can be pinned to.
            cue._dispatch_arrival_ms = frozen_mtc_ms
            cue._stop_requested = False
            cue._go_generation = go_gen
            # Output-commit flag: True once this cue has produced output
            # (reveal, or the DMX scene send). The stop walk cancels only
            # cues that have NOT committed — a playing cue is never cut.
            cue._revealed = False
            # When this pass produced output (monotonic); set by go_threaded.
            # A late cross-node follow cancel reads it (869fc8ytz).
            cue._revealed_at = None
            # Lifecycle flag: True while a GO owns this cue; cleared by
            # disarm() and stop_all_cues(). Unlike _go_generation
            # (increment-only), this is a sound "currently playing" signal —
            # the disable-action path uses it to decide whether disarming
            # would cut live playback.
            cue._playing = True
            # Measurement only: a cue a GO took is not a wasted arm, even if
            # it is stopped before its reveal.
            cue._ever_played = True
            # Stamped with the epoch minted in THIS lock section: reading it
            # back off the cue afterwards could capture a newer unrelated
            # dispatch's epoch and mis-anchor the rejoin.
            self._stamp_skipped_locked(stamp_skipped, chain_epoch, frozen_mtc_ms)

        thread = Thread(
            name=f"GO:{cue.__class__.__name__}:{cue.id}",
            target=self.go_threaded,
            args=[cue, mtc, frozen_mtc_ms, go_gen, chain_epoch, unroll],
            kwargs={"arm_epoch": arm_epoch},
            daemon=True,
        )
        thread.start()

        # Duration-aware lookahead: arm ahead until 2 cues with meaningful
        # playback duration are ready. SYNCHRONOUS on the caller's thread, and
        # arm() can block ~15s per audio cue — so it must not run where the
        # caller is something that cannot afford to wait:
        #   - a continuation's caller is the PREVIOUS cue's thread (dispatch
        #     happens at chain entry now), and blocking it would delay that
        #     cue's own reveal. Chain cues are armed by arm()'s recursion at
        #     load and, failing that, by their own thread in go_threaded.
        #   - a rejoin (unroll=False) is called from the command thread and
        #     the cue is already armed by definition.
        if not is_continuation and unroll:
            self._arm_ahead(cue, arm_epoch=arm_epoch)
        return thread

    def go_from(
        self,
        start_cue: Cue,
        mtc: MtcListener,
        seed_ms: float = None,
        arm_epoch: int = None,
        require_stop_epoch: int = None,
        commit_gate: Callable[[Cue, int], bool] | None = None,
    ) -> Thread | None:
        """Re-enter a post_go='go' chain from start_cue (inclusive), firing THIS
        node's first local+enabled cue at seed_ms + Σ(chain advance of the
        enabled cues skipped on the way). Returns the Thread, or None if no
        local+enabled cue is reachable before a chain break.

        Mirrors NodeEngine.go_script's cross-node walk, but for an IN-CHAIN
        re-entry — specifically an ActionCue 'play' that restarts a sequence
        (e.g. a circular project's loop-back). Plain go() bails on a non-local
        target (returns None), which drops every cue local to a DIFFERENT node
        than the play target: a loop-back 'play' aimed at node01's cue would
        never re-fire the controller's own cues. Walking here makes each engine
        re-enter its own local segment, exactly as it does on a GO.

        arm_epoch / require_stop_epoch / commit_gate are passed to go() when
        set: a cross-node follow dispatches on another thread than the one
        that received it, and a STOP or a cancel in between must win.
        """
        guards = {
            k: v
            for k, v in (
                ("arm_epoch", arm_epoch),
                ("require_stop_epoch", require_stop_epoch),
                ("commit_gate", commit_gate),
            )
            if v is not None
        }
        if seed_ms is None:
            seed_ms = mtc.main_tc.milliseconds_exact
        cue = start_cue
        sigma_ms = 0.0
        walked = 0
        skipped_disabled = []
        while cue is not None and not (
            getattr(cue, "_local", False) and getattr(cue, "enabled", False)
        ):
            if not getattr(cue, "enabled", False):
                # Remember it: enabling it before its slot rejoins this pass.
                skipped_disabled.append(cue)
            if getattr(cue, "post_go", None) != "go":
                # chain break before any local cue — nothing for this node to do
                cue = None
                break
            if getattr(cue, "enabled", False):
                sigma_ms += self._chain_advance_ms(cue)
            cue = getattr(cue, "_target_object", None)
            walked += 1
            if walked > 1024:
                Logger.error("go_from walk hit safety limit; aborting")
                return None
        if cue is None:
            # No local cue to fire, but the disabled ones still get a pass to
            # rejoin, or enabling one mid-show would do nothing at all.
            if skipped_disabled:
                self.stamp_pass(skipped_disabled, seed_ms)
            return None
        if skipped_disabled:
            return self.go(
                cue,
                mtc,
                seed_ms + sigma_ms,
                stamp_skipped=skipped_disabled,
                **guards,
            )
        return self.go(cue, mtc, seed_ms + sigma_ms, **guards)

    def _reveal_wait(self, cue: Cue, mtc: MtcListener, go_gen: int = 0) -> str:
        """Block until live MTC reaches cue._start_mtc; return 'reached' or 'stopped'.

        run_cue() sets a cue up HELD (video invisible / audio not-following /
        action not-yet-run). This gates the reveal on MTC so prewait/postwait
        offsets become real timeline gaps. Cues with no _start_mtc (ActionCue, or
        a CueList used as a target) reveal immediately. DmxCue DOES set _start_mtc
        but self-schedules (its reveal is a no-op), so it merely exits this wait
        once MTC passes start.

        Deliberately does NOT bail on an MTC stall: a recoverable stall
        self-recovers (reveal fires late when timecode resumes); bailing would
        leave the cue permanently held and frees nothing (loop_cue would hang on
        the same stall). STOP always exits via _stop_requested; a newer GO/reload
        exits via _go_generation.
        """
        start = getattr(cue, "_start_mtc", None)
        if start is None:
            return "reached"
        # milliseconds_exact is wrap-accumulated by MtcListener → 24h-safe on
        # long shows (Fable 4.4); rounded would false-trip near a frame boundary.
        return self._wait_mtc(cue, mtc, start.milliseconds_exact, go_gen)

    def _next_local_fire(
        self, cue: Cue, arrival_ms: float, stamp_epoch: int = None
    ) -> tuple["Cue | None", float]:
        """From a just-played cue, walk its post_go='go' chain to THIS node's
        next local+enabled cue and return (that_cue_or_None, its_arrival_ms).

        Under Auto continue the chain advance is zero, so every cue in the
        chain arrives at the same trigger; the accumulator is kept because the
        helper is the one documented place expressing the rule. Cues we skip
        are transparent, and the walk stops at a chain break (post_go != 'go')
        — an explicit hand-off point — bounded against all-disabled/all-remote
        cycles.

        stamp_epoch: when given, disabled cues found on the way are stamped
        with the pass that would have dispatched them, so enabling one later
        can rejoin it at its own slot. Left None (the default) the walk touches
        nothing — callers that only ask "who is next" must stay side-effect
        free.
        """
        acc = arrival_ms + self._chain_advance_ms(cue)
        node = getattr(cue, "_target_object", None)
        walked = 0
        while node is not None:
            if getattr(node, "_local", False) and getattr(node, "enabled", False):
                return node, acc
            if stamp_epoch is not None and not getattr(node, "enabled", False):
                # Disabled: transparent to the timeline, but remember the pass
                # so a re-enable before its slot can still rejoin. Stamped here
                # too when the disabled cue is itself the chain break, which
                # the walk reaches but does not walk past. Guarded so that a
                # caller only asking "who is next" touches nothing at all —
                # including this handler's own attributes, since the walk is
                # also driven with a bare class or a stub as `self`.
                self._stamp_one(node, stamp_epoch, acc)
            if getattr(node, "post_go", None) != "go":
                return None, acc
            if getattr(node, "enabled", False):
                acc += self._chain_advance_ms(node)
            node = getattr(node, "_target_object", None)
            walked += 1
            if walked > 1024:
                Logger.error("post_go fire-walk hit safety limit; aborting")
                return None, acc
        return None, acc

    def _stamp_one(self, cue: Cue, epoch: int | None, arrival_ms: float) -> None:
        """Stamp a single skipped-disabled cue, if stamping was asked for.

        Gated on `epoch is not None` in full — including the lock — because the
        walk is also called with a bare class or a stub as `self`.
        """
        if epoch is None:
            return
        with self._lock:
            self._stamp_skipped_locked([cue], epoch, arrival_ms)

    def go_threaded(
        self,
        cue: Cue,
        mtc: MtcListener,
        frozen_mtc_ms: float = None,
        go_gen: int = 0,
        chain_epoch: int = 0,
        unroll: bool = True,
        arm_epoch: int = None,
    ):
        """Runs a cue based on its properties.

        Args:
            cue: The cue to run
            mtc: The MTC listener (for live MTC)
            frozen_mtc_ms: Optional frozen MTC timestamp in milliseconds.
            go_gen: Generation counter captured at go() time. If the cue's
                    generation has changed by the time the loop ends, another
                    go/stop cycle occurred and this thread must not touch the
                    cue.
            chain_epoch: the pass this cue belongs to, carried unchanged into
                    the continuation dispatch so the whole chain shares it.
            unroll: False suppresses the continuation dispatch (enable-rejoin
                    of a chain that is already unrolled).
            arm_epoch: the STOP/load epoch of the go() that dispatched this
                    cue; every arm this thread makes (its own cue, the
                    go_at_end lookahead) and the continuation it dispatches
                    belong to it, so none of them survives a STOP.
        """
        follow = _FollowOut()
        try:
            self._run_cue_pass(
                cue,
                mtc,
                frozen_mtc_ms,
                go_gen,
                chain_epoch,
                unroll,
                arm_epoch,
                follow,
            )
        except Exception:
            follow.error = True
            raise
        finally:
            self._leave_follow(cue, follow, arm_epoch)

    def _run_cue_pass(
        self,
        cue: Cue,
        mtc: MtcListener,
        frozen_mtc_ms: float,
        go_gen: int,
        chain_epoch: int,
        unroll: bool,
        arm_epoch: int | None,
        follow: _FollowOut,
    ):
        """The body of go_threaded; `follow` records what this pass did about
        its cross-node Auto follow, for go_threaded's exit (869fc8ytz)."""
        # frozen_mtc_ms is this cue's ARRIVAL on the MTC timeline:
        # GO_mtc + Σ(effective durations of preceding cues in the chain).
        # None → manual GO / go_at_end: arrival = live MTC now (so those paths
        # keep their prewait — Fable 1.4).
        if frozen_mtc_ms is None:
            # Used by BaseEngine.timecode = mtc - go_offset for drift; _exact
            # preserves sub-ms precision at NTSC framerates.
            frozen_mtc_ms = mtc.main_tc.milliseconds_exact
            Logger.debug(f"Captured MTC snapshot for cue {cue.id}: {frozen_mtc_ms}ms")

        arrival_ms = frozen_mtc_ms
        if getattr(cue, "_dispatch_arrival_ms", None) is None:
            # go() records this under its accept lock; only the manual-GO /
            # go_at_end paths (no seed) resolve their anchor here.
            cue._dispatch_arrival_ms = arrival_ms
        # Single prewait application point (Fable 1.2): the cue's media and reveal
        # are anchored at start = arrival + prewait. prewait is NO LONGER a
        # wall-clock sleep — _reveal_wait turns it into a real MTC-timeline gap.
        # Under Auto continue every cue in the chain shares `arrival` (the
        # trigger), so this is the ONLY term that separates them.
        start_ms = arrival_ms + cue.prewait.milliseconds_exact

        # Safety net. Auto continue dispatches the whole chain from the trigger
        # and every prewait is >= 0, so a chain cue cannot be reached after its
        # own anchor any more; what can still land here is a go_at_end/manual
        # seed already in the past, or an MTC jump. Silence would mistime a show
        # quietly. Plans/prewait-dispatch-reorder-phase2.md §4.
        self._warn_if_late(cue, mtc, start_ms)

        # Dispatch the rest of the chain NOW, at this cue's arrival — not after
        # its reveal and postwait. Auto continue triggers the whole chain at
        # once, so every cue must be free to park on its own anchor; dispatching
        # at start(k)+postwait(k) reached a cue with a smaller prewait than its
        # predecessor's after its anchor had already passed, and fired it late.
        # Every cue of the pass carries the same chain_epoch, so a stale cascade
        # (loop-back) and a cycle are both refused in go().
        post_go_thread = None
        if unroll and cue.post_go == "go" and not self._chain_continues(cue):
            Logger.info(
                f"stop_project cue {cue.id}: chain break, the cues after it "
                "are not dispatched"
            )
        elif (
            unroll
            and cue.post_go == "go"
            and not cue._stop_requested
            and getattr(cue, "_go_generation", 0) == go_gen
        ):
            try:
                # Walk to THIS node's next local+enabled cue; cues we skip are
                # transparent (the advance is zero), and disabled ones are
                # stamped with this pass so enabling one later can rejoin it.
                next_cue, next_arrival = self._next_local_fire(
                    cue, arrival_ms, stamp_epoch=chain_epoch
                )
                if next_cue is not None:
                    Logger.info(
                        f"Unrolling post_go chain: dispatching {next_cue.id} "
                        f"at the trigger"
                    )
                    post_go_thread = self.go(
                        next_cue,
                        mtc,
                        next_arrival,
                        chain_epoch=chain_epoch,
                        arm_epoch=arm_epoch,
                    )
            except Exception as e:
                # The chain is downstream work; losing it must not cost THIS
                # cue its own reveal.
                Logger.error(f"Chain dispatch from cue {cue.id} failed: {e}")

        if cue._local:
            # A continuation does not arm on the dispatching thread (that is the
            # PREVIOUS cue's thread — a ~15s audio arm there would delay or kill
            # its reveal). It arms here instead, on its own thread, overlapping
            # its own prewait.
            if not getattr(cue, "loaded", False):
                # 869fbyjzx: an arm in flight is waited for, not a fallback.
                in_flight = self.describe_arm_in_flight(cue)
                if in_flight is not None:
                    Logger.info(
                        f"Cue {cue.id} is still being armed ({in_flight}) at "
                        f"dispatch — waiting for the arm in progress"
                    )
                else:
                    Logger.warning(
                        f"Cue {cue.id} not loaded at dispatch — no arm in "
                        f"flight; arming on its own thread before its slot. "
                        f"Re-arming as fallback."
                    )
                wait_report = {}
                self.arm(cue, init=True, epoch=arm_epoch, wait_report=wait_report)
                self._log_arm_wait(cue, "dispatch", in_flight, wait_report)
                if arm_epoch is not None and arm_epoch != self._disarm_epoch:
                    # arm() refused or abandoned the arm: a STOP or a load
                    # landed since this cue was dispatched. Said here, before
                    # the generic failure below, because it is not one.
                    Logger.info(
                        f"Cue {cue.id} was dispatched before a STOP or a "
                        "load; it will not play"
                    )
                    return
                if not getattr(cue, "loaded", False):
                    # The rest of the chain went out at entry, so this failure
                    # costs this cue only.
                    Logger.error(
                        f"{cue.__class__.__name__} {cue.id} could not be armed; "
                        f"it will not play. The rest of the chain is unaffected."
                    )
                    return
                # The check above ran before the arm; an arm longer than this
                # cue's runway makes it late, and that must not be silent.
                self._warn_if_late(cue, mtc, start_ms)
            elif arm_epoch is not None and arm_epoch != self._disarm_epoch:
                # Loaded, but not for this dispatch: the STOP's own re-arm
                # loaded it for the NEXT run while this thread was on its
                # way here. stop_all_cues() could not flag this cue -- it was
                # not armed when the STOP began -- so nothing downstream
                # would hold it back.
                Logger.info(
                    f"Cue {cue.id} was dispatched before a STOP or a load; "
                    "it will not play"
                )
                return

            # Illuminate (sequence-view highlight) at the cue's ARRIVAL — the
            # start of its prewait — NOT at dispatch. Under Auto continue every
            # cue arrives at the trigger, so the whole chain lights together and
            # each cue stays lit for prewait + max(body, postwait).
            if self._wait_mtc(cue, mtc, arrival_ms, go_gen) != "stopped":
                try:
                    self.communications_thread.add_cue(
                        cue.id, str(start_ms), timeout=0.1
                    )
                except Exception:
                    pass

            # Park until the held setup is actually needed. Unrolling the chain
            # at the trigger would otherwise run every cue's run_cue at GO —
            # for video that is N layers decoding invisibly for the whole of
            # their prewaits. A start already inside the window returns at once.
            parked = self._wait_mtc(cue, mtc, start_ms - self._RUN_AHEAD_MS, go_gen)

            if parked != "stopped":
                # Set up HELD at start_ms (video invisible / audio not-following
                # / action not-yet-run / dmx self-scheduled from absolute
                # mtc_time), close enough to the slot that the frame is ready.
                try:
                    run_cue(cue, mtc, start_ms)
                except Exception as e:
                    # Illumination happens above, so a failure here would leave
                    # the cue lit forever with a dead thread behind it.
                    Logger.error(
                        f"run_cue failed for {cue.__class__.__name__} "
                        f"{cue.id}: {e}. The cue will not play."
                    )
                    self._end_illumination(cue)
                    return
                if isinstance(cue, DmxCue):
                    # DMX has no reveal: the scene is already on its way with an
                    # absolute mtc_time and the player self-schedules it. From
                    # here it is committed and the stop walk must not treat it
                    # as still cancellable.
                    with self._lock:
                        cue._revealed = True
                        cue._revealed_at = monotonic()

            # MTC-gated reveal: wait until live MTC reaches start_ms, then reveal
            # (video /visible; audio /mtcfollow; action EXECUTE; dmx no-op). This
            # is what makes prewait/body/postwait real timeline gaps, honored
            # identically on every node. Skipped when the park was cancelled —
            # run_cue never ran, so _start_mtc holds no anchor to wait on.
            if parked != "stopped" and self._reveal_wait(cue, mtc, go_gen) != "stopped":
                # Commit under the same lock the stop walk takes, so a cancel
                # landing inside the 20ms poll gap cannot leak a reveal past it.
                with self._lock:
                    do_reveal = not cue._stop_requested and (
                        getattr(cue, "_go_generation", 0) == go_gen
                    )
                    if do_reveal:
                        cue._revealed = True
                        # Same section: a cross-node follow cancel deciding
                        # "not started yet" must never see _revealed set
                        # without its stamp.
                        cue._revealed_at = monotonic()
                if do_reveal:
                    revealed_at = cue._revealed_at
                    follow.revealed = True
                    reveal_cue(cue, mtc, start_ms)
                    armed_at = getattr(cue, "_armed_at", None)
                    if armed_at is not None:
                        # Both sides of the pre-arm policy in one number: a
                        # small one is a thin margin, a large one an idle
                        # hold. Logged after the reveal, so it never delays it.
                        Logger.info(
                            f"Cue {cue.id} started {revealed_at - armed_at:.3f} s "
                            "after it was armed"
                        )

        # A superseding GO/reload (new _go_generation, without _stop_requested)
        # can arrive during the now-MTC-gated reveal wait — that fresh thread
        # owns the chain. This stale thread must NOT pace postwait or fire the
        # next cue, or the next cue would be go()'d twice (once here with a stale
        # arrival, once by the fresh thread). Cleanup below is already gated by
        # the generation check, so we only guard the outward actions here.
        superseded = getattr(cue, "_go_generation", 0) != go_gen

        # Postwait for AUTO-CONTINUE only: ILLUMINATION hold. It no longer
        # paces anything — the chain left at this cue's arrival, above — but it
        # still holds this thread past the body when post > body, and that is
        # the ONLY thing producing continue's `pre + max(body, post)` highlight
        # (loop_cue never reads postwait, and remove_cue fires after it
        # returns). Removing it would silently shorten the highlight to
        # pre + body. For pause/go_at_end the postwait is a REAL gap AFTER the
        # body — handled by the MTC-gated tail below loop_cue, not by this
        # sleep.
        if (
            cue.post_go == "go"
            and cue.postwait > 0
            and not cue._stop_requested
            and not superseded
        ):
            sleep(cue.postwait.milliseconds_rounded / 1000)

        # Pre-arm go_at_end targets during playback. Runs after
        # run_cue() so current cue is already playing. The arm happens
        # in parallel with the media. go() also calls _arm_ahead but
        # that fires before run_cue — this call catches cues that were
        # disarmed between go() and here (loop passes).
        # Cross-node Auto follow (869fc8ytz): tell the other nodes the follow
        # instant as soon as it is final -- now, or at the start of the last
        # loop -- and before the lookahead below, which can take seconds.
        on_last_loop = None
        if follow.revealed and self._follows_out(cue):
            if cue.loop == 1:
                self._announce_follow(cue, go_gen, start_ms, follow)
            elif cue.loop > 1:
                on_last_loop = partial(
                    self._announce_follow, cue, go_gen, start_ms, follow
                )

        if cue.post_go == "go_at_end":
            self._arm_ahead(cue, arm_epoch=arm_epoch)

        Logger.info(f"Going to loop for {cue.__class__.__name__}:{cue.id}")
        if on_last_loop is None:
            loop_cue(cue, mtc)
        else:
            loop_cue(cue, mtc, on_last_loop=on_last_loop)

        if getattr(cue, "_go_generation", 0) != go_gen:
            Logger.info(
                f"Cue {cue.id} generation changed ({go_gen} →"
                f"{cue._go_generation}), skipping cleanup"
            )
            return

        # ---- pause/go_at_end postwait tail -------------------------------
        # Spec: postwait is a real gap AFTER the body (follow: pre→body→post→
        # next; pause: pre→body→post→standby), and the cue stays illuminated
        # through it. Base end of the body on the MTC timeline; fallbacks for
        # cues that never set _end_mtc (plain ActionCue body=0; aborted A/V
        # setup; CueList) — and guard against a stale _end_mtc left by a
        # previous GO of this object. (FadeCue DOES set a real _end_mtc via
        # _handle_fade_action — used as-is.)
        fire_seed_ms = self._follow_instant_ms(cue, start_ms)

        tail = (
            cue.post_go in ("pause", "go_at_end")
            and cue.postwait > 0
            and not cue._stop_requested
        )
        # Snapshot the follow target's generation BEFORE the tail, so the
        # auto-fire below can YIELD if a newer GO starts the target during the
        # tail — otherwise it would run twice. Only an EXPLICIT selection of the
        # target (setnextcue + GO) does that: a plain GO does not, because
        # Cue.get_next_cue() skips go/go_at_end targets, so next_cue_pointer
        # already points past the whole chain and a plain GO in the tail fires
        # the cue after it while the target still auto-follows. QLab-like; kept
        # deliberately (Ion, 2026-09-26; verified on test2, ClickUp 869ej4x38).
        target_gen0 = (
            getattr(cue._target_object, "_go_generation", 0)
            if cue._target_object
            else 0
        )
        if tail:
            # Hide output at body end WITHOUT disarming (cue must stay in
            # _armed_cues so STOP can reach this wait — disarming here would
            # let the tail expire and ghost-fire after an operator STOP).
            blank_cue(cue, mtc)
            self._wait_mtc(cue, mtc, fire_seed_ms, go_gen)
            if getattr(cue, "_go_generation", 0) != go_gen:
                # Real STOP (stop_all_cues bumps gen) / newer GO owns the cue:
                # no remove_cue (controller's stop already reset status), no
                # fire, no disarm (mirrors the post-loop generation guard).
                Logger.info(
                    f"Cue {cue.id} postwait tail cancelled (stop/newer GO),"
                    " skipping cleanup"
                )
                return
            # A pure _stop_requested exit (action-pause: sets the flag WITHOUT a
            # gen bump) falls through: remove_cue and the follow fire are already
            # guarded on _stop_requested below, and disarm still runs — exactly
            # today's behavior for an action-pause landing mid-body.

        # Notify the controller that the cue finished playing (status → 100).
        # Done here (after loop_cue) so the status only changes to 100 when the
        # cue has actually completed its full duration, not just when playback
        # started.
        # Skipped if the cue was stopped (controller's stop_script already
        # resets to 0).
        if cue._local and not getattr(cue, "_stop_requested", False):
            try:
                self.communications_thread.remove_cue(cue.id, timeout=0.1)
            except Exception:
                pass

        go_at_end_thread = None
        if cue.post_go == "go_at_end" and not self._chain_continues(cue):
            Logger.info(
                f"stop_project cue {cue.id}: chain break, its go_at_end follow "
                "is not fired"
            )
        elif (
            cue.post_go == "go_at_end"
            and cue._target_object
            and not cue._stop_requested
            and getattr(cue._target_object, "_go_generation", 0) == target_gen0
        ):
            Logger.info(f"Running go at end for {cue.__class__.__name__}:{cue.id}")
            # go_from (not go): a non-local follow target is walked past to THIS
            # node's next local cue (go() would bail — same defect class as the
            # play-action loop-back). Explicit seed = body_end + postwait: exact
            # arrival on the shared timeline; go_threaded adds the target's own
            # prewait on top. The target_gen0 compare makes this auto-fire YIELD
            # if a manual GO already started the target during the tail.
            if self._follow_fired(cue, follow, fire_seed_ms, go_gen):
                go_at_end_thread = self.go_from(cue._target_object, mtc, fire_seed_ms)
        elif (
            cue.post_go == "go_at_end"
            and cue._target_object
            and not cue._stop_requested
        ):
            follow.yielded = True
            Logger.info(
                f"go_at_end fire for {cue.id} yielded: "
                "target already started by a newer GO"
            )

        self.disarm(cue, reason="cue_end")

        if cue.post_go == "go_at_end" and go_at_end_thread:
            self.wait_for_cue(go_at_end_thread)

        if cue.post_go == "go" and cue._target_object and not cue._stop_requested:
            if post_go_thread:
                self.wait_for_cue(post_go_thread)

    # ---------------------------
    # Cross-node Auto follow (869fc8ytz)
    # ---------------------------

    @staticmethod
    def _follow_instant_ms(cue: Cue, start_ms: float) -> float:
        """When an Auto follow fires its target: the body's end on the MTC
        timeline plus the postwait. The one expression both the fire and the
        announcement use. Fallbacks for cues that never set _end_mtc (plain
        ActionCue body=0; aborted A/V setup; CueList) and a stale _end_mtc
        left by a previous GO of this object. (FadeCue DOES set a real
        _end_mtc via _handle_fade_action -- used as-is.)"""
        end_attr = getattr(cue, "_end_mtc", None)
        base_end_ms = end_attr.milliseconds_exact if end_attr is not None else start_ms
        if base_end_ms < start_ms:
            base_end_ms = start_ms  # stale from a previous run
        return base_end_ms + cue.postwait.milliseconds_exact

    @staticmethod
    def _chain_continues(cue: Cue) -> bool:
        """False for a stop_project ActionCue (ClickUp 869fhjdq5): the authored
        end of the show is a hard chain break on every node. Nothing after it
        is unrolled at its arrival and no go_at_end follow fires from it, so no
        cue can start in the milliseconds before the controller's STOP lands.
        A DISABLED stop cue never gets here: disabled cues are transparent to
        the chain walk (rehearsal mode)."""
        return not (
            isinstance(cue, ActionCue)
            and getattr(cue, "action_type", None) == "stop_project"
        )

    @staticmethod
    def _follows_out(cue: Cue) -> bool:
        """An Auto follow whose source plays only where it has an output.

        ActionCue, FadeCue, CueList and DmxCue are local on every node, so
        every node fires their follow itself; only AudioCue and VideoCue need
        the other nodes told."""
        return (
            isinstance(cue, (AudioCue, VideoCue))
            and getattr(cue, "_local", False)
            and getattr(cue, "post_go", None) == "go_at_end"
            and getattr(cue, "_target_object", None) is not None
        )

    def _announce_follow(
        self, cue: Cue, go_gen: int, start_ms: float, follow: _FollowOut
    ) -> None:
        """Announce this pass's follow instant to the other nodes, once.

        Decided, recorded and queued in one lock section, so the messages
        leave in the order the decisions were taken; only the pass that still
        owns the cue announces. Never raises: it runs on the cue's thread."""
        relay = self._follow
        if relay is None or follow.record is not None:
            return
        try:
            with self._lock:
                if getattr(cue, "_go_generation", 0) != go_gen or getattr(
                    cue, "_stop_requested", False
                ):
                    return
                if relay.stop_pending:
                    # A STOP or a load was received and has not run yet: this
                    # pass is about to be stopped.
                    return
                run_seq = relay.run_seq
                if run_seq is None:
                    relay.warn_no_run()
                    return
                record = (self._follow_instant_ms(cue, start_ms), run_seq)
                cue._follow_out = record
                follow.record = record
                relay.announce(cue, record[0], run_seq)
        except Exception as e:
            Logger.error(f"Could not announce the follow of cue {cue.id}: {e}")

    def _take_follow_record_locked(self, cue: Cue, record: tuple) -> bool:
        """Compare-and-clear the cue's outstanding announcement. Caller holds
        self._lock. True if `record` was still the outstanding one."""
        if getattr(cue, "_follow_out", None) == record:
            cue._follow_out = None
            return True
        return False

    def _take_follow_on_restart_locked(self, cue: Cue) -> None:
        """go() is starting a cue whose earlier pass announced a follow: that
        follow will not happen. Caller holds self._lock (go()'s commit
        section, after every refusal). The earlier pass's thread may only
        notice at the new pass's end, so the cancel goes out here."""
        record = getattr(cue, "_follow_out", None)
        if record is None:
            return
        cue._follow_out = None
        relay = self._follow
        if relay is None or record[1] != relay.run_seq:
            return
        try:
            relay.cancel(cue, record[0], record[1], "restart")
        except Exception as e:
            Logger.error(f"Could not cancel the follow of cue {cue.id}: {e}")

    def _follow_fired(
        self, cue: Cue, follow: _FollowOut, fire_ms: float, go_gen: int
    ) -> bool:
        """The pass is about to fire its follow: nothing to cancel any more.

        Returns False -- do not fire -- when this pass announced its follow
        and has been replaced since (the cue was started again): go() already
        cancelled the announcement on the other nodes, so firing here would
        play only half of the target. An unannounced follow fires as before."""
        if follow.record is None:
            follow.fired = True
            return True
        with self._lock:
            if getattr(cue, "_go_generation", 0) != go_gen:
                Logger.info(
                    f"Cue {cue.id} was started again before its follow fired; "
                    "the new pass owns the follow"
                )
                return False
            follow.fired = True
            self._take_follow_record_locked(cue, follow.record)
        if abs(fire_ms - follow.record[0]) > 1.0:
            Logger.error(
                f"Cue {cue.id} fires its follow at {fire_ms:.0f}ms but "
                f"announced {follow.record[0]:.0f}ms to the other nodes"
            )
        return True

    def _leave_follow(
        self, cue: Cue, follow: _FollowOut, arm_epoch: int | None
    ) -> None:
        """go_threaded's exit: a pass that announced and did not fire cancels,
        unless every node already knows (a STOP or a load, the yield) or go()
        took the announcement (the cue was started again). Never raises."""
        record = follow.record
        if record is None or follow.fired:
            return
        try:
            with self._lock:
                if not self._take_follow_record_locked(cue, record):
                    return
                if arm_epoch is not None and arm_epoch != self._disarm_epoch:
                    return
                if follow.error:
                    reason = "error"
                elif follow.yielded:
                    return
                elif getattr(cue, "_stop_requested", False):
                    reason = "stop"
                else:
                    Logger.warning(
                        f"Cue {cue.id} ended without firing its follow for no "
                        "known reason; cancelling it on the other nodes"
                    )
                    reason = "error"
                relay = self._follow
                if relay is not None:
                    relay.cancel(cue, record[0], record[1], reason)
        except Exception as e:
            Logger.error(f"Could not cancel the follow of cue {cue.id}: {e}")

    def stop_dispatched_follow(
        self,
        cue: Cue,
        gen: int,
        disarm_epoch: int,
        armed_at: float | None,
        may_cut: bool,
        grace_s: float,
        started_before: float | None = None,
    ) -> tuple[str, float | None]:
        """Stop a cue a cross-node follow dispatched, after the owner
        cancelled that follow (869fc8ytz). Returns (outcome, age):

        - "owned": a newer GO or a STOP has the cue (generation or epoch
          moved); nothing is touched;
        - "stopped": it had not started, or started less than `grace_s` ago
          and `may_cut` (the source was stopped or started again);
        - "left": it started too long ago, or the cancel came from an error
          -- a cue on stage is never cut for an internal reason.

        `age` is seconds since the cue produced output, None if it had not.
        `started_before` is when the cue went on stage in a pass that was
        still playing when the follow dispatched it again (recorded by the
        follow's commit gate): its output counts as started.
        The decision and the stop flags are one lock section; the disarm is
        guarded again (disarm_if_unchanged)."""
        with self._lock:
            if (
                getattr(cue, "_go_generation", 0) != gen
                or self._disarm_epoch != disarm_epoch
            ):
                return "owned", None
            started = getattr(cue, "_revealed_at", None)
            if started is None and getattr(cue, "_revealed", False):
                started = monotonic()
            if started is None:
                started = started_before
            age = None if started is None else monotonic() - started
            if age is not None and (age >= grace_s or not may_cut):
                return "left", age
            cue._stop_requested = True
            new_gen = gen + 1
            cue._go_generation = new_gen
            cue._playing = False
        self.cancel_pending_descendants(cue)
        self._end_illumination(cue)
        # Let the cue's own loop see the flag and leave (polls every 20 ms),
        # as the stop action does.
        sleep(0.1)
        self.disarm_if_unchanged(
            cue, new_gen, disarm_epoch, armed_at, reason="follow_cancelled"
        )
        return "stopped", age

    def disarm_if_unchanged(
        self,
        cue: Cue,
        gen: int,
        disarm_epoch: int,
        armed_at: float | None,
        reason: str,
    ) -> bool:
        """Disarm a cue only if nothing took it since the caller decided to.

        In one lock section: the generation, the STOP/load epoch and the arm
        stamp must be the caller's, the cue still loaded and no arm of it in
        flight; then this thread takes the cue's arm claim and unpublishes it
        (unloaded, out of the armed list). Its player and layers are released
        outside the lock while the claim is held, so no arm of the same cue
        can run meanwhile; the claim is released in a finally (a leaked one
        would make the cue unarmable until the engine restarts). Returns True
        if it disarmed."""
        with self._lock:
            registry = self._arming_registry()
            if (
                getattr(cue, "_go_generation", 0) != gen
                or self._disarm_epoch != disarm_epoch
                or not getattr(cue, "loaded", False)
                or getattr(cue, "_armed_at", None) != armed_at
                or registry.get(cue.id) is not None
            ):
                return False
            claim = _ArmClaim(holder=current_thread().name, started=monotonic())
            registry[cue.id] = claim
            cue.loaded = False
            cue._playing = False
            if cue.id in self._armed_cues_set:
                self._armed_cues.remove(cue)
                self._armed_cues_set.discard(cue.id)
        try:
            Logger.info(f"Disarming cue {cue.id} ({reason})")
            try:
                if isinstance(cue, AudioCue):
                    self.communications_thread.remove_player(
                        f"audioplayer_{cue.id}", timeout=0.1
                    )
                self.communications_thread.remove_cue(cue.id, timeout=0.1)
            except Exception:
                pass
            self._release_cue_resources(cue, reason)
        finally:
            with self._lock:
                if registry.get(cue.id) is claim:
                    del registry[cue.id]
            claim.event.set()
        return True

    def wait_for_cue(self, thread: Thread) -> None:
        """Waits for a cue to finish."""
        Logger.info(f"Waiting for {thread.name} to finish")
        while thread.is_alive():
            sleep(1)
        thread.join()
        Logger.info(f"{thread.name} finished")

    # ---------------------------
    # ---------------------------
    # Action Cue Execution (delegates to ActionHandler)
    # ---------------------------

    def execute_action(
        self,
        cue: ActionCue,
        mtc: MtcListener,
        frozen_mtc_ms: float | None = None,
    ) -> dict:
        """
        Execute an ActionCue against the running show (see ActionHandler).
        """
        from .ActionHandler import ACTION_HANDLER

        return ACTION_HANDLER.execute_action(cue, mtc, frozen_mtc_ms)

    def register_action_hook(
        self,
        phase: str,
        fn,
        *,
        action_types: frozenset | None = None,
    ) -> None:
        """
        Register a cue-layer extension hook; forwards to ``ACTION_HANDLER``.
        """
        from .ActionHandler import ACTION_HANDLER

        ACTION_HANDLER.register_action_hook(
            phase, fn, source="cue_layer", action_types=action_types
        )

    # ---------------------------
    # OSCQuery Message Routing
    # ---------------------------

    def route_audio_message(self, path_parts: list[str], value) -> None:
        """Route audio OSCQuery message to the appropriate handler.

        Args:
            path_parts: Path parts after 'audio' (e.g., ['mixer', '0',
            'master', 'volume']
                        or ['cue', '<uuid>', '0', 'volume'])
            value: The OSC value to set
        """
        if not path_parts:
            Logger.warning("Empty audio path parts")
            return

        if path_parts[0] == "mixer":
            # Route to audio mixer: ['mixer', '<output_index>', '<channel>',
            # 'volume']
            # → /audiomixer/0_mixer/<channel>
            if len(path_parts) >= 3:
                output_index = path_parts[1]
                channel = path_parts[2]
                mixer_cmd = f"/audiomixer/{output_index}_mixer/{channel}"
                mixer_client = PLAYER_HANDLER.get_audio_mixer_client()
                if mixer_client:
                    Logger.debug(f"Routing audio mixer: {mixer_cmd} = {value}")
                    mixer_client.set_value(mixer_cmd, float(value))
                else:
                    Logger.warning("Audio mixer client not available")
            else:
                Logger.warning(f"Invalid mixer path: {path_parts}")

        elif path_parts[0] == "cue":
            # Route to cue player: ['cue', '<uuid>', '<channel>', 'volume']
            # → /vol<channel> on the armed cue's OSC client
            if len(path_parts) >= 3:
                cue_uuid = path_parts[1]
                channel = path_parts[2]
                audio_cmd = f"/vol{channel}"
                cue = self.get_armed_cue_by_id(cue_uuid)
                if cue and hasattr(cue, "_osc") and cue._osc:
                    # UI already sends 0.0-1.0 via sliderToFloat(); just clamp
                    vol_value = max(0.0, min(1.0, float(value)))
                    Logger.debug(
                        f"Routing audio cue {cue_uuid}: {audio_cmd} = {vol_value}"
                    )
                    cue._osc.set_value(audio_cmd, vol_value)
                else:
                    Logger.warning(f"Cue {cue_uuid} not found or has no OSC client")
            else:
                Logger.warning(f"Invalid cue audio path: {path_parts}")
        else:
            Logger.warning(f"Unknown audio path type: {path_parts[0]}")

    def route_dmx_message(self, path_parts: list[str], value) -> None:
        """Route DMX OSCQuery message to the DMX player.

        Args:
            path_parts: Path parts after 'dmx' (e.g., ['mixer', '0', 'channel',
            '1'])
            value: The OSC value to set
        """
        if not path_parts:
            Logger.warning("Empty DMX path parts, skipping routing")
            return
        if "mixer" not in path_parts:
            Logger.warning(
                f'Invalid DMX path (no "mixer" keyword): {path_parts}, skipping routing'
            )
            return

        # Build DMX command from path: find 'mixer' and use everything after it
        # +1 to skip 'mixer' keyword
        mixer_index = path_parts.index("mixer") + 1
        dmx_cmd = "/" + "/".join(path_parts[mixer_index:])
        dmx_client = PLAYER_HANDLER.get_dmx_player_client()
        if dmx_client:
            Logger.debug(f"Routing DMX: {dmx_cmd} = {value}")
            dmx_client.set_value(dmx_cmd, value)
        else:
            Logger.warning("DMX player client not available")

    def get_armed_cue_by_id(self, cue_id: str) -> Cue | None:
        """Returns the armed cue with the given uuid string."""
        with self._lock:
            for cue in self._armed_cues:
                if cue.id == cue_id:
                    return cue
        return None

add_armed_cue(cue)

Adds an armed cue to the list.

Source code in src/cuemsengine/cues/CueHandler.py

def add_armed_cue(self, cue: Cue) -> None:
    """Adds an armed cue to the list."""
    with self._lock:
        self._armed_cues.append(cue)
        self._armed_cues_set.add(cue.id)

arm(cue, init=False, walk=None, epoch=None, wait_report=None)

Arms a cue by appending it to the armed_cues list.

wait_report: optional dict. When this call waits for another thread's arm of the same cue, it sets wait_report["waited_s"] (total seconds waited) and wait_report["holder"] (that thread's name), so a caller can say what really happened instead of guessing from a check made before the call (869fbyjzx).

walk: set only by a background _arm_ahead walk (NodeEngine's PreArm). Its excluded types and should_continue checkpoint also govern this call's own post_go / ActionCue-target recursion, and every cue this call chain actually arms is recorded in walk.armed (869f79ecc).

One arm is in flight per cue id (_arming_registry). An init caller that finds one waits for it, up to _ARM_WAIT_TIMEOUT_S in total, and then re-evaluates: if the cue is still unarmed (the arm it waited on failed or was abandoned) it arms the cue itself instead of giving up.

PUBLISH OR ABANDON. An arm belongs to the STOP/load epoch in which it was REQUESTED: epoch if given, else the walk's, else _disarm_epoch as read here at entry. It is never re-read. If it has moved: - before the cue is claimed (entry, or waking from a wait): return False and arm nothing -- even if the cue is loaded by now, because then the NEW epoch loaded it and a stale caller must not adopt it; - while arm_cue() was running: the arm is ABANDONED at the publish gate. The cue is never marked loaded and never enters the armed list, so no GO can pick it up; what this arm built (a player the STOP may already have killed, video layers) is released here, while this thread still holds the cue's claim. The gate also abandons a cue that was disabled while it was arming. So no thread ever has to disarm a cue after the fact, which is what used to cut a cue the next run was already playing (869f9wqpn).

Source code in src/cuemsengine/cues/CueHandler.py

def arm(
    self,
    cue: Cue,
    init=False,
    walk: _ArmWalk | None = None,
    epoch: int | None = None,
    wait_report: dict | None = None,
) -> bool:
    """Arms a cue by appending it to the armed_cues list.

    wait_report: optional dict. When this call waits for another thread's
    arm of the same cue, it sets wait_report["waited_s"] (total seconds
    waited) and wait_report["holder"] (that thread's name), so a caller
    can say what really happened instead of guessing from a check made
    before the call (869fbyjzx).

    walk: set only by a background _arm_ahead walk (NodeEngine's PreArm).
    Its excluded types and should_continue checkpoint also govern this
    call's own post_go / ActionCue-target recursion, and every cue this
    call chain actually arms is recorded in walk.armed (869f79ecc).

    One arm is in flight per cue id (_arming_registry). An init caller
    that finds one waits for it, up to _ARM_WAIT_TIMEOUT_S in total, and
    then re-evaluates: if the cue is still unarmed (the arm it waited on
    failed or was abandoned) it arms the cue itself instead of giving up.

    PUBLISH OR ABANDON. An arm belongs to the STOP/load epoch in which it
    was REQUESTED: `epoch` if given, else the walk's, else _disarm_epoch
    as read here at entry. It is never re-read. If it has moved:
    - before the cue is claimed (entry, or waking from a wait): return
      False and arm nothing -- even if the cue is loaded by now, because
      then the NEW epoch loaded it and a stale caller must not adopt it;
    - while arm_cue() was running: the arm is ABANDONED at the publish
      gate. The cue is never marked loaded and never enters the armed
      list, so no GO can pick it up; what this arm built (a player the
      STOP may already have killed, video layers) is released here,
      while this thread still holds the cue's claim.
    The gate also abandons a cue that was disabled while it was arming.
    So no thread ever has to disarm a cue after the fact, which is what
    used to cut a cue the next run was already playing (869f9wqpn).
    """
    if cue is None:
        return False
    if epoch is None:
        epoch = (
            walk.epoch
            if walk is not None and walk.epoch is not None
            else self._disarm_epoch
        )
    if walk is not None:
        if walk.skip_types and isinstance(cue, walk.skip_types):
            return False
        if walk.should_continue is not None and not walk.should_continue():
            return False

    wait_deadline = None
    waited_total = 0.0
    while True:
        needs_disarm = False
        claim = None
        pending = None
        stale = False

        with self._lock:
            registry = self._arming_registry()
            in_flight = registry.get(cue.id)
            if epoch != self._disarm_epoch:
                # Checked FIRST, on entry and on every wake.
                stale = True
            elif hasattr(cue, "loaded") and cue.loaded:
                if not cue.enabled:
                    needs_disarm = True
            elif in_flight is not None:
                if not init:
                    # Non-init callers just register; no need to wait
                    return False
                # Another thread is arming — wait for it outside the lock
                pending = in_flight
                pending.waiters += 1
            elif not init:
                if cue.id not in self._armed_cues_set:
                    self._armed_cues.append(cue)
                    self._armed_cues_set.add(cue.id)
            elif cue._local and cue.enabled:
                # Claim the cue inside the lock to block concurrent arm
                # attempts; released in the finally below (outside the
                # lock — intentional: avoids holding it during arm_cue()).
                claim = _ArmClaim(holder=current_thread().name, started=monotonic())
                registry[cue.id] = claim

        if stale:
            Logger.info(
                f"Not arming {type(cue).__name__} {cue.id}: a STOP or a "
                "load happened since this arm was requested"
            )
            return False

        if pending is None:
            break

        if wait_deadline is None:
            wait_deadline = monotonic() + self._ARM_WAIT_TIMEOUT_S
        remaining = wait_deadline - monotonic()
        Logger.debug(
            f"Waiting for in-progress arm of {type(cue).__name__} {cue.id}"
        )
        wait_started = monotonic()
        woke = remaining > 0 and pending.event.wait(timeout=remaining)
        waited_total += monotonic() - wait_started
        if wait_report is not None:
            wait_report["waited_s"] = waited_total
            wait_report["holder"] = pending.holder
        with self._lock:
            pending.waiters -= 1
        if not woke:
            Logger.warning(
                f"Timed out waiting for arm of {cue.id} (held by "
                f"{pending.holder} for {monotonic() - pending.started:.1f}s)"
            )
            return (
                bool(getattr(cue, "loaded", False)) and epoch == self._disarm_epoch
            )
        # Woken: loop and look again. Epoch moved -> give up; loaded ->
        # done; still unarmed -> this thread claims it and arms it itself.

    # Disarm disabled-but-loaded cues outside lock (disarm acquires lock)
    if needs_disarm:
        self.disarm(cue, reason="disabled")
        return False

    if claim is None:
        return not needs_disarm

    abandoned = None
    try:
        Logger.info(f"Arming {type(cue).__name__} {cue.id}")
        arm_started = monotonic()
        arm_cue(cue)
        # The publish gate. Same lock as the epoch bumps (stop_all_cues,
        # disarm_all): either this section comes first and the STOP finds
        # the cue in the armed list and disarms it itself, or the bump
        # comes first and the cue is never published.
        with self._lock:
            if epoch != self._disarm_epoch:
                abandoned = "a STOP or a load happened while it was arming"
            elif not cue.enabled:
                abandoned = "it was disabled while it was arming"
            else:
                cue.loaded = True
                # Measurement only (the armed-to-start and never-played
                # lines): set at every publish, so a cue object reused
                # across runs never carries the previous arm's stamps.
                cue._armed_at = monotonic()
                cue._ever_played = False
                if cue.id not in self._armed_cues_set:
                    self._armed_cues.append(cue)
                    self._armed_cues_set.add(cue.id)
        if abandoned is not None:
            Logger.info(
                f"Abandoned arm of {type(cue).__name__} {cue.id}: {abandoned}"
            )
            self._release_unpublished(cue, "abandoned_arm")
            return False
        # Engine side only: for audio that is the spawn and the JACK port
        # wait; for video it is the OSC sends, and the file open is the
        # videocomposer's own "AsyncVideoLoader: Loaded ... in Nms".
        Logger.info(
            f"Armed {type(cue).__name__} {cue.id} in "
            f"{cue._armed_at - arm_started:.3f} s"
        )
        if walk is not None:
            walk.armed.append(cue)
        if isinstance(cue, AudioCue):
            try:
                self.communications_thread.add_player(
                    f"audioplayer_{cue.id}", None, timeout=0.1
                )
            except Exception:
                pass
    except Exception as e:
        # Fail loud, fail local. _arm_ahead() runs on the GO daemon
        # thread, so letting this propagate would kill a *playing* cue's
        # thread before loop_cue() and skip its generation-tracked
        # cleanup — far worse than one broken cue. Leave the cue
        # unloaded and let go()'s fallback re-arm retry it; if that also
        # fails, go() raises where an operator can see it.
        Logger.error(f"Failed to arm {type(cue).__name__} {cue.id}: {e}")
        Logger.exception(e)
        cue.loaded = False
        # ...and give back whatever it got as far as building (a spawned
        # player, registered layers), still under this cue's claim.
        self._release_unpublished(cue, "arm_failed")
        return False
    finally:
        # Always, and before the recursion below: a leaked claim would
        # block every later arm of this cue id, across reloads too, and a
        # claim held through the recursion would deadlock a go-chain cycle.
        with self._lock:
            registry = self._arming_registry()
            if registry.get(cue.id) is claim:
                del registry[cue.id]
        claim.event.set()

    # Recursive arms — only reached if cue was actually armed.
    # The claim is released by now; the loaded guard prevents re-arm.
    if cue.post_go == "go" and cue._target_object:
        if cue._target_object.enabled:
            if walk is not None:
                self.arm(cue._target_object, init, walk=walk, epoch=epoch)
            else:
                self.arm(cue._target_object, init, epoch=epoch)

    # ActionCue(play) and FadeCue(fade_action) + target = 1 unit. Arm
    # target
    # so it's ready when the action fires (ActionCue has zero duration;
    # FadeCue
    # expects target_cue already armed before reading its OSC cache).
    if isinstance(cue, ActionCue) and cue._action_target_object:
        if cue.action_type in ("play", "fade_action"):
            if walk is not None:
                self.arm(cue._action_target_object, init, walk=walk, epoch=epoch)
            else:
                self.arm(cue._action_target_object, init, epoch=epoch)

    return True

arm_epoch()

The current STOP/load epoch. Whoever requests an arm on one thread and performs it on another captures this at request time and passes it along (arm(epoch=), _ArmWalk.epoch, _arm_ahead(arm_epoch=)).

Source code in src/cuemsengine/cues/CueHandler.py

def arm_epoch(self) -> int:
    """The current STOP/load epoch. Whoever requests an arm on one thread
    and performs it on another captures this at request time and passes
    it along (arm(epoch=), _ArmWalk.epoch, _arm_ahead(arm_epoch=))."""
    return self._disarm_epoch

armed_inventory()

What this node holds armed right now: published (loaded) cues by type, the video layers they hold, and how many of them a GO owns (playing) versus are only held (idle). CueLists are left out: they hold nothing. Measurement only.

Source code in src/cuemsengine/cues/CueHandler.py

def armed_inventory(self) -> dict:
    """What this node holds armed right now: published (loaded) cues by
    type, the video layers they hold, and how many of them a GO owns
    (playing) versus are only held (idle). CueLists are left out: they
    hold nothing. Measurement only."""
    with self._lock:
        held = [
            c
            for c in self._armed_cues
            if getattr(c, "loaded", False) and not isinstance(c, CueList)
        ]
    inv = {
        "cues": len(held),
        "video": 0,
        "layers": 0,
        "audio": 0,
        "dmx": 0,
        "other": 0,
        "playing": 0,
        "idle": 0,
    }
    for cue in held:
        if isinstance(cue, VideoCue):
            inv["video"] += 1
            inv["layers"] += len(getattr(cue, "_layer_ids", None) or [])
        elif isinstance(cue, AudioCue):
            inv["audio"] += 1
        elif isinstance(cue, DmxCue):
            inv["dmx"] += 1
        else:
            inv["other"] += 1
        inv["playing" if getattr(cue, "_playing", False) else "idle"] += 1
    return inv

cancel_parked(cue)

Cancel a cue that was dispatched but has not produced output yet.

Used when a cue is DISABLED after its chain was dispatched. Without this the cue would still fire: the whole chain leaves at the trigger and nothing downstream re-reads enabled. A cue that already revealed is left alone — disabling never cuts live playback.

The cue keeps a stamp of the pass it was cancelled out of, so re-enabling it before its slot puts it back (rejoin_chain).

Source code in src/cuemsengine/cues/CueHandler.py

def cancel_parked(self, cue: Cue) -> bool:
    """Cancel a cue that was dispatched but has not produced output yet.

    Used when a cue is DISABLED after its chain was dispatched. Without
    this the cue would still fire: the whole chain leaves at the trigger
    and nothing downstream re-reads `enabled`. A cue that already
    revealed is left alone — disabling never cuts live playback.

    The cue keeps a stamp of the pass it was cancelled out of, so
    re-enabling it before its slot puts it back (rejoin_chain).
    """
    if getattr(cue, "_revealed", False) or not getattr(cue, "_playing", False):
        return False
    with self._lock:
        self._chain_epoch = self._chain_epoch + 1
        epoch = self._chain_epoch
        cue._stop_requested = True
        cue._go_generation = getattr(cue, "_go_generation", 0) + 1
        cue._go_epoch = epoch
        cue._playing = False
        # Remember where it would have played, so a re-enable can put it
        # back at the same anchor instead of losing the slot.
        arrival = getattr(cue, "_dispatch_arrival_ms", None)
        if arrival is not None:
            cue._chain_pass = (epoch, arrival)
    self._end_illumination(cue)
    Logger.info(f"Cue {cue.id} disabled before its slot — cancelled")
    return True

cancel_pending_descendants(cue)

Cancel the cues an Auto-continue chain dispatched behind cue.

A chained cue only exists because its predecessor ran, so stopping a cue must stop everything scheduled behind it. Under Phase 2 the whole chain is dispatched at the trigger, so by now those cues are parked on their own anchors and nothing else would hold them back — the _stop_requested guard that used to sit in front of the dispatch has moved to chain entry.

Never touches a cue that has already produced output (_revealed): that one is on stage and stays there. The walk continues PAST it, because the cues behind it still depend on the cue being stopped. Non-local cues are skipped too — the node that owns them runs this same walk on its own graph (ActionCues are local everywhere), so the union covers the cluster.

Returns how many cues were cancelled.

Source code in src/cuemsengine/cues/CueHandler.py

def cancel_pending_descendants(self, cue: Cue) -> int:
    """Cancel the cues an Auto-continue chain dispatched behind `cue`.

    A chained cue only exists because its predecessor ran, so stopping a
    cue must stop everything scheduled behind it. Under Phase 2 the whole
    chain is dispatched at the trigger, so by now those cues are parked on
    their own anchors and nothing else would hold them back — the
    `_stop_requested` guard that used to sit in front of the dispatch has
    moved to chain entry.

    Never touches a cue that has already produced output (`_revealed`):
    that one is on stage and stays there. The walk continues PAST it,
    because the cues behind it still depend on the cue being stopped.
    Non-local cues are skipped too — the node that owns them runs this same
    walk on its own graph (ActionCues are local everywhere), so the union
    covers the cluster.

    Returns how many cues were cancelled.
    """
    cancelled = []
    with self._lock:
        # The pass this cue belongs to. Cues that a NEWER pass has since
        # claimed are not ours to cancel — a loop-back can already have
        # re-scheduled them against a fresh trigger.
        target_epoch = getattr(cue, "_go_epoch", 0)
        self._chain_epoch = self._chain_epoch + 1
        epoch = self._chain_epoch
        # Mark the target too: a dispatch of it currently blocked in its
        # own arm must not come back to life after this.
        cue._go_epoch = epoch

        node = (
            getattr(cue, "_target_object", None)
            if getattr(cue, "post_go", None) == "go"
            else None
        )
        # A circular project (A→B→C→A) would otherwise re-mark the same
        # cues on every revolution until the safety limit, reporting ~1024
        # cancellations and firing that many remove_cue sends.
        seen = set()
        while node is not None and id(node) not in seen:
            seen.add(id(node))
            is_local = getattr(node, "_local", False)
            owned = getattr(node, "_go_epoch", 0) <= target_epoch
            if is_local and owned and getattr(node, "enabled", False):
                if not getattr(node, "_revealed", False):
                    node._stop_requested = True
                    node._go_generation = getattr(node, "_go_generation", 0) + 1
                    node._go_epoch = epoch
                    node._playing = False
                    cancelled.append(node)
            elif is_local and owned:
                # Disabled: nothing to cancel, but claim it so enabling it
                # later cannot rejoin a chain that was stopped.
                node._go_epoch = epoch
            if getattr(node, "post_go", None) != "go":
                # A chain break is a hand-off point, not a descendant.
                break
            node = getattr(node, "_target_object", None)
            if len(seen) > 1024:
                Logger.error(
                    "cancel_pending_descendants hit safety limit; aborting"
                )
                break

    # Outside the lock: a cancelled cue's thread leaves through the
    # generation guard, which sits upstream of the usual remove_cue, and
    # the whole chain has been lit since the trigger — so the highlight
    # has to be dropped here or the UI keeps showing dead cues as running.
    for node in cancelled:
        self._end_illumination(node)
    if cancelled:
        # str(): a cue's id is a Uuid, not a str, and join() would raise —
        # inside the one log line that tells an operator this happened.
        Logger.info(
            f"Stop of cue {cue.id} cancelled {len(cancelled)} scheduled "
            f"cue(s) behind it: {', '.join(str(n.id) for n in cancelled)}"
        )
    return len(cancelled)

describe_arm_in_flight(cue)

Who is arming this cue right now and for how long, as text for a log line; None when no arm of it is in flight.

Source code in src/cuemsengine/cues/CueHandler.py

def describe_arm_in_flight(self, cue: Cue) -> str | None:
    """Who is arming this cue right now and for how long, as text for a
    log line; None when no arm of it is in flight."""
    with self._lock:
        claim = self._arming_registry().get(cue.id)
    if claim is None:
        return None
    return f"held by {claim.holder} for {monotonic() - claim.started:.1f}s"

disarm(cue, reason='unspecified')

Disarms a cue by removing it from the armed_cues list.

reason names why (cue_end, stop_action, disabled, project_changed, load, ready_script, shutdown) and is logged: a disarm, the automatic one at cue end included, must never be silent.

Source code in src/cuemsengine/cues/CueHandler.py

def disarm(self, cue: Cue, reason: str = "unspecified") -> bool:
    """Disarms a cue by removing it from the armed_cues list.

    ``reason`` names why (``cue_end``, ``stop_action``, ``disabled``,
    ``project_changed``, ``load``, ``ready_script``, ``shutdown``) and is
    logged: a disarm, the automatic one at cue end included, must never
    be silent.
    """
    cue._playing = False
    if hasattr(cue, "loaded") and cue.loaded:
        self.remove_armed_cue(cue)
        cue.loaded = False
        armed_at = getattr(cue, "_armed_at", None)
        if (
            armed_at is not None
            and not getattr(cue, "_ever_played", False)
            and reason != "cue_end"
            and not isinstance(cue, CueList)
        ):
            # An arm that bought nothing: what a pre-arm policy costs.
            # Not a CueList: go() never dispatches one (the project's
            # root is armed at every load and STOP) and it holds nothing.
            Logger.info(
                f"Cue {cue.id} disarmed after {monotonic() - armed_at:.1f} s "
                f"armed, never played ({reason})"
            )
        try:
            if isinstance(cue, AudioCue):
                self.communications_thread.remove_player(
                    f"audioplayer_{cue.id}", timeout=0.1
                )
            self.communications_thread.remove_cue(cue.id, timeout=0.1)
        except Exception:
            pass

        self._release_cue_resources(cue, reason)
        return True

    return False

disarm_all(reason='unspecified')

Disarms all cues.

Epoch bump, snapshot and clearing the armed list are ONE lock section. Clearing the list after the loop (as this used to) wiped any cue a current-epoch arm published while the loop was killing players: loaded but not in the list, so arm() said "already armed", find_armed_cue() said no, and the cue was unplayable until a reload.

Source code in src/cuemsengine/cues/CueHandler.py

def disarm_all(self, reason: str = "unspecified") -> None:
    """Disarms all cues.

    Epoch bump, snapshot and clearing the armed list are ONE lock
    section. Clearing the list after the loop (as this used to) wiped
    any cue a current-epoch arm published while the loop was killing
    players: loaded but not in the list, so arm() said "already armed",
    find_armed_cue() said no, and the cue was unplayable until a reload.
    """
    self.stop_all_cues()
    with self._lock:
        self._disarm_epoch = self._disarm_epoch + 1
        cues_snapshot = list(self._armed_cues)
        self._armed_cues = []
        self._armed_cues_set.clear()
    for cue in cues_snapshot:
        self.disarm(cue, reason=reason)

disarm_if_unchanged(cue, gen, disarm_epoch, armed_at, reason)

Disarm a cue only if nothing took it since the caller decided to.

In one lock section: the generation, the STOP/load epoch and the arm stamp must be the caller's, the cue still loaded and no arm of it in flight; then this thread takes the cue's arm claim and unpublishes it (unloaded, out of the armed list). Its player and layers are released outside the lock while the claim is held, so no arm of the same cue can run meanwhile; the claim is released in a finally (a leaked one would make the cue unarmable until the engine restarts). Returns True if it disarmed.

Source code in src/cuemsengine/cues/CueHandler.py

def disarm_if_unchanged(
    self,
    cue: Cue,
    gen: int,
    disarm_epoch: int,
    armed_at: float | None,
    reason: str,
) -> bool:
    """Disarm a cue only if nothing took it since the caller decided to.

    In one lock section: the generation, the STOP/load epoch and the arm
    stamp must be the caller's, the cue still loaded and no arm of it in
    flight; then this thread takes the cue's arm claim and unpublishes it
    (unloaded, out of the armed list). Its player and layers are released
    outside the lock while the claim is held, so no arm of the same cue
    can run meanwhile; the claim is released in a finally (a leaked one
    would make the cue unarmable until the engine restarts). Returns True
    if it disarmed."""
    with self._lock:
        registry = self._arming_registry()
        if (
            getattr(cue, "_go_generation", 0) != gen
            or self._disarm_epoch != disarm_epoch
            or not getattr(cue, "loaded", False)
            or getattr(cue, "_armed_at", None) != armed_at
            or registry.get(cue.id) is not None
        ):
            return False
        claim = _ArmClaim(holder=current_thread().name, started=monotonic())
        registry[cue.id] = claim
        cue.loaded = False
        cue._playing = False
        if cue.id in self._armed_cues_set:
            self._armed_cues.remove(cue)
            self._armed_cues_set.discard(cue.id)
    try:
        Logger.info(f"Disarming cue {cue.id} ({reason})")
        try:
            if isinstance(cue, AudioCue):
                self.communications_thread.remove_player(
                    f"audioplayer_{cue.id}", timeout=0.1
                )
            self.communications_thread.remove_cue(cue.id, timeout=0.1)
        except Exception:
            pass
        self._release_cue_resources(cue, reason)
    finally:
        with self._lock:
            if registry.get(cue.id) is claim:
                del registry[cue.id]
        claim.event.set()
    return True

execute_action(cue, mtc, frozen_mtc_ms=None)

Execute an ActionCue against the running show (see ActionHandler).

Source code in src/cuemsengine/cues/CueHandler.py

def execute_action(
    self,
    cue: ActionCue,
    mtc: MtcListener,
    frozen_mtc_ms: float | None = None,
) -> dict:
    """
    Execute an ActionCue against the running show (see ActionHandler).
    """
    from .ActionHandler import ACTION_HANDLER

    return ACTION_HANDLER.execute_action(cue, mtc, frozen_mtc_ms)

find_armed_cue(cue)

Finds an armed cue with the given uuid.

Source code in src/cuemsengine/cues/CueHandler.py

def find_armed_cue(self, cue: Cue) -> Cue | None:
    """Finds an armed cue with the given uuid."""
    with self._lock:
        return cue.id in self._armed_cues_set

get_armed_cue(cue)

Returns the armed cue with the given uuid.

Source code in src/cuemsengine/cues/CueHandler.py

def get_armed_cue(self, cue: Cue) -> Cue | None:
    """Returns the armed cue with the given uuid."""
    try:
        return self.get_armed_cues().index(cue)
    except ValueError:
        return None

get_armed_cue_by_id(cue_id)

Returns the armed cue with the given uuid string.

Source code in src/cuemsengine/cues/CueHandler.py

def get_armed_cue_by_id(self, cue_id: str) -> Cue | None:
    """Returns the armed cue with the given uuid string."""
    with self._lock:
        for cue in self._armed_cues:
            if cue.id == cue_id:
                return cue
    return None

get_armed_cues()

Returns the list of armed cues.

Source code in src/cuemsengine/cues/CueHandler.py

def get_armed_cues(self) -> list[Cue]:
    """Returns the list of armed cues."""
    with self._lock:
        return self._armed_cues

get_next_cue(cue)

Returns the next cue to be played.

Source code in src/cuemsengine/cues/CueHandler.py

def get_next_cue(self, cue: Cue) -> Cue | None:
    """Returns the next cue to be played."""
    return cue._target_object if cue._target_object else None

go(cue, mtc, frozen_mtc_ms=None, chain_epoch=None, unroll=True, stamp_skipped=None, require_stop_epoch=None, arm_epoch=None, commit_gate=None)

Starts a cue in a thread.

Parameters:

Name	Type	Description	Default
cue	Cue	The cue to start	required
mtc	MtcListener	The MTC listener	required
frozen_mtc_ms	float	Optional frozen MTC timestamp for sync with chained	None
chain_epoch	int	None for a FRESH chain entry (manual GO, go_from	None
unroll	bool	False stops go_threaded from dispatching the rest of the	True
stamp_skipped	list	cues the caller's chain walk skipped because they	None
arm_epoch	int	the STOP/load epoch this dispatch belongs to (see	None
arm_epoch()). None for a fresh entry		read here. A continuation	required
commit_gate	Callable[[Cue, int], bool] | None	called inside the commit section, after every other	None

Returns:

Type	Description
Thread | None	Thread running the cue, or None if the cue is disabled, not local
Thread | None	to this node (the node owning the target will run it via its own
Thread | None	GO/post_go dispatch), or the dispatch was refused as stale/after a
Thread | None	STOP.

Source code in src/cuemsengine/cues/CueHandler.py

@logged
def go(
    self,
    cue: Cue,
    mtc: MtcListener,
    frozen_mtc_ms: float = None,
    chain_epoch: int = None,
    unroll: bool = True,
    stamp_skipped: list = None,
    require_stop_epoch: int = None,
    arm_epoch: int = None,
    commit_gate: Callable[[Cue, int], bool] | None = None,
) -> Thread | None:
    """Starts a cue in a thread.

    Args:
        cue: The cue to start
        mtc: The MTC listener
        frozen_mtc_ms: Optional frozen MTC timestamp for sync with chained
        cues
        chain_epoch: None for a FRESH chain entry (manual GO, go_from
        re-entry, enable-rejoin) — mints a new epoch. Otherwise the epoch
        of the pass this dispatch belongs to, which is validated against
        the cue's own high-water mark and the STOP barrier.
        unroll: False stops go_threaded from dispatching the rest of the
        chain — used by the enable-rejoin, where the chain is already
        unrolled and re-dispatching it would churn live cues.
        stamp_skipped: cues the caller's chain walk skipped because they
        are disabled; stamped with this pass so a later enable can rejoin.
        arm_epoch: the STOP/load epoch this dispatch belongs to (see
        arm_epoch()). None for a fresh entry: read here. A continuation
        inherits its chain's. It goes into every arm this dispatch causes
        (the fallback re-arm, the lookahead, the cue's own thread), and a
        dispatch whose epoch has moved is refused.
        commit_gate: called inside the commit section, after every other
        refusal, with the cue and the generation about to be assigned;
        False (or an exception) refuses the dispatch. A cross-node follow
        uses it so that its cancel and its dispatch are decided in one
        place (869fc8ytz).

    Returns:
        Thread running the cue, or None if the cue is disabled, not local
        to this node (the node owning the target will run it via its own
        GO/post_go dispatch), or the dispatch was refused as stale/after a
        STOP.
    """
    if not cue.enabled:
        Logger.info(f"Cue {cue.id} is disabled, skipping execution")
        return None
    if not getattr(cue, "_local", True):
        # Non-local target: handled by the node where it IS local via that
        # node's own go_threaded → post_go chain. Trying to arm/run it
        # locally would fail at re-arm (no local player), raise inside the
        # caller's thread, and kill chained playback (master videocomposer
        # froze after loop 1 when post_go='go' targeted an audio cue local
        # to slave only).
        Logger.info(f"Cue {cue.id} is not local to this node, skipping execution")
        return None
    Logger.info(f"GO command received. Starting cue {cue.id}")

    if arm_epoch is None:
        arm_epoch = self._disarm_epoch
    is_continuation = chain_epoch is not None
    if not is_continuation:
        # Fresh entry: claim the next epoch. It is greater than every
        # recorded stop and every cue mark at this instant, so the checks
        # below can only refuse it if a STOP arrives LATER — while this
        # call is blocked in the re-arm fallback. That is the right
        # outcome: the operator's STOP came after their GO.
        with self._lock:
            self._chain_epoch = self._chain_epoch + 1
            chain_epoch = self._chain_epoch

    if not hasattr(cue, "loaded") or not cue.loaded:
        # 869fbyjzx: say which case this is. An arm still in flight (a
        # pre-arm, an audio player still starting) is not a failure; the
        # old "pre-arm may have failed" said it was. Best-effort: the arm
        # can start or finish between this look and arm() below, so the
        # wait itself is reported from what arm() actually did. True
        # fallbacks keep "not loaded at go() time" and "Re-arming as
        # fallback": the harnesses count them.
        # An arm in flight that a GO waits for is still a LATE cue when the
        # arm is slower than the GO gap (an audio arm is ~0.47 s cold):
        # arming before the hand-off ("Part B") is the only cure and is
        # not built; see AudioMixer.connect_player_to_outputs (869fbyjzx).
        in_flight = self.describe_arm_in_flight(cue)
        if is_continuation:
            Logger.info(
                f"Cue {cue.id} not armed yet at go() time — dispatching; "
                f"its own thread arms it or waits for the arm in progress"
            )
        elif in_flight is not None:
            Logger.info(
                f"Cue {cue.id} is still being armed ({in_flight}) at go() "
                f"time — waiting for the arm in progress"
            )
        else:
            Logger.warning(
                f"Cue {cue.id} not loaded at go() time — no arm in flight "
                f"(never armed, or an earlier arm failed or was "
                f"abandoned). Re-arming as fallback."
            )
        # A continuation does NOT arm here: dispatch now happens at chain
        # entry, so this call runs on the PREVIOUS cue's thread, and an
        # audio arm can block ~15s on its JACK ports — delaying or killing
        # that cue's own reveal. go_threaded arms the cue on its own
        # thread instead, overlapping its own prewait.
        if not is_continuation:
            wait_report = {}
            self.arm(cue, init=True, epoch=arm_epoch, wait_report=wait_report)
            self._log_arm_wait(cue, "go() time", in_flight, wait_report)
            if not hasattr(cue, "loaded") or not cue.loaded:
                if arm_epoch != self._disarm_epoch:
                    # Not an arm failure: the operator's STOP (or a load)
                    # arrived while this cue was being re-armed, and
                    # arm() refused or abandoned it. The STOP wins.
                    Logger.warning(
                        f"Refusing dispatch of cue {cue.id}: a STOP or a "
                        "load arrived while it was being re-armed"
                    )
                    return None
                raise Exception(
                    f"{cue.__class__.__name__} {cue.id} not loaded to go"
                    f"(re-arm failed)"
                )

    with self._lock:
        # Validate + commit atomically. Both checks are deliberately made
        # AFTER any arm above: that is what lets a STOP which landed
        # during a long arm window survive instead of being wiped by the
        # _stop_requested reset below (F4).
        if arm_epoch != self._disarm_epoch:
            # Whether or not the cue is loaded: if it is, the STOP's own
            # re-arm loaded it for the NEXT run, not for this dispatch.
            Logger.warning(
                f"Refusing dispatch of cue {cue.id}: a STOP or a load "
                "arrived while it was being started"
            )
            return None
        if (
            require_stop_epoch is not None
            and self._last_stop_chain_epoch != require_stop_epoch
        ):
            # The caller validated something against the STOP barrier and
            # then had to drop the lock (the enable-rejoin does). A STOP
            # landing in that window must win — otherwise the cue plays
            # after the operator stopped the show.
            Logger.warning(
                f"Refusing dispatch of cue {cue.id}: a STOP arrived while "
                f"it was being put back into its chain"
            )
            return None
        if chain_epoch <= self._last_stop_chain_epoch:
            # For a continuation: the chain was stopped. For a fresh
            # entry: a STOP landed while we were arming — it is newer than
            # this GO, so it wins (the caller must tolerate None).
            Logger.warning(
                f"Refusing dispatch of cue {cue.id}: a STOP ended this "
                f"chain (pass {chain_epoch} <= stop "
                f"{self._last_stop_chain_epoch})"
            )
            return None
        if is_continuation:
            if getattr(cue, "_go_epoch", 0) >= chain_epoch:
                # Either a newer pass already owns the cue (loop-back
                # racing the original unroll), or this same pass already
                # dispatched it — a post_go='go' cycle, which entry-time
                # dispatch would otherwise spin at thread-spawn rate.
                Logger.warning(
                    f"Refusing stale/duplicate chain dispatch of cue "
                    f"{cue.id} (cue pass {getattr(cue, '_go_epoch', 0)} >= "
                    f"dispatch pass {chain_epoch})"
                )
                return None

        go_gen = getattr(cue, "_go_generation", 0) + 1
        if commit_gate is not None:
            try:
                allowed = commit_gate(cue, go_gen)
            except Exception as e:
                Logger.error(f"Commit gate for cue {cue.id} failed: {e}")
                allowed = False
            if not allowed:
                Logger.info(
                    f"Refusing dispatch of cue {cue.id}: its cross-node "
                    "follow was cancelled or replaced"
                )
                return None
        # Only past every refusal: a dispatch that is refused must not
        # cancel the follow of the pass still playing.
        self._take_follow_on_restart_locked(cue)

        cue._go_epoch = chain_epoch
        # Recorded here, not in the spawned thread: a disable arriving in
        # between has to find this cue's anchor to stamp a rejoin with, and
        # in a loop project it would otherwise read the PREVIOUS cycle's.
        # None (manual GO / go_at_end) means "derived from live MTC by the
        # thread", which is not an anchor a rejoin can be pinned to.
        cue._dispatch_arrival_ms = frozen_mtc_ms
        cue._stop_requested = False
        cue._go_generation = go_gen
        # Output-commit flag: True once this cue has produced output
        # (reveal, or the DMX scene send). The stop walk cancels only
        # cues that have NOT committed — a playing cue is never cut.
        cue._revealed = False
        # When this pass produced output (monotonic); set by go_threaded.
        # A late cross-node follow cancel reads it (869fc8ytz).
        cue._revealed_at = None
        # Lifecycle flag: True while a GO owns this cue; cleared by
        # disarm() and stop_all_cues(). Unlike _go_generation
        # (increment-only), this is a sound "currently playing" signal —
        # the disable-action path uses it to decide whether disarming
        # would cut live playback.
        cue._playing = True
        # Measurement only: a cue a GO took is not a wasted arm, even if
        # it is stopped before its reveal.
        cue._ever_played = True
        # Stamped with the epoch minted in THIS lock section: reading it
        # back off the cue afterwards could capture a newer unrelated
        # dispatch's epoch and mis-anchor the rejoin.
        self._stamp_skipped_locked(stamp_skipped, chain_epoch, frozen_mtc_ms)

    thread = Thread(
        name=f"GO:{cue.__class__.__name__}:{cue.id}",
        target=self.go_threaded,
        args=[cue, mtc, frozen_mtc_ms, go_gen, chain_epoch, unroll],
        kwargs={"arm_epoch": arm_epoch},
        daemon=True,
    )
    thread.start()

    # Duration-aware lookahead: arm ahead until 2 cues with meaningful
    # playback duration are ready. SYNCHRONOUS on the caller's thread, and
    # arm() can block ~15s per audio cue — so it must not run where the
    # caller is something that cannot afford to wait:
    #   - a continuation's caller is the PREVIOUS cue's thread (dispatch
    #     happens at chain entry now), and blocking it would delay that
    #     cue's own reveal. Chain cues are armed by arm()'s recursion at
    #     load and, failing that, by their own thread in go_threaded.
    #   - a rejoin (unroll=False) is called from the command thread and
    #     the cue is already armed by definition.
    if not is_continuation and unroll:
        self._arm_ahead(cue, arm_epoch=arm_epoch)
    return thread

go_from(start_cue, mtc, seed_ms=None, arm_epoch=None, require_stop_epoch=None, commit_gate=None)

Re-enter a post_go='go' chain from start_cue (inclusive), firing THIS node's first local+enabled cue at seed_ms + Σ(chain advance of the enabled cues skipped on the way). Returns the Thread, or None if no local+enabled cue is reachable before a chain break.

Mirrors NodeEngine.go_script's cross-node walk, but for an IN-CHAIN re-entry — specifically an ActionCue 'play' that restarts a sequence (e.g. a circular project's loop-back). Plain go() bails on a non-local target (returns None), which drops every cue local to a DIFFERENT node than the play target: a loop-back 'play' aimed at node01's cue would never re-fire the controller's own cues. Walking here makes each engine re-enter its own local segment, exactly as it does on a GO.

arm_epoch / require_stop_epoch / commit_gate are passed to go() when set: a cross-node follow dispatches on another thread than the one that received it, and a STOP or a cancel in between must win.

Source code in src/cuemsengine/cues/CueHandler.py

def go_from(
    self,
    start_cue: Cue,
    mtc: MtcListener,
    seed_ms: float = None,
    arm_epoch: int = None,
    require_stop_epoch: int = None,
    commit_gate: Callable[[Cue, int], bool] | None = None,
) -> Thread | None:
    """Re-enter a post_go='go' chain from start_cue (inclusive), firing THIS
    node's first local+enabled cue at seed_ms + Σ(chain advance of the
    enabled cues skipped on the way). Returns the Thread, or None if no
    local+enabled cue is reachable before a chain break.

    Mirrors NodeEngine.go_script's cross-node walk, but for an IN-CHAIN
    re-entry — specifically an ActionCue 'play' that restarts a sequence
    (e.g. a circular project's loop-back). Plain go() bails on a non-local
    target (returns None), which drops every cue local to a DIFFERENT node
    than the play target: a loop-back 'play' aimed at node01's cue would
    never re-fire the controller's own cues. Walking here makes each engine
    re-enter its own local segment, exactly as it does on a GO.

    arm_epoch / require_stop_epoch / commit_gate are passed to go() when
    set: a cross-node follow dispatches on another thread than the one
    that received it, and a STOP or a cancel in between must win.
    """
    guards = {
        k: v
        for k, v in (
            ("arm_epoch", arm_epoch),
            ("require_stop_epoch", require_stop_epoch),
            ("commit_gate", commit_gate),
        )
        if v is not None
    }
    if seed_ms is None:
        seed_ms = mtc.main_tc.milliseconds_exact
    cue = start_cue
    sigma_ms = 0.0
    walked = 0
    skipped_disabled = []
    while cue is not None and not (
        getattr(cue, "_local", False) and getattr(cue, "enabled", False)
    ):
        if not getattr(cue, "enabled", False):
            # Remember it: enabling it before its slot rejoins this pass.
            skipped_disabled.append(cue)
        if getattr(cue, "post_go", None) != "go":
            # chain break before any local cue — nothing for this node to do
            cue = None
            break
        if getattr(cue, "enabled", False):
            sigma_ms += self._chain_advance_ms(cue)
        cue = getattr(cue, "_target_object", None)
        walked += 1
        if walked > 1024:
            Logger.error("go_from walk hit safety limit; aborting")
            return None
    if cue is None:
        # No local cue to fire, but the disabled ones still get a pass to
        # rejoin, or enabling one mid-show would do nothing at all.
        if skipped_disabled:
            self.stamp_pass(skipped_disabled, seed_ms)
        return None
    if skipped_disabled:
        return self.go(
            cue,
            mtc,
            seed_ms + sigma_ms,
            stamp_skipped=skipped_disabled,
            **guards,
        )
    return self.go(cue, mtc, seed_ms + sigma_ms, **guards)

go_threaded(cue, mtc, frozen_mtc_ms=None, go_gen=0, chain_epoch=0, unroll=True, arm_epoch=None)

Runs a cue based on its properties.

Parameters:

Name	Type	Description	Default
cue	Cue	The cue to run	required
mtc	MtcListener	The MTC listener (for live MTC)	required
frozen_mtc_ms	float	Optional frozen MTC timestamp in milliseconds.	None
go_gen	int	Generation counter captured at go() time. If the cue's generation has changed by the time the loop ends, another go/stop cycle occurred and this thread must not touch the cue.	0
chain_epoch	int	the pass this cue belongs to, carried unchanged into the continuation dispatch so the whole chain shares it.	0
unroll	bool	False suppresses the continuation dispatch (enable-rejoin of a chain that is already unrolled).	True
arm_epoch	int	the STOP/load epoch of the go() that dispatched this cue; every arm this thread makes (its own cue, the go_at_end lookahead) and the continuation it dispatches belong to it, so none of them survives a STOP.	None

Source code in src/cuemsengine/cues/CueHandler.py

def go_threaded(
    self,
    cue: Cue,
    mtc: MtcListener,
    frozen_mtc_ms: float = None,
    go_gen: int = 0,
    chain_epoch: int = 0,
    unroll: bool = True,
    arm_epoch: int = None,
):
    """Runs a cue based on its properties.

    Args:
        cue: The cue to run
        mtc: The MTC listener (for live MTC)
        frozen_mtc_ms: Optional frozen MTC timestamp in milliseconds.
        go_gen: Generation counter captured at go() time. If the cue's
                generation has changed by the time the loop ends, another
                go/stop cycle occurred and this thread must not touch the
                cue.
        chain_epoch: the pass this cue belongs to, carried unchanged into
                the continuation dispatch so the whole chain shares it.
        unroll: False suppresses the continuation dispatch (enable-rejoin
                of a chain that is already unrolled).
        arm_epoch: the STOP/load epoch of the go() that dispatched this
                cue; every arm this thread makes (its own cue, the
                go_at_end lookahead) and the continuation it dispatches
                belong to it, so none of them survives a STOP.
    """
    follow = _FollowOut()
    try:
        self._run_cue_pass(
            cue,
            mtc,
            frozen_mtc_ms,
            go_gen,
            chain_epoch,
            unroll,
            arm_epoch,
            follow,
        )
    except Exception:
        follow.error = True
        raise
    finally:
        self._leave_follow(cue, follow, arm_epoch)

log_armed_inventory(where)

One INFO line with armed_inventory(). Never raises: it is called at the end of a GO and of a PreArm thread, which must not fail on a log line.

Source code in src/cuemsengine/cues/CueHandler.py

def log_armed_inventory(self, where: str) -> None:
    """One INFO line with armed_inventory(). Never raises: it is called
    at the end of a GO and of a PreArm thread, which must not fail on a
    log line."""
    try:
        inv = self.armed_inventory()
        Logger.info(
            f"Armed inventory after {where}: {inv['cues']} cues "
            f"(video {inv['video']} cues / {inv['layers']} layers, "
            f"audio {inv['audio']} players, dmx {inv['dmx']}, "
            f"other {inv['other']}); playing {inv['playing']}; "
            f"idle {inv['idle']}"
        )
    except Exception as e:
        Logger.warning(f"Could not count the armed inventory ({where}): {e}")

register_action_hook(phase, fn, *, action_types=None)

Register a cue-layer extension hook; forwards to ACTION_HANDLER.

Source code in src/cuemsengine/cues/CueHandler.py

def register_action_hook(
    self,
    phase: str,
    fn,
    *,
    action_types: frozenset | None = None,
) -> None:
    """
    Register a cue-layer extension hook; forwards to ``ACTION_HANDLER``.
    """
    from .ActionHandler import ACTION_HANDLER

    ACTION_HANDLER.register_action_hook(
        phase, fn, source="cue_layer", action_types=action_types
    )

rejoin_chain(cue, mtc)

Put a cue enabled mid-pass back into its running chain.

The chain walk stamped this cue with the pass that would have dispatched it while it was disabled. If that pass is still alive and the cue's slot has not gone by, dispatch it at its original anchor so it plays exactly when it was always going to.

Dispatched with unroll=False: the rest of the chain is already running, and re-dispatching it would churn cues that are on stage.

Returns True if the cue was put back.

Source code in src/cuemsengine/cues/CueHandler.py

def rejoin_chain(self, cue: Cue, mtc: MtcListener) -> bool:
    """Put a cue enabled mid-pass back into its running chain.

    The chain walk stamped this cue with the pass that would have
    dispatched it while it was disabled. If that pass is still alive and
    the cue's slot has not gone by, dispatch it at its original anchor so
    it plays exactly when it was always going to.

    Dispatched with unroll=False: the rest of the chain is already
    running, and re-dispatching it would churn cues that are on stage.

    Returns True if the cue was put back.
    """
    if not getattr(cue, "enabled", False) or not getattr(cue, "_local", False):
        return False
    stamp = getattr(cue, "_chain_pass", None)
    if not stamp:
        return False
    epoch, arrival_ms = stamp
    if arrival_ms is None:
        return False

    with self._lock:
        barrier = self._last_stop_chain_epoch
        if epoch <= self._last_stop_chain_epoch:
            Logger.info(
                f"Cue {cue.id} enabled, but the chain it belonged to was "
                f"stopped — it will play on the next GO"
            )
            return False
        if getattr(cue, "_go_epoch", 0) > epoch:
            # A stop action (or a newer pass) claimed this cue after the
            # stamp: a stop takes its scheduled descendants with it, and
            # that outranks re-enabling one of them.
            Logger.info(
                f"Cue {cue.id} enabled, but a stop already cancelled this "
                f"part of the chain — it will play on the next GO"
            )
            return False

    start_ms = arrival_ms + cue.prewait.milliseconds_exact
    now_ms = mtc.main_tc.milliseconds_exact
    if start_ms < now_ms - self._LATE_DISPATCH_TOLERANCE_MS:
        # Firing it now would put it on stage at the wrong moment, which
        # mid-show is worse than not playing it at all.
        Logger.warning(
            f"Cue {cue.id} enabled after its slot (start={start_ms:.0f}ms, "
            f"mtc={now_ms:.0f}ms) — NOT firing it late; it will play on "
            f"the next GO"
        )
        return False

    Logger.info(
        f"Cue {cue.id} enabled before its slot — rejoining the running "
        f"chain at {start_ms:.0f}ms"
    )
    # require_stop_epoch closes the window between the check above and the
    # dispatch: the lock is dropped in between, and a STOP landing there
    # must not be overridden by the fresh epoch this go() mints.
    return (
        self.go(cue, mtc, arrival_ms, unroll=False, require_stop_epoch=barrier)
        is not None
    )

remove_armed_cue(cue)

Removes an armed cue from the list.

Source code in src/cuemsengine/cues/CueHandler.py

def remove_armed_cue(self, cue: Cue) -> bool:
    """Removes an armed cue from the list."""
    with self._lock:
        if cue.id in self._armed_cues_set:
            self._armed_cues.remove(cue)
            self._armed_cues_set.remove(cue.id)
            return True
    return False

route_audio_message(path_parts, value)

Route audio OSCQuery message to the appropriate handler.

Parameters:

Name	Type	Description	Default
path_parts	list[str]	Path parts after 'audio' (e.g., ['mixer', '0',	required
value		The OSC value to set	required

Source code in src/cuemsengine/cues/CueHandler.py

def route_audio_message(self, path_parts: list[str], value) -> None:
    """Route audio OSCQuery message to the appropriate handler.

    Args:
        path_parts: Path parts after 'audio' (e.g., ['mixer', '0',
        'master', 'volume']
                    or ['cue', '<uuid>', '0', 'volume'])
        value: The OSC value to set
    """
    if not path_parts:
        Logger.warning("Empty audio path parts")
        return

    if path_parts[0] == "mixer":
        # Route to audio mixer: ['mixer', '<output_index>', '<channel>',
        # 'volume']
        # → /audiomixer/0_mixer/<channel>
        if len(path_parts) >= 3:
            output_index = path_parts[1]
            channel = path_parts[2]
            mixer_cmd = f"/audiomixer/{output_index}_mixer/{channel}"
            mixer_client = PLAYER_HANDLER.get_audio_mixer_client()
            if mixer_client:
                Logger.debug(f"Routing audio mixer: {mixer_cmd} = {value}")
                mixer_client.set_value(mixer_cmd, float(value))
            else:
                Logger.warning("Audio mixer client not available")
        else:
            Logger.warning(f"Invalid mixer path: {path_parts}")

    elif path_parts[0] == "cue":
        # Route to cue player: ['cue', '<uuid>', '<channel>', 'volume']
        # → /vol<channel> on the armed cue's OSC client
        if len(path_parts) >= 3:
            cue_uuid = path_parts[1]
            channel = path_parts[2]
            audio_cmd = f"/vol{channel}"
            cue = self.get_armed_cue_by_id(cue_uuid)
            if cue and hasattr(cue, "_osc") and cue._osc:
                # UI already sends 0.0-1.0 via sliderToFloat(); just clamp
                vol_value = max(0.0, min(1.0, float(value)))
                Logger.debug(
                    f"Routing audio cue {cue_uuid}: {audio_cmd} = {vol_value}"
                )
                cue._osc.set_value(audio_cmd, vol_value)
            else:
                Logger.warning(f"Cue {cue_uuid} not found or has no OSC client")
        else:
            Logger.warning(f"Invalid cue audio path: {path_parts}")
    else:
        Logger.warning(f"Unknown audio path type: {path_parts[0]}")

route_dmx_message(path_parts, value)

Route DMX OSCQuery message to the DMX player.

Parameters:

Name	Type	Description	Default
path_parts	list[str]	Path parts after 'dmx' (e.g., ['mixer', '0', 'channel',	required
value		The OSC value to set	required

Source code in src/cuemsengine/cues/CueHandler.py

def route_dmx_message(self, path_parts: list[str], value) -> None:
    """Route DMX OSCQuery message to the DMX player.

    Args:
        path_parts: Path parts after 'dmx' (e.g., ['mixer', '0', 'channel',
        '1'])
        value: The OSC value to set
    """
    if not path_parts:
        Logger.warning("Empty DMX path parts, skipping routing")
        return
    if "mixer" not in path_parts:
        Logger.warning(
            f'Invalid DMX path (no "mixer" keyword): {path_parts}, skipping routing'
        )
        return

    # Build DMX command from path: find 'mixer' and use everything after it
    # +1 to skip 'mixer' keyword
    mixer_index = path_parts.index("mixer") + 1
    dmx_cmd = "/" + "/".join(path_parts[mixer_index:])
    dmx_client = PLAYER_HANDLER.get_dmx_player_client()
    if dmx_client:
        Logger.debug(f"Routing DMX: {dmx_cmd} = {value}")
        dmx_client.set_value(dmx_cmd, value)
    else:
        Logger.warning("DMX player client not available")

set_nng_comms(hub_address, node_id)

Set the communications infrastructure

Source code in src/cuemsengine/cues/CueHandler.py

def set_nng_comms(self, hub_address: str, node_id: str):
    """Set the communications infrastructure"""
    from time import sleep

    Logger.info(f"Starting communications for Node {node_id}")
    Logger.info(f"NNG Hub address: {hub_address}")
    self.communications_thread = NodeCommunications(
        hub_address=hub_address, node_id=node_id
    )
    self.communications_thread.start()

    # Wait for NNG thread to initialize (prevents race condition in
    # nni_random)
    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")

stamp_pass(skipped, arrival_ms)

Mint a chain epoch and stamp skipped with it, dispatching nothing.

For the case where a chain walk finds NO local+enabled cue on this node (every local cue in the chain is currently disabled): go() is never called, so nothing would mint an epoch, and enabling one of those cues later could not rejoin the pass. Returns the epoch.

Source code in src/cuemsengine/cues/CueHandler.py

def stamp_pass(self, skipped: list, arrival_ms: float) -> int:
    """Mint a chain epoch and stamp `skipped` with it, dispatching nothing.

    For the case where a chain walk finds NO local+enabled cue on this
    node (every local cue in the chain is currently disabled): go() is
    never called, so nothing would mint an epoch, and enabling one of
    those cues later could not rejoin the pass. Returns the epoch.
    """
    with self._lock:
        self._chain_epoch = self._chain_epoch + 1
        epoch = self._chain_epoch
        self._stamp_skipped_locked(skipped, epoch, arrival_ms)
    return epoch

stop_all_cues()

Signal all armed cues to stop their playback loops.

Also bumps each cue's generation counter so that any still-running go_threaded threads will see a mismatch and skip post-loop cleanup (disarm), which would otherwise undo the re-arm that follows.

Source code in src/cuemsengine/cues/CueHandler.py

def stop_all_cues(self) -> None:
    """Signal all armed cues to stop their playback loops.

    Also bumps each cue's generation counter so that any still-running
    go_threaded threads will see a mismatch and skip post-loop cleanup
    (disarm), which would otherwise undo the re-arm that follows.
    """
    with self._lock:
        # Claim an epoch for this STOP and record it: every continuation
        # of a pass that started at or before it is now refused, including
        # one currently blocked inside go()'s re-arm fallback whose cue is
        # not in _armed_cues yet (F4).
        self._chain_epoch = self._chain_epoch + 1
        self._last_stop_chain_epoch = self._chain_epoch
        # The teardown starts HERE: between this call and disarm_all()
        # the engine resets DMX and video and kills every audio player.
        # Moving the epoch in this same section means an arm publishes
        # either before it (then it is in _armed_cues, flagged just
        # below and disarmed later) or after it (then it is abandoned).
        self._disarm_epoch = self._disarm_epoch + 1
        for cue in self._armed_cues:
            cue._stop_requested = True
            cue._go_generation = getattr(cue, "_go_generation", 0) + 1
            cue._go_epoch = self._chain_epoch
            cue._playing = False

stop_dispatched_follow(cue, gen, disarm_epoch, armed_at, may_cut, grace_s, started_before=None)

Stop a cue a cross-node follow dispatched, after the owner cancelled that follow (869fc8ytz). Returns (outcome, age):

• "owned": a newer GO or a STOP has the cue (generation or epoch moved); nothing is touched;
• "stopped": it had not started, or started less than grace_s ago and may_cut (the source was stopped or started again);
• "left": it started too long ago, or the cancel came from an error -- a cue on stage is never cut for an internal reason.

age is seconds since the cue produced output, None if it had not. started_before is when the cue went on stage in a pass that was still playing when the follow dispatched it again (recorded by the follow's commit gate): its output counts as started. The decision and the stop flags are one lock section; the disarm is guarded again (disarm_if_unchanged).

Source code in src/cuemsengine/cues/CueHandler.py

def stop_dispatched_follow(
    self,
    cue: Cue,
    gen: int,
    disarm_epoch: int,
    armed_at: float | None,
    may_cut: bool,
    grace_s: float,
    started_before: float | None = None,
) -> tuple[str, float | None]:
    """Stop a cue a cross-node follow dispatched, after the owner
    cancelled that follow (869fc8ytz). Returns (outcome, age):

    - "owned": a newer GO or a STOP has the cue (generation or epoch
      moved); nothing is touched;
    - "stopped": it had not started, or started less than `grace_s` ago
      and `may_cut` (the source was stopped or started again);
    - "left": it started too long ago, or the cancel came from an error
      -- a cue on stage is never cut for an internal reason.

    `age` is seconds since the cue produced output, None if it had not.
    `started_before` is when the cue went on stage in a pass that was
    still playing when the follow dispatched it again (recorded by the
    follow's commit gate): its output counts as started.
    The decision and the stop flags are one lock section; the disarm is
    guarded again (disarm_if_unchanged)."""
    with self._lock:
        if (
            getattr(cue, "_go_generation", 0) != gen
            or self._disarm_epoch != disarm_epoch
        ):
            return "owned", None
        started = getattr(cue, "_revealed_at", None)
        if started is None and getattr(cue, "_revealed", False):
            started = monotonic()
        if started is None:
            started = started_before
        age = None if started is None else monotonic() - started
        if age is not None and (age >= grace_s or not may_cut):
            return "left", age
        cue._stop_requested = True
        new_gen = gen + 1
        cue._go_generation = new_gen
        cue._playing = False
    self.cancel_pending_descendants(cue)
    self._end_illumination(cue)
    # Let the cue's own loop see the flag and leave (polls every 20 ms),
    # as the stop action does.
    sleep(0.1)
    self.disarm_if_unchanged(
        cue, new_gen, disarm_epoch, armed_at, reason="follow_cancelled"
    )
    return "stopped", age

wait_for_cue(thread)

Waits for a cue to finish.

Source code in src/cuemsengine/cues/CueHandler.py

def wait_for_cue(self, thread: Thread) -> None:
    """Waits for a cue to finish."""
    Logger.info(f"Waiting for {thread.name} to finish")
    while thread.is_alive():
        sleep(1)
    thread.join()
    Logger.info(f"{thread.name} finished")

arm_cue(cue)

Type-specific logic when arming a cue

Source code in src/cuemsengine/cues/arm_cue.py

@singledispatch
def arm_cue(cue: Cue):
    """
    Type-specific logic when arming a cue
    """
    pass

arm_dmxCue(cue)

Arm a DMX cue by extracting DMX scene data.

The DMX scene data is already loaded in the cue object from the script XML. We extract the universe and channel data from cue.DmxScene and store it in a format suitable for sending as OSC bundles to the local DMX player.

Note: cue._local should be set by check_mappings() based on the output_name. For DMX cues, the output_name format is "{node_uuid}" (just the node UUID). A DMX cue can have multiple outputs (one per target node). check_mappings() should iterate through all outputs and set _local=True if ANY output_name matches the current node UUID. Other outputs are ignored. This function is only called for local cues (checked in CueHandler.arm()).

Source code in src/cuemsengine/cues/arm_cue.py

@arm_cue.register
def arm_dmxCue(cue: DmxCue):
    """Arm a DMX cue by extracting DMX scene data.

    The DMX scene data is already loaded in the cue object from the script XML.
    We extract the universe and channel data from cue.DmxScene and store it
    in a format suitable for sending as OSC bundles to the local DMX player.

    Note: cue._local should be set by check_mappings() based on the
    output_name.
    For DMX cues, the output_name format is "{node_uuid}" (just the node UUID).
    A DMX cue can have multiple outputs (one per target node). check_mappings()
    should iterate through all outputs and set _local=True if ANY output_name
    matches the current node UUID. Other outputs are ignored.
    This function is only called for local cues (checked in CueHandler.arm()).
    """
    # Verify that _local is set (should be set by check_mappings() from
    # output_name)
    is_local = getattr(cue, "_local", True)
    if not is_local:
        Logger.warning(
            f"DMX cue {cue.id} is not local but arm_dmxCue was called. "
            f"This should not happen - check_mappings() should set _local"
            f"from output_name.",
            extra={"caller": cue.__class__.__name__},
        )
        return

    # Get the local DMX player client
    dmx_client = PLAYER_HANDLER.get_dmx_player_client()

    if dmx_client is None:
        Logger.error(
            f"No local DMX player available for cue {cue.id}",
            extra={"caller": cue.__class__.__name__},
        )
        return

    # Assign the local DMX player client to the cue
    cue._osc = dmx_client
    Logger.debug(
        f"DMX cue {cue.id} will use local DMX player (output_name inferred"
        f"_local={is_local})",
        extra={"caller": cue.__class__.__name__},
    )

    # Extract frame data from the DmxScene
    try:
        universe_frames = {}

        # Check if the cue has a DmxScene
        if cue.DmxScene is None:
            Logger.warning(
                f"DMX cue {cue.id} has no DmxScene data",
                extra={"caller": cue.__class__.__name__},
            )
            cue._dmx_frames = {}
            return

        # Extract universe data from the DmxScene
        dmx_universe = cue.DmxScene.DmxUniverse
        if dmx_universe is not None:
            universe_num = dmx_universe.universe_num
            channels_data = {}

            # Extract channel data from dmx_channels list
            if dmx_universe.dmx_channels:
                for dmx_channel in dmx_universe.dmx_channels:
                    channel_num = dmx_channel.channel
                    channel_value = dmx_channel.value
                    channels_data[channel_num] = channel_value

            if channels_data:
                universe_frames[universe_num] = channels_data

        # Store the parsed frame data in the cue for use when running
        cue._dmx_frames = universe_frames

        if universe_frames:
            total_channels = sum(len(channels) for channels in universe_frames.values())
            Logger.info(
                f"DMX cue {cue.id} armed: {len(universe_frames)} universe(s),"
                f"{total_channels} channel(s)",
                extra={"caller": cue.__class__.__name__},
            )
        else:
            Logger.warning(
                f"DMX cue {cue.id} armed but no channel data found in DmxScene",
                extra={"caller": cue.__class__.__name__},
            )

    except Exception as e:
        Logger.error(
            f"Error arming DMX cue {cue.id}: {e}",
            extra={"caller": cue.__class__.__name__},
        )
        Logger.exception(e)
        # Set empty frames to avoid errors when running
        cue._dmx_frames = {}

loop_actionCue(cue, mtc)

Loop an ActionCue

Source code in src/cuemsengine/cues/loop_cue.py

@loop_cue.register
def loop_actionCue(cue: ActionCue, mtc: MtcListener):
    """
    Loop an ActionCue
    """
    pass

loop_audioCue(cue, mtc, on_last_loop=None)

Handle the audio media playback loop.

This method manages the playback loop for audio media, including handling looping behavior and OSC communication for timing control.

Parameters:

Name	Type	Description	Default
ossia		The OSC communication interface.	required
mtc	MtcListener	The MIDI Time Code interface.	required
on_last_loop		called once, at the start of the last of several loops, when the cue's final end is known (cross-node Auto follow, 869fc8ytz). Never for a single or an endless loop.	None

Source code in src/cuemsengine/cues/loop_cue.py

@loop_cue.register
def loop_audioCue(cue: AudioCue, mtc: MtcListener, on_last_loop=None):
    """Handle the audio media playback loop.

    This method manages the playback loop for audio media, including handling
    looping behavior and OSC communication for timing control.

    Args:
        ossia: The OSC communication interface.
        mtc: The MIDI Time Code interface.
        on_last_loop: called once, at the start of the last of several loops,
            when the cue's final end is known (cross-node Auto follow,
            869fc8ytz). Never for a single or an endless loop.
    """
    Logger.info(
        f"Running audio cue loop {cue.id}, cue.loop={cue.loop}"
        f"(type={type(cue.loop).__name__})"
    )

    try:
        loop_counter = 0
        duration = CTimecode(cue.media.duration).return_in_other_framerate(
            mtc.main_tc.framerate
        )
        Logger.info(
            f"Audio duration: {duration}, "
            f"_end_mtc: {cue._end_mtc.milliseconds_rounded}ms, "
            f"current MTC: {mtc.main_tc.milliseconds_rounded}ms"
        )

        while cue.loop < 1 or loop_counter < cue.loop:
            if cue._stop_requested:
                Logger.info(f"Audio loop {cue.id} cancelled by stop request")
                return
            _start_of_last_loop(cue, loop_counter, on_last_loop)
            Logger.info(
                f"Audio loop iteration starting: loop_counter={loop_counter},"
                f"cue.loop={cue.loop}"
            )

            # Future: uncomment to enable percentage progress updates.
            # last_status_update = 0.0
            while mtc.main_tc.milliseconds_rounded < cue._end_mtc.milliseconds_rounded:
                if cue._stop_requested:
                    Logger.info(
                        f"Audio loop {cue.id} cancelled by stop request" f"(inner)"
                    )
                    return
                sleep(0.02)
                # Future: uncomment to enable percentage progress updates.
                # Throttled to CUE_STATUS_UPDATE_HZ (Tier 1 / node-side).
                # _now = time.monotonic()
                # if _now - last_status_update >= 1.0 / CUE_STATUS_UPDATE_HZ:
                #     last_status_update = _now
                # _elapsed = mtc.main_tc.milliseconds_rounded -
                # cue._start_mtc.milliseconds_rounded
                # _total = cue._end_mtc.milliseconds_rounded -
                # cue._start_mtc.milliseconds_rounded
                #     if _total > 0:
                #         _pct = max(1, min(99, int(100 * _elapsed / _total)))
                # CUE_HANDLER.communications_thread.update_cue(cue.id, _pct,
                # timeout=0.1)

            Logger.info(
                f"Audio iteration {loop_counter + 1} finished "
                f"(MTC={mtc.main_tc.milliseconds_rounded}ms reached "
                f"_end_mtc={cue._end_mtc.milliseconds_rounded}ms)"
            )
            loop_counter += 1

            will_loop_again = cue.loop < 1 or loop_counter < cue.loop
            Logger.info(
                f"After increment: loop_counter={loop_counter},"
                f"will_loop_again={will_loop_again}"
            )

            if cue._local and will_loop_again:
                cue._start_mtc = CTimecode(
                    framerate=cue._end_mtc.framerate,
                    frames=cue._end_mtc.frames,
                )
                cue._end_mtc = cue._start_mtc + duration

                # Drift-sensitive: use _exact (float) rather than _rounded
                # (int)
                # to preserve sub-ms precision at NTSC framerates
                # (29.97/23.976).
                offset_to_go = -cue._start_mtc.milliseconds_exact

                Logger.info(
                    f"Loop {loop_counter}: setting offset={offset_to_go}"
                    f" (MTC={mtc.main_tc.milliseconds_rounded}ms,"
                    f" _start_mtc={cue._start_mtc.milliseconds_rounded}ms,"
                    f" _end_mtc={cue._end_mtc.milliseconds_rounded}ms)"
                )

                try:
                    cue._osc.set_value("/offset", offset_to_go)
                    Logger.info(
                        f"Audio offset sent: {offset_to_go}",
                        extra={"caller": cue.__class__.__name__},
                    )
                except Exception as e:
                    Logger.error(
                        f"Audio offset send failed: {e}",
                        extra={"caller": cue.__class__.__name__},
                    )

        Logger.info(
            f"Audio loop FINISHED: loop_counter={loop_counter}, cue.loop={cue.loop}"
        )
        if cue._local:
            try:
                cue._osc.set_value("/mtcfollow", 0)
                Logger.info(
                    "Audio mtcfollow disabled",
                    extra={"caller": cue.__class__.__name__},
                )
            except Exception as e:
                Logger.warning(
                    f"Error disabling mtcfollow: {e}",
                    extra={"caller": cue.__class__.__name__},
                )

    except AttributeError:
        pass

loop_cue(cue, mtc)

Loop a cue based on its type

Source code in src/cuemsengine/cues/loop_cue.py

@singledispatch
def loop_cue(cue: Cue, mtc: MtcListener):
    """
    Loop a cue based on its type
    """
    pass

loop_cueList(cue, mtc)

Loop a CueList

Source code in src/cuemsengine/cues/loop_cue.py

@loop_cue.register
def loop_cueList(cue: CueList, mtc: MtcListener):
    """
    Loop a CueList
    """
    pass

loop_dmxCue(cue, mtc)

Handle the DMX cue duration wait.

DMX scenes are fire-and-forget (sent once in run_dmxCue), so we only wait for the cue duration to elapse to maintain proper script timing. The cue._local guard is maintained for potential future looping implementation.

Parameters:

Name	Type	Description	Default
cue	DmxCue	The DmxCue	required
mtc	MtcListener	The MIDI Time Code interface	required

Source code in src/cuemsengine/cues/loop_cue.py

@loop_cue.register
def loop_dmxCue(cue: DmxCue, mtc: MtcListener):
    """Handle the DMX cue duration wait.

    DMX scenes are fire-and-forget (sent once in run_dmxCue), so we only wait
    for the cue duration to elapse to maintain proper script timing.
    The cue._local guard is maintained for potential future looping
    implementation.

    Args:
        cue: The DmxCue
        mtc: The MIDI Time Code interface
    """
    try:
        # Future: uncomment to enable percentage progress updates.
        # last_status_update = 0.0
        while mtc.main_tc.milliseconds_rounded < cue._end_mtc.milliseconds_rounded:
            if cue._stop_requested:
                Logger.info(f"DMX loop {cue.id} cancelled by stop request")
                return
            sleep(0.02)
            # Future: uncomment to enable percentage progress updates.
            # Throttled to CUE_STATUS_UPDATE_HZ (Tier 1 / node-side).
            # _now = time.monotonic()
            # if _now - last_status_update >= 1.0 / CUE_STATUS_UPDATE_HZ:
            #     last_status_update = _now
            # _elapsed = mtc.main_tc.milliseconds_rounded -
            # cue._start_mtc.milliseconds_rounded
            # _total = cue._end_mtc.milliseconds_rounded -
            # cue._start_mtc.milliseconds_rounded
            #     if _total > 0:
            #         _pct = max(1, min(99, int(100 * _elapsed / _total)))
            # CUE_HANDLER.communications_thread.update_cue(cue.id, _pct,
            # timeout=0.1)

        if cue._local:
            pass

        Logger.debug(f"DMX cue {cue.id} duration elapsed")

    except AttributeError:
        pass

loop_fadeCue(cue, mtc)

Hold a FadeCue in the cue runner for its full duration.

The actual fade is driven by gradient-motiond over OSC; this loop simply blocks until the FadeCue's _end_mtc is reached so general cue lifecycle (auto-disarm of the FadeCue itself in go_threaded's end-of-cue path) only fires after the fade has elapsed. _start_mtc / _end_mtc are set by ActionHandler._handle_fade_action at dispatch time.

Every node walks the chain and runs this loop, but only the node where the action target is local actually dispatches the fade and stamps _end_mtc — on the others the handler bails early and _end_mtc still holds Cue.init's zero CTimecode (or a stale value from a previous GO). Those nodes MUST hold for the same duration anyway: their instant remove_cue would reach the controller first and grey out the cue in the UI (status 1 → 100) while the fade is still running on the owner node. Fall back to now + duration so every node's illumination window matches the fade.

Source code in src/cuemsengine/cues/loop_cue.py

@loop_cue.register
def loop_fadeCue(cue: FadeCue, mtc: MtcListener):
    """Hold a FadeCue in the cue runner for its full duration.

    The actual fade is driven by gradient-motiond over OSC; this loop simply
    blocks until the FadeCue's _end_mtc is reached so general cue lifecycle
    (auto-disarm of the FadeCue itself in go_threaded's end-of-cue path) only
    fires after the fade has elapsed. _start_mtc / _end_mtc are set by
    ActionHandler._handle_fade_action at dispatch time.

    Every node walks the chain and runs this loop, but only the node where the
    action target is local actually dispatches the fade and stamps _end_mtc —
    on the others the handler bails early and _end_mtc still holds
    Cue.__init__'s zero CTimecode (or a stale value from a previous GO). Those
    nodes MUST hold for the same duration anyway: their instant remove_cue
    would reach the controller first and grey out the cue in the UI
    (status 1 → 100) while the fade is still running on the owner node. Fall
    back to now + duration so every node's illumination window matches the
    fade.
    """
    now_ms = mtc.main_tc.milliseconds_rounded
    end_mtc = getattr(cue, "_end_mtc", None)
    if end_mtc is not None and end_mtc.milliseconds_rounded > now_ms:
        end_ms = end_mtc.milliseconds_rounded
    else:
        duration = getattr(cue, "duration", None)
        duration_ms = duration.milliseconds_rounded if duration is not None else 0
        if duration_ms <= 0:
            return
        end_ms = now_ms + duration_ms
        Logger.debug(
            f"FadeCue {cue.id}: no dispatched _end_mtc on this node; holding"
            f" for duration ({duration_ms}ms)"
        )

    while mtc.main_tc.milliseconds_rounded < end_ms:
        if getattr(cue, "_stop_requested", False):
            Logger.info(f"FadeCue {cue.id} loop cancelled by stop request")
            return
        sleep(0.02)

loop_videoCue(cue, mtc, on_last_loop=None)

Handle the video media playback loop.

Manages looping behavior for all layers in cue._layer_ids, updating offset via the single VideoClient in cue._osc. on_last_loop: as for loop_audioCue.

Source code in src/cuemsengine/cues/loop_cue.py

@loop_cue.register
def loop_videoCue(cue: VideoCue, mtc: MtcListener, on_last_loop=None):
    """Handle the video media playback loop.

    Manages looping behavior for all layers in cue._layer_ids,
    updating offset via the single VideoClient in cue._osc.
    on_last_loop: as for loop_audioCue.
    """
    Logger.info(
        f"Running video cue loop {cue.id}, cue.loop={cue.loop}"
        f"(type={type(cue.loop).__name__})"
    )

    try:
        loop_counter = 0
        duration = CTimecode(cue.media.duration).return_in_other_framerate(
            mtc.main_tc.framerate
        )
        Logger.info(
            f"Video duration: {duration}, duration in frames:"
            f"{duration.frame_number} {duration.framerate}"
        )
        Logger.info(
            f"Initial _end_mtc: {cue._end_mtc.milliseconds_rounded}ms,"
            f"current MTC: {mtc.main_tc.milliseconds_rounded}ms"
        )

        layer_ids = getattr(cue, "_layer_ids", [])

        # Tell the videocomposer to wrap frames at the loop boundary (instead
        # of clamping to the last frame) ONLY for a cue that really repeats.
        # A play-once cue must clamp: the videocomposer runs its display
        # latency (33 ms) ahead of MTC, so with wraparound on it reaches end
        # of file before the engine hides the layer and shows frame 0 again
        # (869fa89uh).
        if cue.loop != 1:
            for layer_id in layer_ids:
                try:
                    cue._osc.set_value(f"/videocomposer/layer/{layer_id}/loop", 1)
                except Exception as e:
                    Logger.error(f"Loop enable failed for layer {layer_id}: {e}")

        while cue.loop < 1 or loop_counter < cue.loop:
            if cue._stop_requested:
                Logger.info(f"Video loop {cue.id} cancelled by stop request")
                return
            _start_of_last_loop(cue, loop_counter, on_last_loop)
            # Future: uncomment to enable percentage progress updates.
            # last_status_update = 0.0
            while mtc.main_tc.milliseconds_rounded < cue._end_mtc.milliseconds_rounded:
                if cue._stop_requested:
                    Logger.info(
                        f"Video loop {cue.id} cancelled by stop request (inner)"
                    )
                    return
                sleep(0.02)
                # Future: uncomment to enable percentage progress updates.
                # Throttled to CUE_STATUS_UPDATE_HZ (Tier 1 / node-side).
                # _now = time.monotonic()
                # if _now - last_status_update >= 1.0 / CUE_STATUS_UPDATE_HZ:
                #     last_status_update = _now
                # _elapsed = mtc.main_tc.milliseconds_rounded -
                # cue._start_mtc.milliseconds_rounded
                # _total = cue._end_mtc.milliseconds_rounded -
                # cue._start_mtc.milliseconds_rounded
                #     if _total > 0:
                #         _pct = max(1, min(99, int(100 * _elapsed / _total)))
                # CUE_HANDLER.communications_thread.update_cue(cue.id, _pct,
                # timeout=0.1)

            Logger.info(
                f"Video iteration {loop_counter + 1} finished "
                f"(MTC={mtc.main_tc.milliseconds_rounded}ms reached "
                f"_end_mtc={cue._end_mtc.milliseconds_rounded}ms)"
            )
            loop_counter += 1

            will_loop_again = cue.loop < 1 or loop_counter < cue.loop

            if cue._local and will_loop_again:
                cue._start_mtc = CTimecode(
                    framerate=cue._end_mtc.framerate,
                    frames=cue._end_mtc.frames,
                )
                cue._end_mtc = cue._start_mtc + duration
                offset_change_frames = -cue._start_mtc.frame_number

                Logger.info(
                    f"Loop {loop_counter}: setting offset={offset_change_frames}"
                )

                for layer_id in layer_ids:
                    try:
                        cue._osc.set_value(
                            f"/videocomposer/layer/{layer_id}/offset",
                            int(offset_change_frames),
                        )
                    except Exception as e:
                        Logger.error(f"Offset send failed for layer {layer_id}: {e}")

                # Last pass of a counted loop: stop wrapping (after the new
                # offset, so the layer never clamps mid-boundary) so the end
                # of this pass holds the last frame like a play-once cue.
                if cue.loop >= 1 and loop_counter + 1 >= cue.loop:
                    for layer_id in layer_ids:
                        try:
                            cue._osc.set_value(
                                f"/videocomposer/layer/{layer_id}/loop", 0
                            )
                        except Exception as e:
                            Logger.error(
                                f"Loop disable failed for layer {layer_id}: {e}"
                            )

        Logger.info(f"Loop FINISHED: loop_counter={loop_counter}, cue.loop={cue.loop}")

    except AttributeError:
        pass

blank_cue(cue, mtc)

Hide a cue's output at its body end, WITHOUT disarming it.

Used by the pause/go_at_end postwait tail: loop_videoCue leaves /loop 1 on, so an undisarmed video keeps visibly wrap-looping after the body ends — blank_cue hides it while the cue stays in _armed_cues (STOP-reachable through the whole tail; disarming here instead would make the tail unreachable by stop_all_cues → ghost fire after STOP). Default no-op: audio is already silent (loop_audioCue sends /mtcfollow 0 at loop end), DMX scenes have ended, actions/CueLists have nothing to hide.

Source code in src/cuemsengine/cues/run_cue.py

@singledispatch
def blank_cue(cue: Cue, mtc: MtcListener):
    """Hide a cue's output at its body end, WITHOUT disarming it.

    Used by the pause/go_at_end postwait tail: loop_videoCue leaves /loop 1 on,
    so an undisarmed video keeps visibly wrap-looping after the body ends —
    blank_cue hides it while the cue stays in _armed_cues (STOP-reachable
    through the whole tail; disarming here instead would make the tail
    unreachable by stop_all_cues → ghost fire after STOP). Default no-op:
    audio is already silent (loop_audioCue sends /mtcfollow 0 at loop end),
    DMX scenes have ended, actions/CueLists have nothing to hide.
    """
    pass

reveal_cue(cue, mtc, frozen_mtc_ms=None)

Reveal a held cue at its start_mtc. Default no-op (DmxCue self-schedules).

Source code in src/cuemsengine/cues/run_cue.py

@singledispatch
def reveal_cue(cue: Cue, mtc: MtcListener, frozen_mtc_ms: float = None):
    """Reveal a held cue at its start_mtc. Default no-op (DmxCue self-schedules)."""
    pass

reveal_cueList(cue, mtc, frozen_mtc_ms=None)

Reveal a CueList target by revealing its first enabled child.

Mirrors run_cueList: run_cue set the child up HELD, so reveal must recurse to the same child — otherwise a CueList used as a post_go='go'/'go_at_end' target would leave its child's video invisible / audio silent forever (no error).

run_cueList (earlier in go_threaded) is the arm point, so by reveal the child is normally loaded. If it isn't (run_cueList's re-arm failed — already logged loudly there), we log + SKIP rather than re-arm: a reveal-time arm without run_cue's setup would leave _start_mtc at its default and make reveal_audioCue send a wrong /offset then enable /mtcfollow — actively wrong playback, worse than a clean skip.

Source code in src/cuemsengine/cues/run_cue.py

@reveal_cue.register
def reveal_cueList(cue: CueList, mtc, frozen_mtc_ms: float = None):
    """Reveal a CueList target by revealing its first enabled child.

    Mirrors run_cueList: run_cue set the child up HELD, so reveal must recurse to
    the same child — otherwise a CueList used as a post_go='go'/'go_at_end' target
    would leave its child's video invisible / audio silent forever (no error).

    run_cueList (earlier in go_threaded) is the arm point, so by reveal the child
    is normally loaded. If it isn't (run_cueList's re-arm failed — already logged
    loudly there), we log + SKIP rather than re-arm: a reveal-time arm without
    run_cue's setup would leave _start_mtc at its default and make reveal_audioCue
    send a wrong /offset then enable /mtcfollow — actively wrong playback, worse
    than a clean skip.
    """
    child = _first_enabled_child(cue)
    if child is None:
        return
    if getattr(child, "_local", True) and not getattr(child, "loaded", False):
        Logger.warning(
            f"CueList {cue.id} reveal: child {child.id} not loaded (setup never "
            f"completed) — skipping reveal."
        )
        return
    reveal_cue(child, mtc, frozen_mtc_ms)

run_actionCue(cue, mtc, frozen_mtc_ms=None)

Run an ActionCue by delegating to ActionHandler.execute_action.

Forwards frozen_mtc_ms so a chained 'play' action triggered inside a post_go='go' chain preserves the chain's MTC snapshot — without it, ActionCue-mediated chains capture live MTC inside CueHandler.go and drift relative to the chain's other cues.

Source code in src/cuemsengine/cues/run_cue.py

@run_cue.register
def run_actionCue(cue: ActionCue, mtc: MtcListener, frozen_mtc_ms: float = None):
    """Run an ActionCue by delegating to ActionHandler.execute_action.

    Forwards frozen_mtc_ms so a chained 'play' action triggered inside a
    post_go='go' chain preserves the chain's MTC snapshot — without it,
    ActionCue-mediated chains capture live MTC inside CueHandler.go and
    drift relative to the chain's other cues.
    """
    # Prepare-only: an ActionCue has no media to preload. But it MUST carry a
    # _start_mtc so CueHandler._reveal_wait gates it — otherwise the action would
    # fire at dispatch (a full body early) instead of at its own timeline slot.
    # The action is EXECUTED in reveal_cue() once live MTC reaches _start_mtc.
    if frozen_mtc_ms is not None:
        cue._start_mtc = CTimecode(
            framerate=mtc.main_tc.framerate, start_seconds=frozen_mtc_ms / 1000
        )
    else:
        cue._start_mtc = CTimecode(
            framerate=mtc.main_tc.framerate, frames=mtc.main_tc.frames
        )
    return

run_audioCue(cue, mtc, frozen_mtc_ms=None)

Run an AudioCue

Parameters:

Name	Type	Description	Default
cue	AudioCue	The audio cue to run	required
mtc		The MTC listener (for framerate info)	required
frozen_mtc_ms	float	Optional frozen MTC timestamp for perfect sync with	None

Source code in src/cuemsengine/cues/run_cue.py

@run_cue.register
def run_audioCue(cue: AudioCue, mtc, frozen_mtc_ms: float = None):
    """
    Run an AudioCue

    Args:
        cue: The audio cue to run
        mtc: The MTC listener (for framerate info)
        frozen_mtc_ms: Optional frozen MTC timestamp for perfect sync with
        chained cues
    """
    # Set True only once /offset is sent below. If setup aborts early (e.g. mixer
    # / JACK ports missing), reveal_audioCue sees this False and no-ops instead of
    # asserting mtcfollow on a player that never got set up.
    cue._reveal_ready = False

    # CRITICAL FOR SYNC: Use frozen timestamp if provided (for post_go='go'
    # chains)
    # Otherwise read live MTC. This ensures audio and video cues share the
    # same reference.
    if frozen_mtc_ms is not None:
        mtc_ms = frozen_mtc_ms
        Logger.debug(f"AudioCue {cue.id} using frozen MTC: {mtc_ms}ms")
        # Frozen path: only have a float ms snapshot (CueHandler captured it
        # before this point); reconstruct via canonicalized __init__ which
        # routes through HMSF + tc_to_frames, drop-frame correct.
        cue._start_mtc = CTimecode(
            framerate=mtc.main_tc.framerate, start_seconds=mtc_ms / 1000
        )
    else:
        # Live MTC path: frame-domain construction skips the lossy
        # ms→seconds→frames round-trip entirely (mirrors loop_cue.py:107,224).
        cue._start_mtc = CTimecode(
            framerate=mtc.main_tc.framerate, frames=mtc.main_tc.frames
        )

    # Convert duration to MTC framerate to prevent drift when looping
    duration = CTimecode(cue.media.duration).return_in_other_framerate(
        mtc.main_tc.framerate
    )
    cue._end_mtc = cue._start_mtc + duration

    # Audio player formula: file_position = MTC + offset
    # To play from position 0 when MTC = start_mtc, we need offset = -start_mtc
    offset_to_go = -cue._start_mtc.milliseconds_exact

    # Verify mixer graph; only repair if drifted. Arm already wired it; the
    # unconditional reconnect at GO costs ~21-28 ms (measured) without
    # touching the audio path.
    try:
        mixer = PLAYER_HANDLER.get_audio_mixer()
        if mixer:
            uuid_slug = "".join(str(cue.id).split("-"))
            # Actual JACK client name is Audio_Player-{uuid} with ports
            # "outport 0", "outport 1"
            player_name = f"Audio_Player-{uuid_slug}"

            # Resolve JACK port names from cue output IDs via audio output
            # lookup
            selected_outputs = []
            if hasattr(cue, "outputs") and cue.outputs:
                for output in cue.outputs:
                    output_name = output.get("output_name", "")
                    if len(output_name) > 37:
                        output_id = output_name[37:]
                        port_name = PLAYER_HANDLER.resolve_audio_port(output_id)
                        if port_name:
                            selected_outputs.append(port_name)
                        else:
                            selected_outputs.append(output_id)

            Logger.debug(f"Audio cue {cue.id} selected outputs: {selected_outputs}")

            # If the player's outport 0 is missing, the subprocess died between
            # arm and GO. connect_player_to_outputs would block 15 s in its
            # port-wait loop before failing; abort fast instead.
            channel_0 = f"{player_name}:outport 0"
            if not mixer.conn_man.port_exists(channel_0):
                Logger.error(
                    f"Audio cue {cue.id}: player JACK ports missing at GO "
                    f"({channel_0}); subprocess likely crashed between arm "
                    f"and GO. Aborting cue."
                )
                return

            _verify_audio_wiring_at_go(cue, mixer, player_name, selected_outputs)
    except Exception as e:
        Logger.warning(f"Could not validate/connect player to mixer: {e}")

    # Define the offset - use MTC framerate for consistent timing with video
    try:
        key = "/offset"

        cue._osc.set_value(key, offset_to_go)
        Logger.info(
            f"offset {offset_to_go} to {key}:"
            f"{str(cue._osc.get_node(key).parameter.value)}",
            extra={"caller": cue.__class__.__name__},
        )
    except Exception as e:
        Logger.warning(
            f"Error setting offset in run_audioCue: {e}",
            extra={"caller": cue.__class__.__name__},
        )
    else:
        # Setup reached the offset send → safe for reveal_cue to start following.
        cue._reveal_ready = True

    # /mtcfollow is DEFERRED to reveal_cue() (MTC-gated reveal). The offset above
    # is harmless while not following; reveal_cue turns following on at
    # start_mtc, where the seek position is ~0.
    # NOTE: the deferral used to be REQUIRED — following early with a
    # future-negative offset hit the audioplayer's "Out of file boundaries" path
    # and TERMINATED the cue before it played. Fixed in cuems-audioplayer
    # (869dyufeh): a negative seek now holds silence and self-gates. The deferral
    # is kept as belt-and-suspenders until the fixed player binary is fleet-wide;
    # once it is, this reveal-time /mtcfollow can be dropped (audio becomes
    # symmetric with video).

    # Apply master volume from cue settings
    try:
        master_vol = getattr(cue, "master_vol", None)
        if master_vol is not None:
            # UI uses 0-100 percentage, audioplayer expects 0.0-1.0 gain
            # Convert and clamp to valid range
            vol_value = max(0.0, min(1.0, float(master_vol) / 100.0))
            cue._osc.set_value("/volmaster", vol_value)
            Logger.info(
                f"master_vol {master_vol}% -> {vol_value} set on audio cue {cue.id}",
                extra={"caller": cue.__class__.__name__},
            )
    except Exception as e:
        Logger.warning(
            f"Error setting master volume in run_audioCue: {e}",
            extra={"caller": cue.__class__.__name__},
        )

run_cue(cue, mtc, frozen_mtc_ms=None)

Run a cue based on its type.

Parameters:

Name	Type	Description	Default
cue	Cue	The cue to run	required
mtc	MtcListener	The MTC listener (for framerate info)	required
frozen_mtc_ms	float	Optional frozen MTC timestamp in milliseconds. When provided (e.g., for chained cues with post_go='go'), this timestamp is used instead of reading live MTC. This ensures perfect sync between audio and video cues.	None

Source code in src/cuemsengine/cues/run_cue.py

@singledispatch
def run_cue(cue: Cue, mtc: MtcListener, frozen_mtc_ms: float = None):
    """
    Run a cue based on its type.

    Args:
        cue: The cue to run
        mtc: The MTC listener (for framerate info)
        frozen_mtc_ms: Optional frozen MTC timestamp in milliseconds.
                       When provided (e.g., for chained cues with
                       post_go='go'),
                       this timestamp is used instead of reading live MTC.
                       This ensures perfect sync between audio and video cues.
    """
    pass

run_cueList(cue, mtc, frozen_mtc_ms=None)

Run a CueList by dispatching its first enabled child.

Children bypass CueHandler.go(), so replicate go()'s re-arm safety net here (_ensure_child_loaded) — otherwise an un-armed child fires with only a misleading log and no recovery.

Source code in src/cuemsengine/cues/run_cue.py

@run_cue.register
def run_cueList(cue: CueList, mtc: MtcListener, frozen_mtc_ms: float = None):
    """Run a CueList by dispatching its first enabled child.

    Children bypass CueHandler.go(), so replicate go()'s re-arm safety net here
    (_ensure_child_loaded) — otherwise an un-armed child fires with only a
    misleading log and no recovery.
    """
    child = _first_enabled_child(cue)
    if child is None:
        return
    if not _ensure_child_loaded(cue, child):
        return  # re-arm failed (already logged loudly) — don't dispatch
    run_cue(child, mtc, frozen_mtc_ms)

run_dmxCue(cue, mtc, frozen_mtc_ms=None)

Run a DmxCue

Sends DMX scene bundle directly to the local DMX player. Synchronized with MTC. The scene contains frame data, timing, and fade info. DMX cues have no media duration - duration is inferred from fade times. Only fadein_time is used for now. fade_out defaults to 0

Parameters:

Name	Type	Description	Default
cue	DmxCue	The DMX cue to run	required
mtc		The MTC listener (for framerate info)	required
frozen_mtc_ms	float	Optional frozen MTC timestamp for perfect sync with	None

Source code in src/cuemsengine/cues/run_cue.py

@run_cue.register
def run_dmxCue(cue: DmxCue, mtc, frozen_mtc_ms: float = None):
    """
    Run a DmxCue

    Sends DMX scene bundle directly to the local DMX player.
    Synchronized with MTC. The scene contains frame data, timing, and fade
    info.
    DMX cues have no media duration - duration is inferred from fade times.
    Only fadein_time is used for now. fade_out defaults to 0

    Args:
        cue: The DMX cue to run
        mtc: The MTC listener (for framerate info)
        frozen_mtc_ms: Optional frozen MTC timestamp for perfect sync with
        chained cues
    """
    try:
        # CRITICAL FOR SYNC: Use frozen timestamp if provided (for post_go='go'
        # chains)
        if frozen_mtc_ms is not None:
            mtc_ms = frozen_mtc_ms
            Logger.debug(f"DmxCue {cue.id} using frozen MTC: {mtc_ms}ms")
            # Frozen path: only have a float ms snapshot; canonicalized
            # __init__ routes through HMSF + tc_to_frames.
            cue._start_mtc = CTimecode(
                framerate=mtc.main_tc.framerate, start_seconds=mtc_ms / 1000
            )
        else:
            # Live MTC path: frame-domain construction (no round-trip loss).
            cue._start_mtc = CTimecode(
                framerate=mtc.main_tc.framerate, frames=mtc.main_tc.frames
            )

        # DMX cues have no media - duration is inferred from fade times
        # Duration = fadein_time + fadeout_time (both in milliseconds)
        fadein_ms = getattr(cue, "fadein_time", 0)
        fadeout_ms = getattr(cue, "fadeout_time", 0)
        duration_ms = fadein_ms + fadeout_ms

        # Convert duration to timecode format with explicit framerate
        duration_seconds = duration_ms / 1000.0
        duration = CTimecode(
            framerate=mtc.main_tc.framerate, start_seconds=duration_seconds
        )
        cue._end_mtc = cue._start_mtc + duration

        # Absolute MTC time for this cue (ms). DMX player expects mtc_time as
        # absolute
        # "0:0:S.sss" string so it can schedule m_mtcStart = max(playHead,
        # time).
        offset_milliseconds = cue._start_mtc.milliseconds_exact
        mtc_time_str = f"0:0:{offset_milliseconds / 1000.0}"

        # Get DMX frame data from the cue
        universe_frames = getattr(cue, "_dmx_frames", {})

        if not universe_frames:
            Logger.warning(
                f"DMX cue {cue.id} has no frame data to send",
                extra={"caller": cue.__class__.__name__},
            )
            return

        # Convert fadein_time to seconds for the DMX player (only fadein is
        # used for now)
        fade_time = fadein_ms / 1000.0

        # Check if we have an OSC client
        if cue._osc is None:
            Logger.error(
                f"DMX cue {cue.id} has no OSC client available",
                extra={"caller": cue.__class__.__name__},
            )
            return

        # Enable MTC following so the dmxplayer tracks timecode and stops
        # advancing when MTC stops (e.g. on STOP command).
        cue._osc.enable_mtcfollow()

        # Send DMX scene bundle to local player (mtc_time absolute so no
        # overlap/loss)
        cue._osc.send_dmx_scene(
            universe_frames=universe_frames,
            mtc_time=mtc_time_str,
            fade_time=fade_time,
        )

        Logger.info(
            f"DMX scene sent to local player for cue {cue.id}: "
            f"mtc_time={mtc_time_str} ({offset_milliseconds}ms),"
            f"universes={len(universe_frames)}, fade={fade_time}s",
            extra={"caller": cue.__class__.__name__},
        )

    except Exception as e:
        Logger.error(
            f"Error running DMX cue {cue.id}: {e}",
            extra={"caller": cue.__class__.__name__},
        )
        Logger.exception(e)

run_videoCue(cue, mtc, frozen_mtc_ms=None)

Run a VideoCue.

Sends offset/visible/mtcfollow to all layers in cue._layer_ids via the single VideoClient in cue._osc.

Source code in src/cuemsengine/cues/run_cue.py

@run_cue.register
def run_videoCue(cue: VideoCue, mtc, frozen_mtc_ms: float = None):
    """Run a VideoCue.

    Sends offset/visible/mtcfollow to all layers in cue._layer_ids
    via the single VideoClient in cue._osc.
    """
    Logger.info(f"Running video cue {cue.id}")

    layer_ids = getattr(cue, "_layer_ids", [])

    # The timeline is stamped BEFORE the layer check (869fgq3xv): a cue with
    # no layers on this node (every output dark) still has a body. Downstream
    # reads _start_mtc/_end_mtc: the Auto-follow instant and its cross-node
    # announce (CueHandler._follow_instant_ms), the reveal wait, loop_videoCue.
    # Left at their zero defaults, a go_at_end follow fired at start+postwait
    # cluster-wide and an infinite loop re-based in a tight loop.
    if frozen_mtc_ms is not None:
        mtc_ms = frozen_mtc_ms
        Logger.debug(f"VideoCue {cue.id} using frozen MTC: {mtc_ms}ms")
        # Frozen path: float ms snapshot; canonicalized __init__ handles it.
        cue._start_mtc = CTimecode(
            framerate=mtc.main_tc.framerate, start_seconds=mtc_ms / 1000
        )
    else:
        # Live MTC path: frame-domain construction (no round-trip loss).
        cue._start_mtc = CTimecode(
            framerate=mtc.main_tc.framerate, frames=mtc.main_tc.frames
        )

    duration = CTimecode(cue.media.duration).return_in_other_framerate(
        mtc.main_tc.framerate
    )
    cue._end_mtc = cue._start_mtc + duration
    offset_to_go = -cue._start_mtc.frame_number

    if not layer_ids or getattr(cue, "_osc", None) is None:
        unrouted = getattr(cue, "_unrouted_outputs", None)
        if not layer_ids and unrouted:
            Logger.info(
                f"Video cue {cue.id} is dark on this node (outputs not driven:"
                f" {unrouted}); timeline kept, nothing sent"
            )
        else:
            Logger.error(f"Video cue {cue.id} has no layers or no OSC client")
        return

    client = cue._osc

    # Infinite-loop cues (cue.loop < 1, e.g. loop=-1) must have wraparound
    # enabled on the videocomposer BEFORE the layer starts following MTC below.
    # Otherwise there's a race: the layer follows MTC (mtcfollow=1) from here,
    # but /loop is only sent later by loop_videoCue() — which go_threaded calls
    # AFTER the postwait sleep. In that window the VC has wraparound_=false, so
    # when the media frame overshoots its length it CLAMPS to the last frame
    # (LayerPlayback: adjustedFrame = totalFrames-1) and the video visibly
    # freezes on the final frame until loop_videoCue() finally sends /loop.
    # Sending it here closes the race so the first loop wraps cleanly. Finite
    # loops (cue.loop >= 1) keep the existing loop_videoCue timing untouched.
    loop_early = getattr(cue, "loop", 1) < 1

    # Infinite-loop cues (cue.loop < 1, e.g. loop=-1) must have wraparound
    # enabled on the videocomposer BEFORE the layer starts following MTC below.
    # Otherwise there's a race: the layer follows MTC (mtcfollow=1) from here,
    # but /loop is only sent later by loop_videoCue() — which go_threaded calls
    # AFTER the postwait sleep. In that window the VC has wraparound_=false, so
    # when the media frame overshoots its length it CLAMPS to the last frame
    # (LayerPlayback: adjustedFrame = totalFrames-1) and the video visibly
    # freezes on the final frame until loop_videoCue() finally sends /loop.
    # Sending it here closes the race so the first loop wraps cleanly. Finite
    # loops (cue.loop >= 1) keep the existing loop_videoCue timing untouched.
    loop_early = getattr(cue, "loop", 1) < 1

    # Re-apply position for each layer before making visible (layer may not
    # have been ready when position was set during arm). Each layer's output
    # comes from the arm, by layer id: with an unrouted output skipped, layer
    # i is no longer output i (869fgq3xv).
    layer_outputs = getattr(cue, "_layer_outputs", None) or {}
    media_w, media_h = PLAYER_HANDLER.cue_media_dimensions(cue)

    for layer_id in layer_ids:
        layer_path = f"/videocomposer/layer/{layer_id}"

        # Re-apply canvas position from the output config
        output_name = layer_outputs.get(layer_id)
        if output_name is None:
            Logger.warning(
                f"Video cue {cue.id} layer {layer_id}: no output recorded at arm;"
                " placement re-apply skipped"
            )
        else:
            try:
                output = PLAYER_HANDLER.resolve_video_output_for_cue(cue, output_name)
                x, y = output.get_layer_placement()
                client.set_value(f"{layer_path}/position", [x, y])
                sx, sy = output.get_layer_scale(media_w, media_h)
                client.set_value(f"{layer_path}/scale", [sx, sy])
                # Applied now: clear a failure recorded at arm.
                failed = getattr(cue, "_placement_failed", None)
                if failed and layer_id in failed:
                    failed.remove(layer_id)
            except (KeyError, RuntimeError, ValueError) as e:
                _record_placement_failure(cue, layer_id)
                Logger.error(
                    f'Video cue {cue.id} layer {layer_id} on output "{output_name}":'
                    f" position/scale re-apply at GO NOT applied, either may be"
                    f" left at its default ({type(e).__name__}: {e})"
                )
            except Exception:
                _record_placement_failure(cue, layer_id)
                Logger.exception(
                    f"Unexpected error re-applying position for layer"
                    f' {layer_id} (output "{output_name}"): NOT applied'
                )

        client.set_value(f"{layer_path}/offset", int(offset_to_go))
        # Enable wraparound early for infinite loops, BEFORE mtcfollow, so the
        # first loop wraps instead of freezing on the clamped last frame.
        if loop_early:
            client.set_value(f"{layer_path}/loop", 1)
        # Preload the correct frame while INVISIBLE: offset+mtcfollow let the
        # videocomposer decode the frame at this MTC position, but the layer
        # stays hidden. /visible is DEFERRED to reveal_cue() (MTC-gated reveal
        # at the cue's start_mtc) — this is what makes prewait/postwait gaps real.
        client.set_value(f"{layer_path}/mtcfollow", 1)

    Logger.info(
        f"Video cue {cue.id} set up (held invisible): {len(layer_ids)} layer(s),"
        f" offset={offset_to_go}"
    )